diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,3 @@ +# Changelog for metro-tcp + +## Unreleased changes diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright Lupino (c) 2019 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Lupino nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,3 @@ +# metro-socket + +Socket transport for metro diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/metro-socket.cabal b/metro-socket.cabal new file mode 100644 --- /dev/null +++ b/metro-socket.cabal @@ -0,0 +1,54 @@ +cabal-version: 1.12 + +-- This file has been generated from package.yaml by hpack version 0.33.0. +-- +-- see: https://github.com/sol/hpack +-- +-- hash: dcb41445e65758018ed9b7846b1058ae7edd3e0b6a89cd5d4dca8da443e87798 + +name: metro-socket +version: 0.1.0.0 +synopsis: Socket transport for metro +description: Please see the README on GitHub at +category: Web +homepage: https://github.com/Lupino/metro#readme +bug-reports: https://github.com/Lupino/metro/issues +author: Lupino +maintainer: lmjubuntu@gmail.com +copyright: MIT +license: BSD3 +license-file: LICENSE +build-type: Simple +extra-source-files: + README.md + ChangeLog.md + +source-repository head + type: git + location: https://github.com/Lupino/metro + +library + exposed-modules: + Metro.Socket + Metro.SocketServer + Metro.TCPServer + Metro.TP.Socket + Metro.TP.TCPSocket + Metro.TP.UDPSocket + Metro.UDPServer + other-modules: + Paths_metro_socket + hs-source-dirs: + src + build-depends: + base >=4.7 && <5 + , bytestring + , directory + , hashable + , hslogger + , metro + , mtl + , network + , transformers + , unliftio + default-language: Haskell2010 diff --git a/src/Metro/Socket.hs b/src/Metro/Socket.hs new file mode 100644 --- /dev/null +++ b/src/Metro/Socket.hs @@ -0,0 +1,187 @@ +module Metro.Socket + ( Socket + , close + , listen + , connect + , getHost + , getService + -- udp + , bindTo + , getDatagramAddr + ) where + +import Control.Exception (bracketOnError, throwIO) +import Control.Monad (when) +import Data.List (isPrefixOf) +import Data.Maybe (listToMaybe) +import Network.Socket hiding (bind, connect, listen) +import qualified Network.Socket as S (bind, connect, listen) +import System.Directory (doesFileExist, removeFile) +import System.Exit (exitFailure) +import UnliftIO (tryIO) + +-- Returns the first action from a list which does not throw an exception. +-- If all the actions throw exceptions (and the list of actions is not empty), +-- the last exception is thrown. +-- The operations are run outside of the catchIO cleanup handler because +-- catchIO masks asynchronous exceptions in the cleanup handler. +-- In the case of complete failure, the last exception is actually thrown. +firstSuccessful :: [IO a] -> IO a +firstSuccessful = go Nothing + where + -- Attempt the next operation, remember exception on failure + go _ (p:ps) = + do r <- tryIO p + case r of + Right x -> return x + Left e -> go (Just e) ps + + -- All operations failed, throw error if one exists + go Nothing [] = error "firstSuccessful: empty list" + go (Just e) [] = throwIO e + + +connectTo :: Maybe HostName -> Maybe ServiceName -> IO Socket +connectTo host serv = do + let hints = defaultHints { addrFlags = [AI_ADDRCONFIG] + , addrSocketType = Stream } + addrs <- getAddrInfo (Just hints) host serv + firstSuccessful $ map tryToConnect addrs + where + tryToConnect addr = + bracketOnError + (socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)) + close -- only done if there's an error + (\sock -> do + S.connect sock (addrAddress addr) + return sock + ) + +connectToFile :: FilePath -> IO Socket +connectToFile path = + bracketOnError + (socket AF_UNIX Stream 0) + close + (\sock -> do + S.connect sock (SockAddrUnix path) + return sock + ) + +listenOnFile :: FilePath -> IO Socket +listenOnFile path = + bracketOnError + (socket AF_UNIX Stream 0) + close + (\sock -> do + setSocketOption sock ReuseAddr 1 + S.bind sock (SockAddrUnix path) + S.listen sock maxListenQueue + return sock + ) + +listenOn :: Maybe HostName -> Maybe ServiceName -> IO Socket +listenOn host serv = do + -- We should probably specify addrFamily = AF_INET6 and the filter + -- code below should be removed. AI_ADDRCONFIG is probably not + -- necessary. But this code is well-tested. So, let's keep it. + let hints = defaultHints { addrFlags = [AI_ADDRCONFIG, AI_PASSIVE] + , addrSocketType = Stream + } + addrs <- getAddrInfo (Just hints) host serv + -- Choose an IPv6 socket if exists. This ensures the socket can + -- handle both IPv4 and IPv6 if v6only is false. + let addrs' = filter (\x -> addrFamily x == AF_INET6) addrs + addr = if null addrs' then head addrs else head addrs' + bracketOnError + (socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)) + close + (\sock -> do + setSocketOption sock ReuseAddr 1 + setSocketOption sock NoDelay 1 + S.bind sock (addrAddress addr) + S.listen sock maxListenQueue + return sock + ) + +listen :: String -> IO Socket +listen port = + if "tcp" `isPrefixOf` port then + listenOn (getHost port) (getService port) + else do + let sockFile = dropS port + exists <- doesFileExist sockFile + when exists $ do + e <- tryIO $ connectToFile sockFile + case e of + Left _ -> removeFile sockFile + Right _ -> do + putStrLn "periodicd: bind: resource busy (Address already in use)" + exitFailure + listenOnFile sockFile + +connect :: String -> IO Socket +connect h | "tcp" `isPrefixOf` h = connectTo (getHost h) (getService h) + | otherwise = connectToFile (dropS h) + +getDatagramAddrList :: String -> IO [AddrInfo] +getDatagramAddrList hostPort = getAddrInfo (Just hints) host port + where hints = defaultHints + { addrFlags = [AI_PASSIVE] + , addrSocketType = Datagram + } + + host = getHost hostPort + port = getService hostPort + +getDatagramAddr :: String -> IO (Maybe AddrInfo) +getDatagramAddr hostPort = listToMaybe <$> getDatagramAddrList hostPort + +bindTo :: String -> IO Socket +bindTo hostPort = do + addrs <- getDatagramAddrList hostPort + firstSuccessful $ map tryToConnect addrs + where + tryToConnect addr = + bracketOnError + (socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)) + close -- only done if there's an error + (\sock -> do + setSocketOption sock ReuseAddr 1 + S.bind sock $ addrAddress addr + return sock + ) + + +-- ipv6 fe80::1046:372a:8c3b:94b8%en0:80 + +countColon :: String -> Int +countColon = length . filter (==':') + +-- ipv6 fe80::1046:372a:8c3b:94b8%en0:80 +-- ipv6 fe80::1046:372a:8c3b:94b8%en0 +-- ipv4 127.0.0.1:80 +-- ipv4 127.0.0.1 +-- only port :80 +splitHostPort :: String -> (Maybe String, Maybe String) +splitHostPort hostPort + | colon == 0 = (Just hostPort, Nothing) + | colon == 1 = (takeFst id hostPort, takeSnd id hostPort) + | colon == 5 = (Just hostPort, Nothing) + | colon == 6 = (takeSnd reverse hostPort, takeFst reverse hostPort ) + | otherwise = (Nothing, Nothing) + where colon = countColon hostPort + takeFst f = toMaybe . f . takeWhile (/=':') . f + takeSnd f = toMaybe . f . drop 1 . dropWhile (/=':') . f + +dropS :: String -> String +dropS = drop 3 . dropWhile (/= ':') + +toMaybe :: String -> Maybe String +toMaybe [] = Nothing +toMaybe xs = Just xs + +getHost :: String -> Maybe String +getHost = fst . splitHostPort . dropS + +getService :: String -> Maybe String +getService = snd . splitHostPort . dropS diff --git a/src/Metro/SocketServer.hs b/src/Metro/SocketServer.hs new file mode 100644 --- /dev/null +++ b/src/Metro/SocketServer.hs @@ -0,0 +1,58 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE UndecidableInstances #-} + +module Metro.SocketServer + ( SocketServer + , socketServer + , SSSID (..) + ) where + +import Data.List (isPrefixOf) +import Metro.Class (Servable (..), Transport (..), + TransportConfig) +import Metro.TCPServer +import qualified Metro.TP.Socket as T (Socket, mapTCPSocket, mapUDPSocket) +import Metro.UDPServer +import Network.Socket (SockAddr, Socket) + +data SocketServer = TCP TCPServer + | UDP UDPServer + +data SSSID = TCPSID Socket + | UDPSID SockAddr + +instance Servable SocketServer where + data ServerConfig SocketServer = SSConfig String + type SID SocketServer = SSSID + type STP SocketServer = T.Socket + newServer (SSConfig hostPort) = + if "udp" `isPrefixOf` hostPort then UDP <$> newServer (udpServer hostPort) + else TCP <$> newServer (tcpServer hostPort) + servOnce (TCP s) done = servOnce s $ mapTCPServOnceDone done + servOnce (UDP s) done = servOnce s $ mapUDPServOnceDone done + onConnEnter (TCP s) (TCPSID sid) = onConnEnter s sid + onConnEnter (UDP s) (UDPSID sid) = onConnEnter s sid + onConnEnter _ _ = error "onConnEnter invalid type" + onConnLeave (TCP s) (TCPSID sid) = onConnLeave s sid + onConnLeave (UDP s) (UDPSID sid) = onConnLeave s sid + onConnLeave _ _ = error "onConnLeave invalid type" + servClose (TCP s) = servClose s + servClose (UDP s) = servClose s + +mapTCPServOnceDone + :: (Maybe (SID SocketServer, TransportConfig (STP SocketServer)) -> m ()) + -> Maybe (SID TCPServer, TransportConfig (STP TCPServer)) -> m () +mapTCPServOnceDone done Nothing = done Nothing +mapTCPServOnceDone done (Just (sid, stp)) = done (Just (TCPSID sid, T.mapTCPSocket stp)) + +mapUDPServOnceDone + :: (Maybe (SID SocketServer, TransportConfig (STP SocketServer)) -> m ()) + -> Maybe (SID UDPServer, TransportConfig (STP UDPServer)) -> m () +mapUDPServOnceDone done Nothing = done Nothing +mapUDPServOnceDone done (Just (sid, stp)) = done (Just (UDPSID sid, T.mapUDPSocket stp)) + +socketServer :: String -> ServerConfig SocketServer +socketServer = SSConfig diff --git a/src/Metro/TCPServer.hs b/src/Metro/TCPServer.hs new file mode 100644 --- /dev/null +++ b/src/Metro/TCPServer.hs @@ -0,0 +1,39 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE UndecidableInstances #-} + +module Metro.TCPServer + ( TCPServer + , tcpServer + ) where + +import Control.Monad (void) +import Metro.Class (Servable (..)) +import Metro.Socket (listen) +import Metro.TP.TCPSocket (TCPSocket, tcpSocket_) +import Network.Socket (Socket, SocketOption (KeepAlive), accept, + setSocketOption) +import qualified Network.Socket as Socket (close) +import UnliftIO (async, liftIO) + +newtype TCPServer = TCPServer Socket + +instance Servable TCPServer where + data ServerConfig TCPServer = TCPConfig String + type SID TCPServer = Socket + type STP TCPServer = TCPSocket + newServer (TCPConfig hostPort) = liftIO $ TCPServer <$> listen hostPort + servOnce (TCPServer serv) done = do + (sock, _) <- liftIO $ accept serv + liftIO $ setSocketOption sock KeepAlive 1 + void $ async $ do + done $ Just (sock, tcpSocket_ sock) + liftIO $ Socket.close sock + onConnEnter _ _ = return () + onConnLeave _ _ = return () + servClose (TCPServer serv) = liftIO $ Socket.close serv + +tcpServer :: String -> ServerConfig TCPServer +tcpServer = TCPConfig diff --git a/src/Metro/TP/Socket.hs b/src/Metro/TP/Socket.hs new file mode 100644 --- /dev/null +++ b/src/Metro/TP/Socket.hs @@ -0,0 +1,39 @@ +{-# LANGUAGE TypeFamilies #-} +module Metro.TP.Socket + ( Socket + , socket + , mapTCPSocket + , mapUDPSocket + ) where + +import Data.List (isPrefixOf) +import Metro.Class (Transport (..)) +import Metro.TP.TCPSocket +import Metro.TP.UDPSocket + +data Socket = TCP TCPSocket + | UDP UDPSocket + +instance Transport Socket where + data TransportConfig Socket = + TCPTC (TransportConfig TCPSocket) + | UDPTC (TransportConfig UDPSocket) + newTransport (TCPTC c) = TCP <$> newTransport c + newTransport (UDPTC c) = UDP <$> newTransport c + recvData (TCP tp) = recvData tp + recvData (UDP tp) = recvData tp + sendData (TCP tp) = sendData tp + sendData (UDP tp) = sendData tp + closeTransport (TCP tp) = closeTransport tp + closeTransport (UDP tp) = closeTransport tp + +socket :: String -> TransportConfig Socket +socket hostPort = + if "udp" `isPrefixOf` hostPort then UDPTC $ udpSocket hostPort + else TCPTC $ tcpSocket hostPort + +mapTCPSocket :: TransportConfig TCPSocket -> TransportConfig Socket +mapTCPSocket = TCPTC + +mapUDPSocket :: TransportConfig UDPSocket -> TransportConfig Socket +mapUDPSocket = UDPTC diff --git a/src/Metro/TP/TCPSocket.hs b/src/Metro/TP/TCPSocket.hs new file mode 100644 --- /dev/null +++ b/src/Metro/TP/TCPSocket.hs @@ -0,0 +1,29 @@ +{-# LANGUAGE TypeFamilies #-} +module Metro.TP.TCPSocket + ( TCPSocket + , tcpSocket + , tcpSocket_ + ) where + +import Metro.Class (Transport (..)) +import Metro.Socket (connect) +import Network.Socket (Socket, close) +import Network.Socket.ByteString (recv, sendAll) + +newtype TCPSocket = TCPSocket Socket + +instance Transport TCPSocket where + data TransportConfig TCPSocket = + RawSocket Socket + | SocketUri String + newTransport (RawSocket soc) = pure $ TCPSocket soc + newTransport (SocketUri h) = TCPSocket <$> connect h + recvData (TCPSocket soc) = recv soc + sendData (TCPSocket soc) = sendAll soc + closeTransport (TCPSocket soc) = close soc + +tcpSocket_ :: Socket -> TransportConfig TCPSocket +tcpSocket_ = RawSocket + +tcpSocket :: String -> TransportConfig TCPSocket +tcpSocket = SocketUri diff --git a/src/Metro/TP/UDPSocket.hs b/src/Metro/TP/UDPSocket.hs new file mode 100644 --- /dev/null +++ b/src/Metro/TP/UDPSocket.hs @@ -0,0 +1,51 @@ +{-# LANGUAGE TypeFamilies #-} +module Metro.TP.UDPSocket + ( UDPSocket + , udpSocket + , udpSocket_ + ) where + +import Control.Monad (forever) +import Data.ByteString (empty) +import Metro.Class (Transport (..)) +import Metro.Socket (bindTo, getDatagramAddr) +import Metro.TP.BS (BSTransport, bsTransportConfig, + feed, newBSHandle) +import Network.Socket (addrAddress) +import Network.Socket.ByteString (recvFrom, sendAllTo) +import System.Log.Logger (errorM) +import UnliftIO (Async, async, cancel) + +data UDPSocket = UDPSocket (Maybe (Async ())) BSTransport + +instance Transport UDPSocket where + data TransportConfig UDPSocket = + RawSocket (TransportConfig BSTransport) + | SocketUri String + newTransport (RawSocket h) = UDPSocket Nothing <$> newTransport h + newTransport (SocketUri h) = do + addrInfo <- getDatagramAddr h + case addrInfo of + Nothing -> error $ "Connect UDP Server " ++ h ++ " failed" + Just addrInfo0 -> do + let addr0 = addrAddress addrInfo0 + bsHandle <- newBSHandle empty + sock <- bindTo "udp://0.0.0.0:0" + + io <- async $ forever $ do + (bs, addr1) <- recvFrom sock 4194304 + if addr0 == addr1 then feed bsHandle bs + else errorM "Metro.UDP" $ "Receive unkonw address " ++ show addr1 + + tp <- newTransport $ bsTransportConfig bsHandle $ flip (sendAllTo sock) addr0 + return $ UDPSocket (Just io) tp + + recvData (UDPSocket _ soc) = recvData soc + sendData (UDPSocket _ soc) = sendData soc + closeTransport (UDPSocket io soc) = mapM_ cancel io >> closeTransport soc + +udpSocket :: String -> TransportConfig UDPSocket +udpSocket = SocketUri + +udpSocket_ :: TransportConfig BSTransport -> TransportConfig UDPSocket +udpSocket_ = RawSocket diff --git a/src/Metro/UDPServer.hs b/src/Metro/UDPServer.hs new file mode 100644 --- /dev/null +++ b/src/Metro/UDPServer.hs @@ -0,0 +1,95 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE UndecidableInstances #-} + +module Metro.UDPServer + ( UDPServer + , udpServer + , newClient + ) where + +import Control.Monad (void) +import Data.ByteString (empty) +import Data.Hashable +import Metro.Class (GetPacketId, RecvPacket, + Servable (..), Transport, + TransportConfig) +import Metro.Conn +import Metro.IOHashMap (IOHashMap, newIOHashMap) +import qualified Metro.IOHashMap as HM (delete, insert, lookup) +import Metro.Node (NodeEnv1) +import Metro.Server (ServerT, getServ, handleConn, + serverEnv) +import Metro.Session (SessionT) +import Metro.Socket (bindTo, getDatagramAddr) +import Metro.TP.BS (BSHandle, bsTransportConfig, + closeBSHandle, feed, newBSHandle) +import Metro.TP.UDPSocket (UDPSocket, udpSocket_) +import Network.Socket (SockAddr, Socket, addrAddress) +import qualified Network.Socket as Socket (close) +import Network.Socket.ByteString (recvFrom, sendAllTo) +import System.Log.Logger (errorM) +import UnliftIO + +data UDPServer = UDPServer Socket (IOHashMap String BSHandle) + +instance Servable UDPServer where + data ServerConfig UDPServer = UDPConfig String + type SID UDPServer = SockAddr + type STP UDPServer = UDPSocket + newServer (UDPConfig hostPort) = do + sock <- liftIO $ bindTo hostPort + UDPServer sock <$> newIOHashMap + servOnce us@(UDPServer serv handleList) done = do + (bs, addr) <- liftIO $ recvFrom serv 4194304 + + bsHandle <- HM.lookup handleList $ show addr + case bsHandle of + Just h -> feed h bs + Nothing -> + void . async $ do + h <- newBSHandle bs + config <- newTransportConfig us addr h + done $ Just (addr, config) + closeBSHandle h + + onConnEnter _ _ = return () + onConnLeave (UDPServer _ handleList) addr = HM.delete handleList (show addr) + servClose (UDPServer serv _) = liftIO $ Socket.close serv + +udpServer :: String -> ServerConfig UDPServer +udpServer = UDPConfig + +newTransportConfig + :: (MonadIO m) + => UDPServer + -> SockAddr + -> BSHandle + -> m (TransportConfig UDPSocket) +newTransportConfig (UDPServer sock handleList) addr h = do + HM.insert handleList (show addr) h + return $ udpSocket_ $ bsTransportConfig h $ flip (sendAllTo sock) addr + +newClient + :: (MonadUnliftIO m, Transport tp, Show nid, Eq nid, Hashable nid, Eq k, Hashable k, GetPacketId k rpkt, RecvPacket rpkt) + => (TransportConfig UDPSocket -> TransportConfig tp) + -> String + -> nid + -> u + -> (rpkt -> m Bool) + -> SessionT u nid k rpkt tp m () + -> ServerT UDPServer u nid k rpkt tp m (Maybe (NodeEnv1 u nid k rpkt tp)) +newClient mk hostPort nid uEnv preprocess sess = do + addr <- liftIO $ getDatagramAddr hostPort + case addr of + Nothing -> do + liftIO $ errorM "Metro.UDP" $ "Connect UDP Server " ++ hostPort ++ " failed" + return Nothing + Just addr0 -> do + us <- getServ <$> serverEnv + h <- newBSHandle empty + config <- mk <$> newTransportConfig us (addrAddress addr0) h + connEnv <- initConnEnv config + Just . fst <$> handleConn "Server" (addrAddress addr0) connEnv nid uEnv preprocess sess