packages feed

metro-socket (empty) → 0.1.0.0

raw patch · 12 files changed

+590/−0 lines, 12 filesdep +basedep +bytestringdep +directorysetup-changed

Dependencies added: base, bytestring, directory, hashable, hslogger, metro, mtl, network, transformers, unliftio

Files

+ ChangeLog.md view
@@ -0,0 +1,3 @@+# Changelog for metro-tcp++## Unreleased changes
+ LICENSE view
@@ -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.
+ README.md view
@@ -0,0 +1,3 @@+# metro-socket++Socket transport for metro
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ metro-socket.cabal view
@@ -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 <https://github.com/Lupino/metro/tree/master/metro-socket#readme>+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
+ src/Metro/Socket.hs view
@@ -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
+ src/Metro/SocketServer.hs view
@@ -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
+ src/Metro/TCPServer.hs view
@@ -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
+ src/Metro/TP/Socket.hs view
@@ -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
+ src/Metro/TP/TCPSocket.hs view
@@ -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
+ src/Metro/TP/UDPSocket.hs view
@@ -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
+ src/Metro/UDPServer.hs view
@@ -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