diff --git a/src/websocket-client.ts b/src/websocket-client.ts index 04d8323..6049e23 100644 --- a/src/websocket-client.ts +++ b/src/websocket-client.ts @@ -508,6 +508,7 @@ export class WebsocketClient extends EventEmitter { if (!topics.length) { return; } + const wsMessage = JSON.stringify({ req_id: topics.join(','), op: 'subscribe', @@ -631,16 +632,16 @@ export class WebsocketClient extends EventEmitter { if (isWsPong(msg)) { this.logger.silly('Received pong', { ...loggerCategory, wsKey }); } else { - this.emit('response', msg); + this.emit('response', { ...msg, wsKey }); } return; } if (msg['finalFragment']) { - return this.emit('response', msg); + return this.emit('response', { ...msg, wsKey }); } if (msg?.topic) { - return this.emit('update', msg); + return this.emit('update', { ...msg, wsKey }); } if ( @@ -651,7 +652,7 @@ export class WebsocketClient extends EventEmitter { // usdc options msg?.success === false ) { - return this.emit('errorEvent', msg); + return this.emit('errorEvent', { ...msg, wsKey }); } this.logger.warning('Unhandled/unrecognised ws event message', { @@ -781,6 +782,7 @@ export class WebsocketClient extends EventEmitter { */ public subscribe(wsTopics: WsTopic[] | WsTopic, isPrivateTopic?: boolean) { const topics = Array.isArray(wsTopics) ? wsTopics : [wsTopics]; + topics.forEach((topic) => this.wsStore.addTopic( getWsKeyForTopic(this.options.market, topic, isPrivateTopic), @@ -794,7 +796,9 @@ export class WebsocketClient extends EventEmitter { if ( this.wsStore.isConnectionState(wsKey, WsConnectionStateEnum.CONNECTED) ) { - return this.requestSubscribeTopics(wsKey, topics); + return this.requestSubscribeTopics(wsKey, [ + ...this.wsStore.getTopics(wsKey), + ]); } // start connection process if it hasn't yet begun. Topics are automatically subscribed to on-connect @@ -832,7 +836,9 @@ export class WebsocketClient extends EventEmitter { if ( this.wsStore.isConnectionState(wsKey, WsConnectionStateEnum.CONNECTED) ) { - this.requestUnsubscribeTopics(wsKey, topics); + this.requestUnsubscribeTopics(wsKey, [ + ...this.wsStore.getTopics(wsKey), + ]); } }); } diff --git a/test/unified-margin/ws.public.option.test.ts b/test/unified-margin/ws.public.option.test.ts new file mode 100644 index 0000000..4c179c1 --- /dev/null +++ b/test/unified-margin/ws.public.option.test.ts @@ -0,0 +1,67 @@ +import { + WebsocketClient, + WSClientConfigurableOptions, + WS_KEY_MAP, +} from '../../src'; +import { + logAllEvents, + getSilentLogger, + waitForSocketEvent, + WS_OPEN_EVENT_PARTIAL, +} from '../ws.util'; + +describe('Public Unified Margin Websocket Client (Options)', () => { + let wsClient: WebsocketClient; + + const wsClientOptions: WSClientConfigurableOptions = { + market: 'unifiedOption', + }; + + beforeAll(() => { + wsClient = new WebsocketClient( + wsClientOptions, + getSilentLogger('expectSuccessNoAuth') + ); + wsClient.connectPublic(); + }); + + afterAll(() => { + wsClient.closeAll(); + }); + + it('should open a public ws connection', async () => { + const wsOpenPromise = waitForSocketEvent(wsClient, 'open'); + try { + expect(await wsOpenPromise).toMatchObject({ + event: WS_OPEN_EVENT_PARTIAL, + wsKey: WS_KEY_MAP.unifiedOptionPublic, + }); + } catch (e) { + expect(e).toBeFalsy(); + } + }); + + it('should subscribe to public trade events', async () => { + const wsResponsePromise = waitForSocketEvent(wsClient, 'response'); + // const wsUpdatePromise = waitForSocketEvent(wsClient, 'update'); + + wsClient.subscribe('orderbook.25.BTCUSDT'); + + try { + expect(await wsResponsePromise).toMatchObject({ + success: true, + type: 'COMMAND_RESP', + }); + } catch (e) { + // sub failed + expect(e).toBeFalsy(); + } + + // try { + // expect(await wsUpdatePromise).toStrictEqual('asdfasdf'); + // } catch (e) { + // // no data + // expect(e).toBeFalsy(); + // } + }); +}); diff --git a/test/unified-margin/ws.public.perp.usdt.test.ts b/test/unified-margin/ws.public.perp.usdt.test.ts new file mode 100644 index 0000000..98215d4 --- /dev/null +++ b/test/unified-margin/ws.public.perp.usdt.test.ts @@ -0,0 +1,83 @@ +import { + WebsocketClient, + WSClientConfigurableOptions, + WS_KEY_MAP, +} from '../../src'; +import { + logAllEvents, + getSilentLogger, + waitForSocketEvent, + WS_OPEN_EVENT_PARTIAL, + fullLogger, +} from '../ws.util'; + +describe('Public Unified Margin Websocket Client (Perps - USDT)', () => { + let wsClient: WebsocketClient; + + const wsClientOptions: WSClientConfigurableOptions = { + market: 'unifiedPerp', + }; + + beforeAll(() => { + wsClient = new WebsocketClient( + wsClientOptions, + getSilentLogger('expectSuccessNoAuth') + // fullLogger + ); + // logAllEvents(wsClient); + wsClient.connectPublic(); + }); + + afterAll(() => { + wsClient.closeAll(); + }); + + it('should open a public ws connection', async () => { + const wsOpenPromise = waitForSocketEvent(wsClient, 'open'); + try { + expect(await wsOpenPromise).toMatchObject({ + event: WS_OPEN_EVENT_PARTIAL, + wsKey: expect.stringContaining('unifiedPerpUSD'), + }); + } catch (e) { + expect(e).toBeFalsy(); + } + }); + + it('should subscribe to public trade events through USDT topic', async () => { + const wsResponsePromise = waitForSocketEvent(wsClient, 'response'); + const wsUpdatePromise = waitForSocketEvent(wsClient, 'update'); + + wsClient.subscribe('orderbook.25.BTCUSDT'); + + try { + expect(await wsResponsePromise).toMatchObject({ + op: 'subscribe', + req_id: 'orderbook.25.BTCUSDT', + success: true, + wsKey: WS_KEY_MAP.unifiedPerpUSDTPublic, + }); + } catch (e) { + // sub failed + expect(e).toBeFalsy(); + } + + try { + expect(await wsUpdatePromise).toMatchObject({ + data: { + a: expect.any(Array), + b: expect.any(Array), + s: 'BTCUSDT', + u: expect.any(Number), + }, + topic: 'orderbook.25.BTCUSDT', + ts: expect.any(Number), + type: 'snapshot', + wsKey: WS_KEY_MAP.unifiedPerpUSDTPublic, + }); + } catch (e) { + // no data + expect(e).toBeFalsy(); + } + }); +}); diff --git a/test/ws.util.ts b/test/ws.util.ts index 1fa49da..7d1423e 100644 --- a/test/ws.util.ts +++ b/test/ws.util.ts @@ -105,7 +105,7 @@ export function listenToSocketEvents(wsClient: WebsocketClient) { export function logAllEvents(wsClient: WebsocketClient) { wsClient.on('update', (data) => { - console.log('wsUpdate: ', JSON.stringify(data, null, 2)); + // console.log('wsUpdate: ', JSON.stringify(data, null, 2)); }); wsClient.on('open', (data) => {