import util from 'node:util'; import WS from 'jest-websocket-mock'; import { WebSocket } from 'ws'; import { fireStreamChange, fireStreamAdded, fireStreamRemove, handleConnection, handleSubscription, handleMethod } from './helpers'; import { DDPSDK } from '../src/DDPSDK'; (global as any).WebSocket = (global as any).WebSocket || WebSocket; export let server: WS; const callXTimes = any>(fn: F, times: number): F => { return (async (...args) => { const methods = [].concat(...Array(times)); // eslint-disable-next-line @typescript-eslint/no-unused-vars for (const _ of methods) { await fn(...args); } }) as F; }; beforeEach(async () => { server = new WS('ws://localhost:1234/websocket'); }); afterEach(() => { server.close(); WS.clean(); jest.useRealTimers(); }); it('should handle a stream of messages', async () => { const cb = jest.fn(); const streamName = 'stream'; const streamParams = '123'; const sdk = DDPSDK.create('ws://localhost:1234'); await handleConnection(server, sdk.connection.connect()); const stream = sdk.stream(streamName, streamParams, cb); await handleSubscription(server, stream.id, streamName, streamParams); await stream.ready(); fireStreamChange(server, streamName, streamParams); fireStreamChange(server, streamName, streamParams); fireStreamChange(server, streamName, streamParams); expect(cb).toHaveBeenCalledTimes(3); fireStreamChange(server, streamName, `${streamParams}-another`); expect(cb).toHaveBeenCalledTimes(3); expect(cb).toHaveBeenNthCalledWith(1, 1); sdk.connection.close(); }); it('should ignore messages other from changed', async () => { const cb = jest.fn(); const streamName = 'stream'; const streamParams = '123'; const sdk = DDPSDK.create('ws://localhost:1234'); await handleConnection(server, sdk.connection.connect()); const stream = sdk.stream(streamName, streamParams, cb); await handleSubscription(server, stream.id, streamName, streamParams); await stream.ready(); fireStreamAdded(server, streamName, streamParams); fireStreamRemove(server, streamName, streamParams); expect(cb).toHaveBeenCalledTimes(0); sdk.connection.close(); }); it('should handle streams after reconnect', async () => { const cb = jest.fn(); const streamName = 'stream'; const streamParams = '123'; const sdk = DDPSDK.create('ws://localhost:1234'); await handleConnection(server, sdk.connection.connect()); const result = sdk.stream(streamName, streamParams, cb); expect(result.isReady).toBe(false); expect(sdk.client.subscriptions.size).toBe(1); await handleSubscription(server, result.id, streamName, streamParams); await result.ready(); const fire = callXTimes(fireStreamChange, 3); await fire(server, streamName, streamParams); expect(cb).toHaveBeenCalledTimes(3); // Fake timers are used to avoid waiting for the reconnect timeout jest.useFakeTimers(); server.close(); WS.clean(); server = new WS('ws://localhost:1234/websocket'); const reconnect = new Promise((resolve) => sdk.connection.once('reconnecting', () => resolve(undefined))); const connecting = new Promise((resolve) => sdk.connection.once('connecting', () => resolve(undefined))); const connected = new Promise((resolve) => sdk.connection.once('connected', () => resolve(undefined))); await handleConnection(server, jest.advanceTimersByTimeAsync(1000), reconnect, connecting, connected); // the client should reconnect and resubscribe await Promise.all([handleSubscription(server, result.id, streamName, streamParams), jest.advanceTimersByTimeAsync(1000)]); fire(server, streamName, streamParams); await jest.advanceTimersByTimeAsync(1000); expect(cb).toHaveBeenCalledTimes(6); jest.useRealTimers(); sdk.connection.close(); }); it('should handle an unsubscribe stream after reconnect', async () => { const cb = jest.fn(); const streamName = 'stream'; const streamParams = '123'; const sdk = DDPSDK.create('ws://localhost:1234'); await handleConnection(server, sdk.connection.connect()); const subscription = sdk.stream(streamName, streamParams, cb); expect(subscription.isReady).toBe(false); expect(sdk.client.subscriptions.size).toBe(1); await handleSubscription(server, subscription.id, streamName, streamParams); await expect(subscription.ready()).resolves.toBe(undefined); expect(subscription.isReady).toBe(true); fireStreamChange(server, streamName, streamParams); fireStreamChange(server, streamName, streamParams); fireStreamChange(server, streamName, streamParams); expect(cb).toHaveBeenCalledTimes(3); // Fake timers are used to avoid waiting for the reconnect timeout jest.useFakeTimers(); server.close(); WS.clean(); server = new WS('ws://localhost:1234/websocket'); const reconnect = new Promise((resolve) => sdk.connection.once('reconnecting', () => resolve(undefined))); const connecting = new Promise((resolve) => sdk.connection.once('connecting', () => resolve(undefined))); const connected = new Promise((resolve) => sdk.connection.once('connected', () => resolve(undefined))); await handleConnection(server, jest.advanceTimersByTimeAsync(1000), reconnect, connecting, connected); await handleSubscription(server, subscription.id, streamName, streamParams); expect(subscription.isReady).toBe(true); fireStreamChange(server, streamName, streamParams); subscription.stop(); expect(sdk.client.subscriptions.size).toBe(0); fireStreamChange(server, streamName, streamParams); fireStreamChange(server, streamName, streamParams); jest.advanceTimersByTimeAsync(1000); expect(cb).toHaveBeenCalledTimes(4); expect(sdk.client.subscriptions.size).toBe(0); jest.useRealTimers(); sdk.connection.close(); sdk.connection.close(); }); it('should create and connect to a stream', async () => { const promise = DDPSDK.createAndConnect('ws://localhost:1234'); await handleConnection(server, promise); const sdk = await promise; sdk.connection.close(); }); it.skip('should try to loginWithToken after reconnection', async () => { const sdk = DDPSDK.create('ws://localhost:1234'); await handleConnection(server, sdk.connection.connect()); const messageResult = { id: 123, token: 'token1234', tokenExpires: { $date: 99999999 }, }; const { loginWithToken } = sdk.account; const loginFn = jest.fn((token: string) => loginWithToken.apply(sdk.account, [token])); sdk.account.loginWithToken = loginFn; await handleMethod(server, 'login', [{ resume: 'token' }], JSON.stringify(messageResult), sdk.account.loginWithToken('token')); expect(sdk.account.user?.token).toBe(messageResult.token); expect(loginFn).toHaveBeenCalledTimes(1); // Fake timers are used to avoid waiting for the reconnect timeout jest.useFakeTimers(); server.close(); WS.clean(); server = new WS('ws://localhost:1234/websocket'); const reconnect = new Promise((resolve) => sdk.connection.once('reconnecting', () => resolve(undefined))); const connecting = new Promise((resolve) => sdk.connection.once('connecting', () => resolve(undefined))); const connected = new Promise((resolve) => sdk.connection.once('connected', () => resolve(undefined))); await handleConnection(server, jest.advanceTimersByTimeAsync(1000), reconnect, connecting, connected); jest.advanceTimersByTimeAsync(1000); expect(loginFn).toHaveBeenCalledTimes(2); jest.useRealTimers(); sdk.connection.close(); }); describe('Method call and Disconnection cases', () => { it('should handle properly if the message was sent after disconnection', async () => { const sdk = DDPSDK.create('ws://localhost:1234'); await handleConnection(server, sdk.connection.connect()); const [result] = await handleMethod(server, 'method', ['args1'], '1', sdk.client.callAsync('method', 'args1')); expect(result).toBe(1); // Fake timers are used to avoid waiting for the reconnect timeout jest.useFakeTimers(); server.close(); WS.clean(); server = new WS('ws://localhost:1234/websocket'); const reconnect = new Promise((resolve) => sdk.connection.once('reconnecting', () => resolve(undefined))); const connecting = new Promise((resolve) => sdk.connection.once('connecting', () => resolve(undefined))); const connected = new Promise((resolve) => sdk.connection.once('connected', () => resolve(undefined))); const callResult = sdk.client.callAsync('method', 'args2'); expect(util.inspect(callResult).includes('pending')).toBe(true); await handleConnection(server, jest.advanceTimersByTimeAsync(1000), reconnect, connecting, connected); const [result2] = await handleMethod(server, 'method', ['args2'], '1', callResult); expect(util.inspect(callResult).includes('pending')).toBe(false); expect(result2).toBe(1); sdk.connection.close(); jest.useRealTimers(); }); it.skip('should handle properly if the message was sent before disconnection but got disconnected before receiving the response', async () => { const sdk = DDPSDK.create('ws://localhost:1234'); await handleConnection(server, sdk.connection.connect()); const callResult = sdk.client.callAsync('method', 'args'); expect(util.inspect(callResult).includes('pending')).toBe(true); // Fake timers are used to avoid waiting for the reconnect timeout jest.useFakeTimers(); server.close(); WS.clean(); server = new WS('ws://localhost:1234/websocket'); const reconnect = new Promise((resolve) => sdk.connection.once('reconnecting', () => resolve(undefined))); const connecting = new Promise((resolve) => sdk.connection.once('connecting', () => resolve(undefined))); const connected = new Promise((resolve) => sdk.connection.once('connected', () => resolve(undefined))); await handleConnection(server, jest.advanceTimersByTimeAsync(1000), reconnect, connecting, connected); expect(util.inspect(callResult).includes('pending')).toBe(true); const [result] = await handleMethod(server, 'method', ['args2'], '1', callResult); expect(result).toBe(1); jest.useRealTimers(); sdk.connection.close(); }); });