WebSocket Adapter — adapters/websocket¶

See also: adapters/websocket on pkg.go.dev · Ports — Protocol-Agnostic Wiring · Metrics Observer · Error Handling

Runnable demo: examples/websocket-duplex — a DuplexPort served over a real loopback WebSocket: typed commands in, targeted typed replies out, app-supervised lifecycle, observer metrics (upgrade/frame counts + validation failures) injected once via app.Options.Observer.

adapters/websocket serves typed, codec-validated frame streams over persistent bidirectional connections — completing the HTTP story (IngestAdapter = client→pipeline, SSEAdapter = pipeline→client, WebSocket = both at once). Endpoints are declared once with ports.SocketPattern; every frame passes through the port's codec.

Delivery semantics (read first)¶

• A client that is offline or reconnecting loses frames; a late joiner sees nothing until the next frame. Pair with a LatestPort when "current state on connect" matters.
• Slow clients: each session has a buffered outbound queue (default 16). A full queue drops the frame for that session only — reported as a SocketError wrapping ErrFrameDropped — so one lagging browser tab never blocks the pipeline or other sessions.

Declare the endpoint — ports.SocketPattern¶

var Live = codex.Must(ports.NewDuplexPort[Command, Update]("live",
    commandCodec, updateCodec, ports.PortOptions{Buffer: 8}))
_, err := Live.PluginSocketPattern(ports.SocketPattern{Path: "/live/{room}"})
if err != nil {
    panic(err)
}

Port-type acceptance: SourcePort (inbound-only), SinkPort (broadcast-only — the WS sibling of SSE), DuplexPort (full duplex). Rejected on IOPort/LatestPort/ToolPort — per-message request/reply over a socket is an RPC discipline (ReqReplyPattern territory).

Adapters¶

All three share: - Hub (NewHub(buffer)) — the session registry, constructed in main and passed to the adapter. hub.SessionInfo(session) exposes upgrade-time path vars (which {room} a session joined); hub.Sessions() lists peers. - Upgrader (NewUpgrader(UpgraderOptions{...})) — the gorilla shim. Keepalive is adapter-owned: ping every 30s, pong wait 60s, 1 MiB read limit (override via PingInterval/ReadLimit; CheckOrigin for CORS). - Frame decode failures go to the port's Errors channel — the connection stays open; one bad frame does not disconnect a client.

Targeted replies (DuplexPort)¶

replies := stream.Map(ctx, Live.Inbound(ctx),
    func(f ports.Framed[Command]) (ports.Framed[Update], error) {
        return ports.Framed[Update]{
            Session: f.Session,           // reply to the sender only
            Payload: process(f.Payload),
        }, nil
    }, stream.MapOptions{Name: "ack"})
go Live.Feed(ctx, replies)                // zero Session = broadcast

One struct, one call with connection vars¶

WebSocket now supports declare-once merge fields for both directions. Add InOpts and/or OutOpts on SocketPattern; adapters then merge connection vars (path/query/header from upgrade request) into payload structs automatically on each frame:

patterns := []ports.Pattern{
    ports.SocketPattern{
        Path: "/live/{room}",
        Opts: []rest.RouteOpt{
            rest.PathParam{Name: "room"}.WithCodec(codex.String()),
        },
        InOpts: []ports.SocketInOpt{
            ports.NewRequiredSocketInParam("room", codex.String(),
                func(c Command) string { return c.Room },
                func(c *Command, v string) { c.Room = v }),
        },
        OutOpts: []ports.SocketOutOpt{
            ports.NewRequiredSocketOutParam("room", codex.String(),
                func(u Update) string { return u.Room },
                func(u *Update, v string) { u.Room = v }),
        },
    },
}

Inbound decode and outbound encode keep the same escape hatch: if no InOpts/OutOpts are declared, payloads stay untouched. examples/websocket-duplex shows both approaches side-by-side: auto-merged room via InOpts/OutOpts and manual hub.SessionInfo(session) lookup.

Client-side — dial adapters¶

The Dial* family connects OUT to an external WebSocket endpoint (another go-codex service, a partner API, a feed) and bridges it into the same ports:

NewDialer(DialerOptions{…}) is the gorilla dial shim (same keepalive story as the server side); the URL is baseURL + the handle's path template expanded with vars (declared PathParam codecs validate each value).

Reconnect semantics — no silent loss (by design):

• Auto-reconnect with exponential backoff (250ms → 30s cap, reset after a connection that delivered traffic).
• EVERY failed dial and EVERY drop emits a SocketError (Op "dial" / "read") on the port's Errors channel — consumers KNOW a gap happened and frames may have been missed.
• The session generation (c1, c2, …) advances per connection — a generation change in inbound frames is the visible reconnect marker.
• Outbound frames while the connection is down are DROPPED with ErrFrameDropped (consistent with the server slow-client policy) — including during initial connection establishment: pump or buffer upstream if the first frames matter.

AsyncAPI spec — ports.RegisterSocket¶

OpenAPI cannot express WebSocket frames; AsyncAPI can. RegisterSocket replays a port's SocketPattern against an events.Builder as a channel:

b := events.NewBuilder(events.Info{Title: "Live Ops Socket", Version: "1.0.0"})
b.AddServer("prod", events.Server{URL: "live.example.com", Protocol: "ws"})
_ = ports.RegisterSocket[Command, Update](b, Live)
doc, _ := b.AsyncAPISpec()

• Channel name = the socket path template; {var} placeholders become channel parameters.
• Subscribe operation = frames the application RECEIVES (In); Publish operation = frames it SENDS (Out). One-directional ports emit only their live direction (the struct{} side is skipped).
• Built on events.Builder.AddChannelItem — the escape hatch for channels whose two directions carry different payload types.

(Supplying PortOptions.RESTBuilder still incidentally documents the upgrade route in OpenAPI as a bare GET — harmless endpoint documentation, nothing about frames.)

chi variants¶

adapters/chi mirrors all three server adapters with chi-safe registration (chi.IngestSocketAdapter / chi.BroadcastSocketAdapter / chi.DuplexSocketAdapter): the swap handler is registered at CONSTRUCTOR time (chi's Mux cannot register routes while serving) and the real handler installs atomically at Activate — requests before that get 503. Behaviour is identical; the implementations delegate to the websocket package.

Narrow client interface¶

Adapters accept Upgrader/Socket — small interfaces — never a gorilla type. NewUpgrader adapts gorilla/websocket (already an indirect dependency via mcp-go — zero new modules); socket.go is the only file importing it. Unit tests run against hand-written fakes.

Errors¶

type SocketError struct {
    Path    string        // declared path template
    Session ports.Session // empty for upgrade failures
    Op      string        // "upgrade", "read", "write", "close"
    Err     error
}
var ErrFrameDropped = ...  // sentinel inside SocketError for slow-client drops

Error(), Unwrap(), slog.LogValuer — errors.Is(err, ErrFrameDropped) works through the chain. Frame validation failures wrap the codec chain (per-field observer reports, location "payload").

Observer¶

No new stats extension — the transport-agnostic hooks fit:

• RecordRequest("GET", path, 101|4xx, dur) once per upgrade attempt
• RecordSubscribe(path, success, dur) per inbound frame
• RecordPublish(path, success, dur) per outbound frame
• Path-var failures → RecordValidationError("path", …); frame validation → "payload". Nil observer resolves from ctx.

Error-path ergonomics — ErrorFrame¶

DuplexSocketAdapter and BroadcastSocketAdapter have no synchronous caller to respond to for upstream pipeline errors (received on the port's stream Errors channel) — websocket.ErrorFrame is the duplex/broadcast-socket analogue of events.ErrorChannel, adapted to a persistent multi-session transport: instead of publishing to a declared error topic, the mapped payload is broadcast to every connected session (there is no dedicated error-output channel on a socket — broadcast IS the notification path). Both adapters share the exact same ErrorFrame/ ErrorFrameRule declarative surface — DuplexSocketAdapterOptions.ErrorFrames and BroadcastSocketAdapterOptions.ErrorFrames are both []ErrorFrameRule.

ErrorFrame declares its own codec-backed payload type — independent of the socket's happy-path Out frame type — the same "one-struct-one-call" guarantee rest.ErrorPattern/events.ErrorChannel/reqreply.ErrorPattern/ mcp.ErrorPattern all provide: the payload is validated via its declared codec (all Refine constraints run) before being broadcast, exactly like the happy path.

type ValidationError struct{ Reason string }
func (e ValidationError) Error() string { return "validation: " + e.Reason }

type ErrorPayload struct {
    Code    string
    Message string
}

_ = domain.Live.Bind(ctx, websocket.DuplexSocketAdapter(mux, hub, upgrader, handle,
    websocket.DuplexSocketAdapterOptions{
        ErrorFrames: []websocket.ErrorFrameRule{
            websocket.ErrorFrame[ValidationError, ErrorPayload](errorPayloadCodec,
                func(e ValidationError) (ErrorPayload, error) {
                    return ErrorPayload{Code: "validation", Message: e.Reason}, nil
                },
            ),
        },
    }))

• Direct mode (no map function): E must itself be assignable to the declared payload type.
• Mapped mode (map function provided): converts E into the declared payload type.
• Matching: type-only via errors.As; the first declared ErrorFrame (in declaration order in the ErrorFrames slice) whose type matches wins.
• ErrorFrameRule is NOT parameterized by Out — its payload is independently codec-validated and pre-encoded at declaration time, so DuplexSocketAdapterOptions.ErrorFrames is a plain []websocket.ErrorFrameRule (no type erasure, no runtime type assertion needed at Activate time).

Action model — respond / handle / log¶

A matched rule executes exactly one action, never an implicit chain:

websocket.ErrorFrame[ValidationError, ErrorPayload](errorPayloadCodec, mapFn).
    WithAction(events.ErrorHandle).
    WithHandle(func(err error) { log.Printf("duplex error: %v", err) })

Unmatched errors fall through to the existing default: forwarded unchanged to the port's Errors channel (surfaced via port.Inbound(ctx).Errors). A matched rule whose mapped payload fails its declared codec's constraints is ALSO forwarded to the port's Errors channel (never silently broadcast) — the same terminal-error handling rest.ErrorPattern/events.ErrorChannel use for a mapping/encoding failure.

Not MQTT-over-WebSocket¶

MQTT clients connecting via ws:// tunnel MQTT frames inside WebSocket — a transport option of the MQTT client, already supported by passing a ws:// broker URL to paho (adapters/mqtt/adapters/mqtt5). This adapter is go-codex itself as the WS server speaking its own typed frames; it is not an MQTT broker.

Scope¶

Server-side + client-side + chi + AsyncAPI spec are shipped. Still deferred (awaiting use cases): a ConnectionObserver stats extension and dynamic subprotocol negotiation — recorded in the deferred roadmap.