From c1f114e162349772c94160a214ab03ebd8f89e15 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20Pacho=C5=82ek?= Date: Mon, 28 Sep 2026 00:57:55 +0200 Subject: [PATCH] Add example/nuts-client.js, a replay-aware EventSource helper (#146) A dependency-free ES module that applications copy in. It keeps the cursor of every event that carries one (the connected handshake, messages, and the reset that sets it to 0) in sessionStorage or a storage of your choice, resumes from it after a reload, and opens a new EventSource from it when one gives up for good. Listeners go by event name, which pairs with event_type; raw passes data through for payload_format raw; the EventSource option takes an implementation that can send headers. The demo pages use it. make test-js runs its tests with Node.js's built-in test runner, in CI too, and checks that example_docker keeps the same copy. The README documents it, and the README and website snippets keep their cursor in sessionStorage, one per tab. The source-build demo's start.sh names the page's address, /example/, and /example redirects there so the page's relative import resolves; the image demo pulls the latest image before starting. Closes #146 Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 8 + AGENTS.md | 6 +- CHANGELOG.md | 11 ++ CONTRIBUTING.md | 1 + Caddyfile.test | 4 + Makefile | 10 +- README.md | 95 +++++++++++- example/index.html | 59 +++----- example/nuts-client.js | 177 ++++++++++++++++++++++ example/nuts-client.test.mjs | 242 ++++++++++++++++++++++++++++++ example/start.sh | 2 +- example_docker/Caddyfile | 3 +- example_docker/docker-compose.yml | 1 + example_docker/index.html | 59 +++----- example_docker/nuts-client.js | 177 ++++++++++++++++++++++ example_docker/start.sh | 3 + website/docs/usage.md | 49 +++++- 17 files changed, 813 insertions(+), 94 deletions(-) create mode 100644 example/nuts-client.js create mode 100644 example/nuts-client.test.mjs create mode 100644 example_docker/nuts-client.js diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 145ef73..e95fb44 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,6 +61,14 @@ jobs: - name: Run go vet run: go vet ./... + - name: Set up Node + uses: actions/setup-node@v7 + with: + node-version: '22' + + - name: Run example client tests + run: make test-js + - name: Run govulncheck # Pinned to match the rest of CI (golangci-lint, trivy, action SHAs). # Bump in lockstep with Renovate/Dependabot updates. diff --git a/AGENTS.md b/AGENTS.md index 5c56924..4168447 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -104,7 +104,7 @@ Operational and deployment assets: | [Caddyfile](Caddyfile), [Caddyfile.test](Caddyfile.test) | Reference configurations; `Caddyfile.test` powers the functional stack. | | [Dockerfile](Dockerfile), [Dockerfile.test](Dockerfile.test) | Production and test images (non-root, uid 10001). | | [docker-compose.yml](docker-compose.yml) | Functional-test stack (NATS + NUTS built from source). | -| [example/](example/), [example_docker/](example_docker/) | Interactive demos (source build vs. published image). | +| [example/](example/), [example_docker/](example_docker/) | Interactive demos (source build vs. published image). Both pages use the replay-aware client [example/nuts-client.js](example/nuts-client.js), of which `example_docker/` keeps a copy. | | [ops/prometheus-alerts.yml](ops/prometheus-alerts.yml), [ops/grafana-dashboard.json](ops/grafana-dashboard.json) | Reference alerts and dashboard. | | [scripts/](scripts/) | Helper scripts (e.g. `setup-dev.sh`). | | [docs/](docs/) | All in-depth documentation; see the index in [README.md](README.md#further-documentation). | @@ -129,6 +129,7 @@ go build ./cmd/caddy # equivalent | `make test-unit` | Unit/integration tests with embedded NATS (no Docker). | | `make test-performance` | `TestPerformance_*` plus the named hot-path benchmarks. | | `make test-functional` | Godog BDD scenarios against the Docker Compose stack. | +| `make test-js` | Node.js's built-in test runner on the example JavaScript client, [example/nuts-client.js](example/nuts-client.js) (Node.js 20+, no packages); also checks that `example_docker/` has the same copy. | | `make test-functional-stress FUNCTIONAL_TEST_STRESS_COUNT=N` | Repeats the functional suite N times to catch flakes. | | `make test-functional-matrix` | Runs functional tests against `nats:2.10-alpine` (the supported floor), `nats:2.12-alpine`, `nats:2.14-alpine` (2.14.7+, first release with the multi-filter purge fix) and `nats:2.15-alpine` (matches the embedded `nats-server/v2` major.minor pinned in `go.mod`). | | `make test` | `test-unit` + `test-functional`. | @@ -271,7 +272,8 @@ CI is GitHub Actions ([.github/workflows/ci.yml](.github/workflows/ci.yml), PRs run, in this order: 1. `gofmt`, `go mod tidy` diff check, golangci-lint, unit tests with coverage, - focused race tests, `go vet`, `govulncheck`. + focused race tests, `go vet`, the example JavaScript client's tests, + `govulncheck`. 2. Functional test matrix (`nats:2.10-alpine`, `nats:2.12-alpine`, `nats:2.14-alpine`, `nats:2.15-alpine`) and a 3× functional stress pass. 3. Coverage upload to Codecov. diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f44242..76a30a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- `example/nuts-client.js`, a dependency-free ES module wrapping + EventSource for replay-aware pages (#146). It keeps the cursor of every + event that carries one, the `reset` included, in `sessionStorage` or a + storage of your choice, resumes from it after a reload, and opens a new + EventSource when one gives up for good. The demo pages use it, and + `make test-js` runs its tests with Node.js's built-in test runner. The + README and website snippets now keep their cursor in `sessionStorage`, + one per tab: tabs sharing `localStorage` overwrote each other's. The + source-build demo's `start.sh` now names the page's address, + `/example/` (`/example` redirects there), and the image demo pulls the + latest image before it starts. - `sse_retry ` sends `retry:` with every stream's handshake, setting EventSource's reconnection delay after ordinary disconnects, and `transient_retry ` sets the average of the jittered delay NUTS diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 3d26b84..a947215 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -173,6 +173,7 @@ subset that matches the change when a full local pass would be unreasonable. - [ ] `make test-functional-matrix` passes for NATS compatibility changes. - [ ] `make test-performance` passes for formatter, buffering, replay, or concurrency changes. +- [ ] `make test-js` passes for changes to `example/nuts-client.js`. - [ ] Production Docker image changes build and pass `caddy adapt --config /app/Caddyfile` inside the image. - [ ] Release tooling changes pass `make release-check` and a GoReleaser diff --git a/Caddyfile.test b/Caddyfile.test index a8b4e25..358d333 100644 --- a/Caddyfile.test +++ b/Caddyfile.test @@ -111,6 +111,10 @@ } } + # The demo page (example/) imports ./nuts-client.js, which resolves + # next to the page only under /example/. + redir /example /example/ + handle /example* { root * /srv/example uri strip_prefix /example diff --git a/Makefile b/Makefile index 9103cdb..0f1abff 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: all build test test-unit test-performance test-functional test-functional-dev test-functional-stress test-functional-matrix release-check docker-up wait-functional-stack docker-down docker-logs mutate mutate-pkg mutate-tools clean website-build website-serve website-clean +.PHONY: all build test test-unit test-performance test-functional test-functional-dev test-functional-stress test-functional-matrix test-js release-check docker-up wait-functional-stack docker-down docker-logs mutate mutate-pkg mutate-tools clean website-build website-serve website-clean DOCKER_COMPOSE := $(shell if docker compose version >/dev/null 2>&1; then echo docker compose; elif docker-compose version >/dev/null 2>&1; then echo docker-compose; fi) FUNCTIONAL_TEST_STRESS_COUNT ?= 3 @@ -32,6 +32,13 @@ test-performance: go test -run '^TestPerformance_' -timeout 180s . go test -run '^$$' -bench 'Benchmark(FormatMessageEvent|FormatMessageEventLarge|WriteJSONPayload|IsValidTopic|StreamFeed|SharedFanOut)' -benchmem . +# Run the example JavaScript client's tests with Node.js's built-in runner +# (Node.js 20 or later, no packages), and check that example_docker carries +# the same copy of the client. +test-js: + node --test example/nuts-client.test.mjs + cmp example/nuts-client.js example_docker/nuts-client.js + # Validate GoReleaser config without requiring a local GoReleaser install. release-check: docker run --rm -v "$(CURDIR):/workspace" -w /workspace $(GORELEASER_IMAGE) check @@ -252,6 +259,7 @@ help: @echo " test - Run all tests (unit + functional)" @echo " test-unit - Run unit tests with embedded NATS" @echo " test-performance - Run performance confidence tests and benchmarks" + @echo " test-js - Run the example JavaScript client's tests with Node.js" @echo " test-functional - Run functional/BDD tests with Docker" @echo " test-functional-stress - Run functional/BDD tests repeatedly with Docker" @echo " test-functional-matrix - Run functional/BDD tests against old and current NATS images" diff --git a/README.md b/README.md index 645a0b0..b196bcc 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,7 @@ A Caddy Server module that bridges NATS.io JetStream messages to Server-Sent Eve - [JetStream Setup](#jetstream-setup) - [Client Usage](#client-usage) - [JavaScript EventSource](#javascript-eventsource) + - [Replay-aware JavaScript helper](#replay-aware-javascript-helper) - [Slow Clients And Replay](#slow-clients-and-replay) - [Message Replay with `last-id` or `Last-Event-ID`](#message-replay-with-last-id-or-last-event-id) - [Message Format](#message-format) @@ -1128,7 +1129,9 @@ const events = new EventSource('/events/my-topic'); // Resume from a stored ID after a page reload. EventSource resends this URL, // ?last-id= included, on every auto-reconnect, but its own Last-Event-ID // header is fresher and takes precedence, so reconnects keep moving forward. -const lastId = localStorage.getItem('lastEventId'); +// sessionStorage keeps one ID per tab; tabs sharing localStorage would +// overwrite each other's. +const lastId = sessionStorage.getItem('lastEventId'); const url = lastId ? `/events?topic=notifications&last-id=${encodeURIComponent(lastId)}` : '/events?topic=notifications'; @@ -1139,7 +1142,7 @@ const events = new EventSource(url); // back to 0. function remember(e) { if (e.lastEventId) { - localStorage.setItem('lastEventId', e.lastEventId); + sessionStorage.setItem('lastEventId', e.lastEventId); } } @@ -1168,6 +1171,94 @@ events.onerror = (e) => { }; ``` +### Replay-aware JavaScript helper + +[`example/nuts-client.js`](example/nuts-client.js) wraps `EventSource` with the +replay handling above. It is a dependency-free ES module to copy into your +application, and the demo page in `example/` uses it: + +```javascript +import { subscribe } from './nuts-client.js'; + +const stream = subscribe('/events', { + topics: ['orders', 'invoices'], + on: { + message({ topic, payload }) { + console.log(`[${topic}]`, payload); + }, + reset({ reason }) { + // The stream was recreated or restored; it is replayed next. + console.log('Stream replaced:', reason); + }, + }, + onState(state) { + // connecting, open, retrying, restarting or closed + }, +}); + +// Later: +stream.close(); +``` + +- It takes the cursor from every event that carries one: the `connected` + handshake, messages, and the `reset` event, which sets it back to `0`. +- It keeps the cursor in `sessionStorage`, so a reload resumes after the last + event the tab received. +- When an `EventSource` gives up for good, it opens a new one from the latest + cursor after `restartDelay`. NUTS causes that only with answers that will + not change, such as a `401` or a `400`; a proxy in front of it can cause it + too. + +| Option | Default | Description | +| --- | --- | --- | +| `topics` | `[]` | Topics, each sent as `?topic=`. | +| `on` | `{}` | Listeners by event name, called with the parsed data and the `MessageEvent`. `message` is the default event name; `connected` and `reset` are NUTS' own. | +| `storage` | `sessionStorage` | Where the cursor survives reloads, one per tab. `localStorage` shares one between a site's tabs: a reloaded tab resumes where another one is, and misses what arrived in between. `null` keeps it in memory. | +| `key` | `nuts:` and the URL with its topics | The storage key. | +| `raw` | `false` | Pass message data as it is, for [`payload_format raw`](#payload_format). NUTS' own events are still parsed. | +| `withCredentials` | `false` | Send cookies cross-origin, for `subscriber_jwt_cookie`. | +| `restartDelay` | `10000` | Milliseconds before a new `EventSource` once one gave up. | +| `onState` | none | Called with `connecting`, `open` (NUTS accepted the subscription), `retrying` (`EventSource` reconnects by itself), `restarting` and `closed`. | +| `EventSource` | `globalThis.EventSource` | The implementation to use. | + +With [`event_type`](#event_type) `topic` or `header`, events are named after +their topic or header value, so list every name the route sends in `on`. The +helper does not see the ID of an event without a listener: a reload would +resume from an earlier one and repeat events, though it never skips any. + +Browsers cannot set headers on `EventSource`, so subscriber JWTs travel in a +cookie there ([`subscriber_jwt_cookie`](#subscriber-authentication-and-topic-authorization)). +Elsewhere, pass an implementation that can send headers, such as the +[`eventsource`](https://www.npmjs.com/package/eventsource) package: + +```javascript +import { EventSource } from 'eventsource'; +import { subscribe } from './nuts-client.mjs'; + +class AuthorizedEventSource extends EventSource { + constructor(url, init) { + super(url, { + ...init, + fetch: (input, request) => fetch(input, { + ...request, + headers: { ...request.headers, Authorization: `Bearer ${token}` }, + }), + }); + } +} + +subscribe('https://example.com/events', { + topics: ['orders'], + storage: null, + EventSource: AuthorizedEventSource, + on: { message: (data) => console.log(data) }, +}); +``` + +In Node.js, save it as `nuts-client.mjs`, or in a package with +`"type": "module"`, so that it loads as an ES module. `make test-js` runs its +tests with Node.js's built-in test runner (Node.js 20 or later, no packages). + ### Transient failures and EventSource A browser `EventSource` gives up for good when a connection or reconnection is diff --git a/example/index.html b/example/index.html index d5a99a3..48f3e03 100644 --- a/example/index.html +++ b/example/index.html @@ -144,7 +144,9 @@

🥜 NUTS Demo

both
- diff --git a/example/nuts-client.js b/example/nuts-client.js new file mode 100644 index 0000000..fa1d5b5 --- /dev/null +++ b/example/nuts-client.js @@ -0,0 +1,177 @@ +// nuts-client.js: a small EventSource helper for NUTS streams. +// +// A browser's EventSource already reconnects on its own and sends the last +// event ID it saw. What it does not do is keep that ID across a page reload, +// or start again once it gave up for good. This helper keeps the cursor in a +// storage and resumes from it: +// +// import { subscribe } from './nuts-client.js'; +// +// const stream = subscribe('/events', { +// topics: ['orders', 'invoices'], +// on: { +// message(data) { console.log(data.topic, data.payload); }, +// }, +// }); +// // later: stream.close(); +// +// It takes the cursor from every event that carries one: the connected +// handshake, messages, and the reset event NUTS sends when the stream was +// recreated or restored, which sets it back to 0. With event_type topic or +// header, name every event the route sends in `on`: the ID of an event +// without a listener is not seen, so a reload resumes from an earlier one and +// repeats some events. It never skips any. +// +// No dependencies; it runs wherever there is an EventSource. Copy it into +// your application. + +/** + * Subscribes to a NUTS SSE route. + * + * @param {string} url The route, for example "/events". A query it already + * has is kept; the helper adds last-id itself. + * @param {object} [options] + * @param {string[]} [options.topics] Topics, sent as ?topic=… each. + * @param {{getItem(key: string): (string|null), setItem(key: string, value: string): void}|null} [options.storage] + * Where the cursor survives page reloads. The default, sessionStorage, + * keeps one per tab. localStorage shares one between the tabs of a site: + * a reloaded tab resumes where another one is, and misses what arrived in + * between. null keeps it in memory only. + * @param {string} [options.key] The storage key; by default "nuts:" and the + * URL with its topics. + * @param {Object} [options.on] + * Listeners by event name. "message" is the default name; "connected" and + * "reset" are NUTS' own events. + * @param {boolean} [options.raw] Pass message data as it is, for + * payload_format raw, instead of parsing it as JSON. + * @param {boolean} [options.withCredentials] Send cookies cross-origin, for + * subscriber_jwt_cookie. + * @param {number} [options.restartDelay] Milliseconds to wait before a new + * EventSource once one gave up for good (default 10000). NUTS answers + * passing failures in a way EventSource retries by itself; it gives up only + * on answers that will not change, or when a proxy in front of NUTS fails. + * @param {function(string): void} [options.onState] Told "connecting", + * "open" (NUTS accepted the subscription), "retrying" (EventSource + * reconnects by itself), "restarting" and "closed". + * @param {typeof EventSource} [options.EventSource] An EventSource + * implementation, for tests or clients outside browsers. + * @returns {{close(): void, readonly lastEventId: string}} + */ +export function subscribe(url, options = {}) { + const { + topics = [], + storage = sessionStorageOrNull(), + on = {}, + raw = false, + withCredentials = false, + restartDelay = 10000, + onState = () => {}, + EventSource: EventSourceImpl = globalThis.EventSource, + } = options; + const base = withQuery(url, topics.map((topic) => 'topic=' + encodeURIComponent(topic))); + const key = options.key ?? 'nuts:' + base; + + let cursor = load(); + let source = null; + let timer = null; + let closed = false; + + function load() { + try { + return storage?.getItem(key) || ''; + } catch { + return ''; // storage disabled: keep the cursor in memory + } + } + + function remember(event) { + if (!event.lastEventId) { + return; + } + cursor = event.lastEventId; + try { + storage?.setItem(key, cursor); + } catch { + // storage disabled or full: keep the cursor in memory + } + } + + // parse decodes an event's data. NUTS' own events are always JSON; + // messages are unless payload_format is raw. + function parse(name, data) { + if (raw && name !== 'connected' && name !== 'reset') { + return data; + } + try { + return JSON.parse(data); + } catch { + return data; + } + } + + function start() { + onState('connecting'); + // EventSource keeps this URL on its own reconnects, but NUTS prefers the + // fresher Last-Event-ID header it sends then. + const target = cursor ? withQuery(base, ['last-id=' + encodeURIComponent(cursor)]) : base; + const es = new EventSourceImpl(target, { withCredentials }); + source = es; + for (const name of new Set(['connected', 'reset', 'message', ...Object.keys(on)])) { + es.addEventListener(name, (event) => { + remember(event); + if (name === 'connected') { + onState('open'); + } + const listener = on[name]; + if (listener) { + listener(parse(name, event.data), event); + } + }); + } + es.onerror = () => { + if (closed || source !== es) { + return; + } + if (es.readyState === EventSourceImpl.CLOSED) { + onState('restarting'); + timer = setTimeout(start, restartDelay); + } else { + onState('retrying'); + } + }; + } + + start(); + return { + close() { + if (closed) { + return; + } + closed = true; + clearTimeout(timer); + timer = null; + source.close(); + onState('closed'); + }, + get lastEventId() { + return cursor; + }, + }; +} + +function withQuery(url, params) { + if (params.length === 0) { + return url; + } + return url + (url.includes('?') ? '&' : '?') + params.join('&'); +} + +// sessionStorageOrNull returns the page's sessionStorage, or null where there +// is none or reading it throws (storage disabled, sandboxed frames). +function sessionStorageOrNull() { + try { + return globalThis.sessionStorage ?? null; + } catch { + return null; + } +} diff --git a/example/nuts-client.test.mjs b/example/nuts-client.test.mjs new file mode 100644 index 0000000..6594967 --- /dev/null +++ b/example/nuts-client.test.mjs @@ -0,0 +1,242 @@ +// Tests for nuts-client.js, run by `make test-js` with Node's built-in test +// runner; they need no packages. +import { test, mock } from 'node:test'; +import assert from 'node:assert/strict'; +import { subscribe } from './nuts-client.js'; + +// FakeEventSource records what the helper does with it and dispatches events +// the way a browser would. +class FakeEventSource { + static CONNECTING = 0; + static OPEN = 1; + static CLOSED = 2; + static created = []; + + constructor(url, init) { + this.url = url; + this.init = init; + this.readyState = FakeEventSource.CONNECTING; + this.listeners = new Map(); + this.onerror = null; + this.closed = false; + FakeEventSource.created.push(this); + } + + addEventListener(name, listener) { + const list = this.listeners.get(name) ?? []; + list.push(listener); + this.listeners.set(name, list); + } + + // emit dispatches an event with its data and, as EventSource does, the last + // event ID so far. + emit(name, data, lastEventId = '') { + this.readyState = FakeEventSource.OPEN; + for (const listener of this.listeners.get(name) ?? []) { + listener({ type: name, data, lastEventId }); + } + } + + fail(closed) { + this.readyState = closed ? FakeEventSource.CLOSED : FakeEventSource.CONNECTING; + this.onerror?.({ type: 'error' }); + } + + close() { + this.closed = true; + this.readyState = FakeEventSource.CLOSED; + } +} + +function memoryStorage(initial = {}) { + const items = new Map(Object.entries(initial)); + return { + getItem: (key) => (items.has(key) ? items.get(key) : null), + setItem: (key, value) => items.set(key, String(value)), + }; +} + +function last() { + return FakeEventSource.created[FakeEventSource.created.length - 1]; +} + +// withSessionStorage runs fn with globalThis.sessionStorage defined by +// descriptor, as a browser page would have it. +function withSessionStorage(descriptor, fn) { + const saved = Object.getOwnPropertyDescriptor(globalThis, 'sessionStorage'); + Object.defineProperty(globalThis, 'sessionStorage', { configurable: true, ...descriptor }); + try { + fn(); + } finally { + if (saved) { + Object.defineProperty(globalThis, 'sessionStorage', saved); + } else { + delete globalThis.sessionStorage; + } + } +} + +test('builds the URL from the topics and the stored cursor', () => { + FakeEventSource.created = []; + const storage = memoryStorage({ cursor: '41' }); + subscribe('/events', { topics: ['orders', 'a b'], storage, key: 'cursor', EventSource: FakeEventSource }); + assert.equal(last().url, '/events?topic=orders&topic=a%20b&last-id=41'); + + subscribe('/events', { storage: memoryStorage({ k: '4&x' }), key: 'k', EventSource: FakeEventSource }); + assert.equal(last().url, '/events?last-id=4%26x', 'a stored cursor is encoded'); + assert.deepEqual(last().init, { withCredentials: false }); + + subscribe('/events?topic=x', { storage: null, EventSource: FakeEventSource }); + assert.equal(last().url, '/events?topic=x', 'no cursor, no topics: the URL is kept'); + + subscribe('/events?topic=x', { + topics: ['y'], + storage: memoryStorage({ 'nuts:/events?topic=x&topic=y': '7' }), + withCredentials: true, + EventSource: FakeEventSource, + }); + assert.equal(last().url, '/events?topic=x&topic=y&last-id=7', 'the default key is the URL with its topics'); + assert.deepEqual(last().init, { withCredentials: true }); +}); + +test('keeps the cursor of every event that carries one, reset included', () => { + FakeEventSource.created = []; + const storage = memoryStorage(); + const seen = []; + const states = []; + const stream = subscribe('/events', { + topics: ['orders'], + storage, + key: 'k', + EventSource: FakeEventSource, + onState: (state) => states.push(state), + on: { + connected: (data, event) => seen.push(['connected', data, event.lastEventId]), + message: (data) => seen.push(['message', data]), + orders: (data) => seen.push(['orders', data]), + reset: (data) => seen.push(['reset', data]), + }, + }); + const es = last(); + es.emit('connected', '{"topics":["orders"]}', '41'); + assert.equal(storage.getItem('k'), '41', 'the handshake carries the start position'); + es.emit('message', '{"topic":"orders","payload":{"n":1}}', '42'); + es.emit('orders', '{"topic":"orders"}', '43'); + assert.equal(storage.getItem('k'), '43'); + es.emit('reset', '{"reason":"stream_recreated"}', '0'); + assert.equal(storage.getItem('k'), '0', 'the reset makes a reload replay the new stream'); + assert.equal(stream.lastEventId, '0'); + es.emit('message', 'not json', ''); + assert.equal(storage.getItem('k'), '0', 'an event without an ID keeps the cursor'); + assert.deepEqual(seen, [ + ['connected', { topics: ['orders'] }, '41'], + ['message', { topic: 'orders', payload: { n: 1 } }], + ['orders', { topic: 'orders' }], + ['reset', { reason: 'stream_recreated' }], + ['message', 'not json'], + ]); + assert.deepEqual(states, ['connecting', 'open']); +}); + +test('follows NUTS events and messages without listeners for them', () => { + FakeEventSource.created = []; + const stream = subscribe('/events', { storage: null, EventSource: FakeEventSource, on: { orders() {} } }); + last().emit('message', '{}', '5'); + assert.equal(stream.lastEventId, '5'); + last().emit('reset', '{}', '0'); + assert.equal(stream.lastEventId, '0'); + last().emit('connected', '{}', '3'); + assert.equal(stream.lastEventId, '3'); + last().emit('invoices', '{}', '9'); + assert.equal(stream.lastEventId, '3', 'an event name without a listener is not seen'); +}); + +test('raw payloads are passed as they are, NUTS events still parsed', () => { + FakeEventSource.created = []; + const seen = []; + const on = { connected: (data) => seen.push(data), reset: (data) => seen.push(data), message: (data) => seen.push(data) }; + subscribe('/raw', { raw: true, storage: null, EventSource: FakeEventSource, on }); + last().emit('connected', '{"topics":["a"]}', '1'); + last().emit('message', '{"n":1}', '2'); + last().emit('reset', '{"reason":"stream_rewound"}', '0'); + assert.deepEqual(seen, [{ topics: ['a'] }, '{"n":1}', { reason: 'stream_rewound' }]); +}); + +test('restarts a closed EventSource from the latest cursor, and close stops it', () => { + mock.timers.enable({ apis: ['setTimeout'] }); + try { + FakeEventSource.created = []; + const states = []; + const stream = subscribe('/events', { + topics: ['orders'], + storage: null, + EventSource: FakeEventSource, + onState: (state) => states.push(state), + }); + const first = last(); + first.emit('message', '{}', '7'); + first.fail(false); + assert.equal(FakeEventSource.created.length, 1, 'EventSource retries by itself'); + first.fail(true); + mock.timers.tick(9999); + assert.equal(FakeEventSource.created.length, 1, 'the default delay is 10 seconds'); + mock.timers.tick(1); + assert.equal(FakeEventSource.created.length, 2); + assert.equal(last().url, '/events?topic=orders&last-id=7'); + first.fail(true); // a late error of the old one changes nothing + last().fail(true); + stream.close(); + stream.close(); + mock.timers.tick(20000); + assert.equal(FakeEventSource.created.length, 2, 'a closed subscription does not restart'); + assert.ok(last().closed); + last().fail(true); + assert.deepEqual(states, ['connecting', 'retrying', 'restarting', 'connecting', 'restarting', 'closed']); + } finally { + mock.timers.reset(); + } +}); + +test('restartDelay sets the wait before a new EventSource', () => { + mock.timers.enable({ apis: ['setTimeout'] }); + try { + FakeEventSource.created = []; + subscribe('/events', { restartDelay: 1000, storage: null, EventSource: FakeEventSource }); + last().fail(true); + mock.timers.tick(999); + assert.equal(FakeEventSource.created.length, 1); + mock.timers.tick(1); + assert.equal(FakeEventSource.created.length, 2); + } finally { + mock.timers.reset(); + } +}); + +test('the default storage is the session storage', () => { + FakeEventSource.created = []; + const storage = memoryStorage({ 'nuts:/events?topic=a': '12' }); + withSessionStorage({ value: storage }, () => { + subscribe('/events', { topics: ['a'], EventSource: FakeEventSource }); + }); + assert.equal(last().url, '/events?topic=a&last-id=12'); + last().emit('message', '{}', '13'); + assert.equal(storage.getItem('nuts:/events?topic=a'), '13'); +}); + +test('a storage that throws leaves the cursor in memory', () => { + FakeEventSource.created = []; + const broken = { + getItem() { throw new Error('denied'); }, + setItem() { throw new Error('full'); }, + }; + const stream = subscribe('/events', { storage: broken, EventSource: FakeEventSource }); + assert.equal(last().url, '/events'); + last().emit('message', '{}', '9'); + assert.equal(stream.lastEventId, '9'); + + withSessionStorage({ get() { throw new Error('SecurityError'); } }, () => { + subscribe('/events', { EventSource: FakeEventSource }); + }); + last().emit('message', '{}', '10'); + assert.equal(last().url, '/events', 'without a session storage the cursor lives in memory'); +}); diff --git a/example/start.sh b/example/start.sh index 0137901..9f339f9 100755 --- a/example/start.sh +++ b/example/start.sh @@ -9,6 +9,6 @@ fi echo "" echo "========================================" echo " NUTS is ready!" -echo " Open: http://localhost:8080" +echo " Open: http://localhost:8080/example/" echo "========================================" echo "" diff --git a/example_docker/Caddyfile b/example_docker/Caddyfile index fda81da..1ad7c0e 100644 --- a/example_docker/Caddyfile +++ b/example_docker/Caddyfile @@ -15,7 +15,8 @@ } } - route / { + @static path / /nuts-client.js + route @static { root * /srv/static file_server } diff --git a/example_docker/docker-compose.yml b/example_docker/docker-compose.yml index db9a1e4..7681efb 100644 --- a/example_docker/docker-compose.yml +++ b/example_docker/docker-compose.yml @@ -41,6 +41,7 @@ services: volumes: - ./Caddyfile:/app/Caddyfile:ro - ./index.html:/srv/static/index.html:ro + - ./nuts-client.js:/srv/static/nuts-client.js:ro depends_on: nats-init: condition: service_completed_successfully diff --git a/example_docker/index.html b/example_docker/index.html index d5a99a3..48f3e03 100644 --- a/example_docker/index.html +++ b/example_docker/index.html @@ -144,7 +144,9 @@

🥜 NUTS Demo

both
- diff --git a/example_docker/nuts-client.js b/example_docker/nuts-client.js new file mode 100644 index 0000000..fa1d5b5 --- /dev/null +++ b/example_docker/nuts-client.js @@ -0,0 +1,177 @@ +// nuts-client.js: a small EventSource helper for NUTS streams. +// +// A browser's EventSource already reconnects on its own and sends the last +// event ID it saw. What it does not do is keep that ID across a page reload, +// or start again once it gave up for good. This helper keeps the cursor in a +// storage and resumes from it: +// +// import { subscribe } from './nuts-client.js'; +// +// const stream = subscribe('/events', { +// topics: ['orders', 'invoices'], +// on: { +// message(data) { console.log(data.topic, data.payload); }, +// }, +// }); +// // later: stream.close(); +// +// It takes the cursor from every event that carries one: the connected +// handshake, messages, and the reset event NUTS sends when the stream was +// recreated or restored, which sets it back to 0. With event_type topic or +// header, name every event the route sends in `on`: the ID of an event +// without a listener is not seen, so a reload resumes from an earlier one and +// repeats some events. It never skips any. +// +// No dependencies; it runs wherever there is an EventSource. Copy it into +// your application. + +/** + * Subscribes to a NUTS SSE route. + * + * @param {string} url The route, for example "/events". A query it already + * has is kept; the helper adds last-id itself. + * @param {object} [options] + * @param {string[]} [options.topics] Topics, sent as ?topic=… each. + * @param {{getItem(key: string): (string|null), setItem(key: string, value: string): void}|null} [options.storage] + * Where the cursor survives page reloads. The default, sessionStorage, + * keeps one per tab. localStorage shares one between the tabs of a site: + * a reloaded tab resumes where another one is, and misses what arrived in + * between. null keeps it in memory only. + * @param {string} [options.key] The storage key; by default "nuts:" and the + * URL with its topics. + * @param {Object} [options.on] + * Listeners by event name. "message" is the default name; "connected" and + * "reset" are NUTS' own events. + * @param {boolean} [options.raw] Pass message data as it is, for + * payload_format raw, instead of parsing it as JSON. + * @param {boolean} [options.withCredentials] Send cookies cross-origin, for + * subscriber_jwt_cookie. + * @param {number} [options.restartDelay] Milliseconds to wait before a new + * EventSource once one gave up for good (default 10000). NUTS answers + * passing failures in a way EventSource retries by itself; it gives up only + * on answers that will not change, or when a proxy in front of NUTS fails. + * @param {function(string): void} [options.onState] Told "connecting", + * "open" (NUTS accepted the subscription), "retrying" (EventSource + * reconnects by itself), "restarting" and "closed". + * @param {typeof EventSource} [options.EventSource] An EventSource + * implementation, for tests or clients outside browsers. + * @returns {{close(): void, readonly lastEventId: string}} + */ +export function subscribe(url, options = {}) { + const { + topics = [], + storage = sessionStorageOrNull(), + on = {}, + raw = false, + withCredentials = false, + restartDelay = 10000, + onState = () => {}, + EventSource: EventSourceImpl = globalThis.EventSource, + } = options; + const base = withQuery(url, topics.map((topic) => 'topic=' + encodeURIComponent(topic))); + const key = options.key ?? 'nuts:' + base; + + let cursor = load(); + let source = null; + let timer = null; + let closed = false; + + function load() { + try { + return storage?.getItem(key) || ''; + } catch { + return ''; // storage disabled: keep the cursor in memory + } + } + + function remember(event) { + if (!event.lastEventId) { + return; + } + cursor = event.lastEventId; + try { + storage?.setItem(key, cursor); + } catch { + // storage disabled or full: keep the cursor in memory + } + } + + // parse decodes an event's data. NUTS' own events are always JSON; + // messages are unless payload_format is raw. + function parse(name, data) { + if (raw && name !== 'connected' && name !== 'reset') { + return data; + } + try { + return JSON.parse(data); + } catch { + return data; + } + } + + function start() { + onState('connecting'); + // EventSource keeps this URL on its own reconnects, but NUTS prefers the + // fresher Last-Event-ID header it sends then. + const target = cursor ? withQuery(base, ['last-id=' + encodeURIComponent(cursor)]) : base; + const es = new EventSourceImpl(target, { withCredentials }); + source = es; + for (const name of new Set(['connected', 'reset', 'message', ...Object.keys(on)])) { + es.addEventListener(name, (event) => { + remember(event); + if (name === 'connected') { + onState('open'); + } + const listener = on[name]; + if (listener) { + listener(parse(name, event.data), event); + } + }); + } + es.onerror = () => { + if (closed || source !== es) { + return; + } + if (es.readyState === EventSourceImpl.CLOSED) { + onState('restarting'); + timer = setTimeout(start, restartDelay); + } else { + onState('retrying'); + } + }; + } + + start(); + return { + close() { + if (closed) { + return; + } + closed = true; + clearTimeout(timer); + timer = null; + source.close(); + onState('closed'); + }, + get lastEventId() { + return cursor; + }, + }; +} + +function withQuery(url, params) { + if (params.length === 0) { + return url; + } + return url + (url.includes('?') ? '&' : '?') + params.join('&'); +} + +// sessionStorageOrNull returns the page's sessionStorage, or null where there +// is none or reading it throws (storage disabled, sandboxed frames). +function sessionStorageOrNull() { + try { + return globalThis.sessionStorage ?? null; + } catch { + return null; + } +} diff --git a/example_docker/start.sh b/example_docker/start.sh index 0137901..b115821 100755 --- a/example_docker/start.sh +++ b/example_docker/start.sh @@ -1,5 +1,8 @@ #!/bin/sh cd "$(dirname "$0")" +# The demo shows the latest release: fetch it instead of running whatever +# copy of idcttech/nuts:latest was pulled before. +docker compose pull nuts if docker compose up --help 2>/dev/null | grep -q -- --wait; then docker compose up -d --wait else diff --git a/website/docs/usage.md b/website/docs/usage.md index 7f3c5d1..980b5c9 100644 --- a/website/docs/usage.md +++ b/website/docs/usage.md @@ -362,20 +362,29 @@ const events = new EventSource('/events/my-topic'); ### Handling Messages ```javascript +// Keep the last event ID for replay after a page reload. The handshake and +// every message carry one, and a reset (the stream was recreated or +// restored) sets it back to 0. sessionStorage keeps one per tab. +function remember(e) { + if (e.lastEventId) { + sessionStorage.setItem('lastEventId', e.lastEventId); + } +} + events.addEventListener('connected', (e) => { const { topics } = JSON.parse(e.data); console.log('Connected to:', topics); + remember(e); }); events.addEventListener('message', (e) => { const { topic, payload, time } = JSON.parse(e.data); console.log(`[${topic}] at ${time}:`, payload); - - if (e.lastEventId) { - localStorage.setItem('lastEventId', e.lastEventId); - } + remember(e); }); +events.addEventListener('reset', remember); + events.onerror = (e) => { console.error('SSE error:', e); // EventSource auto-reconnects and sends Last-Event-ID automatically @@ -400,7 +409,7 @@ Clients can resume from where they left off using `last-id` or the standard `Last-Event-ID` header: ```javascript -const lastId = localStorage.getItem('lastEventId'); +const lastId = sessionStorage.getItem('lastEventId'); const url = lastId ? `/events?topic=notifications&last-id=${encodeURIComponent(lastId)}` : '/events?topic=notifications'; @@ -423,6 +432,36 @@ const events = new EventSource(url); > replayed. Cap with `replay_max_messages` and/or `replay_window` for public > or multi-tenant routes. +### Replay-Aware Helper + +[`example/nuts-client.js`](https://github.com/ideaconnect/nuts/blob/main/example/nuts-client.js) +wraps `EventSource` with the replay handling above: a dependency-free ES module +to copy into your application. It keeps the cursor of every event that carries +one (the handshake, messages and the reset) in `sessionStorage`, resumes from +it after a reload, and opens a new `EventSource` when one gives up for good: + +```javascript +import { subscribe } from './nuts-client.js'; + +const stream = subscribe('/events', { + topics: ['orders', 'invoices'], + on: { + message({ topic, payload }) { + console.log(`[${topic}]`, payload); + }, + }, +}); + +// Later: +stream.close(); +``` + +With `event_type topic` or `header`, list every event name the route sends in +`on`: the helper does not see the ID of an event without a listener. The +[README](https://github.com/ideaconnect/nuts#replay-aware-javascript-helper) +describes its options, and how to send subscriber JWTs as headers outside +browsers. + ### Slow Clients Each stream pulls from JetStream only as fast as its client reads, so a client