Stream ownership & UTF-8

Build and run

Requires Java 21+, Clojure CLI and the runtime named below. The example pins the compiler that supports native ^:async and await. Run these commands in a separate terminal from other examples.

git clone https://github.com/OUTCASTGEEK-TECH/fast-twitch.git
cd fast-twitch
git checkout a4eb73f96836d034c7a24fe09dc7d51e4c581f64
cd examples/07-native-transports
clojure -M:build streams
node target/streams.cjs

Expected output

Owned stream cancellation released the reader
Incremental text preserved a split UTF-8 character

Cancellation case

The first stream generates bytes on demand with high-water mark zero. The reducer aborts its controller after reading a chunk. Reduction rejects with the exact reason :finished; the catch block checks that reason. :owned? true causes cancellation of the reader on failure, and finally releases its lock. The example asserts one cancellation and an unlocked stream.

Split text case

The second stream emits the four bytes of 😀 in two pieces. Decoding each piece as a separate complete string would corrupt the character. reduce-stream! with :codec :text retains decoder state between chunks and flushes it at EOF. The final accumulated string must be exactly "😀".

Exercise

Return (reduced total) from the reducer to stop early, and inspect cancellation/lock release. Try :owned? false with an application-owned stream and observe that the reducer releases its lock without cancelling the underlying source. A collection limit belongs to read-stream!; reduction does not accept :max-bytes, so implement a running bound in the reducer when needed.

Reader options and lifecycle · reduce-stream! API.

Complete source and build files

The files below are copied without abbreviation from Fast-Twitch revision a4eb73f96836d034c7a24fe09dc7d51e4c581f64. The walkthrough and API reference use the same revision.

Source: examples/07-native-transports/src/native_demo/streams.cljs

View this exact file on GitHub.

(ns native-demo.streams
  (:require [cljs.core :refer [await]]
            [fast-twitch.client.core :as client])
  (:refer-global :only [AbortController ReadableStream Uint8Array console]))

(defn ^:async -main
  []
  (let [controller (AbortController.)
        cancelled (atom 0)
        stream (ReadableStream. #js {:pull #(.enqueue % (Uint8Array. #js [1 2 3]))
                                     :cancel #(swap! cancelled inc)}
                                #js {:highWaterMark 0})]
    (try (await (client/reduce-stream! stream
                                       (fn [n chunk]
                                         (.abort controller :finished)
                                         (+ n (.-byteLength chunk)))
                                       0
                                       {:signal (.-signal controller) :owned? true}))
         (catch :default reason (assert (= :finished reason))))
    (assert (= 1 @cancelled))
    (assert (not (.-locked stream)))
    (.log console "Owned stream cancellation released the reader"))
  (let [stream (ReadableStream.
                 #js {:start (fn [controller]
                               (.enqueue controller (Uint8Array. #js [240 159]))
                               (.enqueue controller (Uint8Array. #js [152 128]))
                               (.close controller))})
        text (await (client/reduce-stream! stream str "" {:codec :text}))]
    (assert (= "😀" text))
    (.log console "Incremental text preserved a split UTF-8 character")))

(set! *main-cli-fn* -main)

Source: examples/07-native-transports/deps.edn

View this exact file on GitHub.

{:aliases {:build {:main-opts ["-m" "native-demo.build"]}}
 :deps {fast-twitch/fast-twitch {:local/root "../.."}
        org.clojure/clojurescript {:git/sha "56a94ac13bfac22108e4d868dfd4763edaceb864"
                                   :git/url "https://github.com/clojure/clojurescript"}}
 :paths ["src" "dev"]}

Source: examples/07-native-transports/dev/native_demo/build.clj

View this exact file on GitHub.

(ns native-demo.build
  (:require [cljs.build.api :as cljs]))

(def modes
  #{"http" "websocket" "sse" "streams" "ports" "tcp-node" "tcp-deno" "tcp-bun"})

(defn -main
  [mode]
  (doseq [mode (if (= mode "all") (sort modes) [mode])]
    (when-not (modes mode) (throw (ex-info "Choose an example" {:modes modes})))
    (cljs/build "src"
                {:main (symbol (str "native-demo." mode))
                 :target :nodejs
                 :output-to (str "target/" mode ".cjs")
                 :output-dir (str "target/" mode)
                 :optimizations :simple})
    (println mode "COMPILED")))