om-socket (empty) → 0.11.0.3
raw patch · 9 files changed
+1014/−0 lines, 9 filesdep +aesondep +basedep +binarysetup-changed
Dependencies added: aeson, base, binary, binary-conduit, bytestring, conduit, conduit-extra, containers, exceptions, megaparsec, monad-logger, network, om-show, om-socket, stm, text, time, tls, transformers
Files
- LICENSE +20/−0
- README.md +174/−0
- Setup.hs +2/−0
- examples/client.hs +42/−0
- examples/egress.hs +37/−0
- examples/ingress.hs +42/−0
- examples/server.hs +49/−0
- om-socket.cabal +103/−0
- src/OM/Socket.hs +545/−0
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright 2022 Rick Owens++Permission is hereby granted, free of charge, to any person obtaining a+copy of this software and associated documentation files (the "Software"),+to deal in the Software without restriction, including without limitation+the rights to use, copy, modify, merge, publish, distribute, sublicense,+and/or sell copies of the Software, and to permit persons to whom the+Software is furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL+THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR+OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,+ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.+
+ README.md view
@@ -0,0 +1,174 @@+# om-socket++- [Overview](#overview)+- [Examples](#examples)+ - [Open an Ingress service](#open-an-ingress-service)+ - [Open an Egress connection](#open-an-egress-connection)+ - [Start a server process](#start-a-server-process)+ - [Connect a client to a server](#connect-a-client-to-a-server)++## Overview++This package provides some utilities for Haskell programs to communicate raw+binary messages over the network. It includes:++* Opening an "Ingress" service.+ It provides a way for a program to open a socket and accept a+ stream of messages without responding to any of them.++* Open an "Egress" socket.+ It provides a way to connect to an "Ingress" service and dump a stream of+ messages to it.++* Open a bidirectional "server".+ It provides a way to open a "server", which provides your program with a+ stream of requests paired with a way to respond to each request. Responses are+ allowed to be supplied in an order different than that from which the+ corresponding requests were received.++* Open a client to a bidirectional "server".+ It provides a way to connect to an open server and provides a convenient+ `(request -> IO response)` interface to talk to the server.++## Examples++### Open an Ingress service++```haskell+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE OverloadedStrings #-}++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 =+ runConduit $+ openIngress "localhost:9000"+ .| awaitForever (\msg ->+ case msg of+ A -> liftIO $ putStrLn "Got A"+ B -> liftIO $ putStrLn "Got B"+ )+```++ +### Open an Egress connection++```haskell+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE OverloadedStrings #-}++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 =+ runConduit $+ mapM_ yield [A, B, B, A, A, A, B]+ .| openEgress "localhost:9000"++```++### Start a server process+++```haskell+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}++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 =+ 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)+ )++```++### Connect a client to a server++```haskell+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}++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 client. -}+main :: IO ()+main = do+ client <-+ runStdoutLoggingT $+ connectServer "localhost:9000" Nothing+ putStrLn =<< client (EchoRequest "hello")+ putStrLn =<< client (EchoRequest "world")+```
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ examples/client.hs view
@@ -0,0 +1,42 @@+{-# 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")++
+ examples/egress.hs view
@@ -0,0 +1,37 @@+{-# 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"++
+ examples/ingress.hs view
@@ -0,0 +1,42 @@+{-# 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"+ )++
+ examples/server.hs view
@@ -0,0 +1,49 @@+{-# 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)+ )+
+ om-socket.cabal view
@@ -0,0 +1,103 @@+cabal-version: 3.0+name: om-socket+version: 0.11.0.3+synopsis: Socket utilities.+description: Binary ingress server, egress client, and+ bidirectional binarry client/server+homepage: https://github.com/owensmurray/om-socket+license: MIT+author: Rick Owens+maintainer: rick@owensmurray.com+copyright: 2022 Rick Owens+category: Network+build-type: Simple+extra-source-files:+ LICENSE+ README.md++common warnings+ ghc-options:+ -Wmissing-deriving-strategies+ -Wmissing-export-lists+ -Wmissing-import-lists+ -Wredundant-constraints+ -Wall++common dependencies+ build-depends:+ , aeson >= 2.0.3.0 && < 2.1+ , base >= 4.15.0.0 && < 4.16+ , binary >= 0.8.8.0 && < 0.9+ , binary-conduit >= 1.3.1 && < 1.4+ , bytestring >= 0.10.12.1 && < 0.11+ , 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.3+ , monad-logger >= 0.3.37 && < 0.4+ , network >= 3.1.2.7 && < 3.2+ , om-show >= 0.1.2.2 && < 0.2+ , stm >= 2.5.0.0 && < 2.6+ , text >= 1.2.5.0 && < 1.3+ , time >= 1.9.3 && < 1.10+ , tls >= 1.5.8 && < 1.6++library+ import: warnings, dependencies+ exposed-modules: + OM.Socket+ -- other-modules: + -- other-extensions: + 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.6++-- 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+ type: exitcode-stdio-1.0+ hs-source-dirs: examples+ default-language: Haskell2010+ build-depends:+ , om-socket+
+ src/OM/Socket.hs view
@@ -0,0 +1,545 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}++{- | Socket utilities. -}+module OM.Socket (+ -- * Socket Addresses+ AddressDescription(..),+ resolveAddr,++ -- * Ingress-only sockets+ openIngress,++ -- * Egress-only sockets+ openEgress,++ -- * Bidirection request/resposne servers.+ openServer,+ Responded,+ connectServer,+) where+++import Control.Applicative ((<|>))+import Control.Concurrent (Chan, MVar, forkIO, newChan, newEmptyMVar,+ putMVar, readChan, takeMVar, throwTo, writeChan)+import Control.Concurrent.STM (TVar, atomically, newTVar, readTVar,+ retry, writeTVar)+import Control.Exception (SomeException, bracketOnError, throw)+import Control.Monad (join, void, when)+import Control.Monad.Catch (MonadCatch, MonadThrow, throwM, try)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Control.Monad.Logger.CallStack (MonadLoggerIO, askLoggerIO,+ logDebug, logError, logWarn, runLoggingT)+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.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)+import Network.Socket (Family(AF_INET, AF_INET6, AF_UNIX),+ SockAddr(SockAddrInet, SockAddrInet6, SockAddrUnix),+ SocketOption(ReuseAddr), SocketType(Stream), HostName, ServiceName,+ Socket, accept, addrAddress, bind, close, connect, defaultProtocol,+ getAddrInfo, listen, setSocketOption, socket)+import Network.Socket.ByteString (recv)+import Network.Socket.ByteString.Lazy (sendAll)+import Network.TLS (ClientParams, Context, ServerParams, contextNew,+ handshake, recvData, sendData)+import OM.Show (showt)+import Text.Megaparsec (Parsec, eof, 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 Text.Megaparsec as M+++{- |+ Opens an "ingress" socket, which is a socket that accepts a stream of+ messages without responding.+-}+openIngress :: (Binary i, MonadIO m, MonadFail m)+ => AddressDescription+ -> ConduitT () i m ()+openIngress bindAddr = do+ so <- listenSocket =<< resolveAddr bindAddr+ mvar <- liftIO newEmptyMVar+ void . liftIO . forkIO $ acceptLoop so mvar+ mvarToSource 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 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 (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+++{- |+ Open an "egress" socket, which is a socket that sends a stream of messages+ without receiving responses.+-}+openEgress+ :: ( Binary o+ , MonadFail m+ , MonadIO m+ , MonadThrow m+ )+ => AddressDescription+ -> ConduitT o Void m ()+openEgress addr = do+ so <- connectSocket =<< resolveAddr addr+ conduitEncode .| sinkSocket so+++{- | Guess the family of a `SockAddr`. -}+fam :: SockAddr -> Family+fam SockAddrInet {} = AF_INET+fam SockAddrInet6 {} = AF_INET6+fam SockAddrUnix {} = AF_UNIX+++{- | Resolve a host:port address into a 'SockAddr'. -}+resolveAddr :: (MonadIO m, MonadFail m) => AddressDescription -> m SockAddr+resolveAddr addr = do+ (host, port) <- parseAddr addr+ liftIO (getAddrInfo Nothing (Just host) (Just port)) >>= \case+ [] -> fail "Address not found: (host, port)"+ sa:_ -> return (addrAddress sa)+++{- | Parse a host:port address. -}+parseAddr :: (MonadFail m) => AddressDescription -> m (HostName, ServiceName)+parseAddr addr =+ case parse parser "$" (unAddressDescription addr) of+ Left err -> fail (show err)+ Right (host, port) -> return (host, port)+ where+ parser :: Parsec Void Text (HostName, ServiceName)+ parser = do+ host <- M.try ipv6 <|> ipv4+ void $ char ':'+ port <- many (oneOf ("0123456789" :: String))+ eof+ return (host, port)++ ipv6 :: Parsec Void Text HostName+ ipv6 = do+ void $ char '['+ host <- many (satisfy (/= ']'))+ void $ char ']'+ return host++ ipv4 :: Parsec Void Text HostName+ ipv4 = many (satisfy (/= ':'))+++{- | Create a connected socket. -}+connectSocket :: (MonadIO m) => SockAddr -> m Socket+connectSocket addr = liftIO $+ {-+ Make sure to close the socket if an error happens during+ connection, because if not, we could easily run out of file+ descriptors in the case where we rapidly try to send thousands+ of message to the same peer, which could happen when one object+ is a hotspot.+ -}+ bracketOnError+ (socket (fam addr) Stream defaultProtocol)+ close+ (\so -> connect so addr >> return so)+++{- | Create a listening socket. -}+listenSocket :: (MonadIO m) => SockAddr -> m Socket+listenSocket addr = liftIO $ do+ so <- socket (fam addr) Stream defaultProtocol+ setSocketOption so ReuseAddr 1+ bind so addr+ listen so 5+ return so+++{- |+ Open a "server" socket, which is a socket that accepts incoming requests+ and provides a way to respond to those requests.+-}+openServer+ :: ( Binary request+ , Binary response+ , MonadFail m+ , MonadLoggerIO m+ , Show request+ , Show response+ )+ => AddressDescription+ -> Maybe (IO ServerParams)+ -> ConduitT Void (request, response -> m Responded) m ()+openServer bindAddr tls = do+ so <- listenSocket =<< resolveAddr bindAddr+ requestChan <- liftIO newChan+ logging <- askLoggerIO+ void . liftIO . forkIO . (`runLoggingT` logging) $ acceptLoop so requestChan+ chanToSource requestChan+ 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+ (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++ 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)+ where+ output :: Context -> ConduitT ByteString Void IO ()+ output ctx = awaitForever (sendData ctx . BSL.fromStrict)++ input :: Context -> ConduitT Void ByteString IO ()+ input ctx = do+ bytes <- recvData ctx+ if BS.null bytes then+ pure ()+ else do+ yield bytes+ input ctx++ 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+ )+ .| conduitEncode+ .| transPipe liftIO outputSink+ )+++{- |+ Connect to a server. Returns a function in 'MonadIO' that can be used+ to submit requests to (and returns the corresponding response from)+ the server.+-}+connectServer+ :: ( Binary request+ , Binary response+ , MonadIO m+ , MonadLoggerIO n+ , Show response+ )+ => AddressDescription+ -> Maybe ClientParams+ -> n (request -> m response)+connectServer addr tls = do+ logging <- askLoggerIO+ liftIO $ do+ so <- connectSocket =<< resolveAddr addr+ state <- atomically (newTVar ClientState {+ csAlive = True,+ csResponders = Map.empty,+ csMessageId = minBound,+ csMessageQueue = []+ })+ (send, reqSource) <- prepareConnection so+ void . forkIO $ (`runLoggingT` logging) (requestThread send state)+ void . forkIO $ (`runLoggingT` logging) (responseThread reqSource state)+ return (\i -> liftIO $ do+ mvar <- newEmptyMVar+ join . atomically $+ readTVar state >>= \case+ ClientState {csAlive = False} -> return $+ throwM (userError "Server connection died.")+ s@ClientState {csMessageQueue} -> do+ writeTVar state s {+ csMessageQueue = csMessageQueue <> [(i, putMVar mvar)]+ }+ return (takeMVar mvar)+ )+ where+ {- |+ Returns the (output, input) communication channels, either prepared+ for TSL or not depending on the configuration.+ -}+ prepareConnection+ :: (MonadIO m, MonadIO f)+ => Socket+ -> f (BSL.ByteString -> IO (), ConduitT Void ByteString m ())+ prepareConnection so =+ case tls of+ Nothing -> pure (sendAll so, sourceSocket so)+ Just params -> do+ ctx <- contextNew so params+ handshake ctx+ pure (send ctx, reqSource 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++ {- | Receive requests and send them to the server. -}+ requestThread :: (+ Binary i,+ MonadCatch m,+ MonadLoggerIO m+ )+ => (BSL.ByteString -> IO ())+ -> TVar (ClientState i o)+ -> m ()+ requestThread send state =+ join . liftIO . atomically $+ readTVar state >>= \case+ ClientState {csAlive = False} -> pure (pure ())+ ClientState {csMessageQueue = []} -> retry+ s@ClientState {+ csMessageQueue = (m, r):remaining,+ csResponders,+ csMessageId+ }+ -> do+ writeTVar state s {+ csMessageQueue = remaining,+ csResponders = Map.insert csMessageId r csResponders,+ csMessageId = succ csMessageId+ }+ pure $ do+ liftIO $ send (encode (Request csMessageId m))+ requestThread send state++ {- |+ Receive responses from the server and send then them back to the+ client responder.+ -}+ 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)+ )+ ) >>= \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+ ]+++{- | A server endpoint configuration. -}+data Endpoint = Endpoint {+ bindAddr :: AddressDescription,+ tls :: Maybe (IO ServerParams)+ }+ deriving stock (Generic)+++{- | Response to a request. -}+data Response p = Response {+ responseTo :: MessageId,+ response :: p+ }+ deriving stock (Generic, Show)+instance (Binary p) => Binary (Response p)+++{- |+ A description of a socket address on which a socket is or should be+ listening. Supports both IPv4 and IPv6.++ Examples:++ > AddressDescription "[::1]:80" -- IPv6 localhost, port 80+ > AddressDescription "127.0.0.1:80" -- IPv4 localhost, port 80+ > AddressDescription "somehost:80" -- IPv4 or IPv6 (depending on what name resolution returns), port 80+-}+newtype AddressDescription = AddressDescription {+ unAddressDescription :: Text+ }+ deriving stock (Generic)+ deriving newtype+ ( Binary+ , Eq+ , FromJSON+ , FromJSONKey+ , IsString+ , Monoid+ , Ord+ , Semigroup+ , ToJSON+ , ToJSONKey+ )+instance Show AddressDescription where+ show = T.unpack . unAddressDescription+++{- | Client connection state. -}+data ClientState i o = ClientState {+ csAlive :: Bool,+ csResponders :: Map MessageId (o -> IO ()),+ csMessageId :: MessageId,+ csMessageQueue :: [(i, o -> IO ())]+ }+++{- | A Request message type. -}+data Request p = Request {+ messageId :: MessageId,+ payload :: p+ }+ deriving stock (Generic, Show)+instance (Binary p) => Binary (Request p)+++{- | A message identifier. -}+newtype MessageId = MessageId {+ _unMessageId :: Word32+ }+ 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+ implementation to help ensure that you actually responded to the+ request in all cases.+-}+data Responded = Responded++