diff --git a/.claude/skills/fix-flaky-tests/SKILL.md b/.claude/skills/fix-flaky-tests/SKILL.md new file mode 100644 index 00000000..a3399aab --- /dev/null +++ b/.claude/skills/fix-flaky-tests/SKILL.md @@ -0,0 +1,70 @@ +--- +name: fix-flaky-tests +description: Find, reproduce, root-cause and permanently fix flaky (intermittently failing) tests in the hyper-fetch monorepo. Use proactively whenever a test fails once and passes on re-run, when CI is red and then green without code changes, when a `waitFor` times out around 1000ms, when a test depends on sleeps or default mock delays, or when a user says "flaky", "intermittent", "timeout", "works locally", "fix flake". Also use after adding new async tests to prove they are deterministic before committing. +--- + +# Fix Flaky Tests + +A flaky test is a test with a hidden dependency on time, order or shared state. Never "fix" it by re-running, retrying, +widening a timeout, or marking it skipped. Find the dependency and remove it. + +## Workflow + +1. **Reproduce, don't guess.** Run the single spec repeatedly and shuffled before reading any code: + + ```bash + .claude/skills/fix-flaky-tests/scripts/repeat-test.sh packages/react __tests__/features/use-submit/use-submit.queue.spec.ts 10 + ``` + + The script runs the file N times (default 10) with `pnpm exec vitest run` inside the package and prints the failure + rate. Add `--shuffle` to also randomize test order (`--sequence.shuffle`). A 0/10 result after a fix is the minimum + bar; prefer 20 runs for timing-related fixes. + +2. **Confirm it is pre-existing or new.** If the failing test is unrelated to the current change, check it on `main` + with `git stash push -u` → run → `git stash pop`. Pre-existing flakes still get fixed in the same branch (small, + separate commit) - leaving them costs everyone time. + +3. **Read the failure, not the test name.** The message tells you the class of flake: + - `waitFor` / `testLoading` timing out around `1000ms` → the awaited state was already gone or never reached (race). + - "Calling the test function inside another test function" → structural (misplaced braces), not timing. + - `act` warnings / "state update on unmounted component" → missing cleanup or delivery after unsubscribe. + - Different failures each run → shared mutable state between tests. + - Fails only in the full suite, passes alone → order dependence or leaked globals / timers / servers. + +4. **Classify the root cause** (see table) and apply the matching fix pattern. Fix the cause, keep the assertion. + +5. **Verify**: repeat-run the spec (10-20x), run the package suite once, run `pnpm typecheck` in the package. + +6. **Commit with the cause in the message**, e.g. `test(react): make useSubmit queue spec deterministic - mock resolved + before second hook mounted`. + +## Root causes and fixes in this repo + +| Symptom | Root cause | Fix | +| --- | --- | --- | +| Second hook/instance expected to see an in-flight request but sees finished | `mockRequest(request)` default delay is **20ms** (`packages/testing/src/http/http.mock.ts`) - request finishes before the next render/mount | Pass an explicit `delay` (e.g. `{ delay: 500 }`) so the window is wide enough; never rely on the default | +| Assertion right after an async emit | Treating async delivery as sync | Wrap in `await waitFor(() => ...)`; for sockets use `waitForConnection(socket)` before emitting | +| Test uses `await sleep(n)` and asserts a count | Real-timer race | Use `vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] })` + `vi.advanceTimersByTime`, or `waitFor`. Keep `Date`/promises real so mock servers still connect | +| Fake timers break connection setup | mock-socket / msw use timers to connect | Connect with real timers first (`await waitForConnection`), then switch to fake timers inside the test or `beforeEach` | +| Passes alone, fails in suite | Leaked state: `client` cache/dispatchers, socket listeners, servers, spies | Reset in `beforeEach` (`client.clear()`, `vi.resetAllMocks()`), stop servers in `afterEach/afterAll`, `vi.useRealTimers()` in `afterEach` | +| Random test fails each run | Shared module-level fixture mutated by tests | Create fixtures inside `beforeEach`; use unique topics/keys (`getUniqueRequestId`) | +| Only fails on CI | Slower machine amplifies a race | Same fixes as above; never just raise `testTimeout` | +| Timer-based delivery (throttle/batch/debounce) asserts after unsubscribe | Pending timer fires after cleanup | Assert nothing is delivered after dispose; ensure the implementation clears timers on unsubscribe | + +## Rules + +- A flake found while working on something else is still fixed, in its own small commit. Do not leave `it.skip`. +- The fix must make the test deterministic, not merely rarer. If you cannot explain *why* it failed, you have not fixed it. +- Prefer controlling time (fake timers, explicit mock delays) over waiting longer. +- Prefer `waitFor` over `sleep`. Prefer asserting on events over asserting on elapsed time. +- When you touch a flaky spec, also scan its sibling specs for the same pattern (same helper, same default delay) and + fix them in the same commit. +- After fixing, run the whole package suite once - some flakes hide others. + +## Monorepo notes + +- Tests are Vitest; packages expose `pnpm test` and `pnpm typecheck`. Run a single file with + `pnpm exec vitest run ` from the package directory. +- `@hyper-fetch/testing` helpers: `createHttpMockingServer` (`mockRequest(request, { delay, data, status })`), + `createWebsocketMockingServer` (`emitListenerEvent`), `createSseMockingServer`, `waitForConnection`, `sleep`. +- Dependents resolve sibling packages through `dist/` - if types look stale, `pnpm build` the dependency first. diff --git a/.claude/skills/fix-flaky-tests/scripts/repeat-test.sh b/.claude/skills/fix-flaky-tests/scripts/repeat-test.sh new file mode 100755 index 00000000..72c9d7f5 --- /dev/null +++ b/.claude/skills/fix-flaky-tests/scripts/repeat-test.sh @@ -0,0 +1,27 @@ +#!/usr/bin/env bash +# Run one Vitest spec file N times inside a package and report the failure rate. +# Usage: repeat-test.sh [runs=10] [--shuffle] +set -u +PKG_DIR="${1:?package dir required, e.g. packages/react}" +SPEC="${2:?spec path required}" +RUNS="${3:-10}" +SHUFFLE="" +if [[ "${3:-}" == "--shuffle" || "${4:-}" == "--shuffle" ]]; then + SHUFFLE="--sequence.shuffle" + [[ "${3:-}" == "--shuffle" ]] && RUNS=10 +fi + +cd "$PKG_DIR" || exit 1 +fails=0 +for i in $(seq 1 "$RUNS"); do + if out=$(pnpm exec vitest run "$SPEC" $SHUFFLE 2>&1); then + printf "run %2d: pass\n" "$i" + else + fails=$((fails + 1)) + printf "run %2d: FAIL\n" "$i" + echo "$out" | grep -E "×|FAIL|Error|expected|received|Timed out" | head -15 | sed 's/^/ /' + fi +done +echo "----" +echo "failures: $fails / $RUNS ($SPEC)" +[[ "$fails" -eq 0 ]] diff --git a/documentation/docs/guides/react/02-sockets/listening.mdx b/documentation/docs/guides/react/02-sockets/listening.mdx index 09b38aeb..6cfc9f67 100644 --- a/documentation/docs/guides/react/02-sockets/listening.mdx +++ b/documentation/docs/guides/react/02-sockets/listening.mdx @@ -82,6 +82,9 @@ function LiveChat() { In the example above, `onChatMessage` is a pre-defined listener request. You can learn more about creating them in the [sockets guide](/docs/guides/sockets/listening). +If the stream is fast and your component re-renders too often, configure a `delivery` strategy on the listener - see +[Throttling & Batching in React](/docs/guides/react/sockets/throttling). + --- :::success Congratulations! diff --git a/documentation/docs/guides/react/02-sockets/throttling.mdx b/documentation/docs/guides/react/02-sockets/throttling.mdx new file mode 100644 index 00000000..159c4480 --- /dev/null +++ b/documentation/docs/guides/react/02-sockets/throttling.mdx @@ -0,0 +1,177 @@ +--- +sidebar_position: 3 +title: React - Socket Throttling & Batching +sidebar_label: Throttling & Batching +--- + +# Throttling & Batching in React + +With `useListener`, every delivered message is a state update, and every state update is a render. On a stream that +pushes hundreds of messages per second, that is hundreds of renders per second for a screen that refreshes sixty times. + +The fix lives on the listener, not in the component. Configure a `delivery` strategy once, and `useListener` simply +receives fewer, better-shaped updates. No hook options, no manual throttling in effects. + +--- + +:::secondary What you'll learn + +1. How a **`latest`** listener turns a price firehose into one render per frame. +2. How a **`batch`** listener delivers arrays so you can append many rows in one state update. +3. How to let the user **choose the refresh rate** with `setDelivery` and `useMemo`. +4. The one **gotcha** with custom strategies and React re-renders. + +::: + +> Start with the framework-agnostic [Throttling & Batching guide](/docs/guides/sockets/throttling) if you have not +> read it - this page focuses on the React side. + +--- + +## Sampling a price ticker + +Define the listener with a `latest` strategy. `useListener` needs no changes: `data` is simply the newest price in the +last window. + +```ts title="api/socket.ts" +export const onPrice = socket.createListener()({ + topic: "prices", + // highlight-next-line + delivery: { strategy: "latest", interval: 16 }, +}); +``` + +```tsx title="PriceTicker.tsx" +import { useListener } from "@hyper-fetch/react"; +import { onPrice } from "./api/socket"; + +export function PriceTicker() { + // highlight-start + // Renders at most ~60 times per second regardless of how fast the server sends. + // highlight-end + const { data, connected } = useListener(onPrice); + + if (!connected) return Connecting…; + return {data ? data.value.toFixed(2) : "—"}; +} +``` + +--- + +## Batching a trade log + +With `batch`, `data` and the `onEvent` payload are **arrays**, typed automatically as `Trade[]`. Append the whole batch +in a single state update. + +```ts title="api/socket.ts" +export const onTrade = socket.createListener()({ + topic: "trades", + // highlight-next-line + delivery: { strategy: "batch", interval: 100, maxSize: 500 }, +}); +``` + +```tsx title="TradeLog.tsx" +import { useState } from "react"; +import { useListener } from "@hyper-fetch/react"; +import { onTrade } from "./api/socket"; + +export function TradeLog() { + const [trades, setTrades] = useState([]); + const { onEvent } = useListener(onTrade); + + onEvent(({ data }) => { + // highlight-start + // `data` is Trade[] - one setState per window instead of one per trade. + // highlight-end + setTrades((previous) => [...previous, ...data].slice(-1000)); + }); + + return ( +
    + {trades.map((trade) => ( +
  • + {trade.symbol} {trade.price} +
    • + ))} +
+ ); +} +``` + +--- + +## Letting the user pick the refresh rate + +`setDelivery` returns a new listener. Memoize it on the user's choice and pass it to `useListener`; the hook +re-subscribes when the delivery changes and disposes the previous window. + +```tsx title="PriceTicker.tsx" +import { useMemo, useState } from "react"; +import { useListener } from "@hyper-fetch/react"; +import { onPrice } from "./api/socket"; + +export function PriceTicker() { + const [rate, setRate] = useState(16); + // highlight-start + // A new listener only when `rate` changes - not on every render. + // highlight-end + const live = useMemo(() => onPrice.setDelivery({ strategy: "latest", interval: rate }), [rate]); + const { data } = useListener(live); + + return ( + <> + + {data?.value ?? "—"} + + ); +} +``` + +--- + +## The gotcha: custom strategies and identity + +`useListener` decides whether to re-subscribe by comparing the listener's `topic`, `options` and `delivery`: + +- **Preset objects** (`{ strategy: "latest", interval: 16 }`) are compared **by value**. Writing them inline is fine. +- **Custom strategy functions** are compared **by identity**. Define them at module level, next to your listeners. A + strategy created inside a component is a new function on every render and would re-subscribe each time, dropping + whatever was buffered. + +```ts title="api/socket.ts" +// ✅ module level - stable identity +export const summarize = createDeliveryStrategy(/* ... */); +export const onTradeSummary = socket.createListener()({ topic: "trades", delivery: summarize }); +``` + +--- + +:::success Congratulations! + +You can now keep React UIs smooth on the busiest streams. + +- A **`latest`** listener caps renders at the rate you choose. +- A **`batch`** listener gives you arrays and one state update per window. +- **`setDelivery` + `useMemo`** lets the user control the rate. +- Custom strategies live at **module level** to keep a stable identity. + +::: + + + + diff --git a/documentation/docs/guides/sockets/emitting.mdx b/documentation/docs/guides/sockets/emitting.mdx index e6a85585..3cdf47f4 100644 --- a/documentation/docs/guides/sockets/emitting.mdx +++ b/documentation/docs/guides/sockets/emitting.mdx @@ -1,5 +1,5 @@ --- -sidebar_position: 3 +sidebar_position: 4 title: Guide - Socket Emitting sidebar_label: Emitting --- @@ -36,7 +36,7 @@ type ChatMessageType = { //highlight-end const sendChatMessage = socketInstance.createEmitter()({ - endpoint: "chat-message", // topic of the event + topic: "chat-message", // topic of the event }); ``` @@ -86,7 +86,7 @@ Let's create an emitter to send a private message to a specific user. ```typescript const sendPrivateMessage = socketInstance.createEmitter<{ message: string }>()({ - endpoint: "private-message/:userId", + topic: "private-message/:userId", }); ``` @@ -115,7 +115,7 @@ data, or encrypting messages: ```typescript const sendMessage = socketInstance .createEmitter<{ message: string }>()({ - endpoint: "chat-message", + topic: "chat-message", }) .setPayloadMapper((payload) => ({ ...payload, diff --git a/documentation/docs/guides/sockets/listening.mdx b/documentation/docs/guides/sockets/listening.mdx index 6f96c57a..4a7ecb2c 100644 --- a/documentation/docs/guides/sockets/listening.mdx +++ b/documentation/docs/guides/sockets/listening.mdx @@ -25,7 +25,7 @@ that can be easily managed and integrated into your application, including React ## Creating a Listener To start listening for events, you first need to create a listener. This is done using the `socket.createListener` -method, which requires an `endpoint` that corresponds to the event name you want to listen to. It's also a good practice +method, which requires a `topic` that corresponds to the event name you want to listen to. It's also a good practice to provide a type for the expected message payload for type safety. ```ts @@ -34,7 +34,7 @@ to provide a type for the expected message payload for type safety. // This helps to avoid re-creating listeners and attaching multiple handlers for the same event. // highlight-end const onChatMessage = socketInstance.createListener<{ message: string }>()({ - endpoint: "chat-message", + topic: "chat-message", }); ``` @@ -44,7 +44,7 @@ received from this event will be correctly typed. ## Starting and Stopping Once you have a listener, you can start listening for events by calling the `.listen()` method with a callback function. -This function will be executed every time a new message arrives on the specified endpoint. +This function will be executed every time a new message arrives on the specified topic. The `.listen()` method returns an `unlisten` function, which you can call to stop listening for events. It's crucial to call this cleanup function when the listener is no longer needed to prevent memory leaks. @@ -127,6 +127,22 @@ unlistenTwo(); unlistenOne(); ``` +## Dealing with High-Frequency Streams + +If a topic emits hundreds of messages per second, you rarely want every one of them to reach your callbacks. The +`delivery` option lets a listener sample (`latest`) or batch (`batch`) messages before they are delivered, while +`listen()` and `useListener` keep working unchanged: + +```ts +const onPrice = socketInstance.createListener()({ + topic: "prices", + delivery: { strategy: "latest", interval: 16 }, +}); +``` + +Read the [Throttling & Batching guide](/docs/guides/sockets/throttling) for timelines, trade-offs and custom +strategies. + --- :::success Congratulations! diff --git a/documentation/docs/guides/sockets/setup.mdx b/documentation/docs/guides/sockets/setup.mdx index ca2db7f3..532f742b 100644 --- a/documentation/docs/guides/sockets/setup.mdx +++ b/documentation/docs/guides/sockets/setup.mdx @@ -46,7 +46,7 @@ type ChatMessageType = { }; export const onChatMessage = socket.createListener()({ - endpoint: "chat-message", // endpoint of the event + topic: "chat-message", // topic of the event }); ``` @@ -70,7 +70,7 @@ type AcknowledgementResponseType = { }; export const sendChatMessage = socket.createEmitter()({ - endpoint: "chat-message", // endpoint of the event + topic: "chat-message", // topic of the event }); ``` diff --git a/documentation/docs/guides/sockets/throttling.mdx b/documentation/docs/guides/sockets/throttling.mdx new file mode 100644 index 00000000..06ab23fe --- /dev/null +++ b/documentation/docs/guides/sockets/throttling.mdx @@ -0,0 +1,276 @@ +--- +sidebar_position: 3 +title: Socket Throttling & Batching Guide +sidebar_label: Throttling & Batching +--- + +# Throttling & Batching + +Some socket streams are fast. A price ticker can push hundreds of updates per second, a trade feed thousands. Every one +of those messages reaches every `listen` callback, and in React every one of them is a re-render. Most of that work is +wasted: the screen refreshes 60 times a second, and nobody reads 500 intermediate prices. + +Hyper Fetch solves this at the listener level with the **`delivery`** option. You declare *how* messages should reach +your callbacks, and everything built on top of the listener - `listen()`, `useListener`, the SDK - follows along +without any changes. + +--- + +:::secondary What you'll learn + +1. Why high-frequency streams need **delivery control** and where it happens in the pipeline. +2. How to **sample** state-like data with the `latest` strategy. +3. How to **batch** event-like data with the `batch` strategy so nothing is lost. +4. How to **choose** between the two, and the semantics you should know. +5. How to write a **custom strategy** when the presets are not enough. +6. How to **switch delivery at runtime**. + +::: + +## Where delivery happens + +Without `delivery`, a message goes: socket → adapter → your callback. With `delivery`, each `listen()` call gets its +own small controller sitting right before your callback: + +``` +socket ──▶ adapter ──▶ [ delivery controller ] ──▶ your callback + push() deliver() +``` + +- `push` receives every raw message. +- The strategy decides **if and when** to call `deliver`, which is your callback. +- When you unsubscribe, the controller is disposed: timers are cleared and pending messages are dropped. + +Because this lives inside the listener, nothing else has to know about it. The raw stream is still available for +tooling through `socket.events.onListenerEvent`, which is never throttled. + +--- + +## Latest: sampling state-like data + +Use `latest` when only the newest value matters - prices, cursor positions, progress, presence. The first message of an +idle period is delivered immediately, then at most **one message per `interval`**, always the most recent one. +Everything in between is dropped on purpose. + +```ts +// highlight-start +// 16ms ≈ one animation frame - the UI cannot show updates faster than that anyway. +// highlight-end +export const onPrice = socket.createListener()({ + topic: "prices", + delivery: { strategy: "latest", interval: 16 }, +}); + +onPrice.listen(({ data }) => { + // `data` is a single Price - the newest one in the last 16ms window + render(data); +}); +``` + +Timeline for `interval: 100` with messages **A B C** at 0 / 20 / 40 ms and **D** at 300 ms: + +``` +t=0 A arrives → delivered immediately, 100ms window opens +t=20 B arrives → pending +t=40 C arrives → pending (B is dropped) +t=100 window closes → C delivered, next window opens +t=200 window closes → nothing pending → idle +t=300 D arrives → delivered immediately (idle, so it leads again) +``` + +Set `leading: false` if you prefer to wait a full interval before the first delivery: + +```ts +delivery: { strategy: "latest", interval: 100, leading: false } +``` + +--- + +## Batch: collecting event-like data + +Use `batch` when **every message matters** - trades, chat messages, log lines, notifications. The first message opens a +window of `interval` milliseconds; when it closes, every collected message is delivered **at once as an array**, in +arrival order. Nothing is dropped. + +```ts +export const onTrade = socket.createListener()({ + topic: "trades", + // highlight-start + // `maxSize` is a safety valve: a burst of 500 trades is delivered right away instead of growing the buffer. + // highlight-end + delivery: { strategy: "batch", interval: 100, maxSize: 500 }, +}); + +onTrade.listen(({ data }) => { + // highlight-next-line + // `data` is Trade[] - TypeScript infers the array type from the "batch" strategy + appendRows(data); +}); +``` + +Timeline for `interval: 100`: + +``` +t=0 A arrives → buffer [A], 100ms window opens +t=20 B arrives → buffer [A, B] +t=40 C arrives → buffer [A, B, C] +t=100 window closes → [A, B, C] delivered +t=300 D arrives → buffer [D], window opens +t=400 window closes → [D] delivered +``` + +Empty batches are never delivered. + +--- + +## Choosing a strategy + +| | `latest` | `batch` | +| --- | --- | --- | +| **Data is…** | state - the newest value replaces the old one | events - each one must be processed | +| **Losing intermediate messages** | Expected | Never | +| **Callback receives** | `Response` | `Response[]` | +| **Typical interval** | 16-100 ms (render rate) | 50-500 ms (processing rate) | +| **Examples** | prices, positions, progress, presence | trades, chat, logs, notifications | + +If neither fits exactly, write a [custom strategy](#custom-strategies). + +--- + +## Semantics you should know + +- **`extra` is the last message's `extra`.** For the WebSocket adapter `extra` is the raw `MessageEvent`. A batch + delivers the `extra` of its final message; keeping hundreds of them would be pure memory cost. If you need per-item + metadata, use a custom strategy. +- **Nothing fires after unsubscribe.** Calling the function returned by `listen()` disposes the controller and drops + pending messages. This keeps React safe: an unmounted component never receives a late update. +- **Each `listen()` call is independent.** Two subscriptions on the same listener each get their own timers and + buffers, even with the same callback. +- **`delivery` travels with the listener.** `setParams`, `setOptions` and `clone` keep it. `setDelivery` returns a new + listener, like every other setter - see + [Methods Return Clones](/docs/sockets/listener#methods). +- **Zero cost when unused.** Without `delivery`, your callback is registered directly - no wrapper, no timers. +- **Interceptors run first.** `socket.onMessage` modifiers run per raw message, before delivery. + +--- + +## Custom strategies + +A strategy is a factory called once per `listen()`. It receives `deliver` (your callback) and returns an object with +`push` (input) and `dispose` (cleanup). That is the whole contract: one input, one output, one cleanup. + +Declare the output type once with `createDeliveryStrategy` - TypeScript cannot infer it from how you +call `deliver`, the same way it cannot infer the wire type in `createListener()`. + +### Example: debounce (deliver after silence) + +```ts +import { createDeliveryStrategy } from "@hyper-fetch/sockets"; + +export const settle = createDeliveryStrategy(({ deliver }) => { + let timer: ReturnType | undefined; + let last: { data: Tick; extra: any } | undefined; + + return { + push: (message) => { + last = message; + clearTimeout(timer); + // highlight-next-line + timer = setTimeout(() => { + if (last) deliver(last); + last = undefined; + }, 50); + }, + dispose: () => { + clearTimeout(timer); + last = undefined; + }, + }; +}); + +const onTick = socket.createListener()({ topic: "ticks", delivery: settle }); +``` + +### Example: aggregate into a summary + +```ts +type Summary = { count: number; max: number }; + +export const summarize = createDeliveryStrategy(({ deliver }) => { + let count = 0; + let max = -Infinity; + let timer: ReturnType | undefined; + + return { + push: ({ data, extra }) => { + count += 1; + max = Math.max(max, data.price); + timer ??= setTimeout(() => { + timer = undefined; + deliver({ data: { count, max }, extra }); + count = 0; + max = -Infinity; + }, 1000); + }, + dispose: () => clearTimeout(timer), + }; +}); + +const onTradeSummary = socket.createListener()({ topic: "trades", delivery: summarize }); + +onTradeSummary.listen(({ data }) => { + // highlight-next-line + // data: Summary + console.log(`${data.count} trades, max ${data.max}`); +}); +``` + +The built-in presets are implemented with exactly this contract. Their factories, `createLatestDeliveryStrategy` and +`createBatchDeliveryStrategy`, are exported so you can compose them inside your own strategies. + +:::info Define strategies at module level + +Strategy functions are compared by identity. Define them once next to your listeners, not inline in a component, or +React will re-subscribe on every render. Preset objects can be written inline - they are compared by value. + +::: + +--- + +## Switching delivery at runtime + +`setDelivery` returns a clone with the new strategy and the matching callback type. Pass `undefined` to go back to +immediate delivery. + +```ts +const live = onPrice.setDelivery({ strategy: "latest", interval: 16 }); // 60 fps +const relaxed = onPrice.setDelivery({ strategy: "latest", interval: 1000 }); // 1 fps +const everything = onPrice.setDelivery(undefined); // every message +``` + +--- + +:::success Congratulations! + +You can now keep high-frequency streams fast for the server and cheap for the client. + +- You know **where** delivery happens and why it keeps the rest of the API unchanged. +- You can **sample** with `latest` and **batch** with `batch`, and you know which to pick. +- You understand the **semantics**: last `extra`, nothing after unsubscribe, independent subscriptions. +- You can write a **custom strategy** with `push`, `deliver` and `dispose`. + +::: + + + + diff --git a/documentation/docs/react/05-sockets/use-emitter.mdx b/documentation/docs/react/05-sockets/use-emitter.mdx index b49b0ed1..e38806de 100644 --- a/documentation/docs/react/05-sockets/use-emitter.mdx +++ b/documentation/docs/react/05-sockets/use-emitter.mdx @@ -38,7 +38,7 @@ First, let's define an emitter. import { socketInstance } from "./socket"; export const emitMessage = socketInstance.createEmitter()({ - endpoint: "chat-message", // endpoint of the event + topic: "chat-message", // topic of the event }); ``` diff --git a/documentation/docs/react/05-sockets/use-listener.mdx b/documentation/docs/react/05-sockets/use-listener.mdx index ec1a6512..045b5300 100644 --- a/documentation/docs/react/05-sockets/use-listener.mdx +++ b/documentation/docs/react/05-sockets/use-listener.mdx @@ -37,7 +37,7 @@ First, create a `Listener` instance using your socket client. import { socketInstance } from "./socket"; export const onChatMessage = socketInstance.createListener()({ - endpoint: "chat-message", + topic: "chat-message", }); ``` @@ -148,6 +148,58 @@ function App() { --- +## Throttling & Batching + +Every delivered message is a state update, so a fast stream means fast re-rendering. Instead of throttling inside the +component, configure a [delivery strategy](/docs/sockets/listener#delivery-strategies) on the listener. `useListener` +needs no options for this - it simply receives fewer, better-shaped updates, and `data` follows the delivered type. + +```tsx title="Sampling - one render per frame at most" +export const onPrice = socket.createListener()({ + topic: "prices", + delivery: { strategy: "latest", interval: 16 }, +}); + +function PriceTicker() { + const { data } = useListener(onPrice); // data: Price | null + return {data?.value}; +} +``` + +```tsx title="Batching - arrays, one state update per window" +export const onTrade = socket.createListener()({ + topic: "trades", + delivery: { strategy: "batch", interval: 100 }, +}); + +function TradeLog() { + const [trades, setTrades] = useState([]); + const { onEvent } = useListener(onTrade); // data: Trade[] | null + + onEvent(({ data }) => { + setTrades((previous) => [...previous, ...data]); + }); + // ... +} +``` + +To let the user change the rate, memoize a `setDelivery` clone - the hook re-subscribes when the delivery changes: + +```tsx +const live = useMemo(() => onPrice.setDelivery({ strategy: "latest", interval: rate }), [rate]); +const { data } = useListener(live); +``` + +:::info Custom strategies + +Preset objects are compared by value and can be written inline. Custom strategy functions are compared by identity - +define them at module level, or the hook will re-subscribe on every render. See the +[React Throttling & Batching guide](/docs/guides/react/sockets/throttling). + +::: + +--- + ## Options Configuration options for `useListener` must be provided as the second parameter. diff --git a/documentation/docs/sockets/listener.mdx b/documentation/docs/sockets/listener.mdx index e0eb5b89..49d714a9 100644 --- a/documentation/docs/sockets/listener.mdx +++ b/documentation/docs/sockets/listener.mdx @@ -138,6 +138,70 @@ onNoteChange.setParams({ noteId: 2 }).listen(/* ... */); --- +### Delivery Strategies {#delivery-strategies} + +High-frequency streams (prices, trades, logs) can call your callbacks hundreds of times per second. The `delivery` +option controls how messages reach `listen` callbacks, right inside the listener, so `listen()`, `useListener` and the +SDK all benefit without changes. Omit it for immediate delivery of every message - that is the default and has zero +overhead. + +```ts +// Sampling - deliver the first message immediately, then at most one (the newest) per 16ms. +// Best for state-like data where only the latest value matters. +export const onPrice = socket.createListener()({ + topic: "prices", + delivery: { strategy: "latest", interval: 16 }, +}); +onPrice.listen(({ data }) => {}); // data: Price + +// Batching - deliver everything collected during 100ms as one array. Nothing is dropped. +// Best for event-like data where every message must be processed. +export const onTrade = socket.createListener()({ + topic: "trades", + delivery: { strategy: "batch", interval: 100, maxSize: 500 }, +}); +onTrade.listen(({ data }) => {}); // data: Trade[] + +// Custom - full control over buffering and timing. +export const onTick = socket.createListener()({ + topic: "ticks", + delivery: createDeliveryStrategy(({ deliver }) => ({ + push: ({ data, extra }) => { + /* accumulate, then deliver({ data: summary, extra }) */ + }, + dispose: () => { + /* clear timers */ + }, + })), +}); +onTick.listen(({ data }) => {}); // data: Summary +``` + +| | `latest` | `batch` | custom | +| --- | --- | --- | --- | +| Callback `data` | `Response` | `Response[]` | your `Output` type | +| Drops messages | Yes, intermediate ones | Never | up to you | +| Options | `interval`, `leading` (default `true`) | `interval`, `maxSize` | - | + +Semantics shared by all strategies: + +- `extra` is the `extra` of the last message that went into the delivery. +- Unsubscribing disposes the strategy: timers are cleared and pending messages are dropped, so nothing fires after + `unlisten()`. +- Every `listen()` call gets its own buffers and timers. +- `delivery` is kept by `setParams`, `setOptions` and `clone`. Use `setDelivery` to change it (returns a clone); + pass `undefined` to restore immediate delivery. +- `socket.events.onListenerEvent` still receives every raw message - useful for devtools and logging. + +:::info Full guide + +Timelines, how to choose a strategy, and complete custom strategy examples live in the +[Throttling & Batching guide](/docs/guides/sockets/throttling). + +::: + +--- + ### Global Event Subscriptions You can use `socket.events.onListenerEventByTopic` to subscribe to all events for a given listener, regardless of @@ -217,8 +281,8 @@ application. :::danger Methods Return Clones -Methods on a `Listener` instance (e.g., `setParams`, `setOptions`) do not modify the original listener instance in -place. Instead, they return a **new, cloned instance** of the listener with the specified modifications applied. The +Methods on a `Listener` instance (e.g., `setParams`, `setOptions`, `setDelivery`) do not modify the original listener +instance in place. Instead, they return a **new, cloned instance** of the listener with the specified modifications applied. The original listener instance remains unchanged. **Always use the returned new instance for subsequent operations.** This immutable approach ensures isolation between @@ -314,6 +378,31 @@ onChatMessage.listen(({ data }) => { This simple yet powerful feature eliminates guesswork and ensures that your application correctly handles the data structures sent by the server. +### Delivered Type + +A listener carries two types: the **response type** (what arrives on the wire) and the **delivered type** (what your +`listen` callbacks receive). They are the same unless a [delivery strategy](#delivery-strategies) changes the shape - +`batch` delivers `Response[]`, a custom strategy delivers its declared output. The delivered type is inferred from the +`delivery` option, so you never write it by hand. + +```ts +const onTrade = socket.createListener()({ + topic: "trades", + delivery: { strategy: "batch", interval: 100 }, +}); + +type Wire = ExtractListenerResponseType; // Trade +type Delivered = ExtractListenerDeliveredType; // Trade[] + +onTrade.listen(({ data }) => { + // data: Trade[] +}); +``` + +Reusable abstractions keep working: a batched listener still satisfies `ListenerInstance<{ response: Trade }>`. When +you need to constrain on what callbacks receive, use the `delivered` field, e.g. +`ListenerInstance<{ delivered: Trade[] }>`. + --- ## See More diff --git a/documentation/docs/sockets/sdk.mdx b/documentation/docs/sockets/sdk.mdx index 94bf2d1b..140f747b 100644 --- a/documentation/docs/sockets/sdk.mdx +++ b/documentation/docs/sockets/sdk.mdx @@ -107,6 +107,26 @@ solve opposite problems. Picking the right one is the difference between strict | **Omitted fields default to** | Strict, non-`any` (`unknown`, `undefined`, `string`, `false`). | `any` — on purpose, so any concrete `Listener` / `Emitter` satisfies the constraint. | | **Why those defaults?** | Absence means "**there is no payload / no params**". The type should reject mismatches. | Absence means "**I don't care about that field**". The type should accept anything. | +:::info `delivered` field + +`ListenerModel` also accepts an optional `delivered` field - the type `listen` callbacks receive. It defaults to +`response` and only needs to be set when a `delivery` strategy changes the shape, because runtime defaults cannot +change schema types: + +```ts +type Schema = { + trades: { + $listener: ListenerModel<{ response: Trade; topic: "trades"; delivered: Trade[] }>; + }; +}; + +const configured = sdk.$configure({ + trades: { delivery: { strategy: "batch", interval: 100 } }, +}); +``` + +::: + ```ts // MODEL — defining the chat/messages topic exactly: // response is { text, user } @@ -224,6 +244,7 @@ const configured = sdk.$configure({ | Key | Type | Description | |-----|------|-------------| | `options` | `Record` | Adapter-specific options passed to `setOptions` | +| `delivery` | `ListenerDeliveryType` | Delivery strategy passed to `setDelivery` on listeners (`latest` / `batch` preset or a custom strategy). Ignored for emitters. | ### Application Order diff --git a/packages/react/__tests__/features/use-fetch/use-fetch.revalidate.spec.ts b/packages/react/__tests__/features/use-fetch/use-fetch.revalidate.spec.ts index 8a394871..ca844361 100644 --- a/packages/react/__tests__/features/use-fetch/use-fetch.revalidate.spec.ts +++ b/packages/react/__tests__/features/use-fetch/use-fetch.revalidate.spec.ts @@ -3,7 +3,7 @@ import { createHttpMockingServer, sleep } from "@hyper-fetch/testing"; import { act, waitFor } from "@testing-library/react"; import { testSuccessState } from "../../shared"; -import { client, createRequest, renderUseFetch, waitForRender } from "../../utils"; +import { client, createRequest, renderUseFetch } from "../../utils"; const { resetMocks, startServer, stopServer, mockRequest } = createHttpMockingServer(); @@ -74,7 +74,11 @@ describe("useFetch [ refetch ]", () => { }); renderUseFetch(request, { revalidate: false }); - await waitForRender(50); + // Wait for the request to finish instead of sleeping a fixed time - the mock delay can exceed it under load. + await waitFor(() => { + expect(spy).toHaveBeenCalledTimes(1); + }); + await sleep(50); expect(spy).toHaveBeenCalledTimes(1); }); it("should allow to refetch on mount", async () => { diff --git a/packages/react/__tests__/features/use-listener/use-listener.delivery.spec.tsx b/packages/react/__tests__/features/use-listener/use-listener.delivery.spec.tsx new file mode 100644 index 00000000..e763251b --- /dev/null +++ b/packages/react/__tests__/features/use-listener/use-listener.delivery.spec.tsx @@ -0,0 +1,330 @@ +import type { ExtractListenerDeliveredType } from "@hyper-fetch/sockets"; +import { createDeliveryStrategy } from "@hyper-fetch/sockets"; +import { createWebsocketMockingServer, waitForConnection } from "@hyper-fetch/testing"; +import { act, renderHook } from "@testing-library/react"; +import { useListener } from "hooks/use-listener"; +import React, { useMemo, useState } from "react"; +import { expectTypeOf } from "vitest"; + +import { createListener } from "../../utils/listener.utils"; +import { renderUseListener } from "../../utils/use-listener.utils"; + +type Trade = { id: number }; + +describe("useListener [ Delivery ]", () => { + const { startServer, stopServer, emitListenerEvent } = createWebsocketMockingServer(); + let listener = createListener(); + + beforeEach(async () => { + startServer(); + listener = createListener(); + await waitForConnection(listener.socket); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + }); + + afterEach(() => { + vi.useRealTimers(); + stopServer(); + }); + + it("should deliver batched arrays to state and onEvent", async () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + const spy = vi.fn(); + const view = renderUseListener(batched); + act(() => { + view.result.current.onEvent(({ data }) => spy(data)); + }); + + act(() => { + emitListenerEvent(batched, { id: 1 }); + emitListenerEvent(batched, { id: 2 }); + }); + expect(spy).not.toHaveBeenCalled(); + expect(view.result.current.data).toBeNull(); + + act(() => { + vi.advanceTimersByTime(100); + }); + expect(spy).toHaveBeenCalledOnce(); + expect(spy).toHaveBeenCalledWith([{ id: 1 }, { id: 2 }]); + expect(view.result.current.data).toEqual([{ id: 1 }, { id: 2 }]); + }); + + it("should sample with the latest strategy", async () => { + const sampled = listener.setDelivery({ strategy: "latest", interval: 100 }); + const view = renderUseListener(sampled); + + act(() => { + emitListenerEvent(sampled, { id: 1 }); + emitListenerEvent(sampled, { id: 2 }); + emitListenerEvent(sampled, { id: 3 }); + }); + expect(view.result.current.data).toEqual({ id: 1 }); + + act(() => { + vi.advanceTimersByTime(100); + }); + expect(view.result.current.data).toEqual({ id: 3 }); + }); + + it("should not update state after unmount", async () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => null); + const spy = vi.fn(); + const view = renderUseListener(batched); + act(() => { + view.result.current.onEvent(spy); + }); + act(() => { + emitListenerEvent(batched, { id: 1 }); + }); + view.unmount(); + act(() => { + vi.advanceTimersByTime(500); + }); + expect(spy).not.toHaveBeenCalled(); + expect(errorSpy).not.toHaveBeenCalled(); + errorSpy.mockRestore(); + }); + + it("should keep working in StrictMode", async () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + const view = renderHook(() => useListener(batched, { dependencyTracking: false }), { + wrapper: ({ children }) => {children}, + }); + act(() => { + emitListenerEvent(batched, { id: 1 }); + vi.advanceTimersByTime(100); + }); + expect(view.result.current.data).toEqual([{ id: 1 }]); + expect(batched.socket.adapter.listeners.get(batched.topic)?.size).toBe(1); + }); + + it("should re-subscribe when delivery changes", async () => { + const view = renderHook(() => { + const [rate, setRate] = useState(100); + const live = useMemo(() => listener.setDelivery({ strategy: "latest", interval: rate }), [rate]); + const result = useListener(live, { dependencyTracking: false }); + return { ...result, setRate, live }; + }); + + act(() => { + emitListenerEvent(listener, { id: 1 }); + emitListenerEvent(listener, { id: 2 }); + }); + expect(view.result.current.data).toEqual({ id: 1 }); + + act(() => { + view.result.current.setRate(10); + }); + // The previous window (100ms) was disposed with the old subscription - a new one leads immediately. + act(() => { + emitListenerEvent(listener, { id: 3 }); + }); + expect(view.result.current.data).toEqual({ id: 3 }); + expect(listener.socket.adapter.listeners.get(listener.topic)?.size).toBe(1); + }); + + it("should keep one subscription when an equal preset is created inline on every render", () => { + const spy = vi.fn(); + const view = renderHook(() => { + const inline = listener.setDelivery({ strategy: "batch", interval: 100 }); + return useListener(inline, { dependencyTracking: false }); + }); + act(() => { + view.result.current.onEvent(({ data }) => spy(data)); + }); + act(() => { + emitListenerEvent(listener, { id: 1 }); + }); + view.rerender(); + view.rerender(); + act(() => { + emitListenerEvent(listener, { id: 2 }); + }); + expect(listener.socket.adapter.listeners.get(listener.topic)?.size).toBe(1); + act(() => { + vi.advanceTimersByTime(100); + }); + // Buffered message survived the re-renders because the subscription was not recreated. + expect(spy).toHaveBeenCalledOnce(); + expect(spy).toHaveBeenCalledWith([{ id: 1 }, { id: 2 }]); + }); + + it("should re-subscribe when a custom strategy identity changes", () => { + const makeStrategy = () => + createDeliveryStrategy(({ deliver }) => ({ + push: ({ data, extra }) => deliver({ data: data.id * 10, extra }), + dispose: () => null, + })); + const first = makeStrategy(); + const second = makeStrategy(); + const view = renderHook( + ({ strategy }: { strategy: ReturnType }) => + useListener(listener.setDelivery(strategy), { dependencyTracking: false }), + { initialProps: { strategy: first } }, + ); + act(() => { + emitListenerEvent(listener, { id: 1 }); + }); + expect(view.result.current.data).toBe(10); + view.rerender({ strategy: second }); + expect(listener.socket.adapter.listeners.get(listener.topic)?.size).toBe(1); + act(() => { + emitListenerEvent(listener, { id: 2 }); + }); + expect(view.result.current.data).toBe(20); + }); + + it("should render once per flush, not once per message", () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + let renders = 0; + renderHook(() => { + renders += 1; + return useListener(batched, { dependencyTracking: false }); + }); + const initial = renders; + act(() => { + for (let i = 1; i <= 50; i += 1) { + emitListenerEvent(batched, { id: i }); + } + }); + expect(renders).toBe(initial); + act(() => { + vi.advanceTimersByTime(100); + }); + expect(renders).toBe(initial + 1); + }); + + it("should deliver several windows in order to onEvent", () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + const spy = vi.fn(); + const view = renderUseListener(batched); + act(() => { + view.result.current.onEvent(({ data }) => spy(data)); + }); + act(() => { + emitListenerEvent(batched, { id: 1 }); + vi.advanceTimersByTime(100); + emitListenerEvent(batched, { id: 2 }); + emitListenerEvent(batched, { id: 3 }); + vi.advanceTimersByTime(100); + emitListenerEvent(batched, { id: 4 }); + vi.advanceTimersByTime(100); + }); + expect(spy.mock.calls.map(([d]) => d)).toEqual([[{ id: 1 }], [{ id: 2 }, { id: 3 }], [{ id: 4 }]]); + expect(view.result.current.data).toEqual([{ id: 4 }]); + }); + + it("should set extra and timestamp on flush only", () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + const view = renderUseListener(batched); + act(() => { + emitListenerEvent(batched, { id: 1 }); + }); + expect(view.result.current.extra).toBeNull(); + expect(view.result.current.timestamp).toBeNull(); + act(() => { + vi.advanceTimersByTime(100); + }); + expect(view.result.current.extra).toBeObject(); + expect(view.result.current.timestamp).toBeNumber(); + }); + + it("should drop the buffer when listen() is called manually again", () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + const spy = vi.fn(); + const view = renderUseListener(batched); + act(() => { + view.result.current.onEvent(({ data }) => spy(data)); + }); + act(() => { + emitListenerEvent(batched, { id: 1 }); + }); + act(() => { + view.result.current.listen(); + }); + expect(batched.socket.adapter.listeners.get(batched.topic)?.size).toBe(1); + act(() => { + emitListenerEvent(batched, { id: 2 }); + vi.advanceTimersByTime(100); + }); + expect(spy).toHaveBeenCalledOnce(); + expect(spy).toHaveBeenCalledWith([{ id: 2 }]); + }); + + it("should work with a custom strategy", () => { + const summarize = createDeliveryStrategy(({ deliver }) => { + let count = 0; + let timer: ReturnType | undefined; + return { + push: ({ extra }) => { + count += 1; + timer ??= setTimeout(() => { + timer = undefined; + deliver({ data: { count }, extra }); + count = 0; + }, 50); + }, + dispose: () => clearTimeout(timer), + }; + }); + const summarized = listener.setDelivery(summarize); + const view = renderUseListener(summarized); + act(() => { + emitListenerEvent(listener, { id: 1 }); + emitListenerEvent(listener, { id: 2 }); + emitListenerEvent(listener, { id: 3 }); + vi.advanceTimersByTime(50); + }); + expect(view.result.current.data).toEqual({ count: 3 }); + }); + + it("should respect leading: false with the latest strategy", () => { + const sampled = listener.setDelivery({ strategy: "latest", interval: 100, leading: false }); + const view = renderUseListener(sampled); + act(() => { + emitListenerEvent(sampled, { id: 1 }); + emitListenerEvent(sampled, { id: 2 }); + }); + expect(view.result.current.data).toBeNull(); + act(() => { + vi.advanceTimersByTime(100); + }); + expect(view.result.current.data).toEqual({ id: 2 }); + }); + + it("should work with dependency tracking enabled", () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + const view = renderUseListener(batched, { dependencyTracking: true }); + // read data to track it + expect(view.result.current.data).toBeNull(); + act(() => { + emitListenerEvent(batched, { id: 1 }); + vi.advanceTimersByTime(100); + }); + expect(view.result.current.data).toEqual([{ id: 1 }]); + }); + + it("should clean up the adapter registration on unmount", () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + const view = renderUseListener(batched); + expect(batched.socket.adapter.listeners.get(batched.topic)?.size).toBe(1); + view.unmount(); + expect(batched.socket.adapter.listeners.get(batched.topic)?.size).toBe(0); + }); + + it("should type data by the delivered type", () => { + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + expectTypeOf>().toEqualTypeOf(); + const useTyped = () => { + const { data, onEvent } = useListener(batched); + expectTypeOf(data).toEqualTypeOf(); + onEvent(({ data: received }) => expectTypeOf(received).toEqualTypeOf()); + const plain = listener.socket.createListener()({ topic: "plain" }); + const { data: plainData } = useListener(plain); + expectTypeOf(plainData).toEqualTypeOf(); + }; + expect(useTyped).toBeFunction(); + }); +}); diff --git a/packages/react/__tests__/features/use-submit/use-submit.queue.spec.ts b/packages/react/__tests__/features/use-submit/use-submit.queue.spec.ts index fbf6e5e8..17c5878b 100644 --- a/packages/react/__tests__/features/use-submit/use-submit.queue.spec.ts +++ b/packages/react/__tests__/features/use-submit/use-submit.queue.spec.ts @@ -55,7 +55,8 @@ describe("useSubmit [ Queue ]", () => { expect(spy).toHaveBeenCalledTimes(4); }); it("should start in loading mode when request in queue is ongoing", async () => { - mockRequest(request); + // Keep the request in flight long enough for the second hook to mount while it is still ongoing. + mockRequest(request, { delay: 500 }); const previouslyRenderedHook = renderUseSubmit(request); act(() => { @@ -67,7 +68,7 @@ describe("useSubmit [ Queue ]", () => { await testLoading(true, response); }); it("should not start in loading mode when queue is paused", async () => { - mockRequest(request); + mockRequest(request, { delay: 500 }); const previouslyRenderedHook = renderUseSubmit(request); act(() => { diff --git a/packages/react/src/hooks/use-listener/use-listener.hooks.ts b/packages/react/src/hooks/use-listener/use-listener.hooks.ts index cc408b81..adf6c423 100644 --- a/packages/react/src/hooks/use-listener/use-listener.hooks.ts +++ b/packages/react/src/hooks/use-listener/use-listener.hooks.ts @@ -1,5 +1,6 @@ import { useDidUpdate, useWillUnmount } from "@better-hooks/lifecycle"; -import type { ListenerInstance, ExtractListenerResponseType } from "@hyper-fetch/sockets"; +import type { ListenerInstance, ExtractListenerDeliveredType } from "@hyper-fetch/sockets"; +import { getDeliveryKey } from "@hyper-fetch/sockets"; import { useSocketState } from "helpers"; import type { UseListenerOptionsType } from "hooks/use-listener"; import { useProvider } from "provider"; @@ -13,7 +14,10 @@ export const useListener = ( ) => { const { config: globalConfig } = useProvider(); const { dependencyTracking } = { ...globalConfig.useListener, ...options }; - const [state, actions, callbacks, { setRenderKey }] = useSocketState(listener.socket, { dependencyTracking }); + const [state, actions, callbacks, { setRenderKey }] = useSocketState< + ExtractListenerDeliveredType, + ListenerType["socket"] + >(listener.socket, { dependencyTracking }); const removeListenerRef = useRef | null>(null); /** @@ -21,7 +25,7 @@ export const useListener = ( */ const onEventCallback = useRef< - null | ((response: { data: ExtractListenerResponseType; extra: Record }) => void) + null | ((response: { data: ExtractListenerDeliveredType; extra: Record }) => void) >(null); /** @@ -56,7 +60,8 @@ export const useListener = ( () => { listen(); }, - [listener.params, JSON.stringify(listener.options)], + // Topic already contains resolved params. Delivery presets compare by value, custom strategies by identity. + [listener.topic, JSON.stringify(listener.options), getDeliveryKey(listener.delivery)], true, ); diff --git a/packages/sockets/__tests__/e2e/e2e.delivery.spec.ts b/packages/sockets/__tests__/e2e/e2e.delivery.spec.ts new file mode 100644 index 00000000..390e1108 --- /dev/null +++ b/packages/sockets/__tests__/e2e/e2e.delivery.spec.ts @@ -0,0 +1,81 @@ +/** + * @vitest-environment node + */ +import { Socket } from "@hyper-fetch/sockets"; +import { createWebsocketE2EServer, sleep, waitForConnection } from "@hyper-fetch/testing"; + +const wsServer = createWebsocketE2EServer(); + +describe("E2E [ WebSocket Delivery ]", () => { + let url: string; + + beforeAll(async () => { + url = await wsServer.startServer(); + }); + + afterAll(async () => { + await wsServer.stopServer(); + }); + + it("should batch a real stream without losing messages", async () => { + const socket = new Socket({ url }); + await waitForConnection(socket); + const listener = socket.createListener<{ id: number }>()({ + topic: "trades", + delivery: { strategy: "batch", interval: 100 }, + }); + const batches: { id: number }[][] = []; + listener.listen(({ data }) => batches.push(data)); + await sleep(50); + + for (let i = 1; i <= 30; i += 1) { + wsServer.sendToAll("trades", { id: i }); + } + await sleep(400); + + expect(batches.length).toBeGreaterThanOrEqual(1); + expect(batches.length).toBeLessThan(30); + expect(batches.flat().map((m) => m.id)).toEqual(Array.from({ length: 30 }, (_, i) => i + 1)); + await socket.disconnect(); + }); + + it("should sample a real stream and end on the newest message", async () => { + const socket = new Socket({ url }); + await waitForConnection(socket); + const listener = socket.createListener<{ id: number }>()({ + topic: "prices", + delivery: { strategy: "latest", interval: 100 }, + }); + const received: number[] = []; + listener.listen(({ data }) => received.push(data.id)); + await sleep(50); + + for (let i = 1; i <= 30; i += 1) { + wsServer.sendToAll("prices", { id: i }); + } + await sleep(400); + + expect(received.length).toBeLessThan(30); + expect(received[0]).toBe(1); + expect(received.at(-1)).toBe(30); + await socket.disconnect(); + }); + + it("should deliver nothing after unsubscribe", async () => { + const socket = new Socket({ url }); + await waitForConnection(socket); + const listener = socket.createListener<{ id: number }>()({ + topic: "late", + delivery: { strategy: "batch", interval: 100 }, + }); + const spy = vi.fn(); + const stop = listener.listen(spy); + await sleep(50); + wsServer.sendToAll("late", { id: 1 }); + await sleep(20); + stop(); + await sleep(300); + expect(spy).not.toHaveBeenCalled(); + await socket.disconnect(); + }); +}); diff --git a/packages/sockets/__tests__/features/delivery/delivery.batch.spec.ts b/packages/sockets/__tests__/features/delivery/delivery.batch.spec.ts new file mode 100644 index 00000000..94a64897 --- /dev/null +++ b/packages/sockets/__tests__/features/delivery/delivery.batch.spec.ts @@ -0,0 +1,203 @@ +import { createBatchDeliveryStrategy } from "delivery"; + +type Message = { data: number; extra: { id: number } }; + +const message = (data: number): Message => ({ data, extra: { id: data } }); + +describe("Delivery [ Batch ]", () => { + const setup = (options: { interval: number; maxSize?: number }) => { + const deliver = vi.fn(); + const controller = createBatchDeliveryStrategy(options)({ deliver }); + return { deliver, controller }; + }; + + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("should deliver all messages in order when the window closes", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + controller.push(message(2)); + controller.push(message(3)); + expect(deliver).not.toHaveBeenCalled(); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledOnce(); + expect(deliver).toHaveBeenCalledWith({ data: [1, 2, 3], extra: { id: 3 } }); + }); + + it("should measure the window from the first message", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + vi.advanceTimersByTime(60); + controller.push(message(2)); + vi.advanceTimersByTime(40); // t=100 since first + expect(deliver).toHaveBeenCalledOnce(); + expect(deliver).toHaveBeenCalledWith({ data: [1, 2], extra: { id: 2 } }); + }); + + it("should flush early at maxSize and cancel the timer", () => { + const { deliver, controller } = setup({ interval: 100, maxSize: 2 }); + controller.push(message(1)); + controller.push(message(2)); + expect(deliver).toHaveBeenCalledOnce(); + expect(deliver).toHaveBeenCalledWith({ data: [1, 2], extra: { id: 2 } }); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledOnce(); // the timer was cancelled, no empty batch + }); + + it("should open a new window after a flush", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + vi.advanceTimersByTime(100); + controller.push(message(2)); + expect(deliver).toHaveBeenCalledOnce(); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenLastCalledWith({ data: [2], extra: { id: 2 } }); + }); + + it("should never deliver an empty batch", () => { + const { deliver } = setup({ interval: 100 }); + vi.advanceTimersByTime(1000); + expect(deliver).not.toHaveBeenCalled(); + }); + + it("should drop the buffer on dispose", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + controller.push(message(2)); + controller.dispose(); + controller.dispose(); // idempotent + vi.advanceTimersByTime(500); + expect(deliver).not.toHaveBeenCalled(); + }); + + it("should ignore messages pushed after dispose", () => { + const { deliver, controller } = setup({ interval: 100, maxSize: 1 }); + controller.dispose(); + controller.push(message(1)); + expect(deliver).not.toHaveBeenCalled(); + }); + + it("should split a burst into ceil(n / maxSize) batches", () => { + const { deliver, controller } = setup({ interval: 100, maxSize: 10 }); + for (let i = 1; i <= 25; i += 1) { + controller.push(message(i)); + } + expect(deliver).toHaveBeenCalledTimes(2); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledTimes(3); + expect(deliver.mock.calls.map(([{ data }]) => data.length)).toEqual([10, 10, 5]); + }); + + it("should allow disposing from inside the callback", () => { + const deliver = vi.fn(); + const controller = createBatchDeliveryStrategy({ interval: 100 })({ + deliver: (msg) => { + deliver(msg); + controller.dispose(); + }, + }); + controller.push(message(1)); + vi.advanceTimersByTime(100); + controller.push(message(2)); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledOnce(); + }); + + it("should deliver every message immediately with maxSize 1", () => { + const { deliver, controller } = setup({ interval: 100, maxSize: 1 }); + controller.push(message(1)); + controller.push(message(2)); + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliver.mock.calls.map(([m]) => m.data)).toEqual([[1], [2]]); + expect(vi.getTimerCount()).toBe(0); + }); + + it("should flush exactly at the maxSize boundary", () => { + const { deliver, controller } = setup({ interval: 100, maxSize: 3 }); + controller.push(message(1)); + controller.push(message(2)); + expect(deliver).not.toHaveBeenCalled(); + controller.push(message(3)); + expect(deliver).toHaveBeenCalledWith({ data: [1, 2, 3], extra: { id: 3 } }); + controller.push(message(4)); + controller.push(message(5)); + expect(deliver).toHaveBeenCalledOnce(); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenLastCalledWith({ data: [4, 5], extra: { id: 5 } }); + }); + + it("should keep a burst of 1000 messages in one ordered batch without maxSize", () => { + const { deliver, controller } = setup({ interval: 100 }); + const expected: number[] = []; + for (let i = 1; i <= 1000; i += 1) { + controller.push(message(i)); + expected.push(i); + } + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledOnce(); + expect(deliver.mock.calls[0][0].data).toEqual(expected); + }); + + it("should deliver a fresh array for each batch", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + vi.advanceTimersByTime(100); + controller.push(message(2)); + vi.advanceTimersByTime(100); + expect(deliver.mock.calls[0][0].data).not.toBe(deliver.mock.calls[1][0].data); + expect(deliver.mock.calls[0][0].data).toEqual([1]); + expect(deliver.mock.calls[1][0].data).toEqual([2]); + }); + + it("should keep working after the callback throws", () => { + const deliver = vi.fn().mockImplementationOnce(() => { + throw new Error("boom"); + }); + const controller = createBatchDeliveryStrategy({ interval: 100 })({ deliver }); + controller.push(message(1)); + expect(() => vi.advanceTimersByTime(100)).toThrow("boom"); + controller.push(message(2)); + controller.push(message(3)); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenLastCalledWith({ data: [2, 3], extra: { id: 3 } }); + }); + + it("should not leak timers after dispose", () => { + const { controller } = setup({ interval: 100 }); + controller.push(message(1)); + expect(vi.getTimerCount()).toBe(1); + controller.dispose(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("should give independent buffers to controllers created from one factory", () => { + const factory = createBatchDeliveryStrategy({ interval: 100 }); + const first = vi.fn(); + const second = vi.fn(); + const a = factory({ deliver: first }); + const b = factory({ deliver: second }); + a.push(message(1)); + b.push(message(10)); + a.push(message(2)); + vi.advanceTimersByTime(100); + expect(first).toHaveBeenCalledWith({ data: [1, 2], extra: { id: 2 } }); + expect(second).toHaveBeenCalledWith({ data: [10], extra: { id: 10 } }); + }); + + it("should flush pending messages with interval 0 on the next macrotask", () => { + const { deliver, controller } = setup({ interval: 0 }); + controller.push(message(1)); + controller.push(message(2)); + expect(deliver).not.toHaveBeenCalled(); + vi.advanceTimersByTime(0); + expect(deliver).toHaveBeenCalledWith({ data: [1, 2], extra: { id: 2 } }); + }); +}); diff --git a/packages/sockets/__tests__/features/delivery/delivery.latest.spec.ts b/packages/sockets/__tests__/features/delivery/delivery.latest.spec.ts new file mode 100644 index 00000000..cf4e4910 --- /dev/null +++ b/packages/sockets/__tests__/features/delivery/delivery.latest.spec.ts @@ -0,0 +1,254 @@ +import { createLatestDeliveryStrategy } from "delivery"; + +type Message = { data: number; extra: { id: number } }; + +const message = (data: number): Message => ({ data, extra: { id: data } }); + +describe("Delivery [ Latest ]", () => { + const setup = (options: { interval: number; leading?: boolean }) => { + const deliver = vi.fn(); + const controller = createLatestDeliveryStrategy(options)({ deliver }); + return { deliver, controller }; + }; + + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("should deliver the first message immediately", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + expect(deliver).toHaveBeenCalledOnce(); + expect(deliver).toHaveBeenCalledWith(message(1)); + }); + + it("should deliver only the newest message when the window closes", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + controller.push(message(2)); + controller.push(message(3)); + expect(deliver).toHaveBeenCalledOnce(); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenLastCalledWith(message(3)); + }); + + it("should not deliver a trailing message when nothing followed the leading one", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + vi.advanceTimersByTime(500); + expect(deliver).toHaveBeenCalledOnce(); + }); + + it("should open the next window after a trailing delivery", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); // t=0 leading + controller.push(message(2)); // pending + vi.advanceTimersByTime(100); // t=100 trailing → 2, new window + expect(deliver).toHaveBeenLastCalledWith(message(2)); + controller.push(message(3)); // inside the new window → must wait + expect(deliver).toHaveBeenCalledTimes(2); + vi.advanceTimersByTime(100); // t=200 → 3 + expect(deliver).toHaveBeenCalledTimes(3); + expect(deliver).toHaveBeenLastCalledWith(message(3)); + }); + + it("should lead again after going idle", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + vi.advanceTimersByTime(100); // window closes with nothing pending → idle + controller.push(message(2)); + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenLastCalledWith(message(2)); + }); + + it("should wait for the interval when leading is disabled", () => { + const { deliver, controller } = setup({ interval: 100, leading: false }); + controller.push(message(1)); + controller.push(message(2)); + expect(deliver).not.toHaveBeenCalled(); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledOnce(); + expect(deliver).toHaveBeenCalledWith(message(2)); + }); + + it("should deliver the extra of the delivered message", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + controller.push(message(2)); + controller.push(message(3)); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenLastCalledWith({ data: 3, extra: { id: 3 } }); + }); + + it("should drop pending messages on dispose", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + controller.push(message(2)); + controller.dispose(); + controller.dispose(); // idempotent + vi.advanceTimersByTime(500); + expect(deliver).toHaveBeenCalledOnce(); + }); + + it("should ignore messages pushed after dispose", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.dispose(); + controller.push(message(1)); + vi.advanceTimersByTime(500); + expect(deliver).not.toHaveBeenCalled(); + }); + + it("should allow disposing from inside the callback", () => { + const deliver = vi.fn(); + const controller = createLatestDeliveryStrategy({ interval: 100 })({ + deliver: (msg) => { + deliver(msg); + controller.dispose(); + }, + }); + controller.push(message(1)); // leading → disposes + controller.push(message(2)); + vi.advanceTimersByTime(500); + expect(deliver).toHaveBeenCalledOnce(); + }); + + it("should coalesce messages per macrotask with interval 0", () => { + const { deliver, controller } = setup({ interval: 0 }); + controller.push(message(1)); + controller.push(message(2)); + controller.push(message(3)); + expect(deliver).toHaveBeenCalledOnce(); + vi.advanceTimersByTime(0); + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenLastCalledWith(message(3)); + }); + + it("should deliver exactly leading + trailing for a burst of 1000 messages", () => { + const { deliver, controller } = setup({ interval: 100 }); + for (let i = 1; i <= 1000; i += 1) { + controller.push(message(i)); + } + expect(deliver).toHaveBeenCalledOnce(); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenLastCalledWith(message(1000)); + vi.advanceTimersByTime(1000); + expect(deliver).toHaveBeenCalledTimes(2); + }); + + it("should preserve order across many windows", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + controller.push(message(2)); + vi.advanceTimersByTime(100); + controller.push(message(3)); + controller.push(message(4)); + vi.advanceTimersByTime(100); + controller.push(message(5)); + vi.advanceTimersByTime(100); + vi.advanceTimersByTime(100); // idle + controller.push(message(6)); + expect(deliver.mock.calls.map(([m]) => m.data)).toEqual([1, 2, 4, 5, 6]); + }); + + it("should treat a message arriving right after an idle window as leading", () => { + const { deliver, controller } = setup({ interval: 100 }); + controller.push(message(1)); + vi.advanceTimersByTime(100); // closes with nothing pending → idle + controller.push(message(2)); + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenLastCalledWith(message(2)); + }); + + it("should keep working after the callback throws", () => { + const deliver = vi.fn().mockImplementationOnce(() => { + throw new Error("boom"); + }); + const controller = createLatestDeliveryStrategy({ interval: 100 })({ deliver }); + expect(() => controller.push(message(1))).toThrow("boom"); + controller.push(message(2)); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledTimes(2); + expect(deliver).toHaveBeenLastCalledWith(message(2)); + controller.push(message(3)); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenLastCalledWith(message(3)); + }); + + it("should keep working after a trailing callback throws", () => { + const deliver = vi.fn(); + const controller = createLatestDeliveryStrategy({ interval: 100 })({ deliver }); + controller.push(message(1)); + controller.push(message(2)); + deliver.mockImplementationOnce(() => { + throw new Error("boom"); + }); + expect(() => vi.advanceTimersByTime(100)).toThrow("boom"); + controller.push(message(3)); // still inside the window re-armed before the throw + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenLastCalledWith(message(3)); + }); + + it("should go idle and lead again when leading is disabled", () => { + const { deliver, controller } = setup({ interval: 100, leading: false }); + controller.push(message(1)); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledWith(message(1)); + vi.advanceTimersByTime(100); // nothing pending → idle + controller.push(message(2)); + expect(deliver).toHaveBeenCalledOnce(); // not immediate + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenLastCalledWith(message(2)); + }); + + it("should deliver the same object reference that was pushed", () => { + const { deliver, controller } = setup({ interval: 100 }); + const first = message(1); + const last = message(3); + controller.push(first); + controller.push(message(2)); + controller.push(last); + vi.advanceTimersByTime(100); + expect(deliver.mock.calls[0][0]).toBe(first); + expect(deliver.mock.calls[1][0]).toBe(last); + }); + + it("should be a no-op to dispose while idle", () => { + const { deliver, controller } = setup({ interval: 100 }); + expect(() => controller.dispose()).not.toThrow(); + vi.advanceTimersByTime(100); + expect(deliver).not.toHaveBeenCalled(); + }); + + it("should not create timers before the first message", () => { + setup({ interval: 100 }); + expect(vi.getTimerCount()).toBe(0); + }); + + it("should clear all timers on dispose", () => { + const { controller } = setup({ interval: 100 }); + controller.push(message(1)); + expect(vi.getTimerCount()).toBe(1); + controller.dispose(); + expect(vi.getTimerCount()).toBe(0); + }); + + it("should give independent state to controllers created from one factory", () => { + const factory = createLatestDeliveryStrategy({ interval: 100 }); + const first = vi.fn(); + const second = vi.fn(); + const a = factory({ deliver: first }); + const b = factory({ deliver: second }); + a.push(message(1)); + a.push(message(2)); + b.push(message(10)); + vi.advanceTimersByTime(100); + expect(first.mock.calls.map(([m]) => m.data)).toEqual([1, 2]); + expect(second.mock.calls.map(([m]) => m.data)).toEqual([10]); + }); +}); diff --git a/packages/sockets/__tests__/features/delivery/delivery.typescript.spec.ts b/packages/sockets/__tests__/features/delivery/delivery.typescript.spec.ts new file mode 100644 index 00000000..def71b56 --- /dev/null +++ b/packages/sockets/__tests__/features/delivery/delivery.typescript.spec.ts @@ -0,0 +1,143 @@ +import type { ListenerDeliveryType } from "delivery"; +import { createDeliveryStrategy } from "delivery"; +import type { ExtendListener, Listener, ListenerInstance, ListenerModel } from "listener"; +import { Socket } from "socket"; +import type { ExtractListenerDeliveredType, ExtractListenerResponseType } from "types"; +import { expectTypeOf } from "vitest"; + +type Price = { symbol: string; value: number }; +type Summary = { count: number; last: Price }; + +const socket = new Socket({ url: "ws://localhost:3000" }); + +const summarize = createDeliveryStrategy(({ deliver }) => ({ + push: ({ data, extra }) => deliver({ data: { count: 1, last: data }, extra }), + dispose: () => null, +})); + +const plain = socket.createListener()({ topic: "prices" }); +const latest = socket.createListener()({ topic: "prices", delivery: { strategy: "latest", interval: 16 } }); +const batch = socket.createListener()({ topic: "prices", delivery: { strategy: "batch", interval: 100 } }); +const custom = socket.createListener()({ topic: "prices", delivery: summarize }); + +describe("Delivery [ TypeScript ]", () => { + it("should type callback data by strategy", () => { + plain.listen(({ data }) => expectTypeOf(data).toEqualTypeOf()); + latest.listen(({ data }) => expectTypeOf(data).toEqualTypeOf()); + batch.listen(({ data }) => expectTypeOf(data).toEqualTypeOf()); + custom.listen(({ data }) => expectTypeOf(data).toEqualTypeOf()); + expect(true).toBeTrue(); + }); + + it("should keep the response type as the wire type", () => { + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expect(true).toBeTrue(); + }); + + it("should re-type through setDelivery", () => { + const rebatched = plain.setDelivery({ strategy: "batch", interval: 50 }); + expectTypeOf>().toEqualTypeOf(); + const cleared = batch.setDelivery(undefined); + expectTypeOf>().toEqualTypeOf(); + const withParams = socket + .createListener()({ topic: "prices/:symbol", delivery: { strategy: "batch", interval: 50 } }) + .setParams({ symbol: "BTC" }); + expectTypeOf>().toEqualTypeOf(); + expect(true).toBeTrue(); + }); + + it("should satisfy ListenerInstance constraints regardless of delivery", () => { + expectTypeOf(batch).toMatchTypeOf(); + expectTypeOf(batch).toMatchTypeOf>(); + expectTypeOf(custom).toMatchTypeOf>(); + // covariance guard - partial constraints keep matching wider responses + expectTypeOf(plain).toMatchTypeOf>(); + expectTypeOf(batch).toMatchTypeOf>(); + expect(true).toBeTrue(); + }); + + it("should default ListenerModel delivered to response", () => { + type Model = ListenerModel<{ response: Price; topic: "prices" }>; + type Batched = ListenerModel<{ response: Price; topic: "prices"; delivered: Price[] }>; + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expect(true).toBeTrue(); + }); + + it("should follow the response override in ExtendListener", () => { + type Overridden = ExtendListener; + type Kept = ExtendListener; + type Explicit = ExtendListener; + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expect(true).toBeTrue(); + }); + + it("should reject invalid presets", () => { + const invalid = () => { + // @ts-expect-error unknown strategy + socket.createListener()({ topic: "prices", delivery: { strategy: "nope", interval: 1 } }); + // @ts-expect-error interval is required + socket.createListener()({ topic: "prices", delivery: { strategy: "latest" } }); + }; + expect(invalid).toBeFunction(); + }); + + it("should accept preset options", () => { + const withLeading = socket.createListener()({ + topic: "prices", + delivery: { strategy: "latest", interval: 16, leading: false }, + }); + const withMax = socket.createListener()({ + topic: "prices", + delivery: { strategy: "batch", interval: 16, maxSize: 10 }, + }); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + expect(true).toBeTrue(); + }); + + it("should type setDelivery with a custom strategy", () => { + const summarized = plain.setDelivery(summarize); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf>().toEqualTypeOf(); + summarized.listen(({ data }) => expectTypeOf(data).toEqualTypeOf()); + expect(true).toBeTrue(); + }); + + it("should reject a strategy whose input does not match the response", () => { + const wrongInput = createDeliveryStrategy<{ other: string }, number>(({ deliver }) => ({ + push: ({ extra }) => deliver({ data: 1, extra }), + dispose: () => null, + })); + // @ts-expect-error input type mismatch + const invalid = () => socket.createListener()({ topic: "prices", delivery: wrongInput }); + expect(invalid).toBeFunction(); + }); + + it("should keep the delivered type through clone and params", () => { + const cloned = batch.clone(); + expectTypeOf>().toEqualTypeOf(); + const dynamic = socket.createListener()({ topic: "prices/:id", delivery: summarize }); + const bound = dynamic.setParams({ id: 1 }); + expectTypeOf>().toEqualTypeOf(); + bound.listen(({ data }) => expectTypeOf(data).toEqualTypeOf()); + expect(true).toBeTrue(); + }); + + it("should expose delivery with a loose type on the instance", () => { + expectTypeOf(batch.delivery).toMatchTypeOf(); + expectTypeOf(batch.listenerOptions.delivery).toMatchTypeOf(); + expect(true).toBeTrue(); + }); + + it("should type the raw Listener generics", () => { + expectTypeOf>().toMatchTypeOf>(); + expect(true).toBeTrue(); + }); +}); diff --git a/packages/sockets/__tests__/features/delivery/delivery.utils.spec.ts b/packages/sockets/__tests__/features/delivery/delivery.utils.spec.ts new file mode 100644 index 00000000..f3458e09 --- /dev/null +++ b/packages/sockets/__tests__/features/delivery/delivery.utils.spec.ts @@ -0,0 +1,122 @@ +import type { DeliveryControllerType } from "delivery"; +import { + assertDelivery, + createDeliveryController, + createDeliveryStrategy, + getDeliveryKey, + isDeliveryStrategy, +} from "delivery"; + +describe("Delivery [ Utils ]", () => { + describe("createDeliveryStrategy", () => { + it("should return the given strategy untouched", () => { + const strategy = createDeliveryStrategy(({ deliver }) => ({ + push: ({ data, extra }) => deliver({ data: String(data), extra }), + dispose: () => null, + })); + expect(typeof strategy).toBe("function"); + expect(isDeliveryStrategy(strategy)).toBeTrue(); + expect(isDeliveryStrategy({ strategy: "latest", interval: 1 })).toBeFalse(); + }); + }); + + describe("assertDelivery", () => { + it("should accept presets and custom strategies", () => { + expect(() => assertDelivery({ strategy: "latest", interval: 10 })).not.toThrow(); + expect(() => assertDelivery({ strategy: "batch", interval: 10 })).not.toThrow(); + expect(() => assertDelivery(() => ({ push: () => null, dispose: () => null }))).not.toThrow(); + }); + it("should throw for an unknown preset", () => { + expect(() => assertDelivery({ strategy: "nope", interval: 10 } as any)).toThrow( + /Unknown delivery strategy: nope/, + ); + }); + }); + + describe("createDeliveryController", () => { + it("should call a custom strategy with the context", () => { + const controller: DeliveryControllerType = { push: vi.fn(), dispose: vi.fn() }; + const strategy = vi.fn(() => controller); + const deliver = vi.fn(); + const result = createDeliveryController(strategy, { deliver }); + expect(strategy).toHaveBeenCalledOnce(); + expect(strategy).toHaveBeenCalledWith({ deliver }); + expect(result).toBe(controller); + }); + it("should build the latest preset", () => { + const deliver = vi.fn(); + const controller = createDeliveryController({ strategy: "latest", interval: 100 }, { deliver }); + controller.push({ data: 1, extra: null }); + expect(deliver).toHaveBeenCalledWith({ data: 1, extra: null }); + controller.dispose(); + }); + it("should build the batch preset", () => { + vi.useFakeTimers(); + const deliver = vi.fn(); + const controller = createDeliveryController({ strategy: "batch", interval: 100 }, { deliver }); + controller.push({ data: 1, extra: null }); + controller.push({ data: 2, extra: null }); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledWith({ data: [1, 2], extra: null }); + controller.dispose(); + vi.useRealTimers(); + }); + }); + + describe("getDeliveryKey", () => { + it("should compare presets by value and strategies by identity", () => { + const strategy = () => ({ push: () => null, dispose: () => null }); + expect(getDeliveryKey()).toBe(""); + expect(getDeliveryKey({ strategy: "latest", interval: 16 })).toBe( + getDeliveryKey({ strategy: "latest", interval: 16 }), + ); + expect(getDeliveryKey({ strategy: "latest", interval: 16 })).not.toBe( + getDeliveryKey({ strategy: "latest", interval: 32 }), + ); + expect(getDeliveryKey(strategy)).toBe(strategy); + }); + }); + + describe("edge cases", () => { + it("should throw a helpful error when the strategy key is missing", () => { + expect(() => assertDelivery({ interval: 10 } as any)).toThrow(/Unknown delivery strategy: undefined/); + expect(() => assertDelivery({ interval: 10 } as any)).toThrow(/createDeliveryStrategy/); + }); + it("should build latest without leading", () => { + vi.useFakeTimers(); + const deliver = vi.fn(); + const controller = createDeliveryController({ strategy: "latest", interval: 100, leading: false }, { deliver }); + controller.push({ data: 1, extra: null }); + expect(deliver).not.toHaveBeenCalled(); + vi.advanceTimersByTime(100); + expect(deliver).toHaveBeenCalledWith({ data: 1, extra: null }); + controller.dispose(); + vi.useRealTimers(); + }); + it("should build batch with maxSize", () => { + const deliver = vi.fn(); + const controller = createDeliveryController({ strategy: "batch", interval: 100, maxSize: 2 }, { deliver }); + controller.push({ data: 1, extra: null }); + controller.push({ data: 2, extra: null }); + expect(deliver).toHaveBeenCalledWith({ data: [1, 2], extra: null }); + controller.dispose(); + }); + it("should produce different keys for different presets", () => { + expect(getDeliveryKey({ strategy: "latest", interval: 16 })).not.toBe( + getDeliveryKey({ strategy: "batch", interval: 16 }), + ); + expect(getDeliveryKey({ strategy: "batch", interval: 16 })).not.toBe( + getDeliveryKey({ strategy: "batch", interval: 16, maxSize: 5 }), + ); + expect(getDeliveryKey({ strategy: "latest", interval: 16, leading: false })).not.toBe( + getDeliveryKey({ strategy: "latest", interval: 16 }), + ); + }); + it("should produce different keys for different strategy functions", () => { + const a = () => ({ push: () => null, dispose: () => null }); + const b = () => ({ push: () => null, dispose: () => null }); + expect(getDeliveryKey(a)).not.toBe(getDeliveryKey(b)); + expect(getDeliveryKey(a)).toBe(getDeliveryKey(a)); + }); + }); +}); diff --git a/packages/sockets/__tests__/features/listener/listener.delivery.spec.ts b/packages/sockets/__tests__/features/listener/listener.delivery.spec.ts new file mode 100644 index 00000000..70ffa809 --- /dev/null +++ b/packages/sockets/__tests__/features/listener/listener.delivery.spec.ts @@ -0,0 +1,385 @@ +import { createWebsocketMockingServer, waitForConnection } from "@hyper-fetch/testing"; +import { createDeliveryStrategy } from "delivery"; + +import { createListener } from "../../utils/listener.utils"; +import { createSocket } from "../../utils/socket.utils"; + +type DataType = { price: number }; + +describe("Listener [ Delivery ]", () => { + const { startServer, stopServer, emitListenerEvent } = createWebsocketMockingServer(); + let socket = createSocket(); + + beforeEach(async () => { + startServer(); + socket = createSocket(); + await waitForConnection(socket); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + }); + + afterEach(() => { + vi.useRealTimers(); + stopServer(); + }); + + it("should register the original callback when no delivery is configured", () => { + const listener = createListener(socket); + const spy = vi.fn(); + const stop = listener.listen(spy); + expect(socket.adapter.listeners.get(listener.topic)?.has(spy)).toBeTrue(); + stop(); + expect(socket.adapter.listeners.get(listener.topic)?.has(spy)).toBeFalse(); + }); + + it("should register a wrapper and remove it on unsubscribe when delivery is configured", () => { + const listener = createListener(socket, { delivery: { strategy: "latest", interval: 100 } }); + const spy = vi.fn(); + const stop = listener.listen(spy); + const group = socket.adapter.listeners.get(listener.topic); + expect(group?.size).toBe(1); + expect(group?.has(spy)).toBeFalse(); + stop(); + expect(group?.size).toBe(0); + }); + + it("should sample messages with the latest strategy", () => { + const listener = createListener(socket, { delivery: { strategy: "latest", interval: 100 } }); + const spy = vi.fn(); + listener.listen(({ data }) => spy(data)); + + emitListenerEvent(listener, { price: 1 }); + emitListenerEvent(listener, { price: 2 }); + emitListenerEvent(listener, { price: 3 }); + expect(spy).toHaveBeenCalledOnce(); + expect(spy).toHaveBeenCalledWith({ price: 1 }); + + vi.advanceTimersByTime(100); + expect(spy).toHaveBeenCalledTimes(2); + expect(spy).toHaveBeenLastCalledWith({ price: 3 }); + }); + + it("should batch messages with the batch strategy", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const spy = vi.fn(); + listener.listen(({ data, extra }) => spy(data, extra)); + + emitListenerEvent(listener, { price: 1 }); + emitListenerEvent(listener, { price: 2 }); + expect(spy).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(100); + expect(spy).toHaveBeenCalledOnce(); + expect(spy.mock.calls[0][0]).toEqual([{ price: 1 }, { price: 2 }]); + expect(spy.mock.calls[0][1]).toHaveProperty("data"); + }); + + it("should not deliver after unsubscribe", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const spy = vi.fn(); + const stop = listener.listen(spy); + emitListenerEvent(listener, { price: 1 }); + stop(); + vi.advanceTimersByTime(500); + emitListenerEvent(listener, { price: 2 }); + vi.advanceTimersByTime(500); + expect(spy).not.toHaveBeenCalled(); + }); + + it("should run a custom strategy", () => { + const summarize = createDeliveryStrategy(({ deliver }) => { + let count = 0; + let max = -Infinity; + let timer: ReturnType | undefined; + return { + push: ({ data, extra }) => { + count += 1; + max = Math.max(max, data.price); + timer ??= setTimeout(() => { + timer = undefined; + deliver({ data: { count, max }, extra }); + count = 0; + max = -Infinity; + }, 50); + }, + dispose: () => clearTimeout(timer), + }; + }); + const listener = createListener(socket, { delivery: summarize }); + const spy = vi.fn(); + listener.listen(({ data }) => spy(data)); + + emitListenerEvent(listener, { price: 5 }); + emitListenerEvent(listener, { price: 9 }); + emitListenerEvent(listener, { price: 2 }); + vi.advanceTimersByTime(50); + expect(spy).toHaveBeenCalledOnce(); + expect(spy).toHaveBeenCalledWith({ count: 3, max: 9 }); + }); + + it("should give every listen() call its own controller", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const first = vi.fn(); + const second = vi.fn(); + const stopFirst = listener.listen(first); + listener.listen(second); + emitListenerEvent(listener, { price: 1 }); + stopFirst(); + vi.advanceTimersByTime(100); + expect(first).not.toHaveBeenCalled(); + expect(second).toHaveBeenCalledOnce(); + }); + + it("should carry delivery through setParams, setOptions and clone", () => { + const delivery = { strategy: "latest", interval: 16 } as const; + const listener = socket.createListener()({ topic: "prices/:symbol", delivery }); + expect(listener.setParams({ symbol: "BTC" }).delivery).toStrictEqual(delivery); + expect(listener.setOptions(undefined).delivery).toStrictEqual(delivery); + expect(listener.clone().delivery).toStrictEqual(delivery); + }); + + it("should return a clone from setDelivery and allow clearing it", () => { + const listener = createListener(socket); + const batched = listener.setDelivery({ strategy: "batch", interval: 100 }); + expect(batched).not.toBe(listener); + expect(listener.delivery).toBeUndefined(); + expect(batched.delivery).toStrictEqual({ strategy: "batch", interval: 100 }); + expect(batched.setDelivery(undefined).delivery).toBeUndefined(); + }); + + it("should throw on an unknown strategy at creation time", () => { + expect(() => + socket.createListener()({ topic: "prices", delivery: { strategy: "nope", interval: 1 } as any }), + ).toThrow(/Unknown delivery strategy/); + }); + + it("should wrap adapters that call the callback directly (Firebase-style)", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const spy = vi.fn(); + listener.listen(({ data }) => spy(data)); + const [registered] = [...socket.adapter.listeners.get(listener.topic)!.keys()]; + + // Simulate an adapter invoking the registered callback directly, bypassing triggerListeners. + registered({ data: { price: 1 }, extra: null as any }); + registered({ data: { price: 2 }, extra: null as any }); + expect(spy).not.toHaveBeenCalled(); + vi.advanceTimersByTime(100); + expect(spy).toHaveBeenCalledWith([{ price: 1 }, { price: 2 }]); + }); + + it("should run onMessage interceptors per raw message before batching", () => { + const interceptor = vi.fn(({ event }) => event); + socket.onMessage(interceptor); + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + listener.listen(() => null); + emitListenerEvent(listener, { price: 1 }); + emitListenerEvent(listener, { price: 2 }); + expect(interceptor).toHaveBeenCalledTimes(2); + }); + + it("should expose params on the listener", () => { + const listener = socket.createListener()({ topic: "prices/:symbol" }).setParams({ symbol: "BTC" }); + expect(listener.params).toStrictEqual({ symbol: "BTC" }); + expect(listener.topic).toBe("prices/BTC"); + }); + + it("should keep socket.events broadcasts unthrottled", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const callback = vi.fn(); + const byTopic = vi.fn(); + const global = vi.fn(); + listener.listen(callback); + socket.events.onListenerEventByTopic(listener, byTopic); + socket.events.onListenerEvent(global); + + emitListenerEvent(listener, { price: 1 }); + emitListenerEvent(listener, { price: 2 }); + emitListenerEvent(listener, { price: 3 }); + + expect(callback).not.toHaveBeenCalled(); + expect(byTopic).toHaveBeenCalledTimes(3); + expect(global).toHaveBeenCalledTimes(3); + vi.advanceTimersByTime(100); + expect(callback).toHaveBeenCalledOnce(); + }); + + it("should isolate delivery per dynamic topic", () => { + const template = socket.createListener()({ + topic: "prices/:symbol", + delivery: { strategy: "batch", interval: 100 }, + }); + const btc = template.setParams({ symbol: "BTC" }); + const eth = template.setParams({ symbol: "ETH" }); + const btcSpy = vi.fn(); + const ethSpy = vi.fn(); + btc.listen(({ data }) => btcSpy(data)); + eth.listen(({ data }) => ethSpy(data)); + + emitListenerEvent(btc, { price: 1 }); + emitListenerEvent(eth, { price: 10 }); + emitListenerEvent(btc, { price: 2 }); + vi.advanceTimersByTime(100); + + expect(btcSpy).toHaveBeenCalledWith([{ price: 1 }, { price: 2 }]); + expect(ethSpy).toHaveBeenCalledWith([{ price: 10 }]); + }); + + it("should not affect listeners on other topics", () => { + const batched = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const immediate = createListener(socket); + const batchedSpy = vi.fn(); + const immediateSpy = vi.fn(); + batched.listen(batchedSpy); + immediate.listen(immediateSpy); + + emitListenerEvent(batched, { price: 1 }); + emitListenerEvent(immediate, { price: 2 }); + expect(batchedSpy).not.toHaveBeenCalled(); + expect(immediateSpy).toHaveBeenCalledOnce(); + }); + + it("should allow mixing immediate and delivered subscriptions on one topic", () => { + const listener = createListener(socket); + const immediateSpy = vi.fn(); + const batchedSpy = vi.fn(); + listener.listen(immediateSpy); + listener.setDelivery({ strategy: "batch", interval: 100 }).listen(batchedSpy); + + emitListenerEvent(listener, { price: 1 }); + emitListenerEvent(listener, { price: 2 }); + expect(immediateSpy).toHaveBeenCalledTimes(2); + expect(batchedSpy).not.toHaveBeenCalled(); + vi.advanceTimersByTime(100); + expect(batchedSpy).toHaveBeenCalledOnce(); + }); + + it("should create independent subscriptions for the same callback passed twice", () => { + const listener = createListener(socket, { delivery: { strategy: "latest", interval: 100 } }); + const spy = vi.fn(); + const stopFirst = listener.listen(spy); + const stopSecond = listener.listen(spy); + expect(socket.adapter.listeners.get(listener.topic)?.size).toBe(2); + + emitListenerEvent(listener, { price: 1 }); + expect(spy).toHaveBeenCalledTimes(2); + + stopFirst(); + expect(socket.adapter.listeners.get(listener.topic)?.size).toBe(1); + emitListenerEvent(listener, { price: 2 }); + vi.advanceTimersByTime(100); + expect(spy).toHaveBeenCalledTimes(3); + stopSecond(); + expect(socket.adapter.listeners.get(listener.topic)?.size).toBe(0); + }); + + it("should unsubscribe safely from inside the callback", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const spy = vi.fn(); + const stop = listener.listen(({ data }) => { + spy(data); + stop(); + }); + emitListenerEvent(listener, { price: 1 }); + vi.advanceTimersByTime(100); + emitListenerEvent(listener, { price: 2 }); + vi.advanceTimersByTime(100); + expect(spy).toHaveBeenCalledOnce(); + expect(socket.adapter.listeners.get(listener.topic)?.size).toBe(0); + }); + + it("should handle a large burst with the latest strategy", () => { + const listener = createListener(socket, { delivery: { strategy: "latest", interval: 16 } }); + const spy = vi.fn(); + listener.listen(({ data }) => spy(data)); + for (let i = 1; i <= 500; i += 1) { + emitListenerEvent(listener, { price: i }); + } + expect(spy).toHaveBeenCalledOnce(); + vi.advanceTimersByTime(16); + expect(spy).toHaveBeenCalledTimes(2); + expect(spy).toHaveBeenLastCalledWith({ price: 500 }); + }); + + it("should handle a large burst with the batch strategy and maxSize", () => { + const listener = createListener(socket, { + delivery: { strategy: "batch", interval: 100, maxSize: 100 }, + }); + const spy = vi.fn(); + listener.listen(({ data }) => spy(data)); + for (let i = 1; i <= 250; i += 1) { + emitListenerEvent(listener, { price: i }); + } + expect(spy).toHaveBeenCalledTimes(2); + vi.advanceTimersByTime(100); + expect(spy).toHaveBeenCalledTimes(3); + const all = spy.mock.calls.flatMap(([batch]) => batch.map((m: DataType) => m.price)); + expect(all).toEqual(Array.from({ length: 250 }, (_, i) => i + 1)); + }); + + it("should expose delivery on listenerOptions and keep options untouched", () => { + const delivery = { strategy: "batch", interval: 100 } as const; + const listener = createListener(socket, { delivery }); + expect(listener.listenerOptions.delivery).toStrictEqual(delivery); + expect(listener.options).toBeUndefined(); + expect(listener.setOptions({ anything: true } as any).delivery).toStrictEqual(delivery); + }); + + it("should allow overriding delivery through clone", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const cloned = listener.clone({ delivery: { strategy: "latest", interval: 16 } }); + expect(cloned.delivery).toStrictEqual({ strategy: "latest", interval: 16 }); + expect(listener.delivery).toStrictEqual({ strategy: "batch", interval: 100 }); + }); + + it("should accept a custom strategy through setDelivery", () => { + const listener = createListener(socket); + const passthrough = createDeliveryStrategy(({ deliver }) => ({ + push: ({ data, extra }) => deliver({ data: data.price * 2, extra }), + dispose: () => null, + })); + const spy = vi.fn(); + listener.setDelivery(passthrough).listen(({ data }) => spy(data)); + emitListenerEvent(listener, { price: 21 }); + expect(spy).toHaveBeenCalledWith(42); + }); + + it("should call a custom strategy factory once per listen()", () => { + const factory = vi.fn((_context: unknown) => ({ push: vi.fn(), dispose: vi.fn() })); + const listener = createListener(socket, { delivery: factory as any }); + const stopA = listener.listen(() => null); + const stopB = listener.listen(() => null); + expect(factory).toHaveBeenCalledTimes(2); + expect(factory.mock.calls[0][0]).toHaveProperty("deliver"); + stopA(); + stopB(); + expect(factory.mock.results[0].value.dispose).toHaveBeenCalledOnce(); + expect(factory.mock.results[1].value.dispose).toHaveBeenCalledOnce(); + }); + + it("should restore immediate delivery after setDelivery(undefined)", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const spy = vi.fn(); + listener.setDelivery(undefined).listen(spy); + emitListenerEvent(listener, { price: 1 }); + expect(spy).toHaveBeenCalledOnce(); + expect(socket.adapter.listeners.get(listener.topic)?.has(spy)).toBeTrue(); + }); + + it("should keep delivering after the socket reconnects", async () => { + vi.useRealTimers(); + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 20 } }); + const spy = vi.fn(); + listener.listen(({ data }) => spy(data)); + + socket.disconnect(); + socket.connect(); + await waitForConnection(socket); + + emitListenerEvent(listener, { price: 1 }); + emitListenerEvent(listener, { price: 2 }); + await new Promise((resolve) => { + setTimeout(resolve, 60); + }); + expect(spy).toHaveBeenCalledOnce(); + expect(spy).toHaveBeenCalledWith([{ price: 1 }, { price: 2 }]); + }); +}); diff --git a/packages/sockets/__tests__/features/listener/listener.delivery.sse.spec.ts b/packages/sockets/__tests__/features/listener/listener.delivery.sse.spec.ts new file mode 100644 index 00000000..2395005e --- /dev/null +++ b/packages/sockets/__tests__/features/listener/listener.delivery.sse.spec.ts @@ -0,0 +1,55 @@ +import { createSseMockingServer, waitForConnection } from "@hyper-fetch/testing"; +import { ServerSentEventsAdapter } from "adapter-sse/sse-adapter"; + +import { createListener } from "../../utils/listener.utils"; +import { createSocket } from "../../utils/socket.utils"; + +type DataType = { price: number }; + +describe("Listener [ Delivery SSE ]", () => { + const { startServer, emitListenerEvent } = createSseMockingServer(); + let socket = createSocket({ adapter: ServerSentEventsAdapter }); + + beforeEach(async () => { + socket = createSocket({ adapter: ServerSentEventsAdapter }); + startServer(); + await waitForConnection(socket); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("should batch SSE messages", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const spy = vi.fn(); + listener.listen(({ data }) => spy(data)); + emitListenerEvent(listener, { price: 1 }); + emitListenerEvent(listener, { price: 2 }); + expect(spy).not.toHaveBeenCalled(); + vi.advanceTimersByTime(100); + expect(spy).toHaveBeenCalledWith([{ price: 1 }, { price: 2 }]); + }); + + it("should sample SSE messages", () => { + const listener = createListener(socket, { delivery: { strategy: "latest", interval: 100 } }); + const spy = vi.fn(); + listener.listen(({ data }) => spy(data)); + emitListenerEvent(listener, { price: 1 }); + emitListenerEvent(listener, { price: 2 }); + expect(spy).toHaveBeenCalledWith({ price: 1 }); + vi.advanceTimersByTime(100); + expect(spy).toHaveBeenLastCalledWith({ price: 2 }); + }); + + it("should drop pending SSE messages on unsubscribe", () => { + const listener = createListener(socket, { delivery: { strategy: "batch", interval: 100 } }); + const spy = vi.fn(); + const stop = listener.listen(spy); + emitListenerEvent(listener, { price: 1 }); + stop(); + vi.advanceTimersByTime(100); + expect(spy).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/sockets/__tests__/features/sdk/sdk.configuration.spec.ts b/packages/sockets/__tests__/features/sdk/sdk.configuration.spec.ts index 21105f49..6f0216f9 100644 --- a/packages/sockets/__tests__/features/sdk/sdk.configuration.spec.ts +++ b/packages/sockets/__tests__/features/sdk/sdk.configuration.spec.ts @@ -107,6 +107,66 @@ describe("Socket SDK [ Configuration ]", () => { }); }); + describe("delivery defaults", () => { + it("should apply delivery via plain object to listeners only", () => { + const sdk = createSocketSdk(socket); + const configured = sdk.$configure({ + "*": { delivery: { strategy: "batch", interval: 100 } }, + }); + + expect(configured.chat.messages.$listener.delivery).toStrictEqual({ strategy: "batch", interval: 100 }); + expect(configured.notifications.$listener.delivery).toStrictEqual({ strategy: "batch", interval: 100 }); + expect((configured.chat.messages.$emitter as any).delivery).toBeUndefined(); + }); + + it("should apply delivery together with options", () => { + const sdk = createSocketSdk(socket); + const configured = sdk.$configure({ + "chat/messages": { options: { fast: true }, delivery: { strategy: "latest", interval: 16 } }, + }); + + const listener = configured.chat.messages.$listener; + expect(listener.options).toStrictEqual({ fast: true }); + expect(listener.delivery).toStrictEqual({ strategy: "latest", interval: 16 }); + expect(configured.notifications.$listener.delivery).toBeUndefined(); + }); + }); + + describe("delivery via function configuration", () => { + it("should allow setDelivery inside a function configuration", () => { + const sdk = createSocketSdk(socket); + const configured = sdk.$configure({ + "chat.messages.$listener": (instance) => + "setDelivery" in instance ? instance.setDelivery({ strategy: "latest", interval: 16 }) : instance, + }); + expect(configured.chat.messages.$listener.delivery).toStrictEqual({ strategy: "latest", interval: 16 }); + expect(configured.chat.rooms.$listener.delivery).toBeUndefined(); + }); + + it("should apply delivery on a topic prefix", () => { + const sdk = createSocketSdk(socket); + const configured = sdk.$configure({ + "chat/*": { delivery: { strategy: "batch", interval: 50 } }, + }); + expect(configured.chat.messages.$listener.delivery).toStrictEqual({ strategy: "batch", interval: 50 }); + expect(configured.chat.rooms.$listener.delivery).toStrictEqual({ strategy: "batch", interval: 50 }); + expect(configured.notifications.$listener.delivery).toBeUndefined(); + }); + + it("should keep listeners working after SDK delivery configuration", () => { + const sdk = createSocketSdk(socket); + const configured = sdk.$configure({ + "*": { delivery: { strategy: "batch", interval: 100 } }, + }); + const listener = configured.notifications.$listener; + const spy = vi.fn(); + const stop = listener.listen(spy); + expect(socket.adapter.listeners.get(listener.topic)?.size).toBe(1); + stop(); + expect(socket.adapter.listeners.get(listener.topic)?.size).toBe(0); + }); + }); + describe("topic group matching", () => { it("should apply config to an exact topic", () => { const sdk = createSocketSdk(socket); diff --git a/packages/sockets/__tests__/features/sdk/sdk.types.spec.ts b/packages/sockets/__tests__/features/sdk/sdk.types.spec.ts index b73c36ad..1892d053 100644 --- a/packages/sockets/__tests__/features/sdk/sdk.types.spec.ts +++ b/packages/sockets/__tests__/features/sdk/sdk.types.spec.ts @@ -20,28 +20,28 @@ type ChatMessage = { text: string; user: string }; describe("ListenerModel - defaults for omitted fields", () => { it("should default response to unknown", () => { type L = ListenerModel<{ topic: "chat/messages" }>; - type Response = L extends Listener ? R : never; + type Response = L extends Listener ? R : never; expectTypeOf().toEqualTypeOf(); expect(true).toBe(true); }); it("should default topic to string when not provided", () => { type L = ListenerModel<{ response: ChatMessage }>; - type Topic = L extends Listener ? T : never; + type Topic = L extends Listener ? T : never; expectTypeOf().toEqualTypeOf(); expect(true).toBe(true); }); it("should default socket to SocketInstance when not provided", () => { type L = ListenerModel<{ response: ChatMessage; topic: "chat" }>; - type Sock = L extends Listener ? S : never; + type Sock = L extends Listener ? S : never; expectTypeOf().toEqualTypeOf(); expect(true).toBe(true); }); it("should default hasParams to false", () => { type L = ListenerModel<{ topic: "chat" }>; - type HasParams = L extends Listener ? Hp : never; + type HasParams = L extends Listener ? Hp : never; expectTypeOf().toEqualTypeOf(); expect(true).toBe(true); }); @@ -82,8 +82,8 @@ describe("EmitterModel - defaults for omitted fields", () => { describe("ListenerModel / EmitterModel - user-provided fields override defaults", () => { it("should narrow response and topic on a Listener when provided", () => { type L = ListenerModel<{ response: ChatMessage; topic: "chat/messages" }>; - type Response = L extends Listener ? R : never; - type Topic = L extends Listener ? T : never; + type Response = L extends Listener ? R : never; + type Topic = L extends Listener ? T : never; expectTypeOf().toEqualTypeOf(); expectTypeOf().toEqualTypeOf<"chat/messages">(); expect(true).toBe(true); @@ -102,7 +102,7 @@ describe("ListenerModel / EmitterModel - user-provided fields override defaults" describe("ListenerInstance / EmitterInstance - keep `any` defaults (constraint use)", () => { it("should keep ListenerInstance wide so concrete listeners satisfy it", () => { type LI = ListenerInstance; - type Response = LI extends Listener ? R : never; + type Response = LI extends Listener ? R : never; expectTypeOf().toEqualTypeOf(); expect(true).toBe(true); }); @@ -126,7 +126,7 @@ describe("InjectSocket - schema-level socket injection", () => { $listener: ListenerModel<{ response: ChatMessage; topic: "chat" }>; }; type Injected = InjectSocket; - type ResolvedSocket = Injected["$listener"] extends Listener ? S : never; + type ResolvedSocket = Injected["$listener"] extends Listener ? S : never; expectTypeOf().toEqualTypeOf(); expect(true).toBe(true); }); @@ -155,9 +155,10 @@ describe("InjectSocket - schema-level socket injection", () => { }; type Injected = InjectSocket; - type LSocket = Injected["chat"]["messages"]["$listener"] extends Listener ? S : never; + type LSocket = Injected["chat"]["messages"]["$listener"] extends Listener ? S : never; type ESocket = Injected["chat"]["messages"]["$emitter"] extends Emitter ? S : never; - type NestedLSocket = Injected["chat"]["$roomId"]["$listener"] extends Listener ? S : never; + type NestedLSocket = + Injected["chat"]["$roomId"]["$listener"] extends Listener ? S : never; expectTypeOf().toEqualTypeOf(); expectTypeOf().toEqualTypeOf(); @@ -174,8 +175,8 @@ describe("InjectSocket - schema-level socket injection", () => { }; type Injected = InjectSocket; - type LResp = Injected["chat"]["$listener"] extends Listener ? R : never; - type LTopic = Injected["chat"]["$listener"] extends Listener ? T : never; + type LResp = Injected["chat"]["$listener"] extends Listener ? R : never; + type LTopic = Injected["chat"]["$listener"] extends Listener ? T : never; type EPayload = Injected["chat"]["$emitter"] extends Emitter ? P : never; type ETopic = Injected["chat"]["$emitter"] extends Emitter ? T : never; @@ -205,9 +206,9 @@ describe("createSocketSdk - produces a fully resolved SDK type", () => { const sdk = createSocketSdk(socket); type ChatListener = typeof sdk.chat.messages.$listener; - type ResolvedSocket = ChatListener extends Listener ? S : never; - type ResolvedResponse = ChatListener extends Listener ? R : never; - type ResolvedTopic = ChatListener extends Listener ? T : never; + type ResolvedSocket = ChatListener extends Listener ? S : never; + type ResolvedResponse = ChatListener extends Listener ? R : never; + type ResolvedTopic = ChatListener extends Listener ? T : never; expectTypeOf().toEqualTypeOf(); expectTypeOf().toEqualTypeOf(); @@ -233,8 +234,8 @@ describe("createSocketSdk - produces a fully resolved SDK type", () => { const sdk = createSocketSdk(socket); type RoomListener = typeof sdk.chat.$roomId.$listener; - type ResolvedSocket = RoomListener extends Listener ? S : never; - type ResolvedTopic = RoomListener extends Listener ? T : never; + type ResolvedSocket = RoomListener extends Listener ? S : never; + type ResolvedTopic = RoomListener extends Listener ? T : never; expectTypeOf().toEqualTypeOf(); expectTypeOf().toEqualTypeOf<"chat/:roomId">(); @@ -301,3 +302,26 @@ describe("$configure - dot-path callback narrowing", () => { expect(true).toBe(true); }); }); + +describe("ListenerModel - delivered type", () => { + it("should default delivered to response", () => { + type L = ListenerModel<{ response: ChatMessage; topic: "chat/messages" }>; + type Delivered = L extends Listener ? D : never; + expectTypeOf().toEqualTypeOf(); + expect(true).toBe(true); + }); + + it("should keep an explicit delivered type through InjectSocket", () => { + type Schema = { + trades: { + $listener: ListenerModel<{ response: ChatMessage; topic: "trades"; delivered: ChatMessage[] }>; + }; + }; + type Injected = InjectSocket; + type Delivered = Injected["trades"]["$listener"] extends Listener ? D : never; + type Response = Injected["trades"]["$listener"] extends Listener ? R : never; + expectTypeOf().toEqualTypeOf(); + expectTypeOf().toEqualTypeOf(); + expect(true).toBe(true); + }); +}); diff --git a/packages/sockets/src/delivery/delivery.batch.ts b/packages/sockets/src/delivery/delivery.batch.ts new file mode 100644 index 00000000..c72bf261 --- /dev/null +++ b/packages/sockets/src/delivery/delivery.batch.ts @@ -0,0 +1,71 @@ +import type { BatchDeliveryType, DeliveryStrategyType } from "delivery"; + +/** + * Creates the built-in `batch` strategy. + * + * Timeline for `interval: 100` with messages A, B, C at 0/20/40ms and D at 300ms: + * ``` + * t=0 push(A) buffer=[A], open 100ms window + * t=20 push(B) buffer=[A, B] + * t=40 push(C) buffer=[A, B, C] + * t=100 window closes → deliver([A, B, C]) + * t=300 push(D) buffer=[D], open window + * t=400 window closes → deliver([D]) + * ``` + * + * The delivered `extra` is the one of the last message in the batch. + * Exported so custom strategies can compose it. + */ +export const createBatchDeliveryStrategy = ({ + interval, + maxSize = Infinity, +}: Omit): DeliveryStrategyType => { + return ({ deliver }) => { + let timer: ReturnType | undefined; + let buffer: Data[] = []; + let lastExtra: Extra | undefined; + let disposed = false; + + const flush = () => { + if (timer !== undefined) { + clearTimeout(timer); + } + timer = undefined; + if (buffer.length === 0) { + return; + } + const data = buffer; + const extra = lastExtra as Extra; + // Reset before delivering, so a synchronous unsubscribe inside the callback sees clean state. + buffer = []; + lastExtra = undefined; + deliver({ data, extra }); + }; + + return { + push: ({ data, extra }) => { + if (disposed) { + return; + } + buffer.push(data); + lastExtra = extra; + if (buffer.length >= maxSize) { + flush(); + return; + } + if (timer === undefined) { + timer = setTimeout(flush, interval); + } + }, + dispose: () => { + disposed = true; + buffer = []; + lastExtra = undefined; + if (timer !== undefined) { + clearTimeout(timer); + } + timer = undefined; + }, + }; + }; +}; diff --git a/packages/sockets/src/delivery/delivery.latest.ts b/packages/sockets/src/delivery/delivery.latest.ts new file mode 100644 index 00000000..b6e737a7 --- /dev/null +++ b/packages/sockets/src/delivery/delivery.latest.ts @@ -0,0 +1,64 @@ +import type { DeliveryMessageType, DeliveryStrategyType, LatestDeliveryType } from "delivery"; + +/** + * Creates the built-in `latest` (sampling) strategy. + * + * Timeline for `interval: 100` with messages A, B, C at 0/20/40ms and D at 300ms: + * ``` + * t=0 push(A) idle → deliver(A) immediately, open 100ms window + * t=20 push(B) pending = B + * t=40 push(C) pending = C (B is dropped) + * t=100 window closes → deliver(C), open next window + * t=200 window closes → nothing pending → idle + * t=300 push(D) idle → deliver(D) immediately + * ``` + * + * Exported so custom strategies can compose it. + */ +export const createLatestDeliveryStrategy = ({ + interval, + leading = true, +}: Omit): DeliveryStrategyType => { + return ({ deliver }) => { + let timer: ReturnType | undefined; + let pending: DeliveryMessageType | undefined; + let disposed = false; + + const closeWindow = () => { + timer = undefined; + if (!pending) { + // Nothing arrived during the window - go idle, the next message leads again. + return; + } + const message = pending; + pending = undefined; + // Re-arm before delivering, so a synchronous unsubscribe inside the callback clears this timer. + timer = setTimeout(closeWindow, interval); + deliver(message); + }; + + return { + push: (message) => { + if (disposed) { + return; + } + if (timer === undefined) { + timer = setTimeout(closeWindow, interval); + if (leading) { + deliver(message); + return; + } + } + pending = message; + }, + dispose: () => { + disposed = true; + pending = undefined; + if (timer !== undefined) { + clearTimeout(timer); + } + timer = undefined; + }, + }; + }; +}; diff --git a/packages/sockets/src/delivery/delivery.types.ts b/packages/sockets/src/delivery/delivery.types.ts new file mode 100644 index 00000000..7b35a99c --- /dev/null +++ b/packages/sockets/src/delivery/delivery.types.ts @@ -0,0 +1,129 @@ +import type { SocketAdapterInstance } from "adapter"; +import type { ExtractAdapterExtraType } from "types"; + +/** + * A single message flowing through a delivery strategy. + * + * `data` is the payload, `extra` is adapter-specific metadata (for the WebSocket adapter it is the raw `MessageEvent`). + */ +export type DeliveryMessageType = { + data: Data; + extra: Extra; +}; + +/** + * The object a delivery strategy returns for a single `listen()` subscription. + * + * Think of it as a tiny pipe: `push` is the input (every raw message lands here), `deliver` from the + * {@link DeliveryStrategyContextType} is the output, and `dispose` is the cleanup. + */ +export type DeliveryControllerType = { + /** + * Receives every raw message of the subscription. The strategy decides if and when to call `deliver`. + */ + push: (message: DeliveryMessageType) => void; + /** + * Called when the subscription ends (`unlisten()` / component unmount). Clear timers and drop pending + * state here. Nothing may be delivered after `dispose` was called. Must be safe to call more than once. + */ + dispose: () => void; +}; + +/** + * Context handed to a delivery strategy when a subscription starts. + */ +export type DeliveryStrategyContextType = { + /** + * Hands a (possibly aggregated) message to the user's `listen` callback. + */ + deliver: (message: DeliveryMessageType) => void; +}; + +/** + * A custom delivery strategy. It is a factory invoked once per `listen()` call, so every subscription + * owns its own timers and buffers. + * + * `Input` is the raw message type (the listener's response type), `Output` is what `listen` callbacks + * receive. Use {@link createDeliveryStrategy} to author one with full type inference. + * + * @example + * ```ts + * // Deliver after 50ms of silence (debounce) + * const settle = createDeliveryStrategy(({ deliver }) => { + * let timer: ReturnType | undefined; + * let last: DeliveryMessageType | undefined; + * return { + * push: (message) => { + * last = message; + * clearTimeout(timer); + * timer = setTimeout(() => { + * if (last) deliver(last); + * last = undefined; + * }, 50); + * }, + * dispose: () => { + * clearTimeout(timer); + * last = undefined; + * }, + * }; + * }); + * ``` + */ +export type DeliveryStrategyType = ( + context: DeliveryStrategyContextType, +) => DeliveryControllerType; + +/** + * Sampling delivery. The first message of an idle period is delivered immediately (when `leading` is `true`), + * then at most one message per `interval` is delivered - always the most recent one. Intermediate messages + * are dropped. Best for state-like data (prices, positions, cursors) where only the newest value matters. + */ +export type LatestDeliveryType = { + strategy: "latest"; + /** Minimum time in milliseconds between two deliveries. */ + interval: number; + /** + * Deliver the first message of an idle period immediately instead of waiting for the interval to pass. + * @default true + */ + leading?: boolean; +}; + +/** + * Batching delivery. The first message opens a window of `interval` milliseconds; when it closes, every + * message collected in it is delivered at once as an array, in arrival order. Nothing is dropped. + * Best for event-like data (trades, chat messages, logs) where every message must be processed. + */ +export type BatchDeliveryType = { + strategy: "batch"; + /** Window length in milliseconds, measured from the first message of a batch. */ + interval: number; + /** Deliver early as soon as this many messages were collected. Defaults to no limit. */ + maxSize?: number; +}; + +/** Built-in, serializable delivery presets. */ +export type DeliveryPresetType = LatestDeliveryType | BatchDeliveryType; + +/** + * Everything accepted by the listener `delivery` option: a built-in preset or a custom strategy. + */ +export type ListenerDeliveryType = + | DeliveryPresetType + | DeliveryStrategyType>; + +/** + * Resolves what `listen` callbacks receive for a given `delivery` configuration. + * + * - no delivery → `Response` + * - `{ strategy: "latest" }` → `Response` + * - `{ strategy: "batch" }` → `Response[]` + * - custom strategy → its `Output` type + */ +export type ResolveDeliveredType = [Delivery] extends [undefined] + ? Response + : Delivery extends DeliveryStrategyType + ? Output + : Delivery extends { strategy: "batch" } + ? Response[] + : Response; diff --git a/packages/sockets/src/delivery/delivery.utils.ts b/packages/sockets/src/delivery/delivery.utils.ts new file mode 100644 index 00000000..c81e5aa2 --- /dev/null +++ b/packages/sockets/src/delivery/delivery.utils.ts @@ -0,0 +1,96 @@ +import type { + BatchDeliveryType, + DeliveryControllerType, + DeliveryPresetType, + DeliveryStrategyContextType, + DeliveryStrategyType, + LatestDeliveryType, + ListenerDeliveryType, +} from "delivery"; +import { createBatchDeliveryStrategy, createLatestDeliveryStrategy } from "delivery"; + +/** + * Typed identity helper for authoring custom delivery strategies. + * + * `Input` is the listener's response type, `Output` is what `listen` callbacks will receive. TypeScript cannot + * infer `Output` from how you call `deliver`, so declare it here - the same way `createListener()` + * declares the wire type. + * + * @example + * ```ts + * const summarize = createDeliveryStrategy(({ deliver }) => { + * let count = 0; + * let last: Tick | undefined; + * let timer: ReturnType | undefined; + * return { + * push: ({ data, extra }) => { + * count += 1; + * last = data; + * timer ??= setTimeout(() => { + * timer = undefined; + * deliver({ data: { count, last: last! }, extra }); + * count = 0; + * }, 1000); + * }, + * dispose: () => clearTimeout(timer), + * }; + * }); + * + * const onTick = socket.createListener()({ topic: "ticks", delivery: summarize }); + * onTick.listen(({ data }) => data.count); // data: { count: number; last: Tick } + * ``` + */ +export const createDeliveryStrategy = ( + strategy: DeliveryStrategyType, +): DeliveryStrategyType => strategy; + +/** Narrows a `delivery` value to a custom strategy function. */ +export const isDeliveryStrategy = (delivery: ListenerDeliveryType): delivery is DeliveryStrategyType => + typeof delivery === "function"; + +/** + * Validates a `delivery` value. Throws for unknown presets so misconfiguration fails at listener creation, + * not on the first message. + */ +export const assertDelivery = (delivery: ListenerDeliveryType): void => { + if (isDeliveryStrategy(delivery)) { + return; + } + if (delivery.strategy !== "latest" && delivery.strategy !== "batch") { + throw new Error( + `Unknown delivery strategy: ${String((delivery as { strategy: unknown }).strategy)}. Use "latest", "batch" or a custom strategy created with createDeliveryStrategy().`, + ); + } +}; + +/** + * Instantiates a delivery controller for one `listen()` subscription. + */ +export const createDeliveryController = ( + delivery: DeliveryPresetType | DeliveryStrategyType, + context: DeliveryStrategyContextType, +): DeliveryControllerType => { + if (isDeliveryStrategy(delivery)) { + return delivery(context); + } + const { strategy, ...options } = delivery as DeliveryPresetType; + const factory = + strategy === "latest" + ? createLatestDeliveryStrategy(options as Omit) + : createBatchDeliveryStrategy(options as Omit); + return (factory as DeliveryStrategyType)(context); +}; + +/** + * Stable key describing a `delivery` value - presets by value, custom strategies by identity. + * Useful for memoization (e.g. React dependency arrays). + */ +export const getDeliveryKey = (delivery?: ListenerDeliveryType): string | DeliveryStrategyType => { + if (!delivery) { + return ""; + } + if (isDeliveryStrategy(delivery)) { + return delivery; + } + return JSON.stringify(delivery); +}; diff --git a/packages/sockets/src/delivery/index.ts b/packages/sockets/src/delivery/index.ts new file mode 100644 index 00000000..6d14ae7e --- /dev/null +++ b/packages/sockets/src/delivery/index.ts @@ -0,0 +1,4 @@ +export * from "./delivery.types"; +export * from "./delivery.latest"; +export * from "./delivery.batch"; +export * from "./delivery.utils"; diff --git a/packages/sockets/src/index.ts b/packages/sockets/src/index.ts index 59f624c3..e499d8ae 100644 --- a/packages/sockets/src/index.ts +++ b/packages/sockets/src/index.ts @@ -2,6 +2,7 @@ export * from "./socket"; export * from "./adapter"; export * from "./emitter"; export * from "./listener"; +export * from "./delivery"; export * from "./types"; export * from "./utils"; export * from "./adapter-websockets"; diff --git a/packages/sockets/src/listener/listener.ts b/packages/sockets/src/listener/listener.ts index 4ab70d2c..f3486823 100644 --- a/packages/sockets/src/listener/listener.ts +++ b/packages/sockets/src/listener/listener.ts @@ -1,4 +1,6 @@ import type { ExtractUrlParams, ParamsType } from "@hyper-fetch/core"; +import type { DeliveryStrategyContextType, ListenerDeliveryType, ResolveDeliveredType } from "delivery"; +import { assertDelivery, createDeliveryController } from "delivery"; import type { ListenType, ListenerConfigurationType, ListenerOptionsType } from "listener"; import type { SocketInstance } from "socket"; import type { ExtractAdapterListenerOptionsType, ExtractSocketAdapterType } from "types"; @@ -6,24 +8,41 @@ import type { ExtractAdapterListenerOptionsType, ExtractSocketAdapterType } from /** * Represents a socket message listener bound to a specific topic. Use it to subscribe to * typed messages from WebSocket or Server-Sent Events connections via the socket adapter. + * + * `Response` is the raw message type, `Delivered` is what `listen` callbacks receive. They differ only + * when a `delivery` strategy changes the shape (e.g. `batch` delivers `Response[]`). */ export class Listener< Response, Topic extends string, Socket extends SocketInstance, HasParams extends boolean = false, + Delivered = Response, > { readonly topic: Topic; params?: ParamsType = undefined; options?: ExtractAdapterListenerOptionsType> = undefined; + /** + * How messages reach `listen` callbacks. Typed with `any` input on purpose - it keeps `Response` + * covariant so partial `ListenerInstance<{ response: ... }>` constraints keep matching wider listeners. + */ + delivery?: ListenerDeliveryType> = undefined; + /** Stored with an `any` response for the same covariance reason as `delivery`. */ + readonly listenerOptions: ListenerOptionsType>; constructor( readonly socket: Socket, - readonly listenerOptions: ListenerOptionsType>, + listenerOptions: ListenerOptionsType, Response>, ) { - const { topic, options } = listenerOptions; + this.listenerOptions = listenerOptions; + const { topic, options, delivery, params } = listenerOptions as typeof listenerOptions & { params?: ParamsType }; this.topic = topic; this.options = options; + this.params = params; + this.delivery = delivery; + if (delivery) { + assertDelivery(delivery); + } } /** Set adapter-specific listener options. */ @@ -36,11 +55,21 @@ export class Listener< return this.clone({ params }); } + /** + * Set how messages are delivered to `listen` callbacks - a `latest` / `batch` preset or a custom strategy. + * Pass `undefined` to restore immediate delivery. Returns a clone; the callback data type follows the strategy. + */ + setDelivery> | undefined>( + delivery: NewDelivery, + ) { + return this.clone>({ delivery }); + } + /** Create a new listener instance with optional configuration overrides. */ - clone( - options?: ListenerConfigurationType, Topic, Socket>, + clone( + options?: ListenerConfigurationType, Topic, Socket, Response>, ) { - const newInstance = new Listener(this.socket, { + const newInstance = new Listener(this.socket, { ...this.listenerOptions, ...options, topic: this.paramsMapper(options?.params || this.params), @@ -50,13 +79,23 @@ export class Listener< } /** Start listening for messages on the configured topic. Returns a function to unsubscribe. */ - listen: ListenType, Socket> = ( - callback: Parameters, Socket>>[0], + listen: ListenType, Socket> = ( + callback: Parameters, Socket>>[0], ) => { - this.socket.adapter.listen(this, callback); + // With a delivery strategy the adapter receives the controller's `push` instead of the user callback. + // Each listen() call gets its own controller, so subscriptions never share timers or buffers. + const controller = this.delivery + ? createDeliveryController(this.delivery, { + deliver: callback as DeliveryStrategyContextType["deliver"], + }) + : null; + const adapterCallback = controller ? controller.push : callback; + + this.socket.adapter.listen(this, adapterCallback); const removeListener = () => { - this.socket.adapter.removeListener({ topic: this.topic, callback }); + controller?.dispose(); + this.socket.adapter.removeListener({ topic: this.topic, callback: adapterCallback }); }; return removeListener; diff --git a/packages/sockets/src/listener/listener.types.ts b/packages/sockets/src/listener/listener.types.ts index e6588476..d1f3a7fd 100644 --- a/packages/sockets/src/listener/listener.types.ts +++ b/packages/sockets/src/listener/listener.types.ts @@ -1,6 +1,7 @@ /* eslint-disable @typescript-eslint/no-shadow */ import type { ExtractUrlParams, EmptyTypes, TypeWithDefaults } from "@hyper-fetch/core"; import type { SocketAdapterInstance } from "adapter"; +import type { ListenerDeliveryType } from "delivery"; import type { Listener } from "listener"; import type { Socket, SocketInstance } from "socket"; import type { @@ -11,6 +12,7 @@ import type { ExtractAdapterExtraType, ExtractListenerSocketType, ExtractSocketAdapterType, + ExtractListenerDeliveredType, } from "types"; export type ListenerInstanceProperties = { @@ -18,6 +20,11 @@ export type ListenerInstanceProperties = { topic?: string; socket?: SocketInstance; hasParams?: boolean; + /** + * Type received by `listen` callbacks. Differs from `response` only when a `delivery` strategy + * changes the shape (e.g. `batch` delivers `response[]`). + */ + delivered?: any; }; /** @@ -43,7 +50,8 @@ export type ListenerInstance, TypeWithDefaults, TypeWithDefaults, - TypeWithDefaults + TypeWithDefaults, + TypeWithDefaults >; /** @@ -66,6 +74,7 @@ export type ListenerInstance, TypeWithDefaults, TypeWithDefaults, - TypeWithDefaults + TypeWithDefaults, + TypeWithDefaults> >; -export type ListenerOfAdapter = Listener, any>; +export type ListenerOfAdapter = Listener, any, any>; /** Configuration options for creating a Listener instance. */ -export type ListenerOptionsType = { +export type ListenerOptionsType< + Topic extends string, + AdapterType extends SocketAdapterInstance, + Response = any, + Delivery extends ListenerDeliveryType | undefined = + | ListenerDeliveryType + | undefined, +> = { /** The topic/channel name to listen on */ topic: Topic; /** Adapter-specific listener options */ options?: ExtractAdapterListenerOptionsType; + /** + * Controls how incoming messages reach `listen` callbacks. Omit it for immediate delivery of every message. + * + * - `{ strategy: "latest", interval }` - sampling: deliver the first message immediately, then at most one + * (the newest) per `interval` ms. Intermediate messages are dropped. For state-like data such as prices. + * - `{ strategy: "batch", interval, maxSize? }` - deliver every message collected during `interval` ms as one + * array. Nothing is dropped and callbacks receive `Response[]`. For event-like data such as trades. + * - a function created with `createDeliveryStrategy()` - full control over buffering and timing. + * + * Pending messages are dropped on unsubscribe. See the "Throttling & Batching" guide for details. + */ + delivery?: Delivery; }; -export type ListenerConfigurationType = { +export type ListenerConfigurationType = { params?: Params; -} & Partial>>; +} & Partial, Response>>; export type ListenerParamsOptionsType = ExtractListenerHasParamsType extends false @@ -123,14 +152,14 @@ export type ListenerParamsOptionsType = export type ListenType = ExtractUrlParams> extends EmptyTypes ? ( - callback: ListenerCallbackType, ExtractListenerResponseType>, + callback: ListenerCallbackType, ExtractListenerDeliveredType>, ) => () => void : ExtractListenerHasParamsType extends true ? ( - callback: ListenerCallbackType, ExtractListenerResponseType>, + callback: ListenerCallbackType, ExtractListenerDeliveredType>, ) => () => void : ( - callback: ListenerCallbackType, ExtractListenerResponseType>, + callback: ListenerCallbackType, ExtractListenerDeliveredType>, ) => () => void; export type ListenerCallbackType = (response: { @@ -145,10 +174,16 @@ export type ExtendListener< topic?: string; socket?: SocketInstance; hasParams?: true | false; + delivered?: any; }, > = Listener< TypeWithDefaults>, Properties["topic"] extends string ? Properties["topic"] : ExtractListenerTopicType, Properties["socket"] extends SocketInstance ? Properties["socket"] : ExtractListenerSocketType, - Properties["hasParams"] extends true ? true : ExtractListenerHasParamsType + Properties["hasParams"] extends true ? true : ExtractListenerHasParamsType, + TypeWithDefaults< + Properties, + "delivered", + "response" extends keyof Properties ? Properties["response"] : ExtractListenerDeliveredType + > >; diff --git a/packages/sockets/src/sdk/sdk.ts b/packages/sockets/src/sdk/sdk.ts index 4d37f28d..25e7bcc3 100644 --- a/packages/sockets/src/sdk/sdk.ts +++ b/packages/sockets/src/sdk/sdk.ts @@ -1,10 +1,13 @@ +import type { ListenerDeliveryType } from "delivery"; import type { Emitter, EmitterInstance } from "emitter"; -import type { Listener, ListenerInstance } from "listener"; +import { Listener } from "listener"; +import type { ListenerInstance } from "listener"; import type { SocketInstance } from "socket"; import type { ExtractListenerResponseType, ExtractListenerTopicType, ExtractListenerHasParamsType, + ExtractListenerDeliveredType, ExtractEmitterPayloadType, ExtractEmitterTopicType, ExtractEmitterHasPayloadType, @@ -32,7 +35,8 @@ export type InjectSocket, ExtractListenerTopicType extends string ? ExtractListenerTopicType : string, S, - ExtractListenerHasParamsType extends boolean ? ExtractListenerHasParamsType : false + ExtractListenerHasParamsType extends boolean ? ExtractListenerHasParamsType : false, + ExtractListenerDeliveredType > : T extends EmitterInstance ? Emitter< @@ -99,7 +103,10 @@ type SocketSdkConfigurationKeys = | ExtractSocketSdkPaths; export type SocketSdkDefaults = { + /** Adapter-specific options applied with `setOptions` to listeners and emitters. */ options?: Record; + /** Delivery strategy applied with `setDelivery` to listeners. Ignored for emitters. */ + delivery?: ListenerDeliveryType; }; export type SocketSdkConfigurationValue = SocketSdkDefaults | ((instance: SocketSdkLeaf) => SocketSdkLeaf); @@ -142,8 +149,12 @@ const isTopicGroup = (key: string): boolean => { }; const topicMatchesPattern = (topic: string, pattern: string): boolean => { - if (pattern === "*") {return true;} - if (pattern === topic) {return true;} + if (pattern === "*") { + return true; + } + if (pattern === topic) { + return true; + } if (pattern.endsWith("/*")) { const prefix = pattern.slice(0, -1); return topic.startsWith(prefix) || topic === prefix.slice(0, -1); @@ -153,7 +164,12 @@ const topicMatchesPattern = (topic: string, pattern: string): boolean => { const applySocketObjectDefaults = (instance: SocketSdkLeaf, config: SocketSdkDefaults): SocketSdkLeaf => { let result = instance; - if (config.options !== undefined) {result = result.setOptions(config.options);} + if (config.options !== undefined) { + result = result.setOptions(config.options); + } + if (config.delivery !== undefined && result instanceof Listener) { + result = result.setDelivery(config.delivery); + } return result; }; @@ -168,7 +184,9 @@ const applySocketDefaults = ({ sdkPath: string; defaults?: Partial>; }): SocketSdkLeaf => { - if (!defaults) {return instance;} + if (!defaults) { + return instance; + } let result = instance; const entries = Object.entries(defaults); @@ -179,7 +197,9 @@ const applySocketDefaults = ({ for (let i = 0; i < entries.length; i += 1) { const [key, value] = entries[i]; - if (!value) {continue;} + if (!value) { + continue; + } const entry: [string, SocketSdkConfigurationValue] = [key, value]; if (entry[0] === "*") { globalEntries.push(entry); diff --git a/packages/sockets/src/socket/socket.ts b/packages/sockets/src/socket/socket.ts index edbc4d9b..b50bd805 100644 --- a/packages/sockets/src/socket/socket.ts +++ b/packages/sockets/src/socket/socket.ts @@ -3,6 +3,7 @@ import { LoggerManager, AppManager, Time } from "@hyper-fetch/core"; import type { SocketAdapterInstance } from "adapter"; import type { WebsocketAdapterType } from "adapter-websockets/websocket-adapter"; import { WebsocketAdapter } from "adapter-websockets/websocket-adapter"; +import type { ListenerDeliveryType, ResolveDeliveredType } from "delivery"; import type { EmitterInstance, EmitterOptionsType } from "emitter"; import { Emitter } from "emitter"; import EventEmitter from "events"; @@ -238,8 +239,13 @@ export class Socket() => { - return (options: ListenerOptionsType) => { - return new Listener>(this, options); + return | undefined = undefined>( + options: ListenerOptionsType, + ) => { + return new Listener, false, ResolveDeliveredType>( + this, + options, + ); }; }; diff --git a/packages/sockets/src/types/extract.types.ts b/packages/sockets/src/types/extract.types.ts index 0e206b8d..14d4bf7f 100644 --- a/packages/sockets/src/types/extract.types.ts +++ b/packages/sockets/src/types/extract.types.ts @@ -19,10 +19,16 @@ export type ExtractEmitterHasParamsType = E extends Emitter = T extends Listener ? R : never; -export type ExtractListenerTopicType = E extends Listener ? T : never; -export type ExtractListenerSocketType = E extends Listener ? A : never; -export type ExtractListenerHasParamsType = E extends Listener ? P : never; +/** Raw message type of a listener - what arrives on the wire, before any `delivery` strategy. */ +export type ExtractListenerResponseType = T extends Listener ? R : never; +export type ExtractListenerTopicType = E extends Listener ? T : never; +export type ExtractListenerSocketType = E extends Listener ? A : never; +export type ExtractListenerHasParamsType = E extends Listener ? P : never; +/** + * What `listen` callbacks receive. Equals {@link ExtractListenerResponseType} unless the listener has a + * `delivery` strategy (e.g. `batch` delivers `Response[]`). + */ +export type ExtractListenerDeliveredType = T extends Listener ? D : never; // Socket diff --git a/packages/sockets/tsconfig.json b/packages/sockets/tsconfig.json index 2b7b930c..1a0666dc 100644 --- a/packages/sockets/tsconfig.json +++ b/packages/sockets/tsconfig.json @@ -10,6 +10,8 @@ "adapter-sse/*": ["./src/adapter-sse/*"], "adapter-websockets": ["./src/adapter-websockets"], "adapter-websockets/*": ["./src/adapter-websockets/*"], + "delivery": ["./src/delivery"], + "delivery/*": ["./src/delivery/*"], "emitter": ["./src/emitter"], "emitter/*": ["./src/emitter/*"], "listener": ["./src/listener"],