packages feed

dahdit-network (empty) → 0.5.0

raw patch · 4 files changed

+411/−0 lines, 4 filesdep +asyncdep +basedep +bytestring

Dependencies added: async, base, bytestring, dahdit, dahdit-network, network, resourcet, stm, tasty, tasty-hunit, unliftio-core

Files

+ README.md view
@@ -0,0 +1,3 @@+# dahdit-network++Network protocol helpers for Dahdit
+ dahdit-network.cabal view
@@ -0,0 +1,114 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.35.2.+--+-- see: https://github.com/sol/hpack++name:           dahdit-network+version:        0.5.0+synopsis:       Network protocol helpers for Dahdit+description:    Please see the README on GitHub at <https://github.com/ejconlon/dahdit-network#readme>+homepage:       https://github.com/ejconlon/dahdit-network#readme+bug-reports:    https://github.com/ejconlon/dahdit-network/issues+author:         Eric Conlon+maintainer:     ejconlon@gmail.com+copyright:      (c) 2023 Eric Conlon+license:        BSD3+build-type:     Simple+tested-with:+    GHC == 9.2.7+extra-source-files:+    README.md++source-repository head+  type: git+  location: https://github.com/ejconlon/dahdit-network++library+  exposed-modules:+      Dahdit.Network+  other-modules:+      Paths_dahdit_network+  hs-source-dirs:+      src+  default-extensions:+      BangPatterns+      ConstraintKinds+      DataKinds+      DeriveFunctor+      DeriveFoldable+      DeriveGeneric+      DeriveTraversable+      DerivingStrategies+      DerivingVia+      FlexibleContexts+      FlexibleInstances+      FunctionalDependencies+      GADTs+      GeneralizedNewtypeDeriving+      LambdaCase+      KindSignatures+      MultiParamTypeClasses+      Rank2Types+      ScopedTypeVariables+      StandaloneDeriving+      TupleSections+      TypeApplications+      TypeOperators+      TypeFamilies+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints -fno-warn-unused-top-binds+  build-depends:+      base >=4.12 && <5+    , bytestring ==0.11.*+    , dahdit ==0.5.*+    , network ==3.1.*+    , resourcet ==1.2.*+    , unliftio-core ==0.2.*+  default-language: GHC2021++test-suite dahdit-network-test+  type: exitcode-stdio-1.0+  main-is: Main.hs+  other-modules:+      Paths_dahdit_network+  hs-source-dirs:+      test+  default-extensions:+      BangPatterns+      ConstraintKinds+      DataKinds+      DeriveFunctor+      DeriveFoldable+      DeriveGeneric+      DeriveTraversable+      DerivingStrategies+      DerivingVia+      FlexibleContexts+      FlexibleInstances+      FunctionalDependencies+      GADTs+      GeneralizedNewtypeDeriving+      LambdaCase+      KindSignatures+      MultiParamTypeClasses+      Rank2Types+      ScopedTypeVariables+      StandaloneDeriving+      TupleSections+      TypeApplications+      TypeOperators+      TypeFamilies+  ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints -fno-warn-unused-top-binds -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      async ==2.2.*+    , base >=4.12 && <5+    , bytestring ==0.11.*+    , dahdit ==0.5.*+    , dahdit-network+    , network ==3.1.*+    , resourcet ==1.2.*+    , stm ==2.5.*+    , tasty ==1.4.*+    , tasty-hunit ==0.10.*+    , unliftio-core ==0.2.*+  default-language: GHC2021
+ src/Dahdit/Network.hs view
@@ -0,0 +1,240 @@+module Dahdit.Network+  ( Decoder (..)+  , runDecoder+  , Encoder (..)+  , runEncoder+  , Conn (..)+  , HostPort (..)+  , TcpOpts (..)+  , resolveAddr+  , tcpClientConn+  , withTcpClientConn+  , tcpServerConn+  , udpClientConn+  , withUdpClientConn+  , udpServerConn+  , withUdpServerConn+  )+where++import Control.Monad (unless, (>=>))+import Control.Monad.IO.Class (liftIO)+import Control.Monad.IO.Unlift (MonadUnliftIO)+import Dahdit (Binary (..), ByteCount (..), Get, GetError, GetIncCb, GetIncRequest (..), Put, getEnd, getTarget, getTargetInc, putTarget)+import Data.Acquire (Acquire, mkAcquire, withAcquire)+import Data.ByteString (ByteString)+import Data.ByteString qualified as BS+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.Tuple (swap)+import Network.Socket qualified as NS+import Network.Socket.ByteString qualified as NSB++maxRecv :: Int+maxRecv = 65535++maxQueue :: Int+maxQueue = 1024++sockRecvUntil :: NS.Socket -> IORef ByteString -> Int -> IO ()+sockRecvUntil sock ref len = go+ where+  go = do+    lastBs <- readIORef ref+    unless (BS.length lastBs >= len) $ do+      chunkBs <- NSB.recv sock maxRecv+      modifyIORef' ref (<> chunkBs)+      unless (BS.null chunkBs) go++-- Appropriate for TCP connections (uses 'recv' instead of 'recvFrom')+sockGetIncCb :: NS.Socket -> IO (GetIncCb ByteString IO)+sockGetIncCb sock = do+  ref <- newIORef mempty+  pure $ \(GetIncRequest _ (ByteCount off) (ByteCount len)) -> do+    modifyIORef' ref (BS.drop off)+    sockRecvUntil sock ref len+    nextBs <- readIORef ref+    pure (if BS.length nextBs >= len then Nothing else Just nextBs)++newtype Decoder k = Decoder {unDecoder :: forall a. Get a -> IO (k, Either GetError a)}+  deriving stock (Functor)++runDecoder :: (Binary a) => Decoder k -> IO (k, Either GetError a)+runDecoder dec = unDecoder dec get++-- | Decodes a stream of packets incrementally for TCP+streamDecoder :: Maybe ByteCount -> NS.Socket -> IO (Decoder ())+streamDecoder mayLim sock = do+  cb <- sockGetIncCb sock+  pure (Decoder (\getter -> fmap (\(res, _, _) -> ((), res)) (getTargetInc mayLim getter cb)))++-- | Completely decodes one packet at a time for UDP server+datagramServerDecoder :: Maybe ByteCount -> NS.Socket -> Decoder NS.SockAddr+datagramServerDecoder mayLim sock =+  let lim = maybe maxRecv unByteCount mayLim+  in  Decoder $ \getter -> do+        (bs, addr) <- NSB.recvFrom sock lim+        (ea, _) <- getTarget (getEnd getter) bs+        pure (addr, ea)++datagramClientDecoder :: Maybe ByteCount -> NS.Socket -> Decoder ()+datagramClientDecoder mayLim sock =+  let lim = maybe maxRecv unByteCount mayLim+  in  Decoder $ \getter -> do+        bs <- NSB.recv sock lim+        (ea, _) <- getTarget (getEnd getter) bs+        pure ((), ea)++newtype Encoder k = Encoder {unEncoder :: k -> Put -> IO ()}++runEncoder :: (Binary a) => Encoder k -> k -> a -> IO ()+runEncoder enc k = unEncoder enc k . put++streamEncoder :: NS.Socket -> Encoder ()+streamEncoder sock = Encoder (\_ -> putTarget >=> NSB.sendAll sock)++datagramClientEncoder :: NS.Socket -> Encoder ()+datagramClientEncoder sock = Encoder (\_ -> putTarget >=> NSB.sendAll sock)++datagramServerEncoder :: NS.Socket -> Encoder NS.SockAddr+datagramServerEncoder sock = Encoder (\addr -> putTarget >=> flip (NSB.sendAllTo sock) addr)++data Conn k = Conn {connDecoder :: Decoder k, connEncoder :: Encoder k}++data HostPort = HostPort+  { hpHost :: !(Maybe String)+  , hpPort :: !Int+  }+  deriving stock (Eq, Ord, Show)++newtype TcpOpts = TcpOpts {tcoFinTimeoutMs :: Int}+  deriving newtype (Show)+  deriving stock (Eq, Ord)++data SockTy = SockTyTcp | SockTyUdp+  deriving stock (Eq, Ord, Show, Enum, Bounded)++sockTyReal :: SockTy -> NS.SocketType+sockTyReal = \case+  SockTyTcp -> NS.Stream+  SockTyUdp -> NS.Datagram++data Role = RoleServer | RoleClient+  deriving stock (Eq, Ord, Show)++data Target = Target+  { targetHp :: !HostPort+  , targetSockTy :: !SockTy+  , targetRole :: !Role+  }+  deriving stock (Eq, Ord, Show)++resolveAddr :: HostPort -> IO NS.SockAddr+resolveAddr hp@(HostPort host port) = do+  infos <- NS.getAddrInfo Nothing host (Just (show port))+  case infos of+    [] -> fail ("Could not resolve address: " ++ show hp)+    info : _ -> pure (NS.addrAddress info)++targetResolve :: Target -> IO NS.AddrInfo+targetResolve (Target hp@(HostPort host port) sockTy role) = do+  let hints =+        NS.defaultHints+          { NS.addrSocketType = sockTyReal sockTy+          , NS.addrFlags = [NS.AI_PASSIVE | role == RoleServer]+          }+  infos <- NS.getAddrInfo (Just hints) host (Just (show port))+  case infos of+    [] -> fail ("Could not resolve address: " ++ show hp)+    info : _ -> pure info++targetOpen :: Target -> IO (NS.Socket, NS.SockAddr)+targetOpen t = do+  info <- targetResolve t+  sock <- NS.openSocket info+  pure (sock, NS.addrAddress info)++targetConnect :: Target -> IO (NS.Socket, NS.SockAddr)+targetConnect t = do+  p@(sock, addr) <- targetOpen t+  NS.connect sock addr+  pure p++targetBind :: Target -> IO NS.Socket+targetBind t = do+  (sock, addr) <- targetOpen t+  NS.setSocketOption sock NS.ReuseAddr 1+  NS.withFdSocket sock NS.setCloseOnExecIfNeeded+  NS.bind sock addr+  pure sock++tcpClientSock :: HostPort -> TcpOpts -> Acquire (NS.SockAddr, NS.Socket)+tcpClientSock hp (TcpOpts finTo) = mkAcquire acq rel+ where+  acq = fmap swap (targetConnect (Target hp SockTyTcp RoleClient))+  rel (_, sock) = if finTo > 0 then NS.gracefulClose sock finTo else NS.close sock++tcpClientConn :: Maybe ByteCount -> HostPort -> TcpOpts -> Acquire (NS.SockAddr, Conn ())+tcpClientConn mayLim hp to = do+  (addr, sock) <- tcpClientSock hp to+  dec <- liftIO (streamDecoder mayLim sock)+  let enc = streamEncoder sock+  pure (addr, Conn dec enc)++withTcpClientConn :: (MonadUnliftIO m) => Maybe ByteCount -> HostPort -> TcpOpts -> (NS.SockAddr -> Conn () -> m a) -> m a+withTcpClientConn mayLim hp to = withAcquire (tcpClientConn mayLim hp to) . uncurry++tcpServerSock :: HostPort -> Acquire NS.Socket+tcpServerSock hp = mkAcquire acq rel+ where+  acq = do+    sock <- targetBind (Target hp SockTyTcp RoleServer)+    NS.listen sock maxQueue+    pure sock+  rel = NS.close++tcpServerConn :: Maybe ByteCount -> HostPort -> TcpOpts -> Acquire (Acquire (NS.SockAddr, Conn ()))+tcpServerConn mayLim hp to = do+  srvSock <- tcpServerSock hp+  pure $ do+    (addr, cliSock) <- tcpAcceptSock to srvSock+    dec <- liftIO (streamDecoder mayLim cliSock)+    let enc = streamEncoder cliSock+    pure (addr, Conn dec enc)++tcpAcceptSock :: TcpOpts -> NS.Socket -> Acquire (NS.SockAddr, NS.Socket)+tcpAcceptSock (TcpOpts finTo) servSock = mkAcquire acq rel+ where+  acq = fmap swap (NS.accept servSock)+  rel (_, sock) = if finTo > 0 then NS.gracefulClose sock finTo else NS.close sock++udpClientSock :: HostPort -> Acquire (NS.SockAddr, NS.Socket)+udpClientSock hp = mkAcquire acq rel+ where+  acq = fmap swap (targetConnect (Target hp SockTyUdp RoleClient))+  rel = NS.close . snd++udpClientConn :: Maybe ByteCount -> HostPort -> Acquire (NS.SockAddr, Conn ())+udpClientConn mayLim hp = do+  (addr, sock) <- udpClientSock hp+  let dec = datagramClientDecoder mayLim sock+      enc = datagramClientEncoder sock+  pure (addr, Conn dec enc)++withUdpClientConn :: (MonadUnliftIO m) => Maybe ByteCount -> HostPort -> (NS.SockAddr -> Conn () -> m a) -> m a+withUdpClientConn mayLim hp = withAcquire (udpClientConn mayLim hp) . uncurry++udpServerSock :: HostPort -> Acquire NS.Socket+udpServerSock hp = mkAcquire acq rel+ where+  acq = targetBind (Target hp SockTyUdp RoleServer)+  rel = NS.close++udpServerConn :: Maybe ByteCount -> HostPort -> Acquire (Conn NS.SockAddr)+udpServerConn mayLim hp = do+  sock <- udpServerSock hp+  let dec = datagramServerDecoder mayLim sock+      enc = datagramServerEncoder sock+  pure (Conn dec enc)++withUdpServerConn :: (MonadUnliftIO m) => Maybe ByteCount -> HostPort -> (Conn NS.SockAddr -> m a) -> m a+withUdpServerConn mayLim hp = withAcquire (udpServerConn mayLim hp)
+ test/Main.hs view
@@ -0,0 +1,54 @@+module Main+  ( main+  )+where++import Control.Concurrent.Async (async, wait)+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TMVar (newEmptyTMVarIO, putTMVar, takeTMVar)+import Control.Concurrent.STM.TVar (modifyTVar, newTVarIO, readTVarIO)+import Dahdit (Binary, Generic, ViaGeneric (..))+import Dahdit.Network (Conn (..), HostPort (..), runDecoder, runEncoder, withUdpClientConn, withUdpServerConn)+import Test.Tasty (TestTree, defaultMain, testGroup)+import Test.Tasty.HUnit (testCase, (@?=))++data Foo = Foo !Int !Bool+  deriving stock (Eq, Ord, Show, Generic)+  deriving (Binary) via (ViaGeneric Foo)++data Bar = Bar !Char !Int+  deriving stock (Eq, Ord, Show, Generic)+  deriving (Binary) via (ViaGeneric Bar)++testUdp :: TestTree+testUdp = testCase "udp" $ do+  let someFoo = Foo 42 True+      someBar = Bar 'x' 42+  let lim = Just 1024+      hp = HostPort (Just "127.0.0.1") 58764+  readyVar <- newEmptyTMVarIO+  finishVar <- newTVarIO (0 :: Int)+  sx <- async $ withUdpServerConn lim hp $ \(Conn dec enc) -> do+    atomically (putTMVar readyVar ())+    (addr, gotFoo) <- runDecoder dec+    gotFoo @?= Right someFoo+    runEncoder enc addr someBar+    atomically (modifyTVar finishVar (+ 1))+  cx <- async $ withUdpClientConn lim hp $ \_ (Conn dec enc) -> do+    atomically (takeTMVar readyVar)+    runEncoder enc () someFoo+    ((), gotBar) <- runDecoder dec+    gotBar @?= Right someBar+    atomically (modifyTVar finishVar (+ 1))+  wait cx+  wait sx+  numFinished <- readTVarIO finishVar+  numFinished @?= 2++main :: IO ()+main =+  defaultMain $+    testGroup+      "DahditNetwork"+      [ testUdp+      ]