Pure functions for WebSocket authentication logic.
Following FC/IS pattern - all functions are pure (no I/O). JWT token extraction and validation logic. Actual JWT verification delegated to shell/ports.
Pure functions for WebSocket authentication logic. Following FC/IS pattern - all functions are pure (no I/O). JWT token extraction and validation logic. Actual JWT verification delegated to shell/ports.
Pure constructors for cross-instance routing envelopes.
An envelope is plain data describing WHERE a message should go; the bus transports it and a node-local delivery-fn resolves :route against the local registry. No I/O here.
Pure constructors for cross-instance routing envelopes. An envelope is plain data describing WHERE a message should go; the bus transports it and a node-local delivery-fn resolves :route against the local registry. No I/O here.
Pure functions for WebSocket connection state management.
Following FC/IS pattern - all functions are pure (no I/O). Connection lifecycle and authorization logic.
Pure functions for WebSocket connection state management. Following FC/IS pattern - all functions are pure (no I/O). Connection lifecycle and authorization logic.
Pure functions for WebSocket message validation and routing.
Following FC/IS pattern - all functions are pure (no I/O). Message creation, validation, and routing logic.
Pure functions for WebSocket message validation and routing. Following FC/IS pattern - all functions are pure (no I/O). Message creation, validation, and routing logic.
Pure functions for pub/sub topic management.
Implements topic-based message routing where connections can subscribe to topics and receive messages published to those topics. All functions are pure (no I/O, no side effects) following FC/IS pattern.
Key Features:
Data Structure: Subscriptions are represented as a map: {topic-name #{connection-id-1 connection-id-2 ...}}
Pure functions for pub/sub topic management.
Implements topic-based message routing where connections can subscribe to
topics and receive messages published to those topics. All functions are pure
(no I/O, no side effects) following FC/IS pattern.
Key Features:
- Subscribe connection to topic (add to subscription set)
- Unsubscribe connection from topic (remove from set)
- Find all subscribers for topic (set intersection)
- Clean up subscriptions on disconnect (remove all for connection)
Data Structure:
Subscriptions are represented as a map:
{topic-name #{connection-id-1 connection-id-2 ...}}Port definitions for real-time WebSocket communication.
This module defines protocols for WebSocket-based messaging, similar to Phoenix Channels (Elixir) or Socket.io (Node.js). Supports point-to-point, broadcast, and role-based messaging with JWT authentication.
Key Features:
Port definitions for real-time WebSocket communication. This module defines protocols for WebSocket-based messaging, similar to Phoenix Channels (Elixir) or Socket.io (Node.js). Supports point-to-point, broadcast, and role-based messaging with JWT authentication. Key Features: - WebSocket connection management - Message routing (user, role, broadcast, connection-specific) - JWT-based authentication - Connection registry (in-memory or Redis)
Malli schemas for realtime module data structures.
Schemas for:
Malli schemas for realtime module data structures. Schemas for: - Connection state - WebSocket messages - Authentication tokens - Routing metadata - Pub/sub topics and subscriptions
JWT verification adapter that delegates to boundary/user module.
Wraps wagoe.user.shell.auth/validate-jwt-token to provide IJWTVerifier protocol implementation. Avoids direct dependency from core layer to user module.
Responsibilities (Shell/I/O):
JWT verification adapter that delegates to boundary/user module. Wraps wagoe.user.shell.auth/validate-jwt-token to provide IJWTVerifier protocol implementation. Avoids direct dependency from core layer to user module. Responsibilities (Shell/I/O): - Call user module for JWT verification (I/O - external dependency) - Transform user module response to expected format - Handle verification errors
Redis-backed IPubSubManager: topic subscriptions in Redis sets so they are visible cluster-wide.
Keys (optionally prefixed): topic:{t} -> SET of connection-id strings conn:{id} -> SET of topic strings (reverse index)
subscribe / unsubscribe apply both SADD/SREM atomically in a MULTI/EXEC. No explicit DEL on topic sets: Redis auto-removes a set key when its last member is removed, so empty topics disappear and the check-then-act DEL race cannot occur.
Redis-backed IPubSubManager: topic subscriptions in Redis sets so they are
visible cluster-wide.
Keys (optionally prefixed):
topic:{t} -> SET of connection-id strings
conn:{id} -> SET of topic strings (reverse index)
subscribe / unsubscribe apply both SADD/SREM atomically in a MULTI/EXEC.
No explicit DEL on topic sets: Redis auto-removes a set key when its last
member is removed, so empty topics disappear and the check-then-act DEL race
cannot occur.WebSocket connection adapter for Ring/Jetty.
Wraps Ring WebSocket implementation to provide IWebSocketConnection protocol. Handles JSON encoding/decoding and WebSocket frame transmission.
Responsibilities (Shell/I/O):
WebSocket connection adapter for Ring/Jetty. Wraps Ring WebSocket implementation to provide IWebSocketConnection protocol. Handles JSON encoding/decoding and WebSocket frame transmission. Responsibilities (Shell/I/O): - Send messages over WebSocket (I/O operation) - Close WebSocket connections (I/O operation) - JSON encoding (external format transformation) - Error handling and logging
In-memory message bus (single-process / test).
Holds an atom vector of registered delivery-fns. publish invokes each
synchronously and sums their returned counts. This is the default bus for
single-node deployments and the vehicle for the 2-node cross-instance test
(two services sharing one bus instance via the :bus option).
In-memory message bus (single-process / test). Holds an atom vector of registered delivery-fns. `publish` invokes each synchronously and sums their returned counts. This is the default bus for single-node deployments and the vehicle for the 2-node cross-instance test (two services sharing one bus instance via the :bus option).
Redis-backed IMessageBus: routing envelopes travel as Nippy-frozen bytes over a binary Redis pub/sub channel. publish borrows a pooled connection; the subscriber owns ONE dedicated connection (not from the publish pool, so a blocking SUBSCRIBE never starves publishers) on a daemon thread.
Concurrency:
Trust boundary: onMessage thaws Nippy bytes off the channel. Envelopes are produced only by this framework's own publishers on a trusted Redis instance. Do NOT point this at a Redis shared with untrusted writers — Nippy thaw of attacker-controlled bytes is a deserialization risk.
Redis-backed IMessageBus: routing envelopes travel as Nippy-frozen bytes over a binary Redis pub/sub channel. publish borrows a pooled connection; the subscriber owns ONE dedicated connection (not from the publish pool, so a blocking SUBSCRIBE never starves publishers) on a daemon thread. Concurrency: - Singleton subscriber: start-subscriber! is idempotent (running? guard) so a node never holds two channel subscriptions (which would double-deliver). - Reconnect: the daemon loops with backoff, acquiring the connection INSIDE the loop so a Redis outage at startup retries instead of killing the daemon and leaving the node permanently deaf. Gap messages are missed (at-most-once, accepted). Trust boundary: onMessage thaws Nippy bytes off the channel. Envelopes are produced only by this framework's own publishers on a trusted Redis instance. Do NOT point this at a Redis shared with untrusted writers — Nippy thaw of attacker-controlled bytes is a deserialization risk.
In-memory connection registry implementation.
Stores active WebSocket connections in an atom for single-server deployments. For multi-server scaling, swap with Redis-backed registry implementation.
Registry Structure: {connection-id {:connection <Connection record> :ws-adapter <IWebSocketConnection>}}
Responsibilities (Shell/I/O):
In-memory connection registry implementation.
Stores active WebSocket connections in an atom for single-server deployments.
For multi-server scaling, swap with Redis-backed registry implementation.
Registry Structure:
{connection-id {:connection <Connection record>
:ws-adapter <IWebSocketConnection>}}
Responsibilities (Shell/I/O):
- Store and retrieve connection mappings (stateful atom)
- Filter connections by user-id, role (uses core filtering functions)
- Thread-safe updates (atom swap operations)Node-local delivery: resolve a routing envelope to the local node's ws adapters and send. Built as a closure over the local registry + pubsub manager; registered with a message bus via start-subscriber!. Never calls service send methods (no re-publish recursion).
Node-local delivery: resolve a routing envelope to the local node's ws adapters and send. Built as a closure over the local registry + pubsub manager; registered with a message bus via start-subscriber!. Never calls service send methods (no re-publish recursion).
Ring 1.15 WebSocket upgrade handler for wagoe-realtime.
Bridges Ring's map-based Listener (::ring.websocket/listener response) to the IRealtimeService connect/disconnect lifecycle.
Usage: (require '[wagoe.realtime.shell.handlers.ring-websocket :as ws-handler])
;; In your route definitions {:path "/ws" :methods {:get {:handler (ws-handler/websocket-handler realtime-service)}}}
Ring 1.15 WebSocket upgrade handler for wagoe-realtime.
Bridges Ring's map-based Listener (::ring.websocket/listener response)
to the IRealtimeService connect/disconnect lifecycle.
Usage:
(require '[wagoe.realtime.shell.handlers.ring-websocket :as ws-handler])
;; In your route definitions
{:path "/ws"
:methods {:get {:handler (ws-handler/websocket-handler realtime-service)}}}Integrant wiring for the realtime module.
Config key: :wagoe/realtime {:provider :in-memory | :redis ;; redis only: :host "localhost" :port 6379 :password "..." :database 0 ; auth + db selection (production) :timeout 2000 ; socket timeout ms :max-total 8 :max-idle 8 :min-idle 0 ; publish-pool sizing :channel "wagoe:realtime:bus" :key-prefix "realtime" :subscribe-timeout-ms 5000 ; await window for subscription to go live :jwt-verifier <IJWTVerifier ref>}
The local connection registry is in-memory under BOTH providers (sockets are node-local). Only the pub/sub manager and the bus differ. Under :redis the component opens two Jedis pools — one for topic subscriptions (pub/sub manager) and one inside the bus for publish — both closed on halt.
IMPORTANT: the web/WS server component MUST depend on :wagoe/realtime so that start-subscriber! has completed (subscription live) before any WebSocket connection is accepted.
Integrant wiring for the realtime module.
Config key: :wagoe/realtime
{:provider :in-memory | :redis
;; redis only:
:host "localhost" :port 6379
:password "..." :database 0 ; auth + db selection (production)
:timeout 2000 ; socket timeout ms
:max-total 8 :max-idle 8 :min-idle 0 ; publish-pool sizing
:channel "wagoe:realtime:bus"
:key-prefix "realtime"
:subscribe-timeout-ms 5000 ; await window for subscription to go live
:jwt-verifier <IJWTVerifier ref>}
The local connection registry is in-memory under BOTH providers (sockets are
node-local). Only the pub/sub manager and the bus differ. Under :redis the
component opens two Jedis pools — one for topic subscriptions (pub/sub
manager) and one inside the bus for publish — both closed on halt.
IMPORTANT: the web/WS server component MUST depend on :wagoe/realtime so
that start-subscriber! has completed (subscription live) before any WebSocket
connection is accepted.No vars found in this namespace.
In-memory pub/sub topic management (imperative shell).
Implements IPubSubManager protocol using atom-based subscription storage. Coordinates between pure pub/sub core functions and stateful subscription management.
Single-server implementation - subscriptions stored in memory. For multi-server deployments, would need Redis-backed implementation (v0.2.0).
Thread-safe via atom swap operations.
In-memory pub/sub topic management (imperative shell). Implements IPubSubManager protocol using atom-based subscription storage. Coordinates between pure pub/sub core functions and stateful subscription management. Single-server implementation - subscriptions stored in memory. For multi-server deployments, would need Redis-backed implementation (v0.2.0). Thread-safe via atom swap operations.
Realtime service implementation (Shell layer).
Orchestrates WebSocket messaging between core logic and adapters. Implements the imperative shell in the FC/IS architecture pattern.
Responsibilities (Shell/I/O):
Does NOT contain:
Realtime service implementation (Shell layer). Orchestrates WebSocket messaging between core logic and adapters. Implements the imperative shell in the FC/IS architecture pattern. Responsibilities (Shell/I/O): - WebSocket connection lifecycle (open, close) - JWT authentication (delegates to user module) - Message routing (publishes envelopes onto the message bus) - Connection registry management - Pub/sub topic management - Logging and error handling Does NOT contain: - Business logic (lives in core.*) - Database operations (no persistence needed for WebSockets) - Message validation logic (lives in core.message)
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 |