From 824963e15373b8fbf0ba43abe04f32727c1ddc4b Mon Sep 17 00:00:00 2001 From: Maciej Pyrc Date: Wed, 21 Oct 2026 16:52:50 +0200 Subject: [PATCH 1/2] =?UTF-8?q?feat(core):=20=F0=9F=8E=B8=20client.setUrl?= =?UTF-8?q?=20and=20invalidate-all=20cache?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add client.setUrl(url, { abort, invalidate }) to switch the base url at runtime. Cache keys do not include the base url, so abort/invalidate let callers drop in-flight requests and stale data of the previous server. cache.invalidate() called without arguments now invalidates every entry. Refs #148 Co-Authored-By: Claude Opus 5.5 --- documentation/docs/core/cache.mdx | 2 + documentation/docs/core/client.mdx | 24 +++ .../features/cache/cache.base.spec.ts | 16 ++ .../features/client/client.set-url.spec.ts | 149 ++++++++++++++++++ packages/core/src/cache/cache.ts | 7 +- packages/core/src/client/client.ts | 38 ++++- 6 files changed, 231 insertions(+), 5 deletions(-) create mode 100644 packages/core/__tests__/features/client/client.set-url.spec.ts diff --git a/documentation/docs/core/cache.mdx b/documentation/docs/core/cache.mdx index 61bfff89..a5be52e9 100644 --- a/documentation/docs/core/cache.mdx +++ b/documentation/docs/core/cache.mdx @@ -299,6 +299,8 @@ client.cache.invalidate(getNote.setParams({ noteId: 1 }).cacheKey); client.cache.invalidate("CUSTOM_CACHE_KEY"); // OR via RegExp client.cache.invalidate(new RegExp("CUSTOM_CACHE_KEY")); +// OR everything - when called without arguments +client.cache.invalidate(); ``` --- diff --git a/documentation/docs/core/client.mdx b/documentation/docs/core/client.mdx index cdfb266b..e197cfdb 100644 --- a/documentation/docs/core/client.mdx +++ b/documentation/docs/core/client.mdx @@ -427,6 +427,30 @@ Available removal methods: --- +10. ### Changing the base URL + +Use `setUrl` to point the client at a different server at runtime - for example when switching tenants, regions or +environments. Every request sent afterwards uses the new base URL. + +```tsx +import { createClient } from "@hyper-fetch/core"; + +export const client = createClient({ url: "https://eu.example.com" }); + +// highlight-next-line +client.setUrl("https://us.example.com", { abort: true, invalidate: true }); +``` + +Cache keys don't include the base URL, so data cached for the previous server stays in place by default. Pass the +options when the data of both servers should not mix: + +| Option | Default | Description | +|--------|---------|-------------| +| `abort` | `false` | Abort all requests that are still in progress for the previous URL. | +| `invalidate` | `false` | Invalidate the whole cache (and the requests aborted by `abort`), so mounted hooks refetch from the new URL. | + +--- + 6. ### Hydration We can hydrate the client with the data from the server. This is a powerful feature that allows us to pass the data from diff --git a/packages/core/__tests__/features/cache/cache.base.spec.ts b/packages/core/__tests__/features/cache/cache.base.spec.ts index 56e03542..933802ae 100644 --- a/packages/core/__tests__/features/cache/cache.base.spec.ts +++ b/packages/core/__tests__/features/cache/cache.base.spec.ts @@ -109,6 +109,22 @@ describe("Cache [ Base ]", () => { unmount(); }); + it("should invalidate every element when called without arguments", async () => { + const trigger = vi.fn(); + const unmount = cache.events.onInvalidate(trigger); + + cache.set(request.setCacheKey("1"), { ...response, ...details }); + cache.set(request.setCacheKey("2"), { ...response, ...details }); + cache.invalidate(); + + expect(cache.get("1")?.staleTime).toBe(0); + expect(cache.get("2")?.staleTime).toBe(0); + expect(trigger).toHaveBeenCalledTimes(2); + expect(trigger).toHaveBeenCalledWith("1"); + expect(trigger).toHaveBeenCalledWith("2"); + unmount(); + }); + it("should not add to cache when useCache is set to false", async () => { const trigger = vi.fn(); const unmount = cache.events.onDataByKey(request.cacheKey, trigger); diff --git a/packages/core/__tests__/features/client/client.set-url.spec.ts b/packages/core/__tests__/features/client/client.set-url.spec.ts new file mode 100644 index 00000000..6e056b34 --- /dev/null +++ b/packages/core/__tests__/features/client/client.set-url.spec.ts @@ -0,0 +1,149 @@ +import { createHttpMockingServer } from "@hyper-fetch/testing"; +import type { ResponseSuccessType } from "adapter"; +import { getErrorMessage } from "adapter"; +import { Client } from "client"; +import type { HttpAdapterType } from "http-adapter"; +import { xhrExtra } from "http-adapter"; +import type { ResponseDetailsType } from "managers"; +import { http, HttpResponse } from "msw"; + +const { server, resetMocks, startServer, stopServer, mockRequest } = createHttpMockingServer(); + +describe("Client [ setUrl ]", () => { + const response: ResponseSuccessType = { + data: 123, + error: null, + status: 200, + success: true, + extra: xhrExtra, + requestTimestamp: Date.now(), + responseTimestamp: Date.now(), + }; + const details: ResponseDetailsType = { + retries: 0, + requestTimestamp: Date.now(), + responseTimestamp: Date.now(), + addedTimestamp: Date.now(), + triggerTimestamp: Date.now(), + isCanceled: false, + isOffline: false, + willRetry: false, + }; + + let client = new Client({ url: "http://first.test" }); + + beforeAll(() => { + startServer(); + }); + + beforeEach(() => { + client = new Client({ url: "http://first.test" }); + resetMocks(); + vi.resetAllMocks(); + }); + + afterAll(() => { + stopServer(); + }); + + it("should send next requests to the new url", async () => { + server.use( + http.get("http://first.test/users", () => HttpResponse.json({ host: "first" })), + http.get("http://second.test/users", () => HttpResponse.json({ host: "second" })), + ); + const request = client.createRequest<{ response: { host: string } }>()({ + endpoint: "/users", + }); + + const first = await request.send(); + client.setUrl("http://second.test"); + const second = await request.send(); + + expect(client.url).toBe("http://second.test"); + expect(first.data).toStrictEqual({ host: "first" }); + expect(second.data).toStrictEqual({ host: "second" }); + }); + + it("should be chainable", () => { + expect(client.setUrl("http://second.test")).toBe(client); + }); + + it("should do nothing when the url did not change", () => { + const abortSpy = vi.spyOn(client.requestManager, "abortAll"); + const invalidateSpy = vi.spyOn(client.cache, "invalidate"); + + client.setUrl("http://first.test", { abort: true, invalidate: true }); + + expect(client.url).toBe("http://first.test"); + expect(abortSpy).not.toHaveBeenCalled(); + expect(invalidateSpy).not.toHaveBeenCalled(); + }); + + it("should not abort requests nor touch the cache by default", async () => { + const request = client.createRequest<{ response: { value: number } }>()({ + endpoint: "/users", + }); + const data = mockRequest(request, { data: { value: 1 }, delay: 30 }); + client.cache.set(request, { ...response, ...details }); + const staleTime = client.cache.get(request.cacheKey)?.staleTime; + const abortSpy = vi.spyOn(client.requestManager, "abortAll"); + + const pending = request.send(); + client.setUrl("http://second.test"); + const result = await pending; + + expect(abortSpy).not.toHaveBeenCalled(); + expect(result.data).toStrictEqual(data); + expect(client.cache.get(request.cacheKey)?.staleTime).toBe(staleTime); + }); + + it("should abort ongoing requests when abort is enabled", async () => { + const request = client.createRequest<{ response: { value: number } }>()({ + endpoint: "/users", + }); + mockRequest(request, { data: { value: 1 }, delay: 30 }); + + const pending = request.send(); + client.setUrl("http://second.test", { abort: true }); + const result = await pending; + + expect(result.data).toBeNull(); + expect(result.error).toStrictEqual(getErrorMessage("abort")); + expect(result.status).toBe(0); + }); + + it("should invalidate the whole cache when invalidate is enabled", () => { + const usersRequest = client.createRequest()({ endpoint: "/users" }); + const postsRequest = client.createRequest()({ endpoint: "/posts" }); + const trigger = vi.fn(); + const unmount = client.cache.events.onInvalidate(trigger); + + client.cache.set(usersRequest, { ...response, ...details }); + client.cache.set(postsRequest, { ...response, ...details }); + client.setUrl("http://second.test", { invalidate: true }); + unmount(); + + expect(client.cache.get(usersRequest.cacheKey)?.staleTime).toBe(0); + expect(client.cache.get(postsRequest.cacheKey)?.staleTime).toBe(0); + expect(trigger).toHaveBeenCalledWith(usersRequest.cacheKey); + expect(trigger).toHaveBeenCalledWith(postsRequest.cacheKey); + }); + + it("should notify consumers of aborted requests when abort and invalidate are enabled", async () => { + const request = client.createRequest<{ response: { value: number } }>()({ + endpoint: "/users", + }); + mockRequest(request, { data: { value: 1 }, delay: 30 }); + const trigger = vi.fn(); + const unmount = client.cache.events.onInvalidateByKey(request.cacheKey, trigger); + + const pending = request.send(); + client.setUrl("http://second.test", { abort: true, invalidate: true }); + await pending; + unmount(); + + // Nothing was cached yet - the aborted request is reached only by the explicit invalidation + expect(client.cache.get(request.cacheKey)).toBeUndefined(); + expect(trigger).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/core/src/cache/cache.ts b/packages/core/src/cache/cache.ts index 614b818d..f70aa390 100644 --- a/packages/core/src/cache/cache.ts +++ b/packages/core/src/cache/cache.ts @@ -195,9 +195,10 @@ export class Cache { /** * Invalidate cache by cacheKey or partial matching with RegExp * It emits invalidation event for each matching cacheKey and sets staleTime to 0 to indicate out of time cache - * @param key - cacheKey or Request instance or RegExp for partial matching + * Called without arguments it invalidates every cached element. + * @param cacheKeys - cacheKey or Request instance or RegExp for partial matching (defaults to all elements) */ - invalidate = (cacheKeys: string | RegExp | RequestInstance | (string | RegExp | RequestInstance)[]) => { + invalidate = (cacheKeys: string | RegExp | RequestInstance | (string | RegExp | RequestInstance)[] = /.*/) => { this.logger.debug({ title: "Revalidating cache element", type: "system", extra: { cacheKeys } }); const onInvalidate = (key: string | RegExp | RequestInstance) => { @@ -285,7 +286,7 @@ export class Cache { */ getAllKeys = async () => { const keys = await this.lazyStorage?.keys(); - const asyncKeys = [...keys || []]; + const asyncKeys = [...(keys || [])]; const syncKeys = [...this.storage.keys()]; return [...new Set([...asyncKeys, ...syncKeys])]; diff --git a/packages/core/src/client/client.ts b/packages/core/src/client/client.ts index a3631cd9..43a6e5cd 100644 --- a/packages/core/src/client/client.ts +++ b/packages/core/src/client/client.ts @@ -43,7 +43,7 @@ export class Client< GlobalErrorType extends ClientErrorType = Error, Adapter extends AdapterInstance = HttpAdapterType, > { - readonly url: string; + url: string; readonly mode: ClientMode; public debug: boolean; @@ -103,6 +103,38 @@ export class Client< this.submitDispatcher.initialize(this as unknown as ClientInstance<{ adapter: Adapter }>); } + /** + * Set the base url used by the next requests - e.g. when switching tenants, regions or environments at runtime. + * Cache keys don't include the base url, so data cached for the previous url stays in place unless you pass + * `invalidate: true`. Pass `abort: true` to cancel the requests that are still in progress for the previous url. + * @param url new base url + * @param options.abort abort all ongoing requests + * @param options.invalidate invalidate the cache (and requests aborted by `abort`), so active consumers refetch + */ + setUrl = (url: string, options: { abort?: boolean; invalidate?: boolean } = {}): Client => { + if (url === this.url) { + return this; + } + + const { abort = false, invalidate = false } = options; + const runningRequests = this.fetchDispatcher.getAllRunningRequests().map(({ request }) => request); + + this.url = url; + + if (abort) { + this.requestManager.abortAll(); + } + if (invalidate) { + this.cache.invalidate(); + // Aborted requests never land in the cache, so the full invalidation above would not notify their consumers + if (abort && runningRequests.length > 0) { + this.cache.invalidate(runningRequests); + } + } + + return this; + }; + /** * This method enables the logger usage and display the logs in console */ @@ -409,7 +441,9 @@ export class Client< options?: Partial | ((item: HydrateDataType) => Partial), ) => { hydrationData?.forEach((item) => { - if (!item) {return;} + if (!item) { + return; + } const { cacheKey, scope, response, ...fallbackOptions } = item; const defaults = { From 7231aded61409928aa1910b7c16b4464f1f4b89a Mon Sep 17 00:00:00 2001 From: Maciej Pyrc Date: Wed, 21 Oct 2026 16:52:51 +0200 Subject: [PATCH 2/2] =?UTF-8?q?feat(sockets):=20=F0=9F=8E=B8=20socket.setU?= =?UTF-8?q?rl=20and=20encode/decode=20adapter=20options?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add socket.setUrl(url) - applied on the next connection attempt, like setQueryParams. Add flat decode/encode options to the websocket adapter and decode to the SSE adapter, so servers using a custom frame shape (e.g. { type, payload }) work without a custom adapter. Frames decoded to null are ignored; encode/decode errors are reported through onError. Closes #148 Co-Authored-By: Claude Opus 5.5 --- documentation/docs/sockets/adapter.mdx | 25 +++ documentation/docs/sockets/socket.mdx | 14 ++ .../adapter/adapter.sse.codec.spec.ts | 101 +++++++++ .../adapter/adapter.websocket.codec.spec.ts | 208 ++++++++++++++++++ .../features/socket/socket.methods.spec.ts | 40 ++++ .../sockets/src/adapter-sse/sse-adapter.ts | 27 ++- .../src/adapter-sse/sse-adapter.types.ts | 9 + .../adapter-websockets/websocket-adapter.ts | 37 +++- .../websocket-adapter.types.ts | 20 ++ packages/sockets/src/socket/socket.ts | 10 + 10 files changed, 481 insertions(+), 10 deletions(-) create mode 100644 packages/sockets/__tests__/features/adapter/adapter.sse.codec.spec.ts create mode 100644 packages/sockets/__tests__/features/adapter/adapter.websocket.codec.spec.ts diff --git a/documentation/docs/sockets/adapter.mdx b/documentation/docs/sockets/adapter.mdx index e1991df7..118cbdde 100644 --- a/documentation/docs/sockets/adapter.mdx +++ b/documentation/docs/sockets/adapter.mdx @@ -47,6 +47,8 @@ You can pass adapter-specific options via the `adapterOptions` constructor optio | `heartbeatMessage` | `string` | — | The message payload sent for heartbeats. | | `pingTimeout` | `number` | — | Timeout (ms) to wait for a ping response. | | `pongTimeout` | `number` | — | Timeout (ms) to wait for a pong response. | +| `decode` | `(event: MessageEvent) => { topic, data } \| null` | JSON `{ topic, data }` | Turns an incoming frame into a listener event. Return `null` to ignore the frame. | +| `encode` | `({ topic, payload }) => string \| ArrayBufferLike \| Blob \| ArrayBufferView` | `JSON.stringify({ topic, data })` | Turns an emitted message (and the heartbeat) into the frame passed to `websocket.send`. | ```tsx const socket = createSocketClient({ @@ -60,6 +62,28 @@ const socket = createSocketClient({ }); ``` +#### Custom message format + +By default every frame is a JSON `{ topic, data }` object. When your server speaks a different format - for example +`{ type, payload }` - map it with the `decode` and `encode` options instead of writing a custom adapter. Frames decoded +to `null` (pings, unknown messages) are ignored, and errors thrown by either function are reported through `onError`. + +```tsx +const socket = createSocketClient({ + url: "wss://example.com/ws", + adapterOptions: { + // highlight-start + decode: (event) => { + const message = JSON.parse(event.data); + if (message.type === "pong") return null; + return { topic: message.type, data: message.payload }; + }, + encode: ({ topic, payload }) => JSON.stringify({ type: topic, payload }), + // highlight-end + }, +}); +``` + #### Dynamic options Every adapter option can be resolved right before a connection attempt with the `onConnect` interceptor on the socket. @@ -115,6 +139,7 @@ const socket = createSocketClient({ |--------|------|---------|-------------| | `autoConnect` | `boolean` | `true` | Automatically connect when the socket is created. | | `eventSourceInit` | `EventSourceInit` | — | Options passed to the underlying `EventSource` constructor (e.g., `withCredentials`). | +| `decode` | `(event: MessageEvent) => { topic, data } \| null` | JSON `{ topic, data }` | Turns an incoming message into a listener event. Return `null` to ignore the message. | ```tsx const socket = createSocketClient({ diff --git a/documentation/docs/sockets/socket.mdx b/documentation/docs/sockets/socket.mdx index 61b3cf61..f3d6571e 100644 --- a/documentation/docs/sockets/socket.mdx +++ b/documentation/docs/sockets/socket.mdx @@ -125,6 +125,20 @@ attempt, so you never have to keep the socket in sync with your application stat --- +### Changing the URL + +`setUrl` replaces the URL used by the **next** connection attempt - just like `setQueryParams`, call `reconnect()` to +apply it to a live connection. Listeners stay registered, so they keep receiving events from the new server. + +```tsx +export const socket = createSocketClient({ url: "wss://eu.example.com/ws" }); + +socket.setUrl("wss://us.example.com/ws"); +socket.reconnect(); +``` + +--- + ### Authentication To exchange authenticated events, set up the `queryParams` parameter with authentication parameter or resolve the diff --git a/packages/sockets/__tests__/features/adapter/adapter.sse.codec.spec.ts b/packages/sockets/__tests__/features/adapter/adapter.sse.codec.spec.ts new file mode 100644 index 00000000..c66844fd --- /dev/null +++ b/packages/sockets/__tests__/features/adapter/adapter.sse.codec.spec.ts @@ -0,0 +1,101 @@ +import { createSseMockingServer } from "@hyper-fetch/testing"; +import { waitFor } from "@testing-library/dom"; +import type { SSEAdapterOptionsType, ServerSentEventsAdapterType } from "adapter-sse"; +import { ServerSentEventsAdapter } from "adapter-sse"; + +import { createSocket } from "../../utils/socket.utils"; + +const decode: SSEAdapterOptionsType["decode"] = (event) => { + const message = JSON.parse(event.data); + if (message.type === "ping") { + return null; + } + return { topic: message.type, data: message.payload }; +}; + +describe("SSE Adapter [ Decode ]", () => { + const { startServer, getSource } = createSseMockingServer(); + + const send = (data: string) => getSource().emitMessage(new MessageEvent("message", { data })); + + const setup = async (adapterOptions?: SSEAdapterOptionsType) => { + const socket = createSocket({ + adapter: ServerSentEventsAdapter, + adapterOptions, + }); + await startServer(); + return socket; + }; + + beforeEach(() => { + vi.resetAllMocks(); + }); + + it("should receive { topic, data } messages by default", async () => { + const socket = await setup(); + const spy = vi.fn(); + socket + .createListener()({ topic: "user" }) + .listen(({ data }) => spy(data)); + + send(JSON.stringify({ topic: "user", data: { name: "Alice" } })); + + await waitFor(() => { + expect(spy).toHaveBeenCalledWith({ name: "Alice" }); + }); + }); + + it("should map incoming messages to listener topics with custom decode", async () => { + const socket = await setup({ decode }); + const spy = vi.fn(); + socket + .createListener()({ topic: "user" }) + .listen(({ data }) => spy(data)); + + send(JSON.stringify({ type: "user", payload: { name: "Alice" } })); + + await waitFor(() => { + expect(spy).toHaveBeenCalledWith({ name: "Alice" }); + }); + }); + + it("should ignore messages decoded to null", async () => { + const socket = await setup({ decode }); + const eventSpy = vi.fn(); + socket.events.onListenerEvent(eventSpy); + + send(JSON.stringify({ type: "ping" })); + send(JSON.stringify({ type: "user", payload: 1 })); + + await waitFor(() => { + expect(eventSpy).toHaveBeenCalledTimes(1); + }); + expect(eventSpy).toHaveBeenCalledWith(expect.objectContaining({ topic: "user", data: 1 })); + }); + + it("should report decoder errors and keep the connection", async () => { + const error = new Error("Malformed message"); + const socket = await setup({ + decode: (event) => { + if (event.data === "broken") { + throw error; + } + return decode(event); + }, + }); + const errorSpy = vi.fn(); + const spy = vi.fn(); + socket.onError(errorSpy); + socket.createListener()({ topic: "user" }).listen(spy); + + send("broken"); + send(JSON.stringify({ type: "user", payload: 1 })); + + await waitFor(() => { + expect(spy).toHaveBeenCalledTimes(1); + }); + expect(errorSpy).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalledWith({ error }); + expect(socket.adapter.connected).toBe(true); + }); +}); diff --git a/packages/sockets/__tests__/features/adapter/adapter.websocket.codec.spec.ts b/packages/sockets/__tests__/features/adapter/adapter.websocket.codec.spec.ts new file mode 100644 index 00000000..36988640 --- /dev/null +++ b/packages/sockets/__tests__/features/adapter/adapter.websocket.codec.spec.ts @@ -0,0 +1,208 @@ +import { createWebsocketMockingServer, waitForConnection } from "@hyper-fetch/testing"; +import { waitFor } from "@testing-library/dom"; +import type { WebsocketAdapterOptionsType } from "adapter-websockets"; + +import { createSocket } from "../../utils/socket.utils"; + +const decode: WebsocketAdapterOptionsType["decode"] = (event) => { + const message = JSON.parse(event.data); + if (message.type === "ping") { + return null; + } + return { topic: message.type, data: message.payload }; +}; + +const encode: WebsocketAdapterOptionsType["encode"] = ({ topic, payload }) => JSON.stringify({ type: topic, payload }); + +describe("Websocket Adapter [ Encode / Decode ]", () => { + const { getServer, startServer, stopServer } = createWebsocketMockingServer(); + + beforeEach(() => { + vi.resetAllMocks(); + startServer(); + }); + + afterEach(() => { + stopServer(); + }); + + describe("When using the default message format", () => { + it("should receive { topic, data } frames", async () => { + const socket = createSocket(); + await waitForConnection(socket); + const spy = vi.fn(); + socket + .createListener<{ name: string }>()({ topic: "user" }) + .listen(({ data }) => spy(data)); + + getServer().send(JSON.stringify({ topic: "user", data: { name: "Alice" } })); + + await waitFor(() => { + expect(spy).toHaveBeenCalledWith({ name: "Alice" }); + }); + }); + + it("should send { topic, data } frames", async () => { + const socket = createSocket(); + await waitForConnection(socket); + + socket + .createEmitter<{ name: string }>()({ topic: "user" }) + .emit({ payload: { name: "Alice" } }); + + await expect(getServer()).toReceiveMessage(JSON.stringify({ topic: "user", data: { name: "Alice" } })); + }); + }); + + describe("When using custom decode", () => { + it("should map incoming frames to listener topics", async () => { + const socket = createSocket({ adapterOptions: { decode } }); + await waitForConnection(socket); + const spy = vi.fn(); + const other = vi.fn(); + socket + .createListener<{ name: string }>()({ topic: "user" }) + .listen(({ data }) => spy(data)); + socket.createListener()({ topic: "other" }).listen(other); + + getServer().send(JSON.stringify({ type: "user", payload: { name: "Alice" } })); + + await waitFor(() => { + expect(spy).toHaveBeenCalledWith({ name: "Alice" }); + }); + expect(other).not.toHaveBeenCalled(); + }); + + it("should pass the original MessageEvent as extra", async () => { + const socket = createSocket({ adapterOptions: { decode } }); + await waitForConnection(socket); + const spy = vi.fn(); + socket + .createListener()({ topic: "user" }) + .listen(({ extra }) => spy(extra)); + const frame = JSON.stringify({ type: "user", payload: {} }); + + getServer().send(frame); + + await waitFor(() => { + expect(spy).toHaveBeenCalledTimes(1); + }); + expect(spy.mock.calls[0][0].type).toBe("message"); + expect(spy.mock.calls[0][0].data).toBe(frame); + }); + + it("should ignore frames decoded to null", async () => { + const socket = createSocket({ adapterOptions: { decode } }); + await waitForConnection(socket); + const spy = vi.fn(); + const eventSpy = vi.fn(); + socket.events.onListenerEvent(eventSpy); + socket.createListener()({ topic: "user" }).listen(spy); + + getServer().send(JSON.stringify({ type: "ping" })); + getServer().send(JSON.stringify({ type: "user", payload: { name: "Alice" } })); + + await waitFor(() => { + expect(spy).toHaveBeenCalledTimes(1); + }); + expect(eventSpy).toHaveBeenCalledTimes(1); + }); + + it("should report decoder errors and keep the connection", async () => { + const error = new Error("Malformed frame"); + const socket = createSocket({ + adapterOptions: { + decode: (event) => { + if (event.data === "broken") { + throw error; + } + return decode(event); + }, + }, + }); + await waitForConnection(socket); + const errorSpy = vi.fn(); + const spy = vi.fn(); + socket.onError(errorSpy); + socket.createListener()({ topic: "user" }).listen(spy); + + getServer().send("broken"); + getServer().send(JSON.stringify({ type: "user", payload: {} })); + + await waitFor(() => { + expect(spy).toHaveBeenCalledTimes(1); + }); + expect(errorSpy).toHaveBeenCalledTimes(1); + expect(errorSpy).toHaveBeenCalledWith({ error }); + expect(socket.adapter.connected).toBe(true); + }); + + it("should use the decoder returned by onConnect", async () => { + // Registered synchronously - the initial connection attempt runs the interceptors asynchronously + const socket = createSocket().onConnect(({ connection }) => ({ + ...connection, + adapterOptions: { ...connection.adapterOptions, decode }, + })); + await waitForConnection(socket); + const spy = vi.fn(); + socket + .createListener()({ topic: "user" }) + .listen(({ data }) => spy(data)); + + getServer().send(JSON.stringify({ type: "user", payload: 1 })); + + await waitFor(() => { + expect(spy).toHaveBeenCalledWith(1); + }); + }); + }); + + describe("When using custom encode", () => { + it("should encode emitted messages", async () => { + const socket = createSocket({ adapterOptions: { encode } }); + await waitForConnection(socket); + + socket + .createEmitter<{ name: string }>()({ topic: "user" }) + .emit({ payload: { name: "Alice" } }); + + await expect(getServer()).toReceiveMessage(JSON.stringify({ type: "user", payload: { name: "Alice" } })); + }); + + it("should encode heartbeat messages", async () => { + const socket = createSocket({ + adapterOptions: { + encode, + heartbeat: true, + heartbeatMessage: "ping", + pingTimeout: 10, + pongTimeout: 1000, + }, + }); + await waitForConnection(socket); + + await expect(getServer()).toReceiveMessage(JSON.stringify({ type: "heartbeat", payload: "ping" })); + }); + + it("should report encoder errors without sending the frame", async () => { + const error = new Error("Cannot encode"); + const socket = createSocket({ + adapterOptions: { + encode: () => { + throw error; + }, + }, + }); + await waitForConnection(socket); + const errorSpy = vi.fn(); + socket.onError(errorSpy); + const sendSpy = vi.fn(); + getServer().on("message", sendSpy); + + await socket.adapter.emit(socket.createEmitter()({ topic: "user" })); + + expect(errorSpy).toHaveBeenCalledWith({ error }); + expect(sendSpy).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/packages/sockets/__tests__/features/socket/socket.methods.spec.ts b/packages/sockets/__tests__/features/socket/socket.methods.spec.ts index 3c35aedf..fd9b2bdc 100644 --- a/packages/sockets/__tests__/features/socket/socket.methods.spec.ts +++ b/packages/sockets/__tests__/features/socket/socket.methods.spec.ts @@ -1,4 +1,6 @@ import { LoggerManager } from "@hyper-fetch/core"; +import { createWebsocketMockingServer, waitForConnection } from "@hyper-fetch/testing"; +import { waitFor } from "@testing-library/dom"; import { WebsocketAdapter } from "adapter-websockets/websocket-adapter"; import { Socket } from "socket"; @@ -22,6 +24,44 @@ describe("Socket [ Methods ]", () => { expect(queryParams).toBe("testing"); }); + it("should allow to set url without touching the live connection", async () => { + const socket = new Socket({ url: "ws://localhost:1234", adapterOptions: { autoConnect: false } }); + const reconnectSpy = vi.spyOn(socket.adapter, "reconnect"); + + expect(socket.setUrl("ws://localhost:4321")).toBe(socket); + expect(socket.url).toBe("ws://localhost:4321"); + expect(reconnectSpy).not.toHaveBeenCalled(); + }); + + it("should connect to the new url on reconnect and keep listeners", async () => { + // startServer() cleans every mocked server, so the second one is created only after the first started + const first = createWebsocketMockingServer("ws://localhost:1234"); + first.startServer(); + const second = createWebsocketMockingServer("ws://localhost:4321"); + + const socket = new Socket({ url: "ws://localhost:1234" }); + const listener = socket.createListener<{ value: number }>()({ topic: "topic" }); + const spy = vi.fn(); + listener.listen(({ data }) => spy(data)); + await waitForConnection(socket); + await first.getServer().connected; + + socket.setUrl("ws://localhost:4321"); + await socket.reconnect(); + await second.getServer().connected; + await waitForConnection(socket); + + second.emitListenerEvent(listener, { value: 2 }); + + await waitFor(() => { + expect(spy).toHaveBeenCalledWith({ value: 2 }); + }); + + await socket.disconnect(); + first.stopServer(); + second.stopServer(); + }); + it("should allow to activate debug", async () => { const socket = new Socket({ url: "ws://localhost:1234" }); socket.setDebug(true); diff --git a/packages/sockets/src/adapter-sse/sse-adapter.ts b/packages/sockets/src/adapter-sse/sse-adapter.ts index 87057c30..501ca7ea 100644 --- a/packages/sockets/src/adapter-sse/sse-adapter.ts +++ b/packages/sockets/src/adapter-sse/sse-adapter.ts @@ -7,7 +7,7 @@ import type { ListenerCallbackType, ListenerOfAdapter } from "listener"; import { getSocketUrl, parseMessageEvent } from "../utils"; import { getSocketError } from "../utils/socket.errors"; -import type { SSEAdapterOptionsType } from "./sse-adapter.types"; +import type { SSEAdapterOptionsType, SSEDecodedMessageType } from "./sse-adapter.types"; import { getServerSentEventsAdapter } from "./sse-adapter.utils"; /** @@ -60,6 +60,23 @@ export const ServerSentEventsAdapter = (): ServerSentEventsAdapterType => ? socket.adapter.adapterOptions?.autoConnect : true; + /** Maps an incoming message to the listener event, reporting decoder failures instead of throwing */ + const decodeMessage = ( + event: MessageEvent, + decode: SSEAdapterOptionsType["decode"], + ): SSEDecodedMessageType | null => { + try { + if (decode) { + return decode(event); + } + const { topic, data } = parseMessageEvent(event); + return { topic, data }; + } catch (error) { + onError({ error: error as Error }); + return null; + } + }; + const connect = async (): Promise => { const connection = await onConnect(); if (!connection) { @@ -97,10 +114,12 @@ export const ServerSentEventsAdapter = (): ServerSentEventsAdapterType => onError({ error: new Error(error) }); }); + const { decode } = connection.adapterOptions || {}; eventSource.addEventListener("message", (newEvent: MessageEvent) => { - const { topic, data, event } = parseMessageEvent(newEvent); - - onEvent({ topic, data, extra: event }); + const message = decodeMessage(newEvent, decode); + if (message) { + onEvent({ topic: message.topic, data: message.data, extra: newEvent }); + } }); return new Promise((resolve) => { diff --git a/packages/sockets/src/adapter-sse/sse-adapter.types.ts b/packages/sockets/src/adapter-sse/sse-adapter.types.ts index 7be2f494..18f70830 100644 --- a/packages/sockets/src/adapter-sse/sse-adapter.types.ts +++ b/packages/sockets/src/adapter-sse/sse-adapter.types.ts @@ -1,6 +1,15 @@ // Options +/** Listener event decoded from an incoming message */ +export type SSEDecodedMessageType = { topic: string; data: unknown }; + export type SSEAdapterOptionsType = { eventSourceInit?: EventSourceInit; autoConnect?: boolean; + /** + * Turns an incoming message into a listener event - use it when the server does not send `{ topic, data }` + * (e.g. `{ type, payload }`). Return `null` to ignore the message. + * Defaults to parsing the message as JSON and reading its `topic` and `data` fields. + */ + decode?: (event: MessageEvent) => SSEDecodedMessageType | null; }; diff --git a/packages/sockets/src/adapter-websockets/websocket-adapter.ts b/packages/sockets/src/adapter-websockets/websocket-adapter.ts index 2c61274e..06a41f76 100644 --- a/packages/sockets/src/adapter-websockets/websocket-adapter.ts +++ b/packages/sockets/src/adapter-websockets/websocket-adapter.ts @@ -8,7 +8,7 @@ import type { ListenerCallbackType, ListenerOfAdapter } from "listener"; import { getSocketUrl, parseMessageEvent } from "../utils"; import { getSocketError } from "../utils/socket.errors"; -import type { WebsocketAdapterOptionsType } from "./websocket-adapter.types"; +import type { WebsocketAdapterOptionsType, WebsocketDecodedMessageType } from "./websocket-adapter.types"; import { getWebsocketAdapter } from "./websocket-adapter.utils"; /** @@ -120,10 +120,13 @@ export const WebsocketAdapter = (): WebsocketAdapterType => }); newWebsocket.addEventListener("message", (newEvent: MessageEvent) => { - const { topic, data, event } = parseMessageEvent(newEvent); - - onEvent({ topic, data, extra: event }); + // Any frame proves the connection is alive, even the ones ignored by the decoder onHeartbeat(); + + const message = decodeMessage(newEvent); + if (message) { + onEvent({ topic: message.topic, data: message.data, extra: newEvent }); + } }); return new Promise((resolve) => { @@ -209,12 +212,34 @@ export const WebsocketAdapter = (): WebsocketAdapterType => clearTimeout(pongTimer); }; + /** Maps an incoming frame to the listener event, reporting decoder failures instead of throwing */ + const decodeMessage = (event: MessageEvent): WebsocketDecodedMessageType | null => { + try { + if (options?.decode) { + return options.decode(event); + } + const { topic, data } = parseMessageEvent(event); + return { topic, data }; + } catch (error) { + onError({ error: error as Error }); + return null; + } + }; + const sendEventMessage = ({ topic, payload }: Pick) => { if (!websocket) { return false; } - websocket!.send(JSON.stringify({ topic, data: payload })); - return true; + try { + const frame = options?.encode + ? options.encode({ topic, payload }) + : JSON.stringify({ topic, data: payload }); + websocket.send(frame); + return true; + } catch (error) { + onError({ error: error as Error }); + return false; + } }; const onHeartbeat = () => { diff --git a/packages/sockets/src/adapter-websockets/websocket-adapter.types.ts b/packages/sockets/src/adapter-websockets/websocket-adapter.types.ts index 854d29dc..fbddefba 100644 --- a/packages/sockets/src/adapter-websockets/websocket-adapter.types.ts +++ b/packages/sockets/src/adapter-websockets/websocket-adapter.types.ts @@ -1,3 +1,12 @@ +/** Listener event decoded from an incoming frame */ +export type WebsocketDecodedMessageType = { topic: string; data: unknown }; + +/** Emitted message to be encoded into an outgoing frame */ +export type WebsocketEncodeMessageType = { topic: string; payload: unknown }; + +/** Frame accepted by `WebSocket.send` */ +export type WebsocketFrameType = Parameters[0]; + export type WebsocketAdapterOptionsType = { /** * Subprotocols passed to the `WebSocket` constructor (`Sec-WebSocket-Protocol` header). @@ -9,4 +18,15 @@ export type WebsocketAdapterOptionsType = { heartbeatMessage?: string; heartbeat?: boolean; autoConnect?: boolean; + /** + * Turns an incoming frame into a listener event - use it when the server does not send `{ topic, data }` + * (e.g. `{ type, payload }`). Return `null` to ignore the frame (pings, unknown messages). + * Defaults to parsing the frame as JSON and reading its `topic` and `data` fields. + */ + decode?: (event: MessageEvent) => WebsocketDecodedMessageType | null; + /** + * Turns an emitted message (and the heartbeat ping) into the frame passed to `websocket.send`. + * Defaults to `JSON.stringify({ topic, data: payload })`. + */ + encode?: (message: WebsocketEncodeMessageType) => WebsocketFrameType; }; diff --git a/packages/sockets/src/socket/socket.ts b/packages/sockets/src/socket/socket.ts index b50bd805..a73785c8 100644 --- a/packages/sockets/src/socket/socket.ts +++ b/packages/sockets/src/socket/socket.ts @@ -91,6 +91,16 @@ export class Socket { + this.url = url; + return this; + }; + /** * Set the default query params used by the next connection attempts. * Call `reconnect()` to apply them to a live connection, or use `onConnect` to resolve them per attempt.