quic-simple 0.1.0.1 → 0.1.1.0

raw patch · 6 files changed

+409/−86 lines, 6 filesdep +containersdep +textPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: containers, text

API changes (from Hackage documentation)

Files

• CHANGELOG.md +9/−1
• quic-simple.cabal +9/−2
• src/Network/QUIC/Simple.hs +111/−26
• src/Network/QUIC/Simple/Credentials.hs +4/−0
• src/Network/QUIC/Simple/Stream.hs +22/−9
• test/Spec.hs +254/−48

@@ -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. 

@@ -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

@@ -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     )

@@ -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

@@ -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 <-

@@ -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)