Zig 0.17.0

Hub, Sse, Ws

On this page:

  • spider.Hub: The connected SSE or WebSocket clients, and what sends to them: emit, emitTo a channel, broadcast.
  • spider.Ws: One WebSocket connection, as its handler sees it: next() reads a message, send writes one.
  • spider.Sse: One Server-Sent Events stream, as its handler sees it.

spider.Hub

zig
pub const Hub = struct

The open connections of the app and the channels they joined. The server owns its hubs: one for every SSE route together (c.sseHub(), and the hub a spider.every job receives), and one for each WebSocket route (what Ws.broadcast sends through, and what a Server.wsInterval callback receives).

zig
hub.emitTo(channel, "package_updated", .{ .id = pkg.id });

Who receives what:

  • emit: every SSE connection.
  • emitTo, emitHtmlTo: the SSE connections on one channel; recorded for replay.
  • notifyUser: the SSE connections on the channel user:<id>.
  • broadcast, broadcastFmt: every connection, SSE and WebSocket.
  • broadcastToChannel, broadcastToChannelFmt: the connections on one channel, SSE and WebSocket.

None of them returns an error: a connection whose write fails is taken out of the hub and its socket shut down, and the others still get the message. The fields belong to the server; use the methods.

Fields

  • allocator: std.mem.Allocator
  • io: std.Io
  • mutex: std.Io.Mutex
  • connections: std.ArrayListUnmanaged(*ConnectionSlot) = .empty
  • channel_history: std.StringHashMapUnmanaged(ChannelHistory) = .empty
  • next_event_id: u64 = 1: Hub-wide event id counter (guarded by mutex). Global rather than per channel so one connection subscribed to several channels still gets a single, unambiguous Last-Event-ID to resume from.
  • heartbeat_thread: ?std.Thread = null
  • heartbeat_running: std.atomic.Value(bool) = std.atomic.Value(bool).init(false)
  • sweep_thread: ?std.Thread = null
  • sweep_running: std.atomic.Value(bool) = std.atomic.Value(bool).init(false)
  • write_timeout_ms: u32 = 0: Longest one write to a client may block (its socket buffer full: it stopped reading) before the connection is dropped. 0 = unbounded. Server.listen() sets it from Config.stream_write_timeout_ms; only connections registered with a watch entry are bounded.

Hub.default_heartbeat_ms

zig
pub const default_heartbeat_ms: u64 = 30_000

The interval of Server.sseHeartbeat(null): 30 seconds.

Hub.default_sweep_ms

zig
pub const default_sweep_ms: u64 = 60_000

The interval of Server.sseSweep(null): 60 seconds.

Hub.default_retry_ms

zig
pub const default_retry_ms: u64 = 3_000

The reconnection delay every SSE stream announces when it opens: 3 seconds. Sse.setRetry changes it for one connection.

Hub.HistoryEntry

zig
pub const HistoryEntry = struct

One event recorded for replay, as historySince returns it. id is the hub-wide event id sent to the clients; data is the text that went out (JSON from emitTo, HTML from emitHtmlTo).

Fields

  • id: u64
  • event: []const u8
  • data: []const u8
  • timestamp: std.Io.Timestamp

Hub.init

zig
pub fn init(allocator: std.mem.Allocator, io: std.Io) Hub

A hub with no connections. Its writes go through io, which must be the Io of the server that owns the sockets. The server creates the hubs of an app; a test may create its own.

Hub.deinit

zig
pub fn deinit(self: *Hub) void

Stops the heartbeat and sweep threads, closes the stream of every connection still registered and frees what the hub allocated.

Hub.channelCount

zig
pub fn channelCount(self: *Hub, channel: []const u8) usize

Number of live connections subscribed to channel.

Hub.count

zig
pub fn count(self: *Hub) usize

Number of connections registered in this hub, whatever their channel.

Hub.broadcast

zig
pub fn broadcast(self: *Hub, message: []const u8) void

Sends message to every connection of the hub, whatever its channel. A WebSocket client receives it as a text frame; an SSE client as an event named message with the text as data. Not recorded for replay.

Hub.notifyUser

zig
pub fn notifyUser(self: *Hub, user_id: u64, event: []const u8, data: anytype) void

Sends to one user: emitTo on the channel user:<user_id>, the one a stream joins with Sse.joinUser(user_id). Only SSE connections receive it.

Hub.emit

zig
pub fn emit(self: *Hub, event: []const u8, data: anytype) void

Sends the event event, with data as JSON, to every SSE connection of the hub, whatever its channel. WebSocket connections do not receive it. It carries no event id and is not recorded for replay. Nothing is sent when data cannot be serialized.

zig
c.sseHub().emit("post_created", .{ .id = id });

Hub.emitTo

zig
pub fn emitTo(self: *Hub, channel: []const u8, event: []const u8, data: anytype) void

Sends the event event, with data as JSON, to the SSE connections subscribed to channel (Sse.join, Sse.subscribe). WebSocket connections do not receive it. The event gets the next hub-wide id and is recorded in the channel's history (the last 50 events, at most 5 minutes old), so a client that reconnects gets what it missed (Sse.joinWithReplay). It is recorded even when nobody is on the channel. Nothing is sent when data cannot be serialized.

zig
hub.emitTo(channel, "notification_created", .{});

Hub.emitHtmlTo

zig
pub fn emitHtmlTo(self: *Hub, channel: []const u8, event: []const u8, html: []const u8) void

emitTo() for HTML: html goes out as is (not JSON-encoded), one data: line per line, so htmx's SSE extension can swap it in. With htmx 2 the page picks a named event (<div hx-ext="sse" sse-connect="/events" sse-swap="badge">). htmx 4 swaps only a message WITHOUT a name: pass "" as event (<div hx-sse:connect="/events">); a named event becomes a DOM event there (hx-trigger="badge from:body"). Recorded in the channel's history like emitTo(). Render it once for everyone on the channel: it must not carry per-user data.

Hub.historySince

zig
pub fn historySince(self: *Hub, allocator: std.mem.Allocator, channel: []const u8, last_id: u64) ![]HistoryEntry

Copies (caller-owned via allocator) every history entry recorded for channel with id > last_id, in order — what Sse.joinWithReplay() sends on reconnect. Empty slice if the channel has no history or last_id is already caught up.

Hub.broadcastFmt

zig
pub fn broadcastFmt(self: *Hub, comptime fmt: []const u8, args: anytype) void

broadcast with a formatted message: every connection of the hub.

Hub.broadcastToChannelFmt

zig
pub fn broadcastToChannelFmt(self: *Hub, channel: []const u8, comptime fmt: []const u8, args: anytype) void

broadcastToChannel with a formatted message: the connections on channel.

Hub.broadcastToChannel

zig
pub fn broadcastToChannel(self: *Hub, channel: []const u8, message: []const u8) void

Sends message to the connections on channel, SSE and WebSocket alike: a text frame for a WebSocket client, an event named message for an SSE client. Not recorded for replay.

spider.Ws

zig
pub const Ws = struct

One open WebSocket connection. The handler of a Server.ws route receives it after the handshake and reads messages until the client leaves; when the handler returns, the connection is closed.

zig
fn chat(ws: *spider.Ws) !void {
    while (try ws.next()) |message| {
        ws.broadcast(message.data);
    }
}

Each WebSocket route has a hub of its own: broadcast and broadcastTo reach the connections of the same route, the sender included.

Fields

  • channel: []const u8 = "": The channel given to join. Empty before that.
  • params: std.StringHashMapUnmanaged([]const u8) = .{}: The request's params at the time of the handshake: route segments and what the middlewares set. See param.
  • arena: std.mem.Allocator: Lives as long as the connection; received messages are allocated here.
  • io: std.Io: The server's Io.

Ws.next

zig
pub fn next(self: *Ws) !?Message

Waits for the next text or binary message. Null when the connection is over: the client closed it, a read failed, or a frame was not accepted (a payload over 16 MiB, an unknown opcode, a continuation frame). Pings are answered and skipped. The payload is allocated in arena and is not freed before the connection ends.

A message the client split into several frames is not put back together: its first frame comes as a message of its own and the continuation frame that follows ends the connection.

Ws.joinUser

zig
pub fn joinUser(self: *Ws, user_id: u64) !void

join on the channel user:<user_id>: broadcastTo("user:7", ...) then reaches the connections of user 7 on this route. The name is allocated in arena. Hub.notifyUser does not reach it: that one sends to SSE connections only.

Ws.join

zig
pub fn join(self: *Ws, channel: []const u8) !void

Puts this connection on channel, leaving the one it was on: broadcastTo(channel, ...) then reaches it. The name is copied: the caller's buffer may be reused.

Ws.send

zig
pub fn send(self: *Ws, text: []const u8) !void

Sends a text message to this connection only. It takes the same lock as a broadcast from another connection, so the two frames never mix. Fails with the write error when the client is gone.

Ws.broadcast

zig
pub fn broadcast(self: *Ws, text: []const u8) void

Sends a text message to every connection of this route, this one included, whatever their channel. A connection whose write fails is dropped.

Ws.broadcastFmt

zig
pub fn broadcastFmt(self: *Ws, comptime fmt: []const u8, args: anytype) void

broadcast with a formatted message.

Ws.broadcastTo

zig
pub fn broadcastTo(self: *Ws, channel: []const u8, text: []const u8) void

Sends a text message to the connections of this route that joined channel.

Ws.broadcastToFmt

zig
pub fn broadcastToFmt(self: *Ws, channel: []const u8, comptime fmt: []const u8, args: anytype) void

broadcastTo with a formatted message.

Ws.param

zig
pub fn param(self: *Ws, key: []const u8) ?[]const u8

A request param by name: a route segment or a value a middleware set. Null when there is none.

spider.Sse

zig
pub const Sse = struct

One open SSE stream. An SSE handler (Server.sse, Group.sse) receives it after the response headers went out: it picks the channels, then calls wait() so the stream stays open. When the handler returns, the connection is closed.

zig
fn events(sse: *spider.Sse) !void {
    const room = sse.param("room") orelse "lobby";
    const channel = try std.fmt.allocPrint(sse.arena, "room:{s}", .{room});
    try sse.joinWithReplay(channel);
    sse.wait();
}

Events for the channel come from anywhere in the app: c.sseHub().emitTo("room:lobby", "message", .{ .text = "hi" }).

An error returned by the handler ends the stream. Nothing can be answered any more, so the server logs it at warn with the request id and the path; an error that only says the client went away (a failed write, a closed connection) is not logged.

A page of another site can read the stream only when the app lists its origin in Config.sse_allowed_origins: the response then carries Access-Control-Allow-Origin for that origin (* when the list has *).

Fields

  • channel: []const u8 = "": The channel given to join, or the first one given to subscribe. Empty before that.
  • params: std.StringHashMapUnmanaged([]const u8) = .{}: The request's params at the time the stream opened: route segments and what the middlewares set (the _auth_* values). See param.
  • headers: std.StringHashMapUnmanaged([]const u8) = .{}: The request headers. See header for a lookup that ignores case.
  • arena: std.mem.Allocator: Lives as long as the stream. Channel names given to join belong here.
  • io: std.Io: The server's Io.

Sse.send

zig
pub fn send(self: *Sse, event: []const u8, data: anytype) !void

Sends the event event, with data as JSON, to this connection only. It carries no event id and is not recorded for replay. Fails with the write error when the client is gone.

Sse.sendHtml

zig
pub fn sendHtml(self: *Sse, event: []const u8, html: []const u8) !void

send() for HTML: html goes out as is (not JSON-encoded), one data: line per line, for htmx's SSE extension: a named event for htmx 2 (sse-swap), "" as event for htmx 4, which swaps only a message without a name.

Sse.setRetry

zig
pub fn setRetry(self: *Sse, ms: u64) !void

Sends "retry: {ms}\n\n" — tells the client's EventSource how long to wait before reconnecting after a drop. Every stream already announces Hub.default_retry_ms (3 seconds) when it opens; this replaces it. Safe to call more than once (e.g. to change it mid-connection); the client applies whatever the most recently received retry: field said.

Sse.join

zig
pub fn join(self: *Sse, channel: []const u8) !void

Makes channel the only channel of this connection: what Hub.emitTo(channel, ...) sends then reaches it, and the channels it had are dropped. The name is copied: the caller's buffer may be reused.

Sse.open

zig
pub fn open(self: *Sse) void

Sends the response head if nothing was sent yet. The stream calls it by itself once the handler listens (join, subscribe, wait) and before the first event; a handler that does long work before any of those may call it to answer the client sooner.

Sse.joinWithReplay

zig
pub fn joinWithReplay(self: *Sse, channel: []const u8) !void

join() plus replay: if the client reconnected with a Last-Event-ID header, immediately re-sends every event this channel recorded after that id (bounded history: the last 50 events of the channel, at most 5 minutes old), so a dropped connection doesn't silently miss events between reconnects. No-op replay (just a plain join) if there's no Last-Event-ID header or nothing newer is in history.

Sse.subscribe

zig
pub fn subscribe(self: *Sse, channel: []const u8) !void

Adds channel to this connection's subscriptions, keeping the ones it already has (unlike join(), which replaces them). Lets one EventSource carry several channels — browsers cap HTTP/1.1 connections per origin at 6, so one EventSource per channel per tab runs out after a few tabs. The name is copied; the channel field holds the first one given.

Sse.subscribeWithReplay

zig
pub fn subscribeWithReplay(self: *Sse, channels: []const []const u8) !void

subscribe() to every channel, then replay what the client missed across all of them (Last-Event-ID is hub-wide, so one id covers every channel), in the original emit order.

Sse.joinUser

zig
pub fn joinUser(self: *Sse, user_id: u64) !void

join on the channel user:<user_id>, the one Hub.notifyUser(user_id, ...) sends to. Like join, it replaces the channels the connection had.

Sse.param

zig
pub fn param(self: *Sse, key: []const u8) ?[]const u8

A request param by name: a route segment (:room) or a value a middleware set, such as _auth_sub. Null when there is none.

Sse.header

zig
pub fn header(self: *Sse, name: []const u8) ?[]const u8

A request header by name, ignoring case, as Ctx.header does. Null when the request did not send it.

zig
pub fn cookie(self: *Sse, name: []const u8) ?[]const u8

A cookie of the request by name, read from its Cookie header as Ctx.cookie does. The value comes as it was sent, not decoded. Null when there is no such cookie.

Sse.lastEventId

zig
pub fn lastEventId(self: *Sse) ?u64

The Last-Event-ID header the browser sends automatically on reconnect, once it has seen at least one "id:" line — null if absent or not a valid integer.

Sse.wait

zig
pub fn wait(self: *Sse) void

Blocks until the client closes the connection or the hub drops it (a failed write). Call it last: it is what keeps the handler, and so the stream, alive.

What the declarations above take or give back, without a name of its own in spider.

Message

zig
pub const Message = struct

One message received from the client, as Ws.next returns it.

Fields

  • data: []const u8: The payload, allocated in the connection's arena.
  • type: enum { text, binary }: Whether the client sent a text or a binary frame.