SSE producers & consumers

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 sse
node target/sse.cjs
# To also exercise native EventSource: deno run -A target/sse.cjs

Expected result

Every supported runtime prints Fetch stream SSE passed. A runtime with native EventSource additionally prints Native named EventSource passed. The capability check is deliberate: Node and Bun profiles do not supply native EventSource.

Producer lifetime

The handler allocates its events atom inside each request and returns sse/response. Each pull returns one event and advances this request's sequence. A nil event ends the response. Both HTTP requests made by the example receive their own "one", "two" sequence; no shared cursor lets one client steal another client's events.

Fetch consumer

client/request! receives :transport :sse and an event handler. It remains pending until EOF; the event callback collects each decoded map. The assertion checks both :data strings. This client does not reconnect automatically.

Native EventSource consumer

client/connect! receives :transport :event-source, :event-types ["named"] and a callback. It returns a closeable handle immediately. The callback selects the named event and resolves a Promise with its data. finally closes this reconnecting source and the server.

Exercise

Add a third producer event with an :id. Inspect the decoded Fetch event map. Add an abort signal to the Fetch consumer and an :on-cancel hook to the producer; verify cleanup when reading stops. A normal producer EOF also needs application cleanup, because :on-cancel is not the normal-completion hook.

Full SSE guide · SSE and EventSource options.

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/sse.cljs

View this exact file on GitHub.

(ns native-demo.sse
  (:require [cljs.core :refer [await]]
            [fast-twitch.client.core :as client]
            [fast-twitch.server.core :as server]
            [fast-twitch.server.sse :as sse])
  (:refer-global :only [Promise console globalThis]))

(defn ^:async -main
  []
  (let [listened (atom nil)
        ready (Promise. (fn [resolve _]
                          (reset! listened resolve)))
        handler (fn [_]
                  (let [events (atom [{:data "one"} {:data "two" :event "named"}])]
                    (sse/response (fn []
                                    (let [event (first @events)]
                                      (swap! events next)
                                      event)))))
        serving (server/listen! handler {:port 0 :on-listen @listened})
        source (atom nil)]
    (try (let [url (str "http://127.0.0.1:" (.-port (await ready)) "/events")
               events (atom [])]
           (await (client/request! {:url url :request-method :get :headers {}}
                                   {:transport :sse :on-event #(swap! events conj %)}))
           (assert (= ["one" "two"] (mapv :data @events)))
           (.log console "Fetch stream SSE passed")
           (when (aget globalThis "EventSource")
             (let [named (Promise. (fn [resolve _]
                                     (reset! source (client/connect!
                                                      {:url url}
                                                      {:transport :event-source
                                                       :event-types ["named"]
                                                       :on-event
                                                         #(when (= "named" (:event %))
                                                            (resolve (:data %)))}))))]
               (assert (= "two" (await named)))
               (.log console "Native named EventSource passed"))))
         (finally (when @source (client/close! @source))
                  (await (server/close! serving {:force true}))))))

(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")))