The realtime library provides WebSocket-based real-time communication
with JWT authentication, message routing, and topic-based pub/sub.
Client connects with JWT: ws://host/ws?token=<jwt>
Server verifies JWT via the IJWTVerifier the application supplies
Connection record created and registered
Client sends and receives messages
On disconnect: cleanup registry, unsubscribe from all topics
| Type | Target | Use case |
|---|---|---|
| All connections | System announcements |
| Specific user-id | Direct messages |
| Users with a specific role | Admin notifications |
| Specific connection-id | Job progress updates |
(require '[wagoe.realtime.ports :as ports])
;; Broadcast to all
(ports/broadcast service {:type :announcement :text "System maintenance in 5 minutes"})
;; Send to specific user
(ports/send-to-user service user-id {:type :notification :data {...}})
;; Send to all users with a role
(ports/send-to-role service :admin {:type :alert :data {...}})
;; Send to specific connection
(ports/send-to-connection service conn-id {:type :progress :percent 75});; Subscribe connection to a topic
(ports/subscribe-to-topic pubsub-mgr conn-id "order:123")
;; Publish to all topic subscribers
(ports/publish-to-topic service "order:123" {:type :order-updated :payload {...}})
;; Unsubscribe
(ports/unsubscribe-from-topic pubsub-mgr conn-id "order:123")Push rendered Hiccup to a page without writing JavaScript. The stream rides
the web session, so mount it as an ordinary web route, not :public. A client
names its topics in the URL; :authorize-topic decides which it may have, and
without it every topic is refused.
(require '[wagoe.realtime.shell.handlers.sse :as sse])
;; route, in the module's :web routes
["/orders/events"
{:get {:handler (sse/sse-handler svc
:authorize-topic (fn [request topic]
(may-see-order? (:user request) topic)))}}]
;; page
[:div {:hx-ext "sse" :sse-connect "/web/orders/events?topic=order:123"}
[:table [:tbody#order-rows {:sse-swap "message" :hx-swap "beforeend"}]]]
;; from a handler, a job, any replica
(sse/push-fragment! svc "order:123" [:tr [:td "Shipped"]])
(sse/push-fragment! svc "order:123" [:span#badge [:t :orders/count {:n 4}]] {:event "badge"})[:t …] markers resolve in each subscriber’s locale. An element with
:hx-swap-oob swaps by id, so one push can update several places. Under the
:redis provider fragments cross the wire: keep them plain data.
Each open stream holds one server thread for as long as the page is open, and
writes a heartbeat every 15 seconds; a closed tab is cleaned up at the next one.
The :wagoe/realtime key caps streams per node with :max-streams (default
32, under Jetty’s 50 threads); a stream over the cap is told to retry in 10
seconds. A handler accepts at most :max-topics (16) topics per stream.
An open stream does not see a logout by itself. Pass :revalidate, asked every
heartbeat, and the stream closes once it answers false; the browser reconnects
and gets a 401:
(require '[wagoe.user.ports :as user-ports]
'[wagoe.user.shell.middleware :as user-mw])
(sse/sse-handler svc
:authorize-topic may-see-topic?
:revalidate (fn [request]
(when-let [session (some->> (user-mw/extract-session-token request)
(user-ports/validate-session user-service))]
(:active (user-ports/get-user-by-id user-service (:user-id session))))))validate-session also refreshes the session’s last access, so an open page
keeps a sliding session alive.
:wagoe/realtime
{:provider :memory ; or :redis, to share topics across replicas
:max-streams 32 ; SSE streams per node
:jwt-verifier #ig/ref :my-app/jwt-verifier} ; your IJWTVerifier; realtime ships none
;; :redis also takes :host, :port, :password, :database, :channel and :key-prefixSee realtime library for the full reference.
Can you improve this documentation? These fine people already did:
thijscreemers & Thijs CreemersEdit on GitHub
cljdoc builds & hosts documentation for Clojure/Script libraries
| Ctrl+k | Jump to recent docs |
| ← | Move to previous article |
| → | Move to next article |
| Ctrl+/ | Jump to the search field |