diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,15 @@ and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/). -## 0.1.0.1 - Unreleased +## 0.1.1.0 - 2025-12-02 + +- Stream wrappers now use async and will close their streams on exit. + * The exposed stream worker can be used for linking to other threads. + * Client and servers link to their default streams. +- Added `runServerStateful`, a more complex layer under Simple server. +- Added `startClientAsync`, a less complex layer under Simple client. + +## 0.1.0.1 - 2025-11-30 - Added `onException` to `startClientSimple` wrapper, so it can be timed out without leaking its client. diff --git a/quic-simple.cabal b/quic-simple.cabal --- a/quic-simple.cabal +++ b/quic-simple.cabal @@ -5,9 +5,12 @@ -- see: https://github.com/sol/hpack name: quic-simple -version: 0.1.0.1 +version: 0.1.1.0 synopsis: Quick-start wrappers for QUIC -description: . A few layers over QUIC, to get the first bytes out faster. The top level is RPC-like, using Serialise as a codec. Copypaste the code and peel layers as you go to get more power and flexibility. +description: A few layers over QUIC, to get the first bytes out faster. + The top level is RPC-like, using Serialise as a codec. + Copypaste the code and peel layers as you go to get more + power and flexibility. category: Network author: IC Rainbow maintainer: aenor.realm@gmail.com @@ -39,6 +42,7 @@ ImportQualifiedPost LambdaCase OverloadedStrings + DeriveAnyClass ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints build-depends: async @@ -70,11 +74,14 @@ ImportQualifiedPost LambdaCase OverloadedStrings + DeriveAnyClass ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wincomplete-uni-patterns -Wmissing-export-lists -Wmissing-home-modules -Wpartial-fields -Wredundant-constraints -threaded -rtsopts -with-rtsopts=-N build-depends: async , base >=4.7 && <5 , bytestring + , containers , quic-simple , stm + , text default-language: GHC2021 diff --git a/src/Network/QUIC/Simple.hs b/src/Network/QUIC/Simple.hs --- a/src/Network/QUIC/Simple.hs +++ b/src/Network/QUIC/Simple.hs @@ -1,25 +1,31 @@ {-# LANGUAGE CPP #-} module Network.QUIC.Simple - ( -- * Basic wrappers + ( -- $intro + + -- * Basic wrappers runServer , runClient -- * CBOR/Serialise wrappers - , Serialise , runServerSimple , startClientSimple + -- ** More flexible variants + , runServerStateful + , startClientAsync + , Serialise -- * The rest of the QUIC API , module Network.QUIC ) where +import Control.Concurrent.STM import Network.QUIC import Network.QUIC.Simple.Stream import Codec.Serialise (Serialise) -import Control.Concurrent (forkIO, killThread) +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (Async, async, cancel, link, link2) import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) -import Control.Concurrent.STM (atomically, readTBQueue, writeTBQueue, newTBQueueIO) -import Control.Exception (onException) +import Control.Exception (SomeException, handle, onException) import Control.Monad (forever) import Data.IP (IP(..)) import Network.QUIC.Client (ClientConfig(..), defaultClientConfig) @@ -29,6 +35,24 @@ import Network.QUIC.Simple.Credentials (genCredentials) import Network.Socket (HostName, PortNumber, ServiceName) +{- $intro +Check out the tests in the package git for a cookbook. + +If you're unsure, start with the simplest wrapper. +If the wrapper's limitations bother you, replace it with the source code and customize it to suit your needs. +Alternatively, switch to a lower-level implementation to gain more features. + +Don't let wrappers dictate your code structure and protocols — they're just there to get you a QUIC-start! +-} + +{- | Start a server on all of the address-port pairs. + +The server will have an autogenerated set of credentials on each start, just to get the TLS running. +You can use "Network.QUIC.Simple.Credentials.genCredentials" to generate and keep them, +so the clients can pin them after first connection. + +The server will automatically accept the incoming stream before passing it to a (stateless) connection handler. +-} runServer :: [(IP, PortNumber)] -> (Connection -> Stream -> IO ()) -> IO () runServer scAddresses action = do scCredentials <- genCredentials @@ -41,6 +65,11 @@ defaultStream <- acceptStream conn action conn defaultStream +{- | Start a server on the provided host and port and run a stateless CBOR-encoded request-response protocol. + +While it is possible to use `myThreadId` to get some connection identifier and attach connection data on it, +you'd better use `runServerStateful` instead. +-} runServerSimple :: (Serialise q, Serialise r) => IP @@ -48,13 +77,47 @@ -> (q -> IO r) -> IO () runServerSimple host port action = - runServer [(host, port)] \_conn stream0 -> do - (writeQ, readQ) <- streamSerialise stream0 - forever do - query <- atomically (readTBQueue readQ) - reply <- action query - atomically $ writeTBQueue writeQ reply + runServerStateful host port setup teardown handler + where + setup _conn _wq = pure () + teardown _conn _s = pure () + handler s q = do + r <- action q + pure (s, Just r) +{- | Start a server on the provided host and port and run a stateless CBOR-encoded request-response protocol. + +The connection handler is stateful, with the initial state provided by a setup function. +The handler function must provide next connection state, but may opt out of replying. +Throw an exception to terminate the curent connection - the teardown function then can do the clean up. +-} +runServerStateful + :: (Serialise q, Serialise r) + => IP + -> PortNumber + -> (Connection -> TBQueue r -> IO s) + -> (Connection -> s -> IO ()) + -> (s -> q -> IO (s, Maybe r)) + -> IO () +runServerStateful host port setup teardown action = + runServer [(host, port)] \conn stream0 -> do + (codec, (writeQ, readQ)) <- streamSerialise stream0 + link codec + let + loop !s = handle (\(_ :: SomeException) -> teardown conn s) do + query <- atomically (readTBQueue readQ) + (s', reply_) <- action s query + mapM_ (atomically . writeTBQueue writeQ) reply_ + loop s' + setup conn writeQ >>= loop + +{- | Run a client connecting to the provided host/port and auto-request a stream. + +Server validation is disabled. +If you want server authentication, you'd have to do that in your protocol handshake. + +With the @quic@ library >0.2.10 the connection migration will be enabled by default. +-} runClient :: HostName -> ServiceName -> (Connection -> Stream -> IO ()) -> IO () runClient ccServerName ccPortName action = do Client.run cc \conn -> do @@ -71,27 +134,49 @@ #endif } +{- | Start a client wrapper that will wait for a connection. + +When connected, it will provide a way to stop it, and to do a simple blocking call. +There is no call tracking, so the client is not thread-safe. +Which is fine, when used with the 'runServerSimple'. + +Use 'startClientAsync' to expose more functionality. +-} startClientSimple :: (Serialise q, Serialise r) => HostName -> ServiceName -> IO (IO (), q -> IO r) startClientSimple host port = do - client <- newEmptyMVar - tid <- forkIO $ runClient host port \_conn stream0 -> do - requests <- newTBQueueIO 16 - putMVar client requests - (writeQ, readQ) <- streamSerialise stream0 - forever do - (query, handler) <- atomically $ readTBQueue requests - atomically $ writeTBQueue writeQ query - reply <- atomically $ readTBQueue readQ - handler reply - requests <- takeMVar client `onException` killThread tid + (client, _conn, (writeQ, readQ)) <- startClientAsync host port pure - ( killThread tid + ( cancel client , \query -> do - reply <- newEmptyMVar - atomically $ writeTBQueue requests (query, putMVar reply) - takeMVar reply + atomically $ writeTBQueue writeQ query + atomically $ readTBQueue readQ + ) + +{- | Start a client wrapper that will wait for a connection. + +Canceling the exposed worker thread will terminate connection. +The exposed connection can be used to request more streams. +The message queues are running CBOR codec to shuttle the data. +-} +startClientAsync + :: (Serialise q, Serialise r) + => HostName + -> ServiceName + -> IO (Async (), Connection, MessageQueues q r) +startClientAsync host port = do + client <- newEmptyMVar + tid <- async $ runClient host port \conn stream0 -> do + queues <- streamSerialise stream0 + putMVar client (conn, queues) + forever (threadDelay maxBound) + (conn, (codec, queues)) <- takeMVar client `onException` cancel tid + link2 codec tid + pure + ( tid + , conn + , queues ) diff --git a/src/Network/QUIC/Simple/Credentials.hs b/src/Network/QUIC/Simple/Credentials.hs --- a/src/Network/QUIC/Simple/Credentials.hs +++ b/src/Network/QUIC/Simple/Credentials.hs @@ -9,6 +9,10 @@ import Network.TLS qualified as TLS import Time.System qualified as Hourglass +{- | Create a self-signed Ed25519 certificate suitable for TLS connections. + +The certificate will be valid for 365 days if you choose to save it. +-} genCredentials :: IO TLS.Credentials genCredentials = do secret <- Ed25519.generateSecretKey diff --git a/src/Network/QUIC/Simple/Stream.hs b/src/Network/QUIC/Simple/Stream.hs --- a/src/Network/QUIC/Simple/Stream.hs +++ b/src/Network/QUIC/Simple/Stream.hs @@ -6,23 +6,30 @@ import Codec.Serialise (Serialise, serialise, deserialiseIncremental) import Codec.Serialise qualified as IDecode (IDecode(..)) -import Control.Concurrent (forkIO) -import Control.Concurrent.Async (race_) +import Control.Concurrent.Async (Async, async, race_) import Control.Concurrent.STM -import Control.Exception (throwIO) +import Control.Exception (finally, throwIO) import Control.Monad.ST (stToIO) import Data.ByteString qualified as BS import Data.ByteString.Lazy qualified as BSL import Data.IORef import Network.QUIC qualified as QUIC +{- | A pair of bounded queues wrapping a stream. +-} type MessageQueues sendMsg recvMsg = (TBQueue sendMsg, TBQueue recvMsg) +{- | Wrap the stream with the CBOR codec for both incoming and outgoing messages. + +The decoder will perform incremental parsing and emit complete messages. + +No extra framing is required since CBOR is self-delimiting. +-} streamSerialise :: forall sendMsg recvMsg . (Serialise sendMsg, Serialise recvMsg) => QUIC.Stream - -> IO (MessageQueues sendMsg recvMsg) + -> IO (Async (), MessageQueues sendMsg recvMsg) streamSerialise stream = do initial <- stToIO $ deserialiseIncremental @recvMsg state <- newIORef initial @@ -46,16 +53,22 @@ pure ("", Nothing) streamCodec serialise (decode True) stream +{- | Wrap the stream with a codec to provide a TBQueue interface to it. + +The decoder loop is stateless. +But it runs in IO so you can use external state and terminate the stream by erroring out. +-} streamCodec - :: (sendMsg -> BSL.ByteString) - -> (BS.ByteString -> IO (BS.ByteString, Maybe recvMsg)) + :: (sendMsg -> BSL.ByteString) -- ^ Encoder for outgoing messages + -> (BS.ByteString -> IO (BS.ByteString, Maybe recvMsg)) -- ^ Decoder for incomming chunks -> QUIC.Stream - -> IO (MessageQueues sendMsg recvMsg) + -> IO (Async (), MessageQueues sendMsg recvMsg) streamCodec encode decode stream = do readQ <- newTBQueueIO 1024 writeQ <- newTBQueueIO 1024 - _tid <- forkIO $ race_ (reader "" readQ) (writer writeQ) - pure (writeQ, readQ) + worker <- async $ + race_ (reader "" readQ) (writer writeQ) `finally` QUIC.closeStream stream + pure (worker, (writeQ, readQ)) where reader leftovers readQ = do chunk <- diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -1,11 +1,17 @@ -module Main (main) where +module Main where -import Control.Concurrent.Async (race_) -import Control.Concurrent.STM (atomically, readTBQueue, writeTBQueue) -import Control.Monad (forever, replicateM_) +import Control.Concurrent +import Control.Concurrent.STM +import Control.Monad + +import Control.Concurrent.Async (async, cancel, link, race_, replicateConcurrently_) +import Control.Exception (bracket) import Data.ByteString qualified as BS import Data.ByteString.Lazy qualified as BSL import Data.IORef (newIORef, atomicModifyIORef') +import Data.IntMap.Strict qualified as IntMap +import Data.Text (Text) +import Data.Text.Encoding (encodeUtf8, decodeUtf8) import GHC.Generics (Generic) import Network.QUIC.Simple qualified as QUIC import Network.QUIC.Simple.Stream (MessageQueues, streamCodec, streamSerialise) @@ -29,106 +35,306 @@ race_ serverSimple clientSimple putStrLn "" -serverRaw :: IO () -serverRaw = QUIC.runServer [("127.0.0.1", 14443)] \conn stream -> do - putStrLn "Server accepted connection:" - QUIC.getConnectionInfo conn >>= print + putStrLn "Stateful/Async" + race_ serverStateful clientAsync + putStrLn "" - query <- QUIC.recvStream stream 4096 - putStrLn $ "Server got query: " <> show query - QUIC.sendStream stream $ "got yer bytes: " <> query - _ <- QUIC.recvStream stream 4096 - putStrLn "Server quits" +-- * Raw clientRaw :: IO () clientRaw = QUIC.runClient "127.0.0.1" "14443" \conn stream -> do + -- the initial stream comes pre-requested and ready to go putStrLn "Client connected:" QUIC.getConnectionInfo conn >>= print + -- streams have no framing on their own + -- but that's fine, for now... QUIC.sendStream stream "hi there" reply <- QUIC.recvStream stream 4096 putStrLn $ "Client got reply: " <> show reply + QUIC.closeStream stream putStrLn "Client quits" -serverBox :: IO () -serverBox = QUIC.runServer [("127.0.0.1", 14443)] \_conn stream -> do +serverRaw :: IO () +serverRaw = QUIC.runServer [("127.0.0.1", 14443)] \conn stream -> do + -- the initial stream comes pre-accepted and ready to go putStrLn "Server accepted connection:" - (writeQ, readQ) <- dummyCodec stream - forever do - query <- atomically $ readTBQueue readQ - putStrLn $ "Server got query: " <> show query - atomically $ writeTBQueue writeQ $ "got yer bytes: " <> BSL.fromStrict query + QUIC.getConnectionInfo conn >>= print + -- wait until *something* arrives and take it all in + query <- QUIC.recvStream stream 4096 + putStrLn $ "Server got query: " <> show query + QUIC.sendStream stream $ "got yer bytes: " <> query + + -- the final bytes will be "", signalling the connection getting closed + finalBytes <- QUIC.recvStream stream 4096 + putStrLn $ "Server quits after " <> show finalBytes + +-- * Stream queues / framing + +{- | Stateful codec that splits byte stream into text messages + +This allows sending empty messages, but may break if the Text has a NUL in there. +What kind of "text" is that anyway?! +-} +cstringCodec :: QUIC.Stream -> IO (MessageQueues Text Text) +cstringCodec stream = do + previous <- newIORef BSL.empty -- gotta store partial messages somewhere + snd <$> streamCodec encode (decode previous) stream + where + -- add framing + encode msg = BSL.fromChunks [encodeUtf8 msg, "\NUL"] + + -- strip framing + decode previous chunk = + case BS.break (== 0) chunk of + ("", "") -> + -- the connection is closing + pure ("", Nothing) + (partial, "") -> do + -- everything is consumed, but no frame marker is in sight + atomicModifyIORef' previous \prefix -> + (prefix <> BSL.fromStrict partial, ()) + pure ("", Nothing) -- don't emit a message just yet + (suffix, leftovers) -> do + -- found the marker + prefix <- atomicModifyIORef' previous ("",) + pure + ( BS.drop 1 leftovers -- remove the separator + , Just . decodeUtf8 . BSL.toStrict $ + prefix <> BSL.fromStrict suffix -- decode the whole message + ) + clientBox :: IO () clientBox = do QUIC.runClient "127.0.0.1" "14443" \_conn stream -> do - (writeQ, readQ) <- dummyCodec stream + (writeQ, readQ) <- cstringCodec stream + -- send all the messages at once atomically $ writeTBQueue writeQ "hi there" - reply <- atomically $ readTBQueue readQ - putStrLn $ "Client got reply: " <> show reply + -- with the framing in place, this one is a valid message now + atomically $ writeTBQueue writeQ "" + atomically $ writeTBQueue writeQ "and again" + -- read the replies one by one + reply1 <- atomically $ readTBQueue readQ + putStrLn $ "Client got reply 1: " <> show reply1 + reply2 <- atomically $ readTBQueue readQ + putStrLn $ "Client got reply 2: " <> show reply2 + reply3 <- atomically $ readTBQueue readQ + putStrLn $ "Client got reply 3: " <> show reply3 -dummyCodec :: QUIC.Stream -> IO (MessageQueues BSL.ByteString BS.ByteString) -dummyCodec = streamCodec id (\chunk -> pure ("", Just chunk)) +serverBox :: IO () +serverBox = QUIC.runServer [("127.0.0.1", 14443)] \_conn stream -> do + putStrLn "Server accepted connection:" + (writeQ, readQ) <- cstringCodec stream + forever do + -- a linearised logging echo server + query <- atomically $ readTBQueue readQ + putStrLn $ "Server got query: " <> show query + atomically $ writeTBQueue writeQ $ "got yer bytes: " <> query +-- * Serialised messages + data ClientMessage = Hello | Bye - deriving (Eq, Show, Ord, Generic) + deriving (Eq, Show, Ord, Generic) -- derive Generic so Serialise can do its thing +-- using StandAloneDeriving instance QUIC.Serialise ClientMessage data ServerMessage = Ok Int - deriving (Eq, Show, Ord, Generic) + deriving (Eq, Show, Ord, Generic, QUIC.Serialise) -- using DeriveAnyClass -instance QUIC.Serialise ServerMessage +{- XXX: No turn structure is imposed by the protocol. +The messages get delivered in order, but the client and server +have to coordinate implicitly on when to wait for a reply and +when to go without waiting. + +Unless the messages themselves carry call IDs, there's no way to +associate replies with responses besides sending in lockstep. +-} +clientSerialise :: IO () +clientSerialise = do + QUIC.runClient "127.0.0.1" "14443" \_conn stream -> do + (writeQ, readQ) <- snd <$> streamSerialise stream + replicateM_ 5 do + -- send messages one by one + atomically $ writeTBQueue writeQ Hello + -- wait for the reply before sending another one + reply <- atomically $ readTBQueue @ServerMessage readQ + -- here we know that the reply is sent in response to the Hello + putStrLn $ "Client got reply: " <> show reply + -- send, but don't wait for the reply + atomically $ writeTBQueue writeQ Bye + -- the connection is terminated + serverSerialise :: IO () serverSerialise = QUIC.runServer [("127.0.0.1", 14443)] \_conn stream -> do putStrLn "Server accepted connection:" - (writeQ, readQ) <- streamSerialise stream + (writeQ, readQ) <- snd <$> streamSerialise stream let + -- simple state-passing loop loop counter = do query <- atomically (readTBQueue readQ) putStrLn $ "Server got query: " <> show query case query of Hello -> do + -- reply at once before reading the next message atomically $ writeTBQueue writeQ (Ok counter) loop (counter + 1) Bye -> + -- don't reply + -- XXX: the client will hang indefinitely if waiting for the reply here pure () loop 0 -clientSerialise :: IO () -clientSerialise = do - QUIC.runClient "127.0.0.1" "14443" \_conn stream -> do - (writeQ, readQ) <- streamSerialise stream +-- * Sync client-driven RPC + +{- | The wrappers from startClientSimple replace implementation details with a handle. + +It also imposes a sync/linearised interaction -- every call gets a response. +-} +clientSimple :: IO () +clientSimple = + -- since this is a handle pattern, it should be properly scoped + bracket open close \(_stop, call) -> do + -- the flow is the same as in clientSerialise, replicateM_ 5 do - atomically $ writeTBQueue writeQ Hello - reply <- atomically $ readTBQueue @ServerMessage readQ - putStrLn $ "Client got reply: " <> show reply - atomically $ writeTBQueue writeQ Bye + -- no queues, just running a function to get a reply + Ok n <- call Hello + putStrLn $ "Client got reply " <> show n + timeout 1000000 (call Bye) >>= mapM_ \reply -> + putStrLn $ "Shouldn't happen, the server errors out on this: " <> show reply + putStrLn "Stopping" + where + open = QUIC.startClientSimple "127.0.0.1" "14443" + close (stop, _call) = stop +{- | Wrappers in runServerSimple ensure that every call gets a response. +-} serverSimple :: IO () serverSimple = do - counter <- newIORef 0 + counter <- newIORef 0 -- some global state shared by all connections QUIC.runServerSimple "127.0.0.1" 14443 \case Hello -> do - putStrLn "Server got Hello" + -- the connection handler is stateless + self <- myThreadId -- but its thread id can be used as a key in the global state + -- not now, though... + putStrLn $ "Server got Hello from " <> show self + n <- atomicModifyIORef' counter \old -> (old + 1, old) pure $ Ok n Bye -> do putStrLn "Server got Bye" - error "Whelp, the serverSimple must reply, but the protocol must stop. Needs a re-design." + -- trade-offs... + error "Whelp, the serverSimple handler must reply, but the protocol should stop. Needs a re-design..." -clientSimple :: IO () -clientSimple = do - (stop, call) <- QUIC.startClientSimple "127.0.0.1" "14443" - replicateM_ 5 do +-- * DIY async calls/events + +{- | Wrappers from startClientAsync only do the connection setup and provide all the internals. + +The actual messages implicitly wrap ClientMessage/ServerMessage with an optional call id. +This allows the messages to be treated as calls (sync request-response) or casts (no-reply messages). + +The client now has a state of its own to track the requests in flight. +The calls are still sync, but the client now allows calling from multiple threads +without relying on message ordering. +-} +clientAsync :: IO () +clientAsync = do + -- wait for the connection + (client, _conn, (writeQ, readQ)) <- QUIC.startClientAsync "127.0.0.1" "14443" + + -- the casts require no state, but have no response handlers + let cast q = atomically $ writeTBQueue writeQ (Nothing, q) + + -- events, like casts have no call id, but originating from server + -- XXX: this protocol has no notion of server-sent requests though + events <- newTBQueueIO 16 + + calls <- newTVarIO mempty + void $ async do + -- a dedicated thread now reads all messages and does the triage + link client -- exit when the client stops + forever do + atomically (readTBQueue readQ) >>= \case + (Nothing, e) -> + -- no call id -- this is an event + atomically $ writeTBQueue events e + (Just callId, r) -> + -- call id is present -- this is a response + atomically $ modifyTVar' calls $ IntMap.insert callId r + + -- issuing calls from multiple threads requires avoiding getting someone else's reply + counter <- newIORef 0 -- the simplest key is a counter + let + call q = do + callId <- atomicModifyIORef' counter \old -> (old + 1, old) + atomically $ writeTBQueue writeQ (Just callId, q) -- send the message annotated with call id + replyVar <- newTVarIO undefined -- XXX: this will not be read until the reply is arrived + let + popOrRetry = IntMap.alterF \case + Just r -> Nothing <$ writeTVar replyVar r -- pop the reply and store it outside, then return + Nothing -> retry -- the calls map is missing the call id, continue waiting + atomically $ + -- the caller is blocked until the transaction succeeds + readTVar calls >>= popOrRetry callId >>= writeTVar calls + -- the transaction concluded, it is now safe to read the var + readTVarIO replyVar + + -- send all the requests at once, each in its own thread + replicateConcurrently_ 5 do + -- everyone will block until they receive their own response Ok n <- call Hello putStrLn $ "Client got reply " <> show n - timeout 1000000 (call Bye) >>= mapM_ \reply -> - putStrLn $ "Shouldn't happen, the server errors out on this: " <> show reply - putStrLn "Stopping" - stop + + cast Bye -- no reply is even expected + cancel client + -- we have no event handler thread so we read the whole queue + unread <- atomically (flushTBQueue events) + -- the server does not send any events though + putStrLn $ "Client events: " <> show unread + +{- | Wrappers in runServerStateful provide a richer interface and impose less. +-} +serverStateful :: IO () +serverStateful = do + -- some global state + conns <- newTVarIO mempty + connIds <- newIORef 0 + QUIC.runServerStateful "127.0.0.1" 14443 (setup conns connIds) (teardown conns) handler + where + -- every connection has a local state too + setup conns counter _conn writeQ = do + -- generate an explicit key + connId <- atomicModifyIORef' counter \old -> (old + 1, old) + -- don't store ThreadIDs directly! + me <- myThreadId >>= mkWeakThreadId + atomically $ modifyTVar' conns $ + -- this per-connection state will be observable from outside + -- everybody can send messages to this connection without a client to "request" them + IntMap.insert connId (me, writeQ) + pure (connId, writeQ, 0 :: Int) -- with writeQ the request handlers can use async replies too + + -- use the global and local state for cleanup ops + teardown conns _conn (connId, _writeQ, _counter) = + atomically $ modifyTVar' conns $ IntMap.delete connId + + handler connState@(connId, writeQ, counter) msg = do + putStrLn $ "Server got " <> show msg + case msg of + (Just (rid :: Int), Hello) -> do + -- this server replies later... + void $ forkIO do + threadDelay 1000000 + -- with the writeQ being accessible it is possible to send messages at any time + atomically $ writeTBQueue writeQ (Just rid, Ok counter) + -- ... but moves to process the next message ASAP + let connState' = (connId, writeQ, counter + 1) -- update one part, keep the rest + pure (connState', Nothing) + _ -> + -- ignore casted Hellos and any form of Bye + pure (connState, Nothing)