datastar-hs 1.0.1.1 → 1.0.2.0

raw patch · 18 files changed

+1065/−41 lines, 18 filesdep +brotlidep +streaming-commonsdep +zlibdep −taggeddep ~basedep ~bytestringdep ~textnew-component:exe:compression-benchPVP ok

version bump matches the API change (PVP)

Dependencies added: brotli, streaming-commons, zlib, zstd

Dependencies removed: tagged

Dependency ranges changed: base, bytestring, text, wai

API changes (from Hackage documentation)

Files

• CHANGELOG.md +14/−0
• README.md +81/−1
• bench/Main.hs +122/−0
• datastar-hs.cabal +62/−3
• examples/activity-feed.hs +7/−2
• examples/heap-view.hs +18/−14
• examples/hello-world-channel.hs +10/−5
• examples/hello-world-servant.hs +8/−3
• examples/hello-world-warp.hs +6/−1
• src/Hypermedia/Datastar.hs +38/−2
• src/Hypermedia/Datastar/Compression/Brotli.hs +68/−0
• src/Hypermedia/Datastar/Compression/Zlib.hs +68/−0
• src/Hypermedia/Datastar/Compression/Zstd.hs +131/−0
• src/Hypermedia/Datastar/WAI.hs +121/−9
• test/Hypermedia/Datastar/Compression/BrotliSpec.hs +107/−0
• test/Hypermedia/Datastar/Compression/ZlibSpec.hs +109/−0
• test/Hypermedia/Datastar/Compression/ZstdSpec.hs +83/−0
• test/Main.hs +12/−1

@@ -1,5 +1,19 @@ # Changelog +## 1.0.2.0++* Add SSE response compression with `Content-Encoding` negotiation:+  * `sseResponseWith` and `sseResponseWithStrategy`, plus `Compressor` and+    `CompressionStrategy` (`ServerPriority`, `ClientPriority`, `Forced`) — all+    re-exported from `Hypermedia.Datastar`+  * `Hypermedia.Datastar.Compression.Brotli` (`brotli`)+  * `Hypermedia.Datastar.Compression.Zlib` (`gzip`, `deflate`)+  * `Hypermedia.Datastar.Compression.Zstd` (`zstd`) — opt-in behind the `zstd`+    cabal flag (off by default), as it needs an unreleased `hs-zstd`+    ([#3](https://github.com/starfederation/datastar-haskell/issues/3))+* Add a `compression-bench` executable for measuring compression ratios+* Enable stricter GHC warnings (`-Wall` and friends) across all components+ ## 1.0.1.1  * Revert experimental `examples` sub-library introduced in 1.0.1.0 — it broke Hackage doc uploads. Examples remain in the repo as runnable executables.

@@ -22,7 +22,7 @@ Key design decisions:  - **Minimal dependencies** -- the library depends only on `aeson`, `bytestring`,-`http-types`, `text`, and `wai`.+`http-types`, `text`, `wai`, and some compression libraries. - **WAI streaming** -- SSE responses use WAI's native `responseStream`, giving you a `ServerSentEventGenerator` callback with `sendPatchElements`, `sendPatchSignals`, and `sendExecuteScript`.@@ -69,3 +69,83 @@ main :: IO () main = Warp.run 3000 app ```++## Compression++SSE streams can be compressed by negotiating `Content-Encoding` against the+request's `Accept-Encoding`. Pass one or more compressors to `sseResponseWith`+(or `sseResponseWithStrategy`) in preference order:++```haskell+import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Brotli (brotli)+import Hypermedia.Datastar.Compression.Zlib (deflate, gzip)+import Hypermedia.Datastar.Compression.Zstd (zstd)++respond $ sseResponseWith nullLogger [brotli, gzip, deflate] req $ \gen ->+  sendPatchElements gen (patchElements "<div id=\"message\">Hello!</div>")+```++If the client accepts none of the offered encodings, the stream is sent+uncompressed.++### Compression benchmarks++See [bench/Main.hs](bench/Main.hs) for some compression benchmarks.++[Brotli](https://en.wikipedia.org/wiki/Brotli) is outstanding especially when you have+a large blob with small changes.+++```+=== Identical large grid every tick ===+  400 events, ~130.7 KB uncompressed per fragment++  none   :      51.1 MB+  gzip   :       4.0 MB  (   12.8x vs none)+  brotli :       9.0 KB  ( 5779.1x vs none)+  zstd   :      13.6 KB  ( 5779.1x vs none)++=== Fat update: large grid, only the caption changes each tick ===+  400 events, ~130.7 KB uncompressed per fragment++  none   :      51.1 MB+  gzip   :       4.0 MB  (   12.7x vs none)+  brotli :       9.9 KB  ( 5265.5x vs none)+  zstd   :      19.4 KB  ( 5265.5x vs none)++=== Small update: tiny clock div each tick ===+  400 events, ~23 B uncompressed per fragment++  none   :      28.4 KB+  gzip   :       4.4 KB  (    6.5x vs none)+  brotli :       4.8 KB  (    6.0x vs none)+  zstd   :       5.2 KB  (    6.0x vs none)+```++### zstd upstream package++We [added support for flushStream](https://github.com/starfederation/datastar-haskell/issues/3) to+hs-zstd; until we get a new release on hackage, we are pinning the github source using `cabal.project`.++### zstd window size++The zstd compressor uses `ZSTD_initCStream` which sets+the compression level but not the window size, so you get zstd's default+window rather than a large one which would be optimal for "fat updates".++The Haskell wrapper for zstd exposes [compressStream](https://hackage.haskell.org/package/zstd-0.1.3.0/docs/Codec-Compression-Zstd-Base.html#v:compressStream)+but I think we need `ZSTD_compressStream2` to set parameters in `ZSTD_CCTx`?++```C+size_t ZSTD_compressStream2( ZSTD_CCtx* cctx,+                             ZSTD_outBuffer* output,+                             ZSTD_inBuffer* input,+                             ZSTD_EndDirective endOp);+```++<https://facebook.github.io/zstd/zstd_manual.html>:++    Behaves about the same as ZSTD_compressStream, with additional control on end directive.+    - Compression parameters are pushed into CCtx before starting compression, using ZSTD_CCtx_set*()+

@@ -0,0 +1,122 @@+{-# LANGUAGE CPP #-}++{- |+A size benchmark for Datastar SSE compression.++It sends one @PatchElements@ event per "tick" over a single SSE connection and+measures the total bytes that go on the wire, uncompressed vs. gzip vs. brotli —+using the real SDK compressors and streaming path.++Run: @cabal run bench -- [numEvents] [tableRows]@ (defaults: 400 events, 2000 rows).+-}+module Main (main) where++import Control.Concurrent.MVar+import Control.Monad+import Data.ByteString.Builder qualified as BSB+import Data.ByteString.Lazy qualified as BL+import Data.Text (Text)+import Data.Text qualified as T+import System.Environment (getArgs)+import Text.Printf (printf)++import Network.Wai (defaultRequest)+import Network.Wai.Internal (Response (..))++import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Brotli (brotli)+import Hypermedia.Datastar.Compression.Zlib (gzip)+#ifdef ZSTD+import Hypermedia.Datastar.Compression.Zstd (zstd)+#endif++wireBytes :: Response -> IO Int+wireBytes (ResponseStream _ _ body) = do+  nrBytes <- newMVar 0++  let chunkLen = fromIntegral . BL.length . BSB.toLazyByteString++  body+    (\chunk -> modifyMVar_ nrBytes $ \nr -> return $ nr + chunkLen chunk)+    (pure ())++  readMVar nrBytes++wireBytes _ = error "expected a streaming response"++sendAll :: [Text] -> ServerSentEventGenerator -> IO ()+sendAll frags gen = forM_ frags (sendPatchElements gen . patchElements) ++rawBytes :: [Text] -> IO Int+rawBytes frags = wireBytes (sseResponse nullLogger (sendAll frags))++compressedBytes :: Compressor -> [Text] -> IO Int+compressedBytes c frags =+  wireBytes (sseResponseWithStrategy Forced nullLogger [c] defaultRequest (sendAll frags))++bench :: String -> [Text] -> IO ()+bench name frags = do+  let n = length frags+      perEvent = if n == 0 then 0 else T.length (head frags)+  rawN <- rawBytes frags+  gzN <- compressedBytes gzip frags+  brN <- compressedBytes brotli frags++  printf "\n=== %s ===\n" name+  printf "  %d events, ~%s uncompressed per fragment\n\n" n (human perEvent)+  printf "  none   : %12s\n" (human rawN)+  printf "  gzip   : %12s  (%7.1fx vs none)\n" (human gzN) (ratio rawN gzN)+  printf "  brotli : %12s  (%7.1fx vs none)\n" (human brN) (ratio rawN brN)+#ifdef ZSTD+  zstdN <- compressedBytes zstd frags+  printf "  zstd   : %12s  (%7.1fx vs none)\n" (human zstdN) (ratio rawN zstdN)+#endif++ratio :: Int -> Int -> Double+ratio a b = if b == 0 then 0 else fromIntegral a / fromIntegral b++human :: Int -> String+human n+  | n >= 1024 * 1024 = printf "%.1f MB" (fromIntegral n / (1024 * 1024) :: Double)+  | n >= 1024 = printf "%.1f KB" (fromIntegral n / 1024 :: Double)+  | otherwise = printf "%d B" n++tshow :: Int -> Text+tshow = T.pack . show++-- A self-similar table row (lots of intra-fragment repetition).+gridRow :: Int -> Text+gridRow i =+  "<tr><td>" <> tshow i <> "</td><td>item-" <> tshow i <> "</td><td>active</td><td>0.00</td></tr>"++-- A large grid (>32 KB for a couple thousand rows), optionally stamped with a+-- per-tick caption so only a tiny part changes between events.+grid :: Int -> Maybe Int -> Text+grid rows mtick =+  "<table id=\"grid\">"+    <> maybe "" (\t -> "<caption>updated " <> tshow t <> "</caption>") mtick+    <> T.concat (map gridRow [1 .. rows])+    <> "</table>"++clock :: Int -> Text+clock tick = "<div id=\"clock\">" <> tshow tick <> "</div>"++main :: IO ()+main = do+  args <- getArgs+  let n = case args of (a : _) -> read a; _ -> 400+      rows = case args of (_ : b : _) -> read b; _ -> 2000++  putStrLn "Datastar SSE compression benchmark"++  bench+    "Identical large grid every tick"+    (replicate n (grid rows Nothing))++  bench+    "Fat update: large grid, only the caption changes each tick"+    (map (grid rows . Just) [1 .. n])++  bench+    "Small update: tiny clock div each tick"+    (map clock [1 .. n])

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            datastar-hs-version:         1.0.1.1+version:         1.0.2.0 synopsis:        Haskell bindings for Datastar description:   Server-side SDK for building real-time hypermedia applications with@@ -22,10 +22,31 @@   type:     git   location: https://github.com/starfederation/datastar-haskell.git +-- zstd compression depends on an unreleased hs-zstd (streaming flushStream FFI),+-- so it is off by default to keep the Hackage build clean. Enable with -fzstd+-- together with the source-repository-package pin in cabal.project.+-- https://github.com/starfederation/datastar-haskell/issues/3+flag zstd+  description: zstd Content-Encoding compressor (needs unreleased hs-zstd)+  default:     False+  manual:      True++common warnings+  ghc-options:+    -Wall+    -Wcompat+    -Wunused-packages+    -Wredundant-constraints+    -Wincomplete-uni-patterns+    -Wincomplete-record-updates+ library+  import: warnings   exposed-modules:     Hypermedia.Datastar     Hypermedia.Datastar.Attributes+    Hypermedia.Datastar.Compression.Brotli+    Hypermedia.Datastar.Compression.Zlib     Hypermedia.Datastar.ExecuteScript     Hypermedia.Datastar.Logger     Hypermedia.Datastar.PatchElements@@ -36,16 +57,41 @@   default-language: Haskell2010   default-extensions:     ImportQualifiedPost+    LambdaCase     OverloadedStrings   build-depends:     , base       >= 4.14  && < 5     , aeson      >= 2.0   && < 3-    , bytestring >= 0.10  && < 1+    , bytestring >= 0.10.12 && < 1+    , brotli     >= 0.0.0.3 && < 1+    , streaming-commons >= 0.2.3.1 && < 1     , http-types >= 0.12  && < 1     , text       >= 1.2   && < 3     , wai        >= 3.2   && < 4+  if flag(zstd)+    exposed-modules: Hypermedia.Datastar.Compression.Zstd+    build-depends:   zstd >= 0.1.3 && < 1 +executable compression-bench+  import: warnings+  main-is:          Main.hs+  hs-source-dirs:   bench+  default-language: Haskell2010+  default-extensions:+    ImportQualifiedPost+    OverloadedStrings+  build-depends:+    , base+    , bytestring+    , datastar-hs+    , text+    , wai+  if flag(zstd)+    cpp-options: -DZSTD+  ghc-options: -O2+ executable e2e-server+  import: warnings   hs-source-dirs:   e2e-server   main-is:          Main.hs   build-depends:@@ -70,27 +116,37 @@   test-suite datastar-hs-test+  import: warnings   type:             exitcode-stdio-1.0   hs-source-dirs:   test   main-is:          Main.hs   other-modules:+    Hypermedia.Datastar.Compression.BrotliSpec+    Hypermedia.Datastar.Compression.ZlibSpec     Hypermedia.Datastar.ExecuteScriptSpec     Hypermedia.Datastar.PatchElementsSpec     Hypermedia.Datastar.PatchSignalsSpec     Hypermedia.Datastar.SSESpec   build-depends:     , base+    , brotli     , bytestring     , datastar-hs     , hspec     , text     , wai+    , zlib+  if flag(zstd)+    other-modules: Hypermedia.Datastar.Compression.ZstdSpec+    build-depends: zstd+    cpp-options:   -DZSTD   default-language: Haskell2010   default-extensions:     ImportQualifiedPost     OverloadedStrings  executable hello-world-warp+  import: warnings   main-is:          hello-world-warp.hs   hs-source-dirs:   examples   default-language:  Haskell2010@@ -109,6 +165,7 @@   ghc-options: -threaded  executable hello-world-servant+  import: warnings   main-is:          hello-world-servant.hs   hs-source-dirs:   examples   default-language:  Haskell2010@@ -125,7 +182,6 @@     , http-media        >= 0.8   && < 1     , http-types     , servant-server    >= 0.19  && < 1-    , tagged            >= 0.8   && < 1     , text     , wai     , warp              >= 3.2   && < 4@@ -133,6 +189,7 @@   ghc-options: -threaded  executable hello-world-channel+  import: warnings   main-is:          hello-world-channel.hs   hs-source-dirs:   examples   default-language:  Haskell2010@@ -152,6 +209,7 @@   ghc-options: -threaded  executable activity-feed+  import: warnings   main-is:          activity-feed.hs   hs-source-dirs:   examples   default-language:  Haskell2010@@ -171,6 +229,7 @@   ghc-options: -threaded  executable heap-view+  import: warnings   main-is:          heap-view.hs   hs-source-dirs:   examples   default-language:  Haskell2010

@@ -10,6 +10,8 @@ import Data.Time.Clock (getCurrentTime) import Data.Time.Format (defaultTimeLocale, formatTime) import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Brotli (brotli)+import Hypermedia.Datastar.Compression.Zlib (deflate, gzip) import Network.HTTP.Types (status200, status404) import Network.Wai (Application, pathInfo, requestMethod, responseLBS) import Network.Wai qualified as Wai@@ -104,12 +106,15 @@     _ ->       respond $ responseLBS status404 [] "Not found" +compressors :: [Compressor]+compressors = [brotli, gzip, deflate]+ handleGenerate :: Wai.Request -> (Wai.Response -> IO b) -> IO b handleGenerate req respond = do   signalsResult <- readSignals req :: IO (Either String Signals)   case signalsResult of     Left err -> respond $ responseLBS status404 [] (LBS.fromStrict $ BS8.pack $ "Bad signals: " <> err)-    Right signals -> respond $ sseResponse nullLogger $ \gen -> do+    Right signals -> respond $ sseResponseWith nullLogger compressors req $ \gen -> do       sendPatchSignals gen (patchSignals "{\"generating\": true}")        let loop 0 _ _ = pure ()@@ -134,7 +139,7 @@   signalsResult <- readSignals req :: IO (Either String Signals)   case signalsResult of     Left err -> respond $ responseLBS status404 [] (LBS.fromStrict $ BS8.pack $ "Bad signals: " <> err)-    Right signals -> respond $ sseResponse nullLogger $ \gen -> do+    Right signals -> respond $ sseResponseWith nullLogger compressors req $ \gen -> do       let newTotal = _sTotal signals + 1           counterSignals = case status of             Done -> "{\"total\": " <> T.pack (show newTotal) <> ", \"done\": " <> T.pack (show (_sDone signals + 1)) <> "}"

@@ -10,7 +10,6 @@ import Data.ByteString.Lazy qualified as LBS import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map-import Data.Set (Set) import Data.Set qualified as Set import Data.Text (Text) import Data.Text qualified as T@@ -19,6 +18,8 @@ import Data.UUID.V4 qualified as UUID import GHC.Exts.Heap (Box (..), GenClosure (..), asBox, getClosureData) import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Brotli (brotli)+import Hypermedia.Datastar.Compression.Zlib (deflate, gzip) import Network.HTTP.Types (queryToQueryText, status200, status400, status404) import Network.Wai (Application, Request, Response, pathInfo, queryString, requestMethod, responseLBS) import Network.Wai.Handler.Warp qualified as Warp@@ -463,13 +464,13 @@ main = do   args <- getArgs   case args of-    [name] | Just mode <- lookup name allModes -> startServer 3000 mode-    [name, portStr] | Just mode <- lookup name allModes -> startServer (read portStr) mode+    [name'] | Just mode <- lookup name' allModes -> startServer 3000 mode+    [name', portStr] | Just mode <- lookup name' allModes -> startServer (read portStr) mode     _ -> do       hPutStrLn stderr "Usage: heap-view <mode> [port]"       hPutStrLn stderr ""       hPutStrLn stderr "Modes:"-      mapM_ (\(name, mode) -> hPutStrLn stderr $ "  " <> name <> replicate (16 - length name) ' ' <> T.unpack (modeDesc mode)) allModes+      mapM_ (\(name', mode) -> hPutStrLn stderr $ "  " <> name' <> replicate (16 - length name') ' ' <> T.unpack (modeDesc mode)) allModes       exitFailure  startServer :: Int -> Mode -> IO ()@@ -492,14 +493,14 @@       respond $ responseLBS status200 [("Content-Type", "text/html")] (LBS.fromStrict htmlContent)     ("GET", ["heap"]) ->       withSession appState req respond $ \sess ->-        handleHeap appState sess respond+        handleHeap appState sess req respond     ("GET", ["force"]) ->       withSession appState req respond $ \sess ->         handleForce appState sess req respond     ("GET", ["run"])       | Just _ <- modeRun (appMode appState) ->           withSession appState req respond $ \sess ->-            handleRun appState sess respond+            handleRun appState sess req respond     ("GET", ["reset"]) ->       handleReset appState req respond     _ ->@@ -514,9 +515,12 @@   let html = renderHeapTable (sessId sess) (appHasRun appState) desc rootAddr nodes   sendPatchElements gen (patchElements html) -handleHeap :: AppState -> Session -> (Response -> IO b) -> IO b-handleHeap appState sess respond =-  respond $ sseResponse nullLogger $ \gen ->+compressors :: [Compressor]+compressors = [brotli, gzip, deflate]++handleHeap :: AppState -> Session -> Request -> (Response -> IO b) -> IO b+handleHeap appState sess req respond =+  respond $ sseResponseWith nullLogger compressors req $ \gen ->     sendHeapUpdate appState sess gen  handleForce :: AppState -> Session -> Request -> (Response -> IO b) -> IO b@@ -525,7 +529,7 @@   case lookup "addr" params of     Just (Just addr) -> do       forceThunk sess addr-      respond $ sseResponse nullLogger $ \gen ->+      respond $ sseResponseWith nullLogger compressors req $ \gen ->         sendHeapUpdate appState sess gen     _ ->       respond $ responseLBS status400 [] "Missing addr parameter"@@ -541,15 +545,15 @@         Nothing -> newSession appState     Nothing -> newSession appState   modeSetup (appMode appState) sess-  respond $ sseResponse nullLogger $ \gen ->+  respond $ sseResponseWith nullLogger compressors req $ \gen ->     sendHeapUpdate appState sess gen -handleRun :: AppState -> Session -> (Response -> IO b) -> IO b-handleRun appState sess respond = do+handleRun :: AppState -> Session -> Request -> (Response -> IO b) -> IO b+handleRun appState sess req respond = do   case modeRun (appMode appState) of     Just run -> do       run sess-      respond $ sseResponse nullLogger $ \gen -> do+      respond $ sseResponseWith nullLogger compressors req $ \gen -> do         -- Stream live updates every 200ms for ~12 seconds         replicateM_ 60 $ do           sendHeapUpdate appState sess gen

@@ -7,6 +7,8 @@ import Data.ByteString.Lazy qualified as LBS import Data.Text qualified as T import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Brotli (brotli)+import Hypermedia.Datastar.Compression.Zlib (deflate, gzip) import Network.HTTP.Types (status200, status404) import Network.Wai (Application, pathInfo, requestMethod, responseLBS) import Network.Wai qualified as Wai@@ -28,6 +30,9 @@ message :: String message = "Hello, world!" +compressors :: [Compressor]+compressors = [brotli, gzip, deflate]+ main :: IO () main = do   args <- getArgs@@ -47,7 +52,7 @@     ("GET", ["set-delay"]) ->       handleSetDelay state req respond     ("GET", ["hello-world"]) ->-      handleHelloWorld state respond+      handleHelloWorld state req respond     _ ->       respond $ responseLBS status404 [] "Not found" @@ -61,11 +66,11 @@         writeTVar (_delayVar state) (_delay signals)         v <- readTVar (_versionVar state)         writeTVar (_versionVar state) (v + 1)-      respond $ sseResponse nullLogger $ \_ -> pure ()+      respond $ sseResponseWith nullLogger compressors req $ \_ -> pure () -handleHelloWorld :: SharedState -> (Wai.Response -> IO b) -> IO b-handleHelloWorld state respond =-  respond $ sseResponse nullLogger $ \gen ->+handleHelloWorld :: SharedState -> Wai.Request -> (Wai.Response -> IO b) -> IO b+handleHelloWorld state req respond =+  respond $ sseResponseWith nullLogger compressors req $ \gen ->     forever $ do       version <- readTVarIO (_versionVar state)       d <- readTVarIO (_delayVar state)

@@ -6,6 +6,8 @@ import Data.ByteString.Lazy qualified as LBS import Data.Text qualified as T import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Brotli (brotli)+import Hypermedia.Datastar.Compression.Zlib (deflate, gzip) import Network.HTTP.Media ((//)) import Network.HTTP.Types (status404) import Network.Wai qualified as Wai@@ -30,6 +32,9 @@ message :: String message = "Hello, world!" +compressors :: [Compressor]+compressors = [brotli, gzip, deflate]+ type API =   Get '[HTML] LBS.ByteString     :<|> "hello-world" :> Raw@@ -44,11 +49,11 @@   pure $ LBS.fromStrict htmlContent  serveHelloWorld :: Tagged Handler Application-serveHelloWorld = Tagged $ \req respond -> do+serveHelloWorld = Tagged $ \req respond' -> do   signalsResult <- readSignals req :: IO (Either String Signals)   case signalsResult of-    Left _ -> respond $ Wai.responseLBS status404 [] "Bad signals"-    Right signals -> respond $ sseResponse nullLogger $ \gen ->+    Left _ -> respond' $ Wai.responseLBS status404 [] "Bad signals"+    Right signals -> respond' $ sseResponseWith nullLogger compressors req $ \gen ->       mapM_         ( \i -> do             let html = "<div id='message'>" <> T.pack (take i message) <> "</div>"

@@ -6,12 +6,17 @@ import Data.ByteString.Lazy qualified as LBS import Data.Text qualified as T import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Brotli (brotli)+import Hypermedia.Datastar.Compression.Zlib (deflate, gzip) import Network.HTTP.Types (status200, status404) import Network.Wai (Application, pathInfo, requestMethod, responseLBS) import Network.Wai qualified as Wai import Network.Wai.Handler.Warp qualified as Warp import System.Environment (getArgs) +compressors :: [Compressor]+compressors = [brotli, gzip, deflate]+ newtype Signals = Signals {delay :: Int}  instance FromJSON Signals where@@ -46,7 +51,7 @@   signalsResult <- readSignals req :: IO (Either String Signals)   case signalsResult of     Left _ -> respond $ responseLBS status404 [] "Bad signals"-    Right signals -> respond $ sseResponse nullLogger $ \gen ->+    Right signals -> respond $ sseResponseWith nullLogger compressors req $ \gen ->       mapM_         ( \i -> do             let html = "<div id='message'>" <> T.pack (take i message) <> "</div>"

@@ -34,12 +34,33 @@ main = Warp.run 3000 app @ +=== With compression++Use 'sseResponseWith' to negotiate a @Content-Encoding@ from the request's+@Accept-Encoding@ header. Pass the compressors you want to offer, in+preference order — the first one the client accepts wins:++@+import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Brotli (brotli)+import Hypermedia.Datastar.Compression.Zlib (deflate, gzip)++app req respond =+  respond $ sseResponseWith nullLogger [brotli, gzip, deflate] req $ \\gen ->+    sendPatchElements gen (patchElements \"\<div id=\\\"msg\\\"\>Hello!\<\/div\>\")+@++If the client accepts none of them, the stream is sent uncompressed. Use+'sseResponseWithStrategy' to control negotiation with a 'CompressionStrategy'+(e.g. prefer the client's order, or force a single encoding).+ === Module guide  * "Hypermedia.Datastar.PatchElements" — send HTML to morph into the DOM * "Hypermedia.Datastar.PatchSignals" — update the browser's reactive signals * "Hypermedia.Datastar.ExecuteScript" — run JavaScript in the browser * "Hypermedia.Datastar.WAI" — SSE streaming, signal decoding, request helpers+* "Hypermedia.Datastar.Compression.Brotli", "Hypermedia.Datastar.Compression.Zlib" — @Content-Encoding@ compressors * "Hypermedia.Datastar.Types" — protocol types and defaults  === Further reading@@ -75,8 +96,19 @@   , readSignals   , isDatastarRequest +    -- * Compression+    --+    -- | Negotiate @Content-Encoding@ for an SSE stream. Pass one or more+    -- compressors from "Hypermedia.Datastar.Compression.Brotli" or+    -- "Hypermedia.Datastar.Compression.Zlib". 'sseResponseWith' uses+    -- 'ServerPriority'; 'sseResponseWithStrategy' lets you choose.+  , Compressor+  , CompressionStrategy (..)+  , sseResponseWith+  , sseResponseWithStrategy+     -- * Logger-  , DatastarLogger (..)+  , DatastarLogger   , nullLogger   , stderrLogger   )@@ -88,11 +120,15 @@ import Hypermedia.Datastar.PatchSignals (PatchSignals (..), patchSignals) import Hypermedia.Datastar.Types (ElementNamespace (..), ElementPatchMode (..), EventType (..)) import Hypermedia.Datastar.WAI-  ( ServerSentEventGenerator+  ( CompressionStrategy (..)+  , Compressor+  , ServerSentEventGenerator   , isDatastarRequest   , readSignals   , sendExecuteScript   , sendPatchElements   , sendPatchSignals   , sseResponse+  , sseResponseWith+  , sseResponseWithStrategy   )

@@ -0,0 +1,68 @@+module Hypermedia.Datastar.Compression.Brotli+  ( brotli+  , brotliWith+  , defaultBrotliParams+  )+where++import Codec.Compression.Brotli qualified as B+import Data.ByteString qualified as BS+import Data.ByteString.Builder qualified as BSB+import Data.ByteString.Lazy qualified as BL+import Control.Concurrent.MVar++import Hypermedia.Datastar.WAI (Compressor (..))++brotli :: Compressor+brotli = brotliWith defaultBrotliParams++brotliWith :: B.CompressParams -> Compressor+brotliWith params = Compressor "br" wrap+  where+    wrap rawWrite rawFlush = do+      csM <- B.compressIO params >>= newMVar++      let write builder = do+            let bs = BL.toStrict $ BSB.toLazyByteString builder+            if BS.null bs+              then pure ()+              else modifyMVar_ csM $ \case +                     B.CompressInputRequired _ supply ->+                       supply bs >>= doRawWrites rawWrite+                     other -> pure other++          flush = do+            modifyMVar_ csM $ \case+              B.CompressInputRequired doFlush _ ->+                doFlush >>= doRawWrites rawWrite+              other -> pure other+            rawFlush++          finish = do+            modifyMVar_ csM $ \case+              B.CompressInputRequired _ supply ->+                supply BS.empty >>= doRawWrites rawWrite+              other -> pure other+            rawFlush++      return (write, flush, finish)++doRawWrites+  :: (BSB.Builder -> IO ())+  -> B.CompressStream IO+  -> IO (B.CompressStream IO)+doRawWrites rawWrite cs = case cs of+  B.CompressOutputAvailable chunk nextAction -> do+    rawWrite $ BSB.byteString chunk+    cs' <- nextAction+    doRawWrites rawWrite cs'+  _ -> pure cs++defaultBrotliParams :: B.CompressParams+defaultBrotliParams =+  B.defaultCompressParams+    { B.compressMode = B.CompressionModeText+    , B.compressLevel = B.CompressionLevel5+    , B.compressWindowSize = B.CompressionWindowBits24+    }+

@@ -0,0 +1,68 @@+module Hypermedia.Datastar.Compression.Zlib+  ( gzip+  , gzipWith+  , deflate+  , deflateWith+  , defaultCompressionLevel+  )+where++import Control.Exception (throwIO)++import Data.ByteString qualified as BS+import Data.ByteString.Builder qualified as BSB+import Data.ByteString.Lazy qualified as BL++import Data.Streaming.Zlib+  ( PopperRes (..)+  , WindowBits (..)+  , feedDeflate+  , finishDeflate+  , flushDeflate+  , initDeflate+  )++import Hypermedia.Datastar.WAI (Compressor (..))++defaultCompressionLevel :: Int+defaultCompressionLevel = 6++-- | A gzip 'Compressor' (@Content-Encoding: gzip@) at 'defaultCompressionLevel'.+gzip :: Compressor+gzip = gzipWith defaultCompressionLevel++-- | A gzip 'Compressor' at an explicit zlib level (0–9).+gzipWith :: Int -> Compressor+gzipWith level = zlibCompressor "gzip" level (WindowBits 31)++-- | A zlib/deflate 'Compressor' (@Content-Encoding: deflate@) at 'defaultCompressionLevel'.+deflate :: Compressor+deflate = deflateWith defaultCompressionLevel++-- | A zlib/deflate 'Compressor' at an explicit zlib level (0–9).+deflateWith :: Int -> Compressor+deflateWith level = zlibCompressor "deflate" level (WindowBits 15)++-- WindowBits 31 selects gzip framing, 15 selects zlib/deflate framing.+zlibCompressor :: BS.ByteString -> Int -> WindowBits -> Compressor+zlibCompressor enc level wbits = Compressor enc wrap+  where +    wrap rawWrite rawFlush = do+        def <- initDeflate level wbits++        let drain popper = do+              res <- popper+              case res of+                PRDone -> pure ()+                PRNext bs -> rawWrite (BSB.byteString bs) >> drain popper+                PRError e -> throwIO e++            write builder = do+              popper <- feedDeflate def $ BL.toStrict $ BSB.toLazyByteString builder+              drain popper++            flush = drain (flushDeflate def) >> rawFlush++            finish = drain (finishDeflate def) >> rawFlush++        pure (write, flush, finish)

@@ -0,0 +1,131 @@+{- |+A 'Compressor' that compresses an SSE stream with zstd (@Content-Encoding: zstd@),+driving libzstd's streaming API directly so each event can be flushed.+-}+module Hypermedia.Datastar.Compression.Zstd+  ( zstd+  , zstdWith+  , defaultZstdLevel+  )+where++import Control.Exception (Exception, throwIO)+import Control.Monad (unless, when)++import Data.ByteString qualified as BS+import Data.ByteString.Builder qualified as BSB+import Data.ByteString.Lazy qualified as BL+import Data.ByteString.Unsafe qualified as BU+import Data.Word (Word8)++import Foreign.ForeignPtr+  ( ForeignPtr+  , finalizeForeignPtr+  , mallocForeignPtrBytes+  , newForeignPtr+  , withForeignPtr+  )+import Foreign.Marshal.Alloc (finalizerFree, malloc)+import Foreign.Ptr (Ptr, castPtr, nullPtr)+import Foreign.Storable (peek, poke)++import Codec.Compression.Zstd.FFI+  ( Buffer (..)+  , In+  , Out+  , compressStream+  , createCStream+  , cstreamOutSize+  , endStream+  , flushStream+  , getErrorName+  , initCStream+  , isError+  , p_freeCStream+  )++import Hypermedia.Datastar.WAI (Compressor (..))++newtype ZstdError = ZstdError String+  deriving (Show)++instance Exception ZstdError++-- | zstd's default compression level (3).+defaultZstdLevel :: Int+defaultZstdLevel = 3++-- | A zstd 'Compressor' (@Content-Encoding: zstd@) at 'defaultZstdLevel'.+zstd :: Compressor+zstd = zstdWith defaultZstdLevel++-- | A zstd 'Compressor' at an explicit compression level (1–22).+zstdWith :: Int -> Compressor+zstdWith level =+  Compressor+    { compressorEncoding = "zstd"+    , compressorWrap = \rawWrite rawFlush -> do+        let out = rawWrite . BSB.byteString+            outSize = fromIntegral cstreamOutSize :: Int++        csPtr <- createCStream+        when (csPtr == nullPtr) $ throwIO (ZstdError "ZSTD_createCStream returned NULL")+        initRet <- initCStream csPtr (fromIntegral level)+        when (isError initRet) $+          throwIO (ZstdError ("ZSTD_initCStream: " <> getErrorName initRet))+        csFp <- newForeignPtr p_freeCStream csPtr++        inFp <- newForeignPtr finalizerFree =<< (malloc :: IO (Ptr (Buffer In)))+        outFp <- newForeignPtr finalizerFree =<< (malloc :: IO (Ptr (Buffer Out)))+        scratchFp <- mallocForeignPtrBytes outSize :: IO (ForeignPtr Word8)++        let+          drainOut outBuf scratch = do+            produced <- fromIntegral . bufPos <$> peek outBuf+            when (produced > 0) $+              out =<< BS.packCStringLen (castPtr scratch, produced)++          resetOut outBuf scratch =+            poke outBuf (Buffer scratch (fromIntegral outSize) 0)++          checkRet ctx ret =+            when (isError ret) $ throwIO (ZstdError (ctx <> ": " <> getErrorName ret))++          write builder =+            let bs = BL.toStrict (BSB.toLazyByteString builder)+             in unless (BS.null bs) $+                  BU.unsafeUseAsCStringLen bs $ \(srcPtr, len) ->+                    withForeignPtr csFp $ \cs ->+                      withForeignPtr inFp $ \inBuf ->+                        withForeignPtr outFp $ \outBuf ->+                          withForeignPtr scratchFp $ \scratch -> do+                            poke inBuf (Buffer (castPtr srcPtr) (fromIntegral len) 0)+                            let loop = do+                                  resetOut outBuf scratch+                                  checkRet "ZSTD_compressStream" =<< compressStream cs outBuf inBuf+                                  drainOut outBuf scratch+                                  consumed <- fromIntegral . bufPos <$> peek inBuf+                                  when (consumed < len) loop+                            loop++          pump name op =+            withForeignPtr csFp $ \cs ->+              withForeignPtr outFp $ \outBuf ->+                withForeignPtr scratchFp $ \scratch -> do+                  let loop = do+                        resetOut outBuf scratch+                        remaining <- op cs outBuf+                        checkRet name remaining+                        drainOut outBuf scratch+                        when (remaining > 0) loop+                  loop++          flush = pump "ZSTD_flushStream" flushStream >> rawFlush++          finish = do+            pump "ZSTD_endStream" endStream+            finalizeForeignPtr csFp+            rawFlush++        pure (write, flush, finish)+    }

@@ -47,8 +47,14 @@ import Data.Aeson (FromJSON) import Data.Aeson qualified as A +import Data.ByteString qualified as BS import Data.ByteString.Builder qualified as BSB+import Data.ByteString.Char8 qualified as BS8 +import           Data.List  (find)+import           Data.Maybe (listToMaybe)++ import Network.HTTP.Types qualified as WAI import Network.Wai qualified as WAI @@ -58,7 +64,75 @@ import Hypermedia.Datastar.Logger qualified as Logger import Hypermedia.Datastar.PatchElements qualified as PE import Hypermedia.Datastar.PatchSignals qualified as PS+data Compressor = Compressor +  { compressorEncoding :: BS.ByteString+  -- ^ The @Content-Encoding@ token, e.g. @"br"@ for Brotli compression. +  , compressorWrap :: (BSB.Builder -> IO ())+                   -> IO ()+                   -> IO (BSB.Builder -> IO (), IO (), IO ())+  -- ^ @compressorWrap rawWrite rawFlush@ returns @(write, flush, finish)@.+  --+  -- * @write@ compresses and forwards bytes for an event;+  -- * @flush@ flushes the compression encoded and then the underlying raw transport;+  -- * @finish@ closes the compression stream.+  }++data CompressionStrategy+  = ServerPriority+  | ClientPriority+  | Forced+  deriving (Eq, Show)++-- | Faithful port of the Go reference parser: split on ',', trim surrounding+-- whitespace, take the token before ';', drop empties. No q-value handling,+-- no case folding, no wildcard logic.+parseEncodings :: BS.ByteString -> [BS.ByteString]+parseEncodings header =+  [ token+  | part <- BS8.split ',' header+  , let token = BS8.takeWhile (/= ';') (strip part)+  , not (BS8.null token)+  ]+  where+    strip   = BS8.dropWhile isOWS . BS8.dropWhileEnd isOWS+    isOWS c = c == ' ' || c == '\t'++-- | The encoding tokens of a request's @Accept-Encoding@ header, in order.+-- An absent or empty header yields [].+clientEncodings :: WAI.Request -> [BS.ByteString]+clientEncodings req +  = maybe+     [] +     parseEncodings+     (lookup WAI.hAcceptEncoding $ WAI.requestHeaders req)++{- | Pick a compressor for a request's @Accept-Encoding@ header using the given+'CompressionStrategy', or 'Nothing' if none applies. Token matching is exact+(case-sensitive), mirroring the Go reference.+-}+negotiateWith :: CompressionStrategy -> [Compressor] -> WAI.Request -> Maybe Compressor+negotiateWith strategy compressors req =+  case strategy of+    ServerPriority ->+      -- first server compressor whose token the client sent+      find (\c -> compressorEncoding c `elem` accepted) compressors++    ClientPriority ->+      -- first client token (header order) that the server offers+      listToMaybe+        [ c+        | enc <- accepted+        , c   <- compressors+        , compressorEncoding c == enc+        ]++    Forced ->+      -- first configured compressor, regardless of Accept-Encoding+      listToMaybe compressors+  where+    accepted = clientEncodings req+ {- | An opaque handle for sending SSE events to the browser.  Obtain one from the callback passed to 'sseResponse'. The handle is@@ -86,7 +160,32 @@ @ -} sseResponse :: Logger.DatastarLogger -> (ServerSentEventGenerator -> IO ()) -> WAI.Response-sseResponse logger callback =+sseResponse logger = sseResponse' logger Nothing++sseResponseWith+  :: Logger.DatastarLogger+  -> [Compressor]+  -> WAI.Request+  -> (ServerSentEventGenerator -> IO ())+  -> WAI.Response+sseResponseWith = sseResponseWithStrategy ServerPriority++sseResponseWithStrategy+  :: CompressionStrategy+  -> Logger.DatastarLogger+  -> [Compressor]+  -> WAI.Request+  -> (ServerSentEventGenerator -> IO ())+  -> WAI.Response+sseResponseWithStrategy strategy logger compressors req =+  sseResponse' logger (negotiateWith strategy compressors req)++sseResponse'+  :: Logger.DatastarLogger+  -> Maybe Compressor+  -> (ServerSentEventGenerator -> IO ())+  -> WAI.Response+sseResponse' logger chosen callback =   WAI.responseStream     WAI.status200     headers@@ -97,16 +196,29 @@     , ("Content-Type", "text/event-stream")     , ("Connection", "keep-alive")     ]+      <> case chosen of+        Just c -> [("Content-Encoding", compressorEncoding c)]+        Nothing -> [] -  action write flush = do+  action rawWrite rawFlush = do+    (write, flush, finish) <- case chosen of+      Nothing -> pure (rawWrite, rawFlush, pure ())+      Just c -> compressorWrap c rawWrite rawFlush     lock <- newMVar ()-    callback $-      ServerSentEventGenerator-        { sseWrite = write-        , sseFlush = flush-        , sseLock = lock-        , sseLogger = logger-        }+    let gen =+          ServerSentEventGenerator+            { sseWrite = write+            , sseFlush = flush+            , sseLock = lock+            , sseLogger = logger+            }+    callback gen `finally` ignoreIOErrors finish++ignoreIOErrors :: IO () -> IO ()+ignoreIOErrors act = act `catch` handler+ where+  handler :: IOException -> IO ()+  handler _ = pure ()  send :: ServerSentEventGenerator -> DatastarEvent -> IO () send gen event = do

@@ -0,0 +1,107 @@+module Hypermedia.Datastar.Compression.BrotliSpec (spec) where++import Test.Hspec++import Codec.Compression.Brotli qualified as B+import Data.ByteString.Builder qualified as BSB+import Data.ByteString.Lazy qualified as BL+import Data.IORef++import Network.Wai (defaultRequest, requestHeaders)+import Network.Wai.Internal (Response (..))++import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Brotli (brotli)+import Hypermedia.Datastar.Logger (nullLogger)+import Hypermedia.Datastar.WAI (Compressor (..), compressorWrap, negotiateWith)++-- | Drive a streaming WAI response to completion, returning its response headers+-- and the full raw body.+runStream (ResponseStream _status headers body) = do+  ref <- newIORef mempty+  body (\chunk -> modifyIORef' ref (<> chunk)) (pure ())+  bytes <- BSB.toLazyByteString <$> readIORef ref+  pure (headers, bytes)+runStream _ = error "expected a streaming response"++-- | A compressor that only carries an encoding token, for negotiation tests.+-- Its 'compressorWrap' is never invoked by 'negotiateWith'.+fakeCompressor :: BL.ByteString -> Compressor+fakeCompressor enc = Compressor (BL.toStrict enc) (\_ _ -> error "unused")++spec :: Spec+spec = describe "Hypermedia.Datastar.Compression.Brotli" $ do+  let sendEvents gen = do+        sendPatchElements gen (patchElements "<div id=\"a\">1</div>")+        sendPatchElements gen (patchElements "<div id=\"b\">2</div>")+        sendPatchSignals gen (patchSignals "{\"count\":42}")+      withAccept enc = defaultRequest {requestHeaders = [("Accept-Encoding", enc)]}++  it "round-trips: the br stream decompresses to the uncompressed stream" $ do+    (_, reference) <- runStream (sseResponse nullLogger sendEvents)+    (headers, compressed) <-+      runStream (sseResponseWith nullLogger [brotli] (withAccept "br") sendEvents)++    headers `shouldSatisfy` elem ("Content-Encoding", "br")+    B.decompress compressed `shouldBe` reference++  it "declines compression when the client does not accept br" $ do+    (_, reference) <- runStream (sseResponse nullLogger sendEvents)+    (headers, body) <-+      runStream (sseResponseWith nullLogger [brotli] (withAccept "gzip") sendEvents)++    filter ((== "Content-Encoding") . fst) headers `shouldBe` []+    body `shouldBe` reference++  it "emits output incrementally on flush, not buffered until finish" $ do+    ref <- newIORef mempty+    let rawWrite c = modifyIORef' ref (<> c)+        sizeSoFar = fromIntegral . BL.length . BSB.toLazyByteString <$> readIORef ref+    (write, flush, finish) <- compressorWrap brotli rawWrite (pure ())++    write (BSB.byteString "event: datastar-patch-elements\ndata: elements <div>1</div>\n\n")+    flush+    afterFirst <- sizeSoFar++    write (BSB.byteString "event: datastar-patch-elements\ndata: elements <div>2</div>\n\n")+    flush+    afterSecond <- sizeSoFar++    finish+    afterFinish <- sizeSoFar++    -- Each flush must push bytes onto the wire; a compressor that buffered+    -- everything until finish would leave afterFirst == afterSecond == 0.+    afterFirst `shouldSatisfy` (> (0 :: Int))+    afterSecond `shouldSatisfy` (> afterFirst)+    afterFinish `shouldSatisfy` (>= afterSecond)++    -- And the accumulated stream is still a valid, complete brotli stream.+    whole <- BSB.toLazyByteString <$> readIORef ref+    B.decompress whole+      `shouldBe` "event: datastar-patch-elements\ndata: elements <div>1</div>\n\n\+                 \event: datastar-patch-elements\ndata: elements <div>2</div>\n\n"++  describe "negotiateWith" $ do+    let br = fakeCompressor "br"+        gz = fakeCompressor "gzip"+        enc = fmap compressorEncoding+        req h = defaultRequest {requestHeaders = [("Accept-Encoding", h)]}++    it "ServerPriority picks the first server compressor the client accepts" $+      enc (negotiateWith ServerPriority [br, gz] (req "gzip, br")) `shouldBe` Just "br"++    it "ClientPriority picks the client's most-preferred offered encoding" $+      enc (negotiateWith ClientPriority [br, gz] (req "gzip, br")) `shouldBe` Just "gzip"++    it "Forced ignores Accept-Encoding entirely" $+      enc (negotiateWith Forced [br, gz] (req "identity")) `shouldBe` Just "br"++    it "ignores q-value params and whitespace around each token" $+      enc (negotiateWith ClientPriority [br, gz] (req "  gzip;q=0.1 , br ")) `shouldBe` Just "gzip"++    it "returns Nothing when no server encoding is accepted" $+      enc (negotiateWith ServerPriority [br, gz] (req "deflate")) `shouldBe` Nothing++    it "returns Nothing when the Accept-Encoding header is absent" $+      enc (negotiateWith ServerPriority [br, gz] defaultRequest) `shouldBe` Nothing

@@ -0,0 +1,109 @@+module Hypermedia.Datastar.Compression.ZlibSpec (spec) where++import Test.Hspec++import Control.Monad (forM_)++import Codec.Compression.GZip qualified as GZip+import Codec.Compression.Zlib qualified as Zlib+import Data.ByteString qualified as BS+import Data.ByteString.Builder qualified as BSB+import Data.ByteString.Char8 qualified as BS8+import Data.ByteString.Lazy qualified as BL+import Data.IORef++import Network.Wai (defaultRequest, requestHeaders)+import Network.Wai.Internal (Response (..))++import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Zlib (deflate, gzip)+import Hypermedia.Datastar.Logger (nullLogger)+import Hypermedia.Datastar.WAI (compressorWrap)++-- | Drive a streaming WAI response to completion, returning its response headers+-- and the full raw body.+runStream (ResponseStream _status headers body) = do+  ref <- newIORef mempty+  body (\chunk -> modifyIORef' ref (<> chunk)) (pure ())+  bytes <- BSB.toLazyByteString <$> readIORef ref+  pure (headers, bytes)+runStream _ = error "expected a streaming response"++spec :: Spec+spec = describe "Hypermedia.Datastar.Compression.Zlib" $ do+  let sendEvents gen = do+        sendPatchElements gen (patchElements "<div id=\"a\">1</div>")+        sendPatchElements gen (patchElements "<div id=\"b\">2</div>")+        sendPatchSignals gen (patchSignals "{\"count\":42}")+      withAccept enc = defaultRequest {requestHeaders = [("Accept-Encoding", enc)]}++      -- (encoding token, compressor, matching decompressor)+      codecs :: [(BS.ByteString, Compressor, BL.ByteString -> BL.ByteString)]+      codecs =+        [ ("gzip", gzip, GZip.decompress)+        , ("deflate", deflate, Zlib.decompress)+        ]++  forM_ codecs $ \(enc, comp, decompress) -> do+    let name = BS8.unpack enc++    it (name <> ": round-trips to the uncompressed stream") $ do+      (_, reference) <- runStream (sseResponse nullLogger sendEvents)+      (headers, compressed) <-+        runStream (sseResponseWith nullLogger [comp] (withAccept enc) sendEvents)+      headers `shouldSatisfy` elem ("Content-Encoding", enc)+      decompress compressed `shouldBe` reference++    it (name <> ": emits output incrementally on flush, not buffered until finish") $ do+      ref <- newIORef mempty+      let rawWrite c = modifyIORef' ref (<> c)+          sizeSoFar = fromIntegral . BL.length . BSB.toLazyByteString <$> readIORef ref+      (write, flush, finish) <- compressorWrap comp rawWrite (pure ())++      write (BSB.byteString "event: datastar-patch-elements\ndata: elements <div>1</div>\n\n")+      flush+      afterFirst <- sizeSoFar++      write (BSB.byteString "event: datastar-patch-elements\ndata: elements <div>2</div>\n\n")+      flush+      afterSecond <- sizeSoFar++      finish+      afterFinish <- sizeSoFar++      -- Each flush must push bytes onto the wire; a compressor that buffered+      -- everything until finish would leave afterFirst == afterSecond == 0.+      afterFirst `shouldSatisfy` (> (0 :: Int))+      afterSecond `shouldSatisfy` (> afterFirst)+      afterFinish `shouldSatisfy` (>= afterSecond)++      -- And the accumulated stream is still a valid, complete stream.+      whole <- BSB.toLazyByteString <$> readIORef ref+      decompress whole+        `shouldBe` "event: datastar-patch-elements\ndata: elements <div>1</div>\n\n\+                   \event: datastar-patch-elements\ndata: elements <div>2</div>\n\n"++  it "declines when the client accepts neither" $ do+    (_, reference) <- runStream (sseResponse nullLogger sendEvents)+    (headers, body) <-+      runStream (sseResponseWith nullLogger [gzip, deflate] (withAccept "br") sendEvents)+    filter ((== "Content-Encoding") . fst) headers `shouldBe` []+    body `shouldBe` reference++  -- With several server compressors offered and a client that accepts all of+  -- them, ServerPriority (the default for sseResponseWith) picks the *server's*+  -- first — so list order, not header order, decides. Verified end-to-end via+  -- the chosen Content-Encoding and that the matching codec decompresses it.+  it "picks the server's first compressor when the client accepts several (gzip first)" $ do+    (_, reference) <- runStream (sseResponse nullLogger sendEvents)+    (headers, compressed) <-+      runStream (sseResponseWith nullLogger [gzip, deflate] (withAccept "deflate, gzip") sendEvents)+    headers `shouldSatisfy` elem ("Content-Encoding", "gzip")+    GZip.decompress compressed `shouldBe` reference++  it "picks the server's first compressor when the client accepts several (deflate first)" $ do+    (_, reference) <- runStream (sseResponse nullLogger sendEvents)+    (headers, compressed) <-+      runStream (sseResponseWith nullLogger [deflate, gzip] (withAccept "deflate, gzip") sendEvents)+    headers `shouldSatisfy` elem ("Content-Encoding", "deflate")+    Zlib.decompress compressed `shouldBe` reference

@@ -0,0 +1,83 @@+module Hypermedia.Datastar.Compression.ZstdSpec (spec) where++import Test.Hspec++import Codec.Compression.Zstd.Lazy qualified as Zstd+import Data.ByteString.Builder qualified as BSB+import Data.ByteString.Lazy qualified as BL+import Data.IORef++import Network.Wai (defaultRequest, requestHeaders)+import Network.Wai.Internal (Response (..))++import Hypermedia.Datastar+import Hypermedia.Datastar.Compression.Zlib (gzip)+import Hypermedia.Datastar.Compression.Zstd (zstd)+import Hypermedia.Datastar.Logger (nullLogger)+import Hypermedia.Datastar.WAI (compressorWrap)++-- | Drive a streaming WAI response to completion, returning its response headers+-- and the full raw body.+runStream (ResponseStream _status headers body) = do+  ref <- newIORef mempty+  body (\chunk -> modifyIORef' ref (<> chunk)) (pure ())+  bytes <- BSB.toLazyByteString <$> readIORef ref+  pure (headers, bytes)+runStream _ = error "expected a streaming response"++spec :: Spec+spec = describe "Hypermedia.Datastar.Compression.Zstd" $ do+  let sendEvents gen = do+        sendPatchElements gen (patchElements "<div id=\"a\">1</div>")+        sendPatchElements gen (patchElements "<div id=\"b\">2</div>")+        sendPatchSignals gen (patchSignals "{\"count\":42}")+      withAccept enc = defaultRequest {requestHeaders = [("Accept-Encoding", enc)]}++  it "round-trips to the uncompressed stream" $ do+    (_, reference) <- runStream (sseResponse nullLogger sendEvents)+    (headers, compressed) <-+      runStream (sseResponseWith nullLogger [zstd] (withAccept "zstd") sendEvents)+    headers `shouldSatisfy` elem ("Content-Encoding", "zstd")+    Zstd.decompress compressed `shouldBe` reference++  it "declines when the client does not accept zstd" $ do+    (_, reference) <- runStream (sseResponse nullLogger sendEvents)+    (headers, body) <-+      runStream (sseResponseWith nullLogger [zstd] (withAccept "gzip") sendEvents)+    filter ((== "Content-Encoding") . fst) headers `shouldBe` []+    body `shouldBe` reference++  it "is chosen ahead of gzip when offered first and both are accepted" $ do+    (_, reference) <- runStream (sseResponse nullLogger sendEvents)+    (headers, compressed) <-+      runStream (sseResponseWith nullLogger [zstd, gzip] (withAccept "gzip, zstd") sendEvents)+    headers `shouldSatisfy` elem ("Content-Encoding", "zstd")+    Zstd.decompress compressed `shouldBe` reference++  it "emits output incrementally on flush, not buffered until finish" $ do+    ref <- newIORef mempty+    let rawWrite c = modifyIORef' ref (<> c)+        sizeSoFar = fromIntegral . BL.length . BSB.toLazyByteString <$> readIORef ref+    (write, flush, finish) <- compressorWrap zstd rawWrite (pure ())++    write (BSB.byteString "event: datastar-patch-elements\ndata: elements <div>1</div>\n\n")+    flush+    afterFirst <- sizeSoFar++    write (BSB.byteString "event: datastar-patch-elements\ndata: elements <div>2</div>\n\n")+    flush+    afterSecond <- sizeSoFar++    finish+    afterFinish <- sizeSoFar++    -- Each flush must push bytes onto the wire; a compressor that buffered+    -- everything until finish would leave afterFirst == afterSecond == 0.+    afterFirst `shouldSatisfy` (> (0 :: Int))+    afterSecond `shouldSatisfy` (> afterFirst)+    afterFinish `shouldSatisfy` (>= afterSecond)++    whole <- BSB.toLazyByteString <$> readIORef ref+    Zstd.decompress whole+      `shouldBe` "event: datastar-patch-elements\ndata: elements <div>1</div>\n\n\+                 \event: datastar-patch-elements\ndata: elements <div>2</div>\n\n"

@@ -1,8 +1,14 @@+{-# LANGUAGE CPP #-}+ module Main where  import Test.Hspec -import Hypermedia.Datastar qualified+import Hypermedia.Datastar.Compression.BrotliSpec qualified+import Hypermedia.Datastar.Compression.ZlibSpec qualified+#ifdef ZSTD+import Hypermedia.Datastar.Compression.ZstdSpec qualified+#endif import Hypermedia.Datastar.ExecuteScriptSpec qualified import Hypermedia.Datastar.PatchElementsSpec qualified import Hypermedia.Datastar.PatchSignalsSpec qualified@@ -10,6 +16,11 @@  main :: IO () main = hspec $ do+  Hypermedia.Datastar.Compression.BrotliSpec.spec+  Hypermedia.Datastar.Compression.ZlibSpec.spec+#ifdef ZSTD+  Hypermedia.Datastar.Compression.ZstdSpec.spec+#endif   Hypermedia.Datastar.ExecuteScriptSpec.spec   Hypermedia.Datastar.PatchElementsSpec.spec   Hypermedia.Datastar.PatchSignalsSpec.spec