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