om-socket 0.11.0.5 → 1.0.0.0

raw patch · 12 files changed

+667/−495 lines, 12 filesdep +hspecdep +om-forkdep +streamingdep −binary-conduitdep −conduitdep −conduit-extraPVP ok

version bump matches the API change (PVP)

Dependencies added: hspec, om-fork, streaming, streaming-binary, streaming-bytestring, unliftio-core

Dependencies removed: binary-conduit, conduit, conduit-extra, time, transformers

API changes (from Hackage documentation)

Files

• README.md +79/−36
• examples/Client.hs +39/−0
• examples/Egress.hs +34/−0
• examples/Ingress.hs +41/−0
• examples/Server.hs +55/−0
• examples/client.hs +0/−42
• examples/egress.hs +0/−37
• examples/ingress.hs +0/−42
• examples/server.hs +0/−49
• om-socket.cabal +31/−63
• src/OM/Socket.hs +263/−226
• test/test.hs +125/−0

@@ -40,30 +40,40 @@ {-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE OverloadedStrings #-} -module Main (main) where+module Ingress (main) where -import Conduit ((.|), awaitForever, runConduit) import Control.Monad.IO.Class (MonadIO(liftIO)) import Data.Binary (Binary)+import Data.Function ((&)) import GHC.Generics (Generic)+import OM.Fork (runRace) import OM.Socket (openIngress)+import Prelude (($), IO, putStrLn)+import qualified Streaming.Prelude as Stream -{- | The messages that arrive on the socket. -}+{- |+  The messages that arrive on the socket.++  This type would typically be shared by both the Ingress and Egress side,+  or at least the binary encoding must be identical.+-} data Msg   = A   | B   deriving stock (Generic)   deriving anyclass (Binary) + main :: IO () main =-  runConduit $+  runRace $     openIngress "localhost:9000"-    .| awaitForever (\msg ->-         case msg of-           A -> liftIO $ putStrLn "Got A"-           B -> liftIO $ putStrLn "Got B"-       )+    & Stream.mapM_+       (\msg ->+           case msg of+             A -> liftIO $ putStrLn "Got A"+             B -> liftIO $ putStrLn "Got B"+         ) ```    @@ -75,25 +85,33 @@ {-# LANGUAGE DerivingStrategies #-} {-# LANGUAGE OverloadedStrings #-} -module Main (main) where+module Egress (main) where -import Conduit ((.|), runConduit, yield) import Data.Binary (Binary)+import Data.Function ((&)) import GHC.Generics (Generic) import OM.Socket (openEgress)+import Prelude (IO)+import qualified Streaming.Prelude as Stream -{- | The messages that arrive on the socket. -}+{- |+  The messages that arrive on the socket.++  This type would typically be shared between the client and the server,+  or at least the binary representation must be identical (as, for+  example, if the client is not written in Haskell).+-} data Msg   = A   | B   deriving stock (Generic)   deriving anyclass (Binary) + main :: IO () main =-  runConduit $-    mapM_ yield [A, B, B, A, A, A, B]-    .| openEgress "localhost:9000"+  Stream.each [A, B, B, A, A, A, B]+  & openEgress "localhost:9000"  ``` @@ -105,39 +123,55 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings #-} -module Main (main) where -import Conduit ((.|), awaitForever, runConduit)+module Server (main) where+ import Control.Monad.Logger (runStdoutLoggingT)-import Control.Monad.Trans.Class (MonadTrans(lift)) import Data.Binary (Binary)+import Data.Function ((&))+import OM.Fork (runRace) import OM.Socket (openServer)+import Prelude (Maybe(Nothing), ($), IO, Show, String)+import qualified Streaming.Prelude as Stream -{- | The requests accepted by the server. -}++{-|+  The requests accepted by the server.++  This type would typically be shared between client and server.+-} newtype Request = EchoRequest String   deriving newtype (Binary, Show)  -{- | The response sent back to the client. -}+{-|+  The response sent back to the client.++  This type would typically be shared between client and server.+-} newtype Responsee = EchoResponse String   deriving newtype (Binary, Show)  -{- | Simple echo resposne server. -}+{-| Simple echo resposne server. -} main :: IO () main =-  runStdoutLoggingT . runConduit $-    pure ()-    .| openServer "localhost:9000" Nothing-    .| awaitForever (\(EchoRequest str, respond) ->-        {--          You don't necessarily have to respond right away if you don't-          want to. You can cache the responder away in some state and-          get back to it at some later time if you like.-        -}-        lift $ respond (EchoResponse str)+  runRace+  $ runStdoutLoggingT+  $ (+      openServer "localhost:9000" Nothing+      & Stream.mapM_+          (\ (EchoRequest str, respond) ->+            {-+              You don't necessarily have to respond right away if+              you don't want to. You can cache the responder away in+              some state and get back to it at some later time if you+              like. `openServer` and `connectServer` have a mechnamism+              for handling out of order responses.+            -}+            respond (EchoResponse str)+          )     )- ```  ### Connect a client to a server@@ -147,23 +181,32 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings #-} -module Main (main) where+module Client (main) where  import Control.Monad.Logger (runStdoutLoggingT) import Data.Binary (Binary) import OM.Socket (connectServer) -{- | The requests accepted by the server. -}+{-|+  The requests accepted by the server.++  This would typically need to be shared between the client and the+  server. Certainly the binary encoding must be identical.+-} newtype Request = EchoRequest String   deriving newtype (Binary, Show)  -{- | The response sent back to the client. -}+{-|+  The response sent back to the client.++  Likewise, this would be shared between client and server.+-} newtype Responsee = EchoResponse String   deriving newtype (Binary, Show)  -{- | Simple echo resposne client. -}+{-| Simple echo resposne client. -} main :: IO () main = do   client <-

@@ -0,0 +1,39 @@+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}++module Client (main) where++import Control.Monad.Logger (runStdoutLoggingT)+import Data.Binary (Binary)+import OM.Socket (connectServer)++{-|+  The requests accepted by the server.++  This would typically need to be shared between the client and the+  server. Certainly the binary encoding must be identical.+-}+newtype Request = EchoRequest String+  deriving newtype (Binary, Show)+++{-|+  The response sent back to the client.++  Likewise, this would be shared between client and server.+-}+newtype Responsee = EchoResponse String+  deriving newtype (Binary, Show)+++{-| Simple echo resposne client. -}+main :: IO ()+main = do+  client <-+    runStdoutLoggingT $+      connectServer "localhost:9000" Nothing+  putStrLn =<< client (EchoRequest "hello")+  putStrLn =<< client (EchoRequest "world")++

@@ -0,0 +1,34 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE OverloadedStrings #-}++module Egress (main) where++import Data.Binary (Binary)+import Data.Function ((&))+import GHC.Generics (Generic)+import OM.Socket (openEgress)+import Prelude (IO)+import qualified Streaming.Prelude as Stream++{- |+  The messages that arrive on the socket.++  This type would typically be shared between the client and the server,+  or at least the binary representation must be identical (as, for+  example, if the client is not written in Haskell).+-}+data Msg+  = A+  | B+  deriving stock (Generic)+  deriving anyclass (Binary)+++main :: IO ()+main =+  Stream.each [A, B, B, A, A, A, B]+  & openEgress "localhost:9000"++

@@ -0,0 +1,41 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE OverloadedStrings #-}++module Ingress (main) where++import Control.Monad.IO.Class (MonadIO(liftIO))+import Data.Binary (Binary)+import Data.Function ((&))+import GHC.Generics (Generic)+import OM.Fork (runRace)+import OM.Socket (openIngress)+import Prelude (($), IO, putStrLn)+import qualified Streaming.Prelude as Stream++{- |+  The messages that arrive on the socket.++  This type would typically be shared by both the Ingress and Egress side,+  or at least the binary encoding must be identical.+-}+data Msg+  = A+  | B+  deriving stock (Generic)+  deriving anyclass (Binary)+++main :: IO ()+main =+  runRace $+    openIngress "localhost:9000"+    & Stream.mapM_+       (\msg ->+           case msg of+             A -> liftIO $ putStrLn "Got A"+             B -> liftIO $ putStrLn "Got B"+         )++

@@ -0,0 +1,55 @@+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}+++module Server (main) where++import Control.Monad.Logger (runStdoutLoggingT)+import Data.Binary (Binary)+import Data.Function ((&))+import OM.Fork (runRace)+import OM.Socket (openServer)+import Prelude (Maybe(Nothing), ($), IO, Show, String)+import qualified Streaming.Prelude as Stream+++{-|+  The requests accepted by the server.++  This type would typically be shared between client and server.+-}+newtype Request = EchoRequest String+  deriving newtype (Binary, Show)+++{-|+  The response sent back to the client.++  This type would typically be shared between client and server.+-}+newtype Responsee = EchoResponse String+  deriving newtype (Binary, Show)+++{-| Simple echo resposne server. -}+main :: IO ()+main =+  runRace+  $ runStdoutLoggingT+  $ (+      openServer "localhost:9000" Nothing+      & Stream.mapM_+          (\ (EchoRequest str, respond) ->+            {-+              You don't necessarily have to respond right away if+              you don't want to. You can cache the responder away in+              some state and get back to it at some later time if you+              like. `openServer` and `connectServer` have a mechnamism+              for handling out of order responses.+            -}+            respond (EchoResponse str)+          )+    )++

@@ -1,42 +0,0 @@-{-# LANGUAGE DerivingStrategies #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedStrings #-}--{-# OPTIONS_GHC -Wwarn #-}--module Main (main) where--import Control.Monad.Logger (runStdoutLoggingT)-import Data.Binary (Binary)-import OM.Socket (connectServer)--{- | The requests accepted by the server. -}-newtype Request = EchoRequest String-  deriving newtype (Binary, Show)---{- | The response sent back to the client. -}-newtype Responsee = EchoResponse String-  deriving newtype (Binary, Show)---{- | Simple echo resposne server. -}-main :: IO ()-main = do-  {--    Don't actually call sendRequests, because there is no server running to-    connect to, which will cause an error, which will cause the test to fail.-  -}-  -- sendRequests-  pure ()---sendRequests :: IO ()-sendRequests = do-  client <--    runStdoutLoggingT $-      connectServer "localhost:9000" Nothing-  putStrLn =<< client (EchoRequest "hello")-  putStrLn =<< client (EchoRequest "world")--

@@ -1,37 +0,0 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE DerivingStrategies #-}-{-# LANGUAGE OverloadedStrings #-}--{-# OPTIONS_GHC -Wwarn #-}--module Main (main) where--import Conduit ((.|), runConduit, yield)-import Data.Binary (Binary)-import GHC.Generics (Generic)-import OM.Socket (openEgress)--{- | The messages that arrive on the socket. -}-data Msg-  = A-  | B-  deriving stock (Generic)-  deriving anyclass (Binary)--main :: IO ()-main =-  {--    Don't actually call sendMessages, because there is no server running to-    connect to, which will cause an error, which will cause the test to fail.-  -}-  -- sendMessages-  pure ()--sendMessages :: IO ()-sendMessages = do-  runConduit $-    mapM_ yield [A, B, B, A, A, A, B]-    .| openEgress "localhost:9000"--

@@ -1,42 +0,0 @@-{-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE DerivingStrategies #-}-{-# LANGUAGE OverloadedStrings #-}--{-# OPTIONS_GHC -Wwarn #-}--module Main (main) where--import Conduit ((.|), awaitForever, runConduit)-import Control.Monad.IO.Class (MonadIO(liftIO))-import Data.Binary (Binary)-import GHC.Generics (Generic)-import OM.Socket (openIngress)--{- | The messages that arrive on the socket. -}-data Msg-  = A-  | B-  deriving stock (Generic)-  deriving anyclass (Binary)--main :: IO ()-main =-  {--    Don't actually call serveForever, because the "test" we are using to make-    sure this compiles will never finish running!-  -}-  -- serveForever-  pure ()--serveForever :: IO ()-serveForever =-  runConduit $-    openIngress "localhost:9000"-    .| awaitForever (\msg ->-         case msg of-           A -> liftIO $ putStrLn "Got A"-           B -> liftIO $ putStrLn "Got B"-       )--  

@@ -1,49 +0,0 @@-{-# LANGUAGE DerivingStrategies #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedStrings #-}--{-# OPTIONS_GHC -Wwarn #-}--module Main (main) where--import Conduit ((.|), awaitForever, runConduit)-import Control.Monad.Logger (runStdoutLoggingT)-import Control.Monad.Trans.Class (MonadTrans(lift))-import Data.Binary (Binary)-import OM.Socket (openServer)--{- | The requests accepted by the server. -}-newtype Request = EchoRequest String-  deriving newtype (Binary, Show)---{- | The response sent back to the client. -}-newtype Responsee = EchoResponse String-  deriving newtype (Binary, Show)---{- | Simple echo resposne server. -}-main :: IO ()-main = do-  {--    Don't actually call server, because the "test" we are using to make-    sure this compiles will never finish running!-  -}-  -- server-  pure ()---server :: IO ()-server =-  runStdoutLoggingT . runConduit $-    pure ()-    .| openServer "localhost:9000" Nothing-    .| awaitForever (\(EchoRequest str, respond) ->-        {--          You don't necessarily have to respond right away if you don't-          want to. You can cache the responder away in some state and-          get back to it at some later time if you like.-        -}-        lift $ respond (EchoResponse str)-    )-

@@ -1,6 +1,6 @@ cabal-version:       3.0 name:                om-socket-version:             0.11.0.5+version:             1.0.0.0 synopsis:            Socket utilities. description:         Binary ingress server, egress client, and                      bidirectional binarry client/server@@ -25,23 +25,24 @@  common dependencies   build-depends:-    , aeson          >= 2.0.3.0   && < 2.3-    , base           >= 4.15.1.0  && < 4.20-    , binary         >= 0.8.8.0   && < 0.9-    , binary-conduit >= 1.3.1     && < 1.4-    , bytestring     >= 0.10.12.1 && < 0.13-    , conduit        >= 1.3.4.3   && < 1.4-    , conduit-extra  >= 1.3.6     && < 1.4-    , containers     >= 0.6.4.1   && < 0.7-    , exceptions     >= 0.10.4    && < 0.11-    , megaparsec     >= 9.2.2     && < 9.7-    , monad-logger   >= 0.3.37    && < 0.4-    , network        >= 3.1.2.7   && < 3.2-    , om-show        >= 0.1.2.6   && < 0.2-    , stm            >= 2.5.0.0   && < 2.6-    , text           >= 1.2.5.0   && < 2.2-    , time           >= 1.9.3     && < 1.13-    , tls            >= 1.5.8     && < 1.10+    , aeson                >= 2.0.3.0   && < 2.3+    , base                 >= 4.15.1.0  && < 4.20+    , binary               >= 0.8.8.0   && < 0.9+    , bytestring           >= 0.10.12.1 && < 0.13+    , containers           >= 0.6.4.1   && < 0.7+    , exceptions           >= 0.10.4    && < 0.11+    , megaparsec           >= 9.2.2     && < 9.7+    , monad-logger         >= 0.3.37    && < 0.4+    , network              >= 3.1.2.7   && < 3.2+    , om-fork              >= 0.7.1.9   && < 0.8+    , om-show              >= 0.1.2.6   && < 0.2+    , stm                  >= 2.5.0.0   && < 2.6+    , streaming            >= 0.2.4.0   && < 0.3+    , streaming-binary     >= 0.3.0.1   && < 0.4+    , streaming-bytestring >= 0.3.2     && < 0.4+    , text                 >= 1.2.5.0   && < 2.2+    , tls                  >= 1.5.8     && < 1.10+    , unliftio-core        >= 0.2.1.0   && < 0.3  library   import: warnings, dependencies@@ -52,52 +53,19 @@   hs-source-dirs: src   default-language: Haskell2010 --- This isn't really a test, it is just used to make sure the example--- compiles without clusttering up the package with a bunch of--- executables.-test-suite ingress-example-  import: warnings, dependencies-  main-is: ingress.hs-  type: exitcode-stdio-1.0-  hs-source-dirs: examples-  default-language: Haskell2010-  build-depends:-    , om-socket---- This isn't really a test, it is just used to make sure the example--- compiles without clusttering up the package with a bunch of--- executables.-test-suite egress-example-  import: warnings, dependencies-  main-is: egress.hs-  type: exitcode-stdio-1.0-  hs-source-dirs: examples-  default-language: Haskell2010-  build-depends:-    , om-socket---- This isn't really a test, it is just used to make sure the example--- compiles without clusttering up the package with a bunch of--- executables.-test-suite server-example-  import: warnings, dependencies-  main-is: server.hs-  type: exitcode-stdio-1.0-  hs-source-dirs: examples-  default-language: Haskell2010-  build-depends:-    , om-socket-    , transformers >= 0.5.6.2 && < 0.7---- This isn't really a test, it is just used to make sure the example--- compiles without clusttering up the package with a bunch of--- executables.-test-suite client-example-  import: warnings, dependencies-  main-is: client.hs+test-suite test+  import: dependencies, warnings+  main-is: test.hs   type: exitcode-stdio-1.0-  hs-source-dirs: examples+  other-modules:+    Client+    Egress+    Ingress+    Server+  hs-source-dirs:+    , test+    , examples   default-language: Haskell2010   build-depends:     , om-socket-+    , hspec >= 2.11.1 && < 2.12

@@ -4,6 +4,7 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}  {- | Socket utilities. -} module OM.Socket (@@ -24,30 +25,25 @@ ) where  -import Control.Applicative (Alternative((<|>)))-import Control.Concurrent (Chan, MVar, forkIO, newChan, newEmptyMVar,-  putMVar, readChan, takeMVar, throwTo, writeChan)-import Control.Concurrent.STM (TVar, atomically, newTVar, readTVar,+import Control.Concurrent (MVar, forkIO, newEmptyMVar, putMVar, takeMVar)+import Control.Concurrent.STM (TVar, atomically, newTVarIO, readTVar,   retry, writeTVar) import Control.Exception (SomeException, bracketOnError, throw)-import Control.Monad (join, void, when)+import Control.Monad (when) import Control.Monad.Catch (MonadThrow(throwM), MonadCatch, try)-import Control.Monad.IO.Class (MonadIO(liftIO))+import Control.Monad.IO.Unlift (MonadUnliftIO) import Control.Monad.Logger.CallStack (LoggingT(runLoggingT),-  MonadLoggerIO(askLoggerIO), logDebug, logError, logWarn)+  MonadLoggerIO(askLoggerIO), NoLoggingT(runNoLoggingT), MonadLogger,+  logDebug, logError, logWarn) import Data.Aeson (FromJSON, FromJSONKey, ToJSON, ToJSONKey) import Data.Binary (Binary(get), encode) import Data.Binary.Get (Decoder(Done, Fail, Partial), pushChunk,   runGetIncremental) import Data.ByteString (ByteString)-import Data.Conduit ((.|), ConduitT, awaitForever, runConduit, transPipe,-  yield)-import Data.Conduit.Network (sinkSocket, sourceSocket)-import Data.Conduit.Serialization.Binary (conduitDecode, conduitEncode)+import Data.Foldable (traverse_) import Data.Map (Map) import Data.String (IsString) import Data.Text (Text)-import Data.Time (diffUTCTime, getCurrentTime) import Data.Void (Void) import Data.Word (Word32) import GHC.Generics (Generic)@@ -60,58 +56,71 @@ import Network.Socket.ByteString.Lazy (sendAll) import Network.TLS (ClientParams, Context, ServerParams, contextNew,   handshake, recvData, sendData)+import OM.Fork (Race, race) import OM.Show (showt) import Prelude (Applicative(pure), Bool(False, True), Bounded(minBound),   Either(Left, Right), Enum(succ), Eq((/=)), Functor(fmap), Maybe(Just,-  Nothing), Monad((>>), (>>=), return), MonadFail(fail), Semigroup((<>)),-  Show(show), ($), (++), (.), (=<<), IO, Monoid, Num, Ord, String,-  sequence_, snd, userError)+  Nothing), Monad((>>), (>>=), return), MonadFail(fail), Monoid(mempty),+  Semigroup((<>)), Show(show), ($), (++), (.), (=<<), IO, Num, Ord,+  String, flip, snd, userError)+import Streaming (Alternative((<|>)), MFunctor(hoist), MonadIO(liftIO),+  MonadTrans(lift), Of, Stream, join, void)+import Streaming.Binary (decoded)+import Streaming.ByteString (ByteStream, reread) import Text.Megaparsec (MonadParsec(eof), Parsec, many, oneOf, parse,   satisfy) import Text.Megaparsec.Char (char) import qualified Data.ByteString as BS import qualified Data.ByteString.Lazy as BSL-import qualified Data.Conduit.List as CL import qualified Data.Map as Map import qualified Data.Text as T+import qualified Streaming.Prelude as S import qualified Text.Megaparsec as M  -{- |-  Opens an "ingress" socket, which is a socket that accepts a stream of-  messages without responding.+{-|+  Opens an "ingress" socket, which is a socket that accepts a stream+  of messages without responding. In particular, we listen on a socket,+  accepting new connections, an each connection concurrently reads its+  elements off the socket and pushes them onto the stream. -}-openIngress :: (Binary i, MonadIO m, MonadFail m)+openIngress+  :: forall i m never_returns.+     ( Binary i+     , MonadFail m+     , MonadIO m+     , Race+     )   => AddressDescription-  -> ConduitT () i m ()+  -> Stream (Of i) m never_returns openIngress bindAddr = do     so <- listenSocket =<< resolveAddr bindAddr     mvar <- liftIO newEmptyMVar-    void . liftIO . forkIO $ acceptLoop so mvar-    mvarToSource mvar+    liftIO+      . runNoLoggingT+      . race "ingress accept loop"+      . liftIO+      $ acceptLoop so mvar+    mvarToStream mvar   where-    mvarToSource :: (MonadIO m) => MVar a -> ConduitT () a m ()-    mvarToSource mvar = do-      liftIO (takeMVar mvar) >>= yield-      mvarToSource mvar--    acceptLoop :: (Binary i) => Socket -> MVar i -> IO ()+    acceptLoop :: Socket -> MVar i -> IO ()     acceptLoop so mvar = do       (conn, _) <- accept so       void . forkIO $ feed (runGetIncremental get) conn mvar       acceptLoop so mvar -    feed :: (Binary i) => Decoder i -> Socket -> MVar i -> IO ()-+    feed+      :: Decoder i+      -> Socket+      -> MVar i+      -> IO ()     feed (Done leftover _ i) conn mvar = do       putMVar mvar i       feed (runGetIncremental get `pushChunk` leftover) conn mvar-     feed (Partial k) conn mvar = do       bytes <- recv conn 4096       when (BS.null bytes) (fail "Socket closed by peer.")       feed (k (Just bytes)) conn mvar-     feed (Fail _ _ err) _conn _chan =       fail $ "Socket crashed. Decoding error: " ++ show err @@ -124,13 +133,18 @@   :: ( Binary o      , MonadFail m      , MonadIO m-     , MonadThrow m      )   => AddressDescription-  -> ConduitT o Void m ()-openEgress addr = do+  -> Stream (Of o) m r+  -> m r+openEgress addr stream = do   so <- connectSocket =<< resolveAddr addr-  conduitEncode .| sinkSocket so+  result <-+    S.mapM_+      (liftIO . sendAll so . encode)+      stream+  liftIO (close so)+  pure result   {- | Guess the family of a `SockAddr`. -}@@ -206,118 +220,86 @@   and provides a way to respond to those requests. -} openServer-  :: ( Binary request+  :: forall request response m never_returns.+     ( Binary request      , Binary response+     , MonadLogger m+     , MonadCatch m      , MonadFail m-     , MonadLoggerIO m-     , Show request-     , Show response+     , MonadUnliftIO m+     , Race      )   => AddressDescription   -> Maybe (IO ServerParams)-  -> ConduitT Void (request, response -> m Responded) m ()+  -> Stream (Of (request, response -> m Responded)) m never_returns openServer bindAddr tls = do     so <- listenSocket =<< resolveAddr bindAddr-    requestChan <- liftIO newChan-    logging <- askLoggerIO-    void . liftIO . forkIO . (`runLoggingT` logging) $ acceptLoop so requestChan-    chanToSource requestChan+    requestMVar <- liftIO newEmptyMVar+    lift+      . race "server accept loop"+      $ acceptLoop so requestMVar+    mvarToStream requestMVar   where     acceptLoop-      :: ( Binary i-         , Binary o-         , MonadIO m-         , MonadLoggerIO n-         , Show i-         , Show o-         )-      => Socket-      -> Chan (i, o -> m Responded)-      -> n ()-    acceptLoop so requestChan = do+      :: Socket+      -> MVar (request, response -> m Responded)+      -> m void+    acceptLoop so requestMVar = do       (conn, ra) <- liftIO (accept so)-      (inputSource, outputSink) <- prepareConnection conn       logDebug $ "New connection: " <>  showt ra-      responseChan <- liftIO newChan-      logging <- askLoggerIO-      rtid <--        liftIO-        . forkIO-        . (`runLoggingT` logging)-        $ responderThread responseChan outputSink-      void . liftIO . forkIO . (`runLoggingT` logging) $ do-        result <- try $ runConduit (-            pure ()-            .| transPipe liftIO inputSource-            .| conduitDecode-            .| awaitForever (\req@Request {messageId, payload} -> do-                logDebug $ showt ra <> ": Got request: " <> showt req-                start <- liftIO getCurrentTime-                yield (-                    payload,-                    \res -> do-                      liftIO . writeChan responseChan . Response messageId $ res-                      end <- liftIO getCurrentTime-                      liftIO . (`runLoggingT` logging) . logDebug-                        $ showt ra <> ": Responded to " <> showt messageId <> " in ("-                        <> showt (diffUTCTime end start) <> ")"-                      pure Responded-                  )-              )-            .| CL.mapM_ (liftIO . writeChan requestChan)-          )-        case result of-          Left err -> liftIO $ throwTo rtid (err :: SomeException)-          Right () -> return ()-        logDebug $ "Closed connection: " <>  showt ra-      acceptLoop so requestChan+      (input, send) <- prepareConnection conn+      void . liftIO . forkIO $ handleConnection input send requestMVar+      acceptLoop so requestMVar -    prepareConnection-      :: (MonadIO m)-      => Socket-      -> m (ConduitT Void ByteString IO (), ConduitT ByteString Void IO ())-    prepareConnection conn =-        case tls of-          Nothing -> pure (sourceSocket conn, sinkSocket conn)-          Just getParams ->-            liftIO $ do-              ctx <- contextNew conn =<< getParams-              handshake ctx-              pure (input ctx, output ctx)+    handleConnection+      :: ByteStream IO () {-^ raw bytes input from the socket -}+      -> (BSL.ByteString -> m ()) {-^ How to send bytes back -}+      -> MVar (request, response -> m Responded)+         {-^ how we stream (req, respond) tuples to the client code. -}+      -> IO ()+    handleConnection input send requestMVar =+        void $+          S.mapM_+            sendRequestToMVar+            (decoded input)       where-        output :: Context -> ConduitT ByteString Void IO ()-        output ctx = awaitForever (sendData ctx . BSL.fromStrict)+        sendRequestToMVar :: Request request -> IO ()+        sendRequestToMVar Request { messageId , payload } =+          putMVar+            requestMVar+            ( payload+            , respond messageId+            ) -        input :: Context -> ConduitT Void ByteString IO ()-        input ctx = do-          bytes <- recvData ctx-          if BS.null bytes then-            pure ()-          else do-            yield bytes-            input ctx+        respond :: MessageId -> response -> m Responded+        respond responseTo response = do+          send . encode $ Response { responseTo , response }+          pure Responded -    responderThread :: (-          Binary p,-          MonadLoggerIO m,-          MonadThrow m,-          Show p-        )-      => Chan (Response p)-      -> ConduitT ByteString Void IO ()-      -> m ()-    responderThread chan outputSink = runConduit (-        pure ()-        .| chanToSource chan-        .| awaitForever (\res@Response {responseTo, response} -> do-            logDebug-              $ "Responding to " <> showt responseTo-              <> " with: " <> showt response-            yield res+    {- Maybe make a TLS connection. -}+    prepareConnection+      :: Socket+      -> m+          ( ByteStream IO ()+          , BSL.ByteString -> m ()           )-        .| conduitEncode-        .| transPipe liftIO outputSink-      )+    prepareConnection conn =+      case tls of+        Nothing ->+          pure+            ( rereadNull+                (flip recv 4096)+                conn+            , liftIO . sendAll conn+            )+        Just getParams ->+          liftIO $ do+            ctx <- contextNew conn =<< getParams+            handshake ctx+            pure+              ( rereadNull recvData ctx+              , sendData ctx+              )   {- |@@ -326,7 +308,8 @@   the server. -} connectServer-  :: ( Binary request+  :: forall n request m response.+     ( Binary request      , Binary response      , MonadIO m      , MonadLoggerIO n@@ -339,24 +322,26 @@     logging <- askLoggerIO     liftIO $ do       so <- connectSocket =<< resolveAddr addr-      state <- atomically (newTVar ClientState {-          csAlive = True,-          csResponders = Map.empty,-          csMessageId = minBound,-          csMessageQueue = []-        })-      (send, reqSource) <- prepareConnection so+      state <-+        newTVarIO+          ClientState+            { csServerAlive = True+            , csResponders = Map.empty+            , csMessageId = minBound+            , csRequestQueue = []+            }+      (send, responseSource) <- prepareConnection so       void . forkIO $ (`runLoggingT` logging) (requestThread send state)-      void . forkIO $ (`runLoggingT` logging) (responseThread reqSource state)+      void . forkIO $ (`runLoggingT` logging) (responseThread responseSource state)       return (\i -> liftIO $ do           mvar <- newEmptyMVar           join . atomically $             readTVar state >>= \case-              ClientState {csAlive = False} -> return $+              ClientState {csServerAlive = False} -> return $                 throwM (userError "Server connection died.")-              s@ClientState {csMessageQueue} -> do+              s@ClientState {csRequestQueue} -> do                 writeTVar state s {-                    csMessageQueue = csMessageQueue <> [(i, putMVar mvar)]+                    csRequestQueue = csRequestQueue <> [(i, putMVar mvar)]                   }                 return (takeMVar mvar)         )@@ -366,51 +351,55 @@       for TSL or not depending on the configuration.     -}     prepareConnection-      :: (MonadIO m, MonadIO f)-      => Socket-      -> f (BSL.ByteString -> IO (), ConduitT Void ByteString m ())+      :: Socket+      -> IO+           ( BSL.ByteString -> IO ()+           , ByteStream IO ()+           )     prepareConnection so =         case tls of-          Nothing -> pure (sendAll so, sourceSocket so)+          Nothing ->+            pure+              ( sendAll so+              , rereadNull+                  (flip recv 4096)+                  so+              )           Just params -> do             ctx <- contextNew so params             handshake ctx-            pure (send ctx, reqSource ctx)+            pure (send ctx, resSource ctx)       where         send :: Context -> BSL.ByteString -> IO ()         send = sendData -        reqSource :: (MonadIO m) => Context -> ConduitT Void ByteString m ()-        reqSource ctx = do-          bytes <- recvData ctx-          if BS.null bytes-            then pure ()-            else do-              yield bytes-              reqSource ctx+        resSource+          :: Context+          -> ByteStream IO ()+        resSource = do+          rereadNull recvData -    {- | Receive requests and send them to the server. -}-    requestThread :: (-          Binary i,-          MonadCatch m,-          MonadLoggerIO m-        )-      => (BSL.ByteString -> IO ())-      -> TVar (ClientState i o)-      -> m ()+    {- |+      Receive requests from the client request function and send them+      to the server.+    -}+    requestThread+      :: (BSL.ByteString -> IO ())+      -> TVar (ClientState request response)+      -> LoggingT IO ()     requestThread send state =       join . liftIO . atomically $         readTVar state >>= \case-          ClientState {csAlive = False} -> pure (pure ())-          ClientState {csMessageQueue = []} -> retry+          ClientState {csServerAlive = False} -> pure (pure ())+          ClientState {csRequestQueue = []} -> retry           s@ClientState {-                csMessageQueue = (m, r):remaining,+                csRequestQueue = (m, r):remaining,                 csResponders,                 csMessageId               }             -> do               writeTVar state s {-                  csMessageQueue = remaining,+                  csRequestQueue = remaining,                   csResponders = Map.insert csMessageId r csResponders,                   csMessageId = succ csMessageId                 }@@ -420,47 +409,64 @@      {- |       Receive responses from the server and send then them back to the-      client responder.+      client request function.     -}-    responseThread :: (-          Binary o,-          MonadLoggerIO m,-          MonadCatch m,-          Show o-        )-      => ConduitT Void ByteString m ()-      -> TVar (ClientState i o)-      -> m ()-    responseThread reqSource state = do-      (try . runConduit) (-          pure ()-          .| reqSource-          .| conduitDecode-          .| CL.mapM_ (\r@Response {responseTo, response} ->-              join . liftIO . atomically $-                readTVar state >>= \ ClientState {csResponders} ->-                  case Map.lookup responseTo csResponders of-                    Nothing -> return $-                      logWarn $ "Unexpected server response: " <> showt r-                    Just respond -> return $ liftIO (respond response)+    responseThread+      :: ByteStream IO ()+      -> TVar (ClientState request response)+      -> LoggingT IO ()+    responseThread resSource stateT = do+        try+            (+              void $+                S.mapM_+                  handleResponse+                  (hoist liftIO (decoded resSource))             )-        ) >>= \case-          Left err ->-            logError-              $ "Socket receive error: "-              <> showt (err :: SomeException)-          Right () -> return ()-      join . liftIO . atomically $-        readTVar state >>= \s@ClientState {csResponders, csMessageQueue} -> do-          writeTVar state s {csAlive = False}-          return . liftIO . sequence_ $ [-              r (throw (userError "Remote connection died."))-              | r <--                  fmap snd (Map.toList csResponders)-                  <> fmap snd csMessageQueue-            ]+          >>= \case+            Left err -> do+              logError+                $ "Socket receive error: "+                <> showt (err :: SomeException)+              throw err+            Right () ->+              pure ()+        closeClientState+      where+        closeClientState =+          join . liftIO . atomically $ do+            state <- readTVar stateT+            writeTVar stateT state+              { csServerAlive = False+              , csResponders = mempty+              , csRequestQueue = mempty+              } +            pure . traverse_ liftIO $+              [ respond (throw (userError "Remote connection died."))+              | respond <-+                  Map.elems (csResponders state)+                  <>  fmap snd (csRequestQueue state)+              ] +        handleResponse :: Response response -> LoggingT IO ()+        handleResponse+            responsePackage@Response+              { responseTo+              , response+              }+          = do+              join . lift . atomically $ do+                state <- readTVar stateT+                case deleteFind responseTo (csResponders state) of+                  Nothing ->+                    pure . logWarn $+                      "Unexpected server response: " <> showt responsePackage+                  Just (respond, newResponders) -> do+                    writeTVar stateT state {csResponders = newResponders}+                    pure . lift $ respond response++ {- | A server endpoint configuration. -} data Endpoint = Endpoint {     bindAddr :: AddressDescription,@@ -470,9 +476,9 @@   {- | Response to a request. -}-data Response p = Response {-    responseTo :: MessageId,-      response :: p+data Response p = Response+  { responseTo :: MessageId+  ,   response :: p   }   deriving stock (Generic, Show) instance (Binary p) => Binary (Response p)@@ -509,18 +515,18 @@   {- | Client connection state. -}-data ClientState i o = ClientState {-    csAlive :: Bool,-    csResponders :: Map MessageId (o -> IO ()),-    csMessageId :: MessageId,-    csMessageQueue :: [(i, o -> IO ())]+data ClientState i o = ClientState+  {  csServerAlive :: Bool+  ,   csResponders :: Map MessageId (o -> IO ())+  ,    csMessageId :: MessageId+  , csRequestQueue :: [(i, o -> IO ())]   }   {- | A Request message type. -}-data Request p = Request {-    messageId :: MessageId,-      payload :: p+data Request p = Request+  { messageId :: MessageId+  ,   payload :: p   }   deriving stock (Generic, Show) instance (Binary p) => Binary (Request p)@@ -533,13 +539,6 @@   deriving newtype (Binary, Num, Bounded, Eq, Ord, Show, Enum)  -{- | Construct a coundiut source by reading forever from a 'Chan'. -}-chanToSource :: (MonadIO m) => Chan a -> ConduitT Void a m ()-chanToSource chan = do-  yield =<< liftIO (readChan chan)-  chanToSource chan-- {- |   Proof that a response function was called on the server. Mainly   useful for including in a type signature somewhere in your server@@ -547,5 +546,43 @@   request in all cases. -} data Responded = Responded+++mvarToStream+  :: (MonadIO m)+  => MVar i+  -> Stream (Of i) m never_returns+mvarToStream mvar = do+  liftIO (takeMVar mvar) >>= S.yield+  mvarToStream mvar+++rereadNull+  :: (Monad m)+  => (c -> m ByteString)+  -> c+  -> ByteStream m ()+rereadNull f =+  reread+    (\c -> do+      bytes <- f c+      pure $ if BS.null bytes then Nothing else Just bytes+    )+++{-|+  If the key exists in the map, delete it and return its value along+  with the new map.+-}+deleteFind+  :: (Ord k)+  => k+  -> Map k v+  -> Maybe (v, Map k v)+deleteFind key m =+  case Map.lookup key m of+    Nothing -> Nothing+    Just v ->+      Just (v, Map.delete key m)  

@@ -0,0 +1,125 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar,+  threadDelay)+import Control.Monad (void)+import Control.Monad.Catch (MonadCatch)+import Control.Monad.IO.Unlift (MonadIO(liftIO), MonadUnliftIO)+import Control.Monad.Logger (MonadLogger, runStdoutLoggingT)+import Data.Function ((&))+import Data.Text (Text)+import OM.Fork (Race, runRace)+import OM.Show (showt)+import OM.Socket (Responded, connectServer, openEgress, openIngress,+  openServer)+import Prelude (Applicative(pure), Foldable(length), Functor(fmap),+  Maybe(Nothing), Traversable(sequence), ($), (.), (<$>), (=<<), IO,+  Int, MonadFail, fst, print, reverse, seq, sequence_)+import Streaming.Prelude (Of, Stream)+import Test.Hspec (hspec, it, shouldBe)+import qualified Client+import qualified Egress+import qualified Ingress+import qualified Server+import qualified Streaming.Prelude as Stream+++main :: IO ()+main =+  hspec $ do+    it "ingress/egress" $ do+      let+        expected :: [Int]+        expected = [1, 2, 3, 4]+      block <- newEmptyMVar+      void . forkIO $ do {- Ingress thread -}+        let+          instream :: (Race) => Stream (Of Int) IO ()+          instream = openIngress "localhost:9933"+        {-+          The instream never closes, because an ingress server is meant+          to listen forever, so we consume only the expected number+          of items from the stream+        -}+        actual <-+          runRace $+            Stream.take (length expected) instream & Stream.toList_+        putMVar block actual++      {- Push the expected items over the network.  -}+      Stream.each expected & openEgress "localhost:9933"+      actual <- takeMVar block+      print actual+      actual `shouldBe` expected++    it "client/server" $ do+      let+        requestStream+          :: ( MonadCatch m+             , MonadFail m+             , MonadLogger m+             , MonadUnliftIO m+             , Race+             )+          => Stream (Of (Int, Text -> m Responded)) m void+        requestStream = openServer "localhost:9934" Nothing++        requests :: [Int]+        requests = [1..4]++        expected :: [Text]+        expected = showt <$> requests++      block <- newEmptyMVar+      void . forkIO $ runRace $ runStdoutLoggingT $ do+        {-+          Handle the requests in reverse order to test out of order+          handling.+        -}+        reqs <-+          fmap reverse+          . Stream.toList_+          . Stream.take (length expected)+          $ requestStream+        liftIO (print (fst <$> reqs))+        sequence_+          [ respond (showt request)+          | (request, respond) <- reqs+          ]+        liftIO (putMVar block ())+      call <- runStdoutLoggingT (connectServer "localhost:9934" Nothing)+      calls <-+        sequence+          [ do+              threadDelay 100_000+              response <- newEmptyMVar+              void . forkIO $+                putMVar response =<< call i+              pure response+          | i <- requests+          ]+      responses <-+        sequence+          [ takeMVar response+          | response <- calls+          ]+      print responses+      responses `shouldBe` expected+      takeMVar block++    it "make sure the examples compile" $+      {-+        We don't actually _run_ the examples. Really it is the imports+        that makes sure the examples compile. We `seq` them here just+        so we don't get warnings about unused imports.+      -}+      Client.main+      `seq` Server.main+      `seq` Ingress.main+      `seq` Egress.main+      `seq` (pure () :: IO ())+