Skip to content

sse

Directory actions

More options

Directory actions

More options

Latest commit

 

History

History

sse

Folders and files

NameName
Last commit message
Last commit date

parent directory

..
src
src
 
 
README.md
README.md
 
 
TECHNICAL.md
TECHNICAL.md
 
 
package.json
package.json
 
 
tsconfig.json
tsconfig.json
 
 
tsdown.config.ts
tsdown.config.ts
 
 
verify.config.ts
verify.config.ts
 
 

README.md

@kyneta/sse-network-adapter

SSE (Server-Sent Events) adapter for @kyneta/exchange — client, server, and Express integration. Provides real-time sync using SSE for server→client messages and HTTP POST for client→server messages, both encoded with the @kyneta/wire text protocol (JSON codec + text framing + text fragmentation).

Subpath Exports

Export Entry point Environment
@kyneta/sse-network-adapter/client ./dist/client.js Browser, Bun, Node.js
@kyneta/sse-network-adapter/server ./dist/server.js Bun, Node.js
@kyneta/sse-network-adapter/express ./dist/express.js Node.js (Express)

Server Setup

Express (recommended)

Use createSseExpressRouter for zero-boilerplate integration with Express:

import { Exchange } from "@kyneta/exchange"
import { SseServerAdapter } from "@kyneta/sse-network-adapter/server"
import { createSseExpressRouter } from "@kyneta/sse-network-adapter/express"
import express from "express"

const app = express()

const serverAdapter = new SseServerAdapter()

const exchange = new Exchange({
  identity: { peerId: "server", name: "server", type: "service" },
  adapters: [() => serverAdapter],
})

app.use("/sse", createSseExpressRouter(serverAdapter, {
  syncPath: "/sync",
  eventsPath: "/events",
  heartbeatInterval: 30000,
}))

app.listen(3000)

Hono

For Hono or other frameworks, use parseTextPostBody and SseServerAdapter.registerConnection directly:

import { SseServerAdapter } from "@kyneta/sse-network-adapter/server"
import { parseTextPostBody } from "@kyneta/sse-network-adapter/express"
import { Hono } from "hono"
import { streamSSE } from "hono/streaming"

const sseAdapter = new SseServerAdapter()

const app = new Hono()

app.get("/sse/events", async (c) => {
  const peerId = c.req.query("peerId")
  if (!peerId) return c.json({ error: "peerId required" }, 400)

  return streamSSE(c, async (stream) => {
    const connection = sseAdapter.registerConnection(peerId)

    // sendFn receives pre-encoded text frame strings
    connection.setSendFunction((textFrame) => {
      stream.writeSSE({ data: textFrame })
    })

    stream.onAbort(() => {
      sseAdapter.unregisterConnection(peerId)
    })

    await new Promise(() => {}) // keep alive
  })
})

app.post("/sse/sync", async (c) => {
  const peerId = c.req.header("x-peer-id")
  if (!peerId) return c.json({ error: "x-peer-id required" }, 400)

  const connection = sseAdapter.getConnection(peerId)
  if (!connection) return c.json({ error: "Not connected" }, 404)

  const body = await c.req.text()
  const result = parseTextPostBody(connection.reassembler, body)

  if (result.type === "messages") {
    for (const msg of result.messages) {
      connection.receive(msg)
    }
  }

  return c.json(result.response.body, result.response.status)
})

Client Setup

Browser

Use createSseClient for browser-to-server connections:

import { Exchange } from "@kyneta/exchange"
import { createSseClient } from "@kyneta/sse-network-adapter/client"

const exchange = new Exchange({
  identity: { peerId: "browser-client", name: "Alice", type: "user" },
  adapters: [createSseClient({
    postUrl: "/sse/sync",
    eventSourceUrl: (peerId) => `/sse/events?peerId=${peerId}`,
    reconnect: { enabled: true },
  })],
})

Connection Lifecycle

The client connection lifecycle is a Program<SseClientMsg, SseClientState, SseClientEffect> from @kyneta/machine — a pure Mealy machine with data effects. Unlike the WebSocket adapter, SSE has no separate "ready" signal — the connection is usable as soon as EventSource.onopen fires, giving a 4-state lifecycle.

The SseClientTransport class is a thin imperative shell that interprets data effects as I/O (FC/IS design). The program itself is deterministically testable — every state × event combination is covered by pure data tests (no EventSource, no timing, never flaky).

disconnected → connecting → connected
                   ↓            ↓
              reconnecting ← ─ ─┘
                   ↓
              connecting (retry)
                   ↓
              disconnected (max retries)
State Description
disconnected No active connection. Optional reason field describes why.
connecting EventSource being created. Tracks attempt number.
connected EventSource open — protocol messages can flow.
reconnecting Connection lost, scheduling next attempt. Tracks attempt and nextAttemptMs.

Connection Handshake

  1. Client creates EventSource, transitions to connecting
  2. EventSource.onopen fires, transitions to connected
  3. Client creates its channel, calls establishChannel()
  4. Synchronizer exchanges establish-request / establish-response

EventSource Error Handling

On EventSource.onerror, the program produces a close-event-source effect and transitions to reconnecting (or disconnected if retries are exhausted). The imperative shell closes the EventSource immediately, preventing the browser's built-in EventSource reconnection from running. This gives the program full control over backoff timing and attempt counting — reconnect delays are computed purely as data effects (start-reconnect-timer).

POST Retry

Client→server messages are sent via HTTP POST. POST failures (network errors, non-2xx responses) are retried with exponential backoff at the transport level — this is an imperative-shell concern, not part of the connection lifecycle program.

Observing State

The public observation API is powered by createObservableProgram from @kyneta/machine:

import { createSseClient } from "@kyneta/sse-network-adapter/client"

const adapter = createSseClient({
  postUrl: "/sse/sync",
  eventSourceUrl: (peerId) => `/sse/events?peerId=${peerId}`,
  lifecycle: {
    onStateChange: ({ from, to }) => console.log(`${from.status} → ${to.status}`),
    onDisconnect: (reason) => console.log("disconnected:", reason.type),
    onReconnecting: (attempt, nextMs) => console.log(`retry #${attempt} in ${nextMs}ms`),
    onReconnected: () => console.log("reconnected"),
  },
})

// Or subscribe to transitions programmatically
const unsub = adapter.subscribeToTransitions(({ from, to }) => {
  console.log(`${from.status} → ${to.status}`)
})

await adapter.waitForStatus("connected", { timeoutMs: 5000 })

Wire Format

Asymmetric wire encoding — text frames downstream (SSE), binary CBOR upstream (POST):

Direction Transport Wire format
Client → Server HTTP POST (application/octet-stream) Binary CBOR frame
Server → Client SSE data: event Text frame (["2c", seq, <payload>])

Text Frames

Every message is wrapped in a text frame — a JSON array with a 2-character prefix followed by the per-direction seq:

Complete frame:  ["2c", 42, {"type":"present","docs":[{"docId":"doc-1"}]}]
Fragment frame:  ["2f", 42, 0, 3, 1500, "{\"type\":\"offer\"..."]

The "2c" prefix means "version 2, complete, no hash". seq is the per-direction message id (it groups a message's fragments). Fragments use "2f" and carry seq, index, total, totalSize, and a JSON substring chunk.

Why Text Instead of Binary?

The old @loro-extended/adapter-sse used an asymmetric format: binary CBOR for POST, ad-hoc JSON for SSE. The new adapter uses uniform text encoding because:

  • Single code path for encode/decode on both client and server
  • Human-readable POST bodies and SSE events for debugging
  • No need for express.raw() with application/octet-stream
  • Text fragmentation works in both directions

The ~33% bandwidth overhead of base64 for binary payloads (vs. native CBOR byte strings) is acceptable for SSE's use case (chat, presence, signaling). For bandwidth-sensitive workloads, use the WebSocket adapter.

Configuration

Client Options

Option Default Description
postUrl — POST URL. String or (peerId) => string function.
eventSourceUrl — SSE URL. String or (peerId) => string function.
reconnect.enabled true Enable automatic reconnection.
reconnect.maxAttempts 10 Maximum reconnection attempts.
reconnect.baseDelay 1000 Base delay in ms for exponential backoff.
reconnect.maxDelay 30000 Maximum delay cap in ms.
postRetry.maxAttempts 3 Maximum POST retry attempts.
postRetry.baseDelay 1000 Base delay in ms for POST retry backoff.
postRetry.maxDelay 10000 Maximum POST retry delay in ms.
fragmentThreshold 60000 Character threshold for text fragmentation.

Server Options

Option Default Description
fragmentThreshold 60000 Character threshold for text fragmentation.

Express Router Options

Option Default Description
syncPath "/sync" Path for POST endpoint.
eventsPath "/events" Path for SSE endpoint.
heartbeatInterval 30000 Heartbeat interval in ms.
getPeerIdFromSyncRequest reads x-peer-id header Custom peerId extraction for POST.
getPeerIdFromEventsRequest reads peerId query param Custom peerId extraction for SSE.

Heartbeat

The Express router sends SSE comment heartbeats (: heartbeat\n\n) at the configured interval. SSE comments are silently ignored by EventSource clients. This keeps connections alive through proxies and load balancers that terminate idle connections.

Custom Framework Integration

The parseTextPostBody function provides a framework-agnostic handler for POST requests:

import { parseTextPostBody } from "@kyneta/sse-network-adapter/express"

// In your framework's request handler
const result = parseTextPostBody(connection.reassembler, bodyAsString)

if (result.type === "messages") {
  for (const msg of result.messages) {
    connection.receive(msg)
  }
}

// Send response
response.status(result.response.status).json(result.response.body)

Response Types

Result Type HTTP Status Meaning
messages 200 Message(s) decoded successfully
pending 202 Fragment received, waiting for more
error 400 Decode or reassembly error

The sendFn Pattern

SseConnection.send() handles encoding and fragmentation internally. The injected sendFn receives pre-encoded text frame strings — the framework integration just wraps them in transport syntax:

// Express
connection.setSendFunction((textFrame) => {
  res.write(`data: ${textFrame}\n\n`)
})

// Hono
connection.setSendFunction((textFrame) => {
  stream.writeSSE({ data: textFrame })
})

Architecture

┌──────────────────────────────────────────────────────────┐
│                        Client                            │
│  ┌──────────────────┐        ┌───────────────────┐       │
│  │ SseClientAdapter │        │ EventSource       │       │
│  │ (text POST)      │───────▶│ (text receive)    │       │
│  └──────────────────┘        └───────────────────┘       │
└──────────────────────────────────────────────────────────┘
         │                             ▲
         │ HTTP POST                   │ SSE
         │ (text wire frame)           │ (text wire frame)
         ▼                             │
┌──────────────────────────────────────────────────────────┐
│                        Server                            │
│  ┌──────────────────┐        ┌───────────────────┐       │
│  │ Express Router   │        │ SSE Writer        │       │
│  │ (parseTextPost)  │───────▶│ (sendFn)          │       │
│  └──────────────────┘        └───────────────────┘       │
│           │                             ▲                │
│           ▼                             │                │
│  ┌───────────────────────────────────────────────────┐   │
│  │          SseServerAdapter                         │   │
│  │  ┌────────────────────────────────────────────┐   │   │
│  │  │ SseConnection (per peer)                   │   │   │
│  │  │ - Pipeline (asymmetric encode/decode)    │   │   │
│  │  │ - Pipeline handles send/receive framing  │   │   │
│  │  │ - Channel reference                        │   │   │
│  │  └────────────────────────────────────────────┘   │   │
│  └───────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────┘

Peer Dependencies

{
  "peerDependencies": {
    "@kyneta/exchange": ">=0.0.1",
    "@kyneta/wire": ">=0.0.1"
  }
}

Express is an optional peer dependency — only needed if using @kyneta/sse-network-adapter/express.

License

MIT