Streams, bounds & cancellation

Pick collection or reduction

(ns application.reading
  (:require [cljs.core :refer [await]]
            [fast-twitch.client.core :as client]))

read-body! reads (:body message) from a request/response map. read-stream! reads a native readable stream directly. Both collect the full result before resolving. reduce-stream! reads incrementally and awaits each reducer result, so downstream work participates in backpressure.

TCP's configured :streaming permits the ordinary collection options but rejects BYOB and any :chunk-bytes key. Fetch SSE has its own line/event limit options, not this collector map. Passing one transport's settings to another rejects rather than silently ignoring them.

Bound a collected response

(defn ^:async fetch-text! [url]
  (let [response (await (client/request!
                         {:url url :request-method :get :headers {}}))]
    {:status (:status response)
     :text (await (client/read-body! response
                   {:codec :text :max-bytes 65536}))}))

Here HTTP uses its native codec and the reader owns subsequent consumption. The shorter alternative is request! with {:codec :text :streaming {:max-bytes 65536}}. Choose one read path: streams are single-use resources. A declared Content-Length is not a substitute for counting received bytes. A collector rejects when accumulated bytes exceed its configured limit.

Text uses strict UTF-8 decoding. JSON first reads UTF-8 text and then decodes string-keyed data. A missing stream collects as empty bytes/text; empty input is not valid JSON. Do not expect a default byte reader to yield a string.

Reduce without collecting everything

(defn ^:async count-bytes! [stream]
  (await (client/reduce-stream!
           stream
           (fn [total chunk]
             (let [next (+ total (.-byteLength chunk))]
               (when (> next 65536)
                 (throw (ex-info "Application byte limit" {:bytes next})))
               next))
           0)))

This computes a count without retaining chunks. Reduction has no :max-bytes option: place an application limit in the reducer when one is needed. Returning (reduced result) terminates early; the final resolved value is the unwrapped result. Reducer functions can return Promises, which are awaited before the next read. A reducer that deliberately accumulates all chunks still consumes memory proportional to the input.

With :codec :text, one incremental decoder preserves characters split between chunks and flushes at EOF. Invalid UTF-8 rejects. This is different from decoding each byte chunk as an independent string. Runnable split-character example.

Reader ownership

The reader always releases its lock when the operation settles. With the default :owned? true, failure, abort or early reduced termination also cancels the reader. A producer's cancel hook can then release upstream work. Normal EOF completes naturally.

With :owned? false, failure or early termination releases the lock without cancelling the underlying stream. Use this only when the caller owns the stream's continued lifetime. Reading bytes has already consumed them; releasing the lock does not rewind the source. Another reader may continue from what remains, not from the beginning.

Cancellation has separate boundaries

For an HTTP request, :request-init {:signal ...} controls Fetch. For later native-body reading, pass :signal to the reader. For decoded HTTP collection put reader settings under :streaming. A TCP connection's top-level signal controls connecting, while a :streaming signal controls subsequent receiving. Fetch SSE's top-level signal controls both its HTTP connection and consumer loop.

An already-aborted reader rejects with the signal's reason. Aborting during an outstanding read races that read; owned readers are cancelled and locks are released. The reason may be arbitrary data. Catch it with catch :default, not an Error-only assumption. This does not define a network retry or application deadline by itself.

BYOB

Use {:mode :byob :chunk-bytes 4096} only for native byte streams supporting BYOB readers. A generic ReadableStream does not become a byte stream because this option is supplied. The configured buffer size is bounded to 1–1048576 bytes, defaulting to 65536. Native .getReader may reject unsupported modes. TCP's configured adapters do not expose this option.

Return a stream from a handler

Return the native stream as :body with an appropriate content type. Do not collect it before returning if incremental delivery is the goal. A normal response map, native Response and a Promise of either are all supported. HEAD/conditional wrappers can dispose a body they remove; timeout middleware bounds response creation rather than the entire body lifetime.

Pull-based SSE producer · All reader signatures · Native reader source.