quic-simple 0.1.1.0 → 0.1.2.0

raw patch · 6 files changed

+370/−47 lines, 6 filesdep +ramdep +time-hourglassdep −hourglassdep −memorydep ~quicPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: ram, time-hourglass

Dependencies removed: hourglass, memory

Dependency ranges changed: quic

API changes (from Hackage documentation)

Files

• CHANGELOG.md +13/−0
• README.md +9/−2
• quic-simple.cabal +5/−5
• src/Network/QUIC/Simple.hs +16/−9
• src/Network/QUIC/Simple/Stream.hs +155/−22
• test/Spec.hs +172/−9

@@ -6,6 +6,19 @@ and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/). +## 0.1.2.0 - 2026-09-03++- Servers and clients are datagram-enabled by default.+- Added server input queue to runServerStateful setup callback.+  * Can be used to inject datagrams as server queries.+- startClientAsync no longer inherits the caller's exception mask, so stopping a client started inside a bracket cannot deadlock on the quic watchdog thread.++- `Network.QUIC.Simple.Stream` is now meant for qualified import: `streamSerialise` and `streamCodec` became `Stream.serialise` and `Stream.codec`.+- Added `Stream.open` and `Stream.acceptLoop` for running multiple protocols over one connection, each stream announced by a header message.+- Added `Stream.sendMessage` and `Stream.recvMessage` for one-shot CBOR exchanges on bare streams, e.g. handshakes and stream headers.+- Added `Stream.serialiseFrom` and `Stream.codecFrom` to pick up a stream after a header, and `Stream.consume` for raw byte payloads.+- Stream workers now finish when the peer closes the stream instead of spinning on the end of stream.+ ## 0.1.1.0 - 2025-12-02  - Stream wrappers now use async and will close their streams on exit.

@@ -4,8 +4,15 @@  - `QUIC.runServer [("127.0.0.1", 14443)] \conn stream ->` -- start a generic server with random TLS credentials and auto-accept the first stream. - `QUIC.runClient "127.0.0.1" "14443" \conn stream ->` -- start a generic client and request an initial stream.-- `(writeQ, readQ) <- QUIC.streamCodec encode decodeIncremental stream` -- convert a stream to a pair of queues.-- `(writeQ, readQ) <- QUIC.streamSerialise stream` -- run a CBOR codec over a stream.+- `(writeQ, readQ) <- Stream.codec encode decodeIncremental stream` -- convert a stream to a pair of queues.+- `(writeQ, readQ) <- Stream.serialise stream` -- run a CBOR codec over a stream.++Handshakes and multiple streams per connection, with `import Network.QUIC.Simple.Stream qualified as Stream`:++- `Stream.sendMessage stream hello` / `(reply, leftovers) <- Stream.recvMessage stream` -- exchange one-shot CBOR messages on a bare stream.+- `stream <- Stream.open conn header` -- request a stream and announce its protocol with a header message.+- `Stream.acceptLoop conn \header leftovers stream ->` -- accept the peer's streams and dispatch them by header, each in its own thread.+- `Stream.serialiseFrom leftovers stream` / `Stream.consume leftovers stream sink` -- continue a stream after its header with CBOR messages or raw bytes.  An extra-simple pair of wrappers for QUIC and dirty RPC: 

@@ -1,11 +1,11 @@ cabal-version: 2.2 --- This file has been generated from package.yaml by hpack version 0.38.1.+-- This file has been generated from package.yaml by hpack version 0.39.6. -- -- see: https://github.com/sol/hpack  name:           quic-simple-version:        0.1.1.0+version:        0.1.2.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.@@ -50,13 +50,13 @@     , bytestring     , crypton     , crypton-x509-    , hourglass     , iproute-    , memory     , network-    , quic >=0.2.7+    , quic >=0.3.2+    , ram     , serialise     , stm+    , time-hourglass     , tls   default-language: GHC2021 

@@ -19,11 +19,10 @@  import Control.Concurrent.STM import Network.QUIC-import Network.QUIC.Simple.Stream  import Codec.Serialise (Serialise) import Control.Concurrent (threadDelay)-import Control.Concurrent.Async (Async, async, cancel, link, link2)+import Control.Concurrent.Async (Async, asyncWithUnmask, cancel, link, link2) import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) import Control.Exception (SomeException, handle, onException) import Control.Monad (forever)@@ -33,6 +32,7 @@ import Network.QUIC.Server (ServerConfig(..), defaultServerConfig) import Network.QUIC.Server qualified as Server import Network.QUIC.Simple.Credentials (genCredentials)+import Network.QUIC.Simple.Stream qualified as Stream import Network.Socket (HostName, PortNumber, ServiceName)  {- $intro@@ -60,6 +60,9 @@     sc = defaultServerConfig       { scCredentials       , scAddresses+#if MIN_VERSION_quic(0,3,2)+      , scMaxDatagramFrameSize = 0xFFFF -- advertise support+#endif       }   Server.run sc \conn -> do     defaultStream <- acceptStream conn@@ -79,7 +82,7 @@ runServerSimple host port action =   runServerStateful host port setup teardown handler   where-    setup _conn _wq = pure ()+    setup _conn _rq _wq = pure ()     teardown _conn _s = pure ()     handler s q = do       r <- action q@@ -95,13 +98,13 @@   :: (Serialise q, Serialise r)   => IP   -> PortNumber-  -> (Connection -> TBQueue r -> IO s)+  -> (Connection -> TBQueue q -> 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+    (codec, (writeQ, readQ)) <- Stream.serialise stream0     link codec     let       loop !s = handle (\(_ :: SomeException) -> teardown conn s) do@@ -109,7 +112,7 @@         (s', reply_) <- action s query         mapM_ (atomically . writeTBQueue writeQ) reply_         loop s'-    setup conn writeQ >>= loop+    setup conn readQ writeQ >>= loop  {- | Run a client connecting to the provided host/port and auto-request a stream. @@ -132,6 +135,9 @@       , ccSockConnected = True       , ccWatchDog = True #endif+#if MIN_VERSION_quic(0,3,2)+      , ccMaxDatagramFrameSize = 1200 -- a conservative request as datagrams must not be fragmented+#endif       }  {- | Start a client wrapper that will wait for a connection.@@ -166,11 +172,11 @@   :: (Serialise q, Serialise r)   => HostName   -> ServiceName-  -> IO (Async (), Connection, MessageQueues q r)+  -> IO (Async (), Connection, Stream.MessageQueues q r) startClientAsync host port = do   client <- newEmptyMVar-  tid <- async $ runClient host port \conn stream0 -> do-    queues <- streamSerialise stream0+  tid <- asyncWithUnmask \unmask -> unmask $ runClient host port \conn stream0 -> do+    queues <- Stream.serialise stream0     putMVar client (conn, queues)     forever (threadDelay maxBound)   (conn, (codec, queues)) <- takeMVar client `onException` cancel tid@@ -180,3 +186,4 @@     , conn     , queues     )+

@@ -1,20 +1,45 @@ module Network.QUIC.Simple.Stream-  ( MessageQueues-  , streamSerialise-  , streamCodec+  ( -- $intro++    -- * Message queues+    MessageQueues+  , serialise+  , serialiseFrom+  , codec+  , codecFrom+    -- * One-shot messages+  , sendMessage+  , recvMessage+    -- * Raw bytes+  , consume+    -- * Multiple streams+  , open+  , acceptLoop   ) where -import Codec.Serialise (Serialise, serialise, deserialiseIncremental)+import Codec.Serialise (Serialise, deserialiseIncremental)+import Codec.Serialise qualified as CBOR import Codec.Serialise qualified as IDecode (IDecode(..))-import Control.Concurrent.Async (Async, async, race_)+import Control.Concurrent.Async (Async, async, cancel, link, poll, race_) import Control.Concurrent.STM import Control.Exception (finally, throwIO)+import Control.Monad (filterM, forever, unless) import Control.Monad.ST (stToIO) import Data.ByteString qualified as BS import Data.ByteString.Lazy qualified as BSL import Data.IORef+import Data.Maybe (isNothing) import Network.QUIC qualified as QUIC +{- $intro+This module is meant to be imported qualified:++> import Network.QUIC.Simple.Stream qualified as Stream++The connection wrappers in "Network.QUIC.Simple" hand out the first stream.+The functions here run protocols over it, or open and accept more streams.+-}+ {- | A pair of bounded queues wrapping a stream. -} type MessageQueues sendMsg recvMsg = (TBQueue sendMsg, TBQueue recvMsg)@@ -25,12 +50,23 @@  No extra framing is required since CBOR is self-delimiting. -}-streamSerialise+serialise+  :: (Serialise sendMsg, Serialise recvMsg)+  => QUIC.Stream+  -> IO (Async (), MessageQueues sendMsg recvMsg)+serialise = serialiseFrom ""++{- | Same as 'serialise', but starts decoding from the bytes already received.++Use with the leftovers from 'recvMessage' when a stream starts with a header.+-}+serialiseFrom   :: forall sendMsg recvMsg   . (Serialise sendMsg, Serialise recvMsg)-  => QUIC.Stream+  => BS.ByteString+  -> QUIC.Stream   -> IO (Async (), MessageQueues sendMsg recvMsg)-streamSerialise stream = do+serialiseFrom leftovers stream = do   initial <- stToIO $ deserialiseIncremental @recvMsg   state <- newIORef initial   let@@ -39,49 +75,146 @@       case decoder of         IDecode.Fail _leftovers _offset err ->           throwIO err -- crash writer (thus the stream, and the reader/writer etc)-        IDecode.Done leftovers _consumed msg -> do+        IDecode.Done leftovers' _consumed msg -> do           stToIO deserialiseIncremental >>= writeIORef state -- restart decoder-          pure (leftovers, Just msg)-        IDecode.Partial consume -> do+          pure (leftovers', Just msg)+        IDecode.Partial consume_ -> do           -- want more data (initial state?)-          stToIO (consume $ Just chunk) >>= writeIORef state -- step decoder+          stToIO (consume_ $ Just chunk) >>= writeIORef state -- step decoder           if starting then             -- re-check if done             decode False ""           else             -- suspend and wait for next chunk             pure ("", Nothing)-  streamCodec serialise (decode True) stream+  codecFrom CBOR.serialise (decode True) leftovers 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.++The worker finishes when the peer closes the stream, or when either the codec or the stream fails.+Cancel it to close the stream from this side.+Messages still queued for sending at that moment are dropped. -}-streamCodec+codec   :: (sendMsg -> BSL.ByteString) -- ^ Encoder for outgoing messages   -> (BS.ByteString -> IO (BS.ByteString, Maybe recvMsg)) -- ^ Decoder for incomming chunks   -> QUIC.Stream   -> IO (Async (), MessageQueues sendMsg recvMsg)-streamCodec encode decode stream = do+codec encode decode = codecFrom encode decode ""++{- | Same as 'codec', but starts decoding from the bytes already received.+-}+codecFrom+  :: (sendMsg -> BSL.ByteString) -- ^ Encoder for outgoing messages+  -> (BS.ByteString -> IO (BS.ByteString, Maybe recvMsg)) -- ^ Decoder for incomming chunks+  -> BS.ByteString -- ^ Bytes received before the codec took over+  -> QUIC.Stream+  -> IO (Async (), MessageQueues sendMsg recvMsg)+codecFrom encode decode leftovers stream = do   readQ <- newTBQueueIO 1024   writeQ <- newTBQueueIO 1024   worker <- async $-    race_ (reader "" readQ) (writer writeQ) `finally` QUIC.closeStream stream+    race_ (reader leftovers readQ) (writer writeQ) `finally` QUIC.closeStream stream   pure (worker, (writeQ, readQ))   where-    reader leftovers readQ = do+    reader buffered readQ = do       chunk <--        if BS.null leftovers then+        if BS.null buffered then           QUIC.recvStream stream 4096         else-          pure leftovers-      (leftovers', message_) <- decode chunk-      mapM_ (atomically . writeTBQueue readQ) message_-      reader leftovers' readQ+          pure buffered+      unless (BS.null chunk) do+        (buffered', message_) <- decode chunk+        mapM_ (atomically . writeTBQueue readQ) message_+        reader buffered' readQ      writer writeQ = do       message <- atomically $ readTBQueue writeQ       let chunks = BSL.toChunks $ encode message       QUIC.sendStreamMany stream chunks       writer writeQ++{- | Send a single CBOR message to a stream.+-}+sendMessage :: Serialise msg => QUIC.Stream -> msg -> IO ()+sendMessage stream = QUIC.sendStreamMany stream . BSL.toChunks . CBOR.serialise++{- | Receive a single CBOR message from a stream.++Bytes received past the end of the message are returned along with it.+They belong to whatever comes next on the stream,+so feed them to 'serialiseFrom' or 'consume'.+In a lockstep exchange there will be none.++Throws a decoding error if the stream ends or gets garbled before the message is complete.+-}+recvMessage :: forall msg. Serialise msg => QUIC.Stream -> IO (msg, BS.ByteString)+recvMessage stream = stToIO (deserialiseIncremental @msg) >>= step+  where+    step = \case+      IDecode.Done leftovers _offset msg ->+        pure (msg, leftovers)+      IDecode.Fail _leftovers _offset err ->+        throwIO err+      IDecode.Partial consume_ -> do+        chunk <- QUIC.recvStream stream 4096+        stToIO (consume_ $ if BS.null chunk then Nothing else Just chunk) >>= step++{- | Feed the incoming bytes to a sink until the peer closes the stream.++The bytes received before, e.g. the leftovers from 'recvMessage', go first.+-}+consume :: BS.ByteString -> QUIC.Stream -> (BS.ByteString -> IO ()) -> IO ()+consume leftovers stream sink = do+  unless (BS.null leftovers) $ sink leftovers+  drain+  where+    drain = do+      chunk <- QUIC.recvStream stream 65536+      unless (BS.null chunk) do+        sink chunk+        drain++{- | Request a new bidirectional stream and announce its purpose with a header message.++The peer is expected to run 'acceptLoop' with a matching header type.+Anything sent right after the header will arrive as leftovers on the other side.+-}+open :: Serialise header => QUIC.Connection -> header -> IO QUIC.Stream+open conn header = do+  new <- QUIC.stream conn+  sendMessage new header+  pure new++{- | Accept streams opened by the peer and dispatch them by their header message.++Each stream gets a thread of its own that closes the stream when the handler returns.+The handler receives the bytes that arrived along with the header,+to be passed along to 'serialiseFrom' or 'consume'.++A crashing handler takes down the accept loop, and thus the connection.+Leaving the loop, by cancelling it or by connection failure, cancels the handlers still running.+-}+acceptLoop+  :: Serialise header+  => QUIC.Connection+  -> (header -> BS.ByteString -> QUIC.Stream -> IO ())+  -> IO ()+acceptLoop conn handler = do+  running <- newTVarIO []+  let+    cancelAll = readTVarIO running >>= mapM_ cancel+    prune = readTVarIO running >>= filterM (fmap isNothing . poll) >>= atomically . writeTVar running+    handleStream new = do+      (header, leftovers) <- recvMessage new+      handler header leftovers new+    accept = forever do+      new <- QUIC.acceptStream conn+      worker <- async $ handleStream new `finally` QUIC.closeStream new+      link worker+      prune+      atomically $ modifyTVar' running (worker :)+  accept `finally` cancelAll

@@ -4,17 +4,17 @@ import Control.Concurrent.STM import Control.Monad -import Control.Concurrent.Async (async, cancel, link, race_, replicateConcurrently_)-import Control.Exception (bracket)+import Control.Concurrent.Async (async, cancel, concurrently_, link, race_, replicateConcurrently_, wait)+import Control.Exception (bracket, throwIO) import Data.ByteString qualified as BS import Data.ByteString.Lazy qualified as BSL-import Data.IORef (newIORef, atomicModifyIORef')+import Data.IORef (newIORef, atomicModifyIORef', readIORef) 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)+import Network.QUIC.Simple.Stream qualified as Stream import System.Timeout (timeout)  main :: IO ()@@ -27,6 +27,10 @@   race_ serverBox clientBox   putStrLn "" +  putStrLn "Datagram"+  race_ serverDatagram clientDatagram+  putStrLn ""+   putStrLn "Serialise"   race_ serverSerialise clientSerialise   putStrLn ""@@ -39,6 +43,10 @@   race_ serverStateful clientAsync   putStrLn "" +  putStrLn "Multistream"+  race_ serverMulti clientMulti+  putStrLn ""+ -- * Raw  clientRaw :: IO ()@@ -78,10 +86,10 @@ 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 :: QUIC.Stream -> IO (Stream.MessageQueues Text Text) cstringCodec stream = do   previous <- newIORef BSL.empty -- gotta store partial messages somewhere-  snd <$> streamCodec encode (decode previous) stream+  snd <$> Stream.codec encode (decode previous) stream   where     -- add framing     encode msg = BSL.fromChunks [encodeUtf8 msg, "\NUL"]@@ -133,6 +141,47 @@     putStrLn $ "Server got query: " <> show query     atomically $ writeTBQueue writeQ $ "got yer bytes: " <> query +-- * Unreliable datagrams++clientDatagram :: IO ()+clientDatagram = 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"+  QUIC.sendDatagram conn "yo"+  yo <- QUIC.recvDatagram conn+  putStrLn $ "Client got datagram: " <> show yo+  reply <- QUIC.recvStream stream 4096+  putStrLn $ "Client got reply: " <> show reply++  QUIC.closeStream stream+  putStrLn "Client quits"++serverDatagram :: IO ()+serverDatagram = QUIC.runServer [("127.0.0.1", 14443)] \conn stream -> do+  -- the initial stream comes pre-accepted and ready to go+  putStrLn "Server accepted connection:"+  QUIC.getConnectionInfo conn >>= print++  -- wait until *something* arrives and take it all in+  query <- QUIC.recvStream stream 4096+  putStrLn $ "Server got query: " <> show query++  yo <- QUIC.recvDatagram conn+  putStrLn $ "Server got datagram: " <> show yo+  QUIC.sendDatagram conn yo++  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++ -- * Serialised messages  data ClientMessage@@ -159,7 +208,7 @@ clientSerialise :: IO () clientSerialise = do   QUIC.runClient "127.0.0.1" "14443" \_conn stream -> do-    (writeQ, readQ) <- snd <$> streamSerialise stream+    (writeQ, readQ) <- snd <$> Stream.serialise stream     replicateM_ 5 do       -- send messages one by one       atomically $ writeTBQueue writeQ Hello@@ -174,7 +223,7 @@ serverSerialise :: IO () serverSerialise = QUIC.runServer [("127.0.0.1", 14443)] \_conn stream -> do   putStrLn "Server accepted connection:"-  (writeQ, readQ) <- snd <$> streamSerialise stream+  (writeQ, readQ) <- snd <$> Stream.serialise stream   let     -- simple state-passing loop     loop counter = do@@ -308,7 +357,7 @@   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+    setup conns counter _conn _readQ writeQ = do       -- generate an explicit key       connId <- atomicModifyIORef' counter \old -> (old + 1, old)       -- don't store ThreadIDs directly!@@ -338,3 +387,117 @@         _ ->           -- ignore casted Hellos and any form of Bye           pure (connState, Nothing)++-- * Handshake and multiple streams++{- | The first stream carries only the handshake.++Nothing else can be said before the session is established+and nothing about the handshake can be said after.+-}+data ClientHello = ClientHello { version :: Int, token :: Text }+  deriving (Show, Generic, QUIC.Serialise)++data ServerHello = Welcome Int | Rejected Text+  deriving (Show, Generic, QUIC.Serialise)++-- | Proof of a completed handshake, carrying whatever it produced.+newtype Session = Session { sessionId :: Int }++{- | Every stream opened afterwards starts with a header naming its protocol.++The rest of the stream is whatever that protocol says: CBOR messages, raw bytes, anything.+-}+data StreamHeader+  = Rpc+  | Upload Text Int+  deriving (Show, Generic, QUIC.Serialise)++data Received = Received Int+  deriving (Show, Generic, QUIC.Serialise)++clientMulti :: IO ()+clientMulti = QUIC.runClient "127.0.0.1" "14443" \conn hello -> do+  Stream.sendMessage hello ClientHello{version = 1, token = "letmein"}+  -- the handshake is lockstep, so there are no leftovers to worry about+  (reply, _) <- Stream.recvMessage hello+  session <- case reply of+    Welcome sid -> pure (Session sid)+    Rejected why -> throwIO (userError (show why))+  QUIC.closeStream hello+  putStrLn $ "Client got session " <> show (sessionId session)++  -- the streams run their protocols concurrently over one connection+  concurrently_ (rpc conn) (upload conn)+  putStrLn "Client quits"+  where+    rpc conn = do+      stream <- Stream.open conn Rpc+      -- the RPC stream is CBOR messages after the header+      (codec, (writeQ, readQ)) <- Stream.serialise stream+      link codec+      replicateM_ 3 do+        atomically $ writeTBQueue writeQ Hello+        Ok n <- atomically $ readTBQueue readQ+        putStrLn $ "Client got RPC reply " <> show n+      -- the replies are in, so the codec has nothing left to send+      cancel codec++    upload conn = do+      let+        pattern = BS.pack [0 .. 255]+        block = BS.concat (replicate 256 pattern)+        blocks = replicate 64 block+        size = sum (map BS.length blocks)+      -- the upload stream is raw bytes after the header, no framing needed+      stream <- Stream.open conn (Upload "blob" size)+      QUIC.sendStreamMany stream blocks+      -- finish sending, but keep the stream open for the reply+      QUIC.shutdownStream stream+      (Received got, _) <- Stream.recvMessage stream+      QUIC.closeStream stream+      unless (got == size) $+        throwIO (userError $ "Upload size mismatch: " <> show (got, size))+      putStrLn $ "Client uploaded " <> show got <> " bytes"++serverMulti :: IO ()+serverMulti = QUIC.runServer [("127.0.0.1", 14443)] \conn hello -> do+  (ClientHello{version, token}, _) <- Stream.recvMessage hello+  session <-+    if version == 1 && token == "letmein" then do+      Stream.sendMessage hello (Welcome 42)+      pure (Session 42)+    else do+      Stream.sendMessage hello (Rejected "bad credentials")+      throwIO (userError "handshake failed")+  QUIC.closeStream hello+  putStrLn $ "Server established session " <> show (sessionId session)++  -- from now on the peer announces what each stream is for+  Stream.acceptLoop conn \header leftovers stream -> case header of+    Rpc ->+      rpc session leftovers stream+    Upload name size ->+      upload name size leftovers stream+  where+    rpc Session{sessionId} leftovers stream = do+      -- the first message may have arrived together with the header+      (codec, (writeQ, readQ)) <- Stream.serialiseFrom leftovers stream+      counter <- newIORef sessionId+      let+        serve = forever do+          Hello <- atomically $ readTBQueue readQ+          n <- atomicModifyIORef' counter \old -> (old + 1, old)+          atomically $ writeTBQueue writeQ (Ok n)+      -- the codec finishes when the client closes the stream+      race_ (wait codec) serve+      putStrLn "Server RPC stream closed"++    upload name size leftovers stream = do+      received <- newIORef 0+      -- the bytes after the header are the payload+      Stream.consume leftovers stream \chunk ->+        atomicModifyIORef' received \old -> (old + BS.length chunk, ())+      total <- readIORef received+      putStrLn $ "Server received " <> show name <> ": " <> show total <> " of " <> show size+      Stream.sendMessage stream (Received total)