From 3f5039ef8bf265fbe50abf138dbb60ff5d6d03ad Mon Sep 17 00:00:00 2001 From: tiagosiebler Date: Thu, 15 Sep 2022 12:20:39 +0100 Subject: [PATCH] cleaning around websocket client --- examples/ws-private.ts | 4 +- examples/ws-public.ts | 34 +++++---- src/index.ts | 2 +- src/types/websockets.ts | 9 +-- src/util/WsStore.ts | 37 ++++----- src/util/websocket-util.ts | 87 ++++++++++++++++------ src/websocket-client.ts | 149 +++++++++++++------------------------ 7 files changed, 164 insertions(+), 158 deletions(-) diff --git a/examples/ws-private.ts b/examples/ws-private.ts index 51ce8d7..59623f3 100644 --- a/examples/ws-private.ts +++ b/examples/ws-private.ts @@ -1,5 +1,5 @@ import { DefaultLogger } from '../src'; -import { WebsocketClient, wsKeySpotPublic } from '../src/websocket-client'; +import { WebsocketClient } from '../src/websocket-client'; // or // import { DefaultLogger, WebsocketClient } from 'bybit-api'; @@ -33,6 +33,8 @@ import { WebsocketClient, wsKeySpotPublic } from '../src/websocket-client'; logger ); + // wsClient.subscribePublicSpotOrderbook('test', 'full'); + wsClient.on('update', (data) => { console.log('raw message received ', JSON.stringify(data, null, 2)); }); diff --git a/examples/ws-public.ts b/examples/ws-public.ts index 130c87c..2aa1cd7 100644 --- a/examples/ws-public.ts +++ b/examples/ws-public.ts @@ -1,8 +1,7 @@ -import { DefaultLogger } from '../src'; -import { WebsocketClient, wsKeySpotPublic } from '../src/websocket-client'; +import { DefaultLogger, WS_KEY_MAP, WebsocketClient } from '../src'; // or -// import { DefaultLogger, WebsocketClient } from 'bybit-api'; +// import { DefaultLogger, WS_KEY_MAP, WebsocketClient } from 'bybit-api'; (async () => { const logger = { @@ -10,13 +9,16 @@ import { WebsocketClient, wsKeySpotPublic } from '../src/websocket-client'; // silly: () => {}, }; - const wsClient = new WebsocketClient({ - // key: key, - // secret: secret, - // market: 'inverse', - // market: 'linear', - market: 'spot', - }, logger); + const wsClient = new WebsocketClient( + { + // key: key, + // secret: secret, + market: 'linear', + // market: 'inverse', + // market: 'spot', + }, + logger + ); wsClient.on('update', (data) => { console.log('raw message received ', JSON.stringify(data, null, 2)); @@ -25,7 +27,7 @@ import { WebsocketClient, wsKeySpotPublic } from '../src/websocket-client'; wsClient.on('open', (data) => { console.log('connection opened open:', data.wsKey); - if (data.wsKey === wsKeySpotPublic) { + if (data.wsKey === WS_KEY_MAP.spotPublic) { // Spot public. // wsClient.subscribePublicSpotTrades('BTCUSDT'); // wsClient.subscribePublicSpotTradingPair('BTCUSDT'); @@ -40,16 +42,20 @@ import { WebsocketClient, wsKeySpotPublic } from '../src/websocket-client'; console.log('ws automatically reconnecting.... ', wsKey); }); wsClient.on('reconnected', (data) => { - console.log('ws has reconnected ', data?.wsKey ); + console.log('ws has reconnected ', data?.wsKey); }); // Inverse // wsClient.subscribe('trade'); // Linear - // wsClient.subscribe('trade.BTCUSDT'); + wsClient.subscribe('trade.BTCUSDT'); + + setTimeout(() => { + console.log('unsubscribing'); + wsClient.unsubscribe('trade.BTCUSDT'); + }, 5 * 1000); // For spot, request public connection first then send required topics on 'open' // wsClient.connectPublic(); - })(); diff --git a/src/index.ts b/src/index.ts index 1325998..4f8c4c4 100644 --- a/src/index.ts +++ b/src/index.ts @@ -10,6 +10,6 @@ export * from './usdc-perpetual-client'; export * from './unified-margin-client'; export * from './websocket-client'; export * from './util/logger'; +export * from './util'; export * from './types'; -export * from './util/WsStore'; export * from './constants/enum'; diff --git a/src/types/websockets.ts b/src/types/websockets.ts index a2a1cdf..856d537 100644 --- a/src/types/websockets.ts +++ b/src/types/websockets.ts @@ -1,4 +1,4 @@ -import { RestClientOptions } from '../util'; +import { RestClientOptions, WS_KEY_MAP } from '../util'; export type APIMarket = 'inverse' | 'linear' | 'spot'; //| 'v3'; @@ -69,12 +69,7 @@ export type WsPrivateTopic = export type WsTopic = WsPublicTopics | WsPrivateTopic; /** This is used to differentiate between each of the available websocket streams (as bybit has multiple websockets) */ -export type WsKey = - | 'inverse' - | 'linearPrivate' - | 'linearPublic' - | 'spotPrivate' - | 'spotPublic'; +export type WsKey = typeof WS_KEY_MAP[keyof typeof WS_KEY_MAP]; export interface WSClientConfigurableOptions { key?: string; diff --git a/src/util/WsStore.ts b/src/util/WsStore.ts index 7c421ce..9539079 100644 --- a/src/util/WsStore.ts +++ b/src/util/WsStore.ts @@ -1,4 +1,5 @@ import WebSocket from 'isomorphic-ws'; +import { WsKey } from '../types'; import { DefaultLogger } from './logger'; @@ -44,9 +45,9 @@ export default class WsStore { } /** Get WS stored state for key, optionally create if missing */ - get(key: string, createIfMissing?: true): WsStoredState; - get(key: string, createIfMissing?: false): WsStoredState | undefined; - get(key: string, createIfMissing?: boolean): WsStoredState | undefined { + get(key: WsKey, createIfMissing?: true): WsStoredState; + get(key: WsKey, createIfMissing?: false): WsStoredState | undefined; + get(key: WsKey, createIfMissing?: boolean): WsStoredState | undefined { if (this.wsState[key]) { return this.wsState[key]; } @@ -56,11 +57,11 @@ export default class WsStore { } } - getKeys(): string[] { - return Object.keys(this.wsState); + getKeys(): WsKey[] { + return Object.keys(this.wsState) as WsKey[]; } - create(key: string): WsStoredState | undefined { + create(key: WsKey): WsStoredState | undefined { if (this.hasExistingActiveConnection(key)) { this.logger.warning( 'WsStore setConnection() overwriting existing open connection: ', @@ -74,7 +75,7 @@ export default class WsStore { return this.get(key); } - delete(key: string) { + delete(key: WsKey) { if (this.hasExistingActiveConnection(key)) { const ws = this.getWs(key); this.logger.warning( @@ -88,15 +89,15 @@ export default class WsStore { /* connection websocket */ - hasExistingActiveConnection(key: string) { + hasExistingActiveConnection(key: WsKey) { return this.get(key) && this.isWsOpen(key); } - getWs(key: string): WebSocket | undefined { + getWs(key: WsKey): WebSocket | undefined { return this.get(key)?.ws; } - setWs(key: string, wsConnection: WebSocket): WebSocket { + setWs(key: WsKey, wsConnection: WebSocket): WebSocket { if (this.isWsOpen(key)) { this.logger.warning( 'WsStore setConnection() overwriting existing open connection: ', @@ -109,7 +110,7 @@ export default class WsStore { /* connection state */ - isWsOpen(key: string): boolean { + isWsOpen(key: WsKey): boolean { const existingConnection = this.getWs(key); return ( !!existingConnection && @@ -117,37 +118,37 @@ export default class WsStore { ); } - getConnectionState(key: string): WsConnectionStateEnum { + getConnectionState(key: WsKey): WsConnectionStateEnum { return this.get(key, true)!.connectionState!; } - setConnectionState(key: string, state: WsConnectionStateEnum) { + setConnectionState(key: WsKey, state: WsConnectionStateEnum) { this.get(key, true)!.connectionState = state; } - isConnectionState(key: string, state: WsConnectionStateEnum): boolean { + isConnectionState(key: WsKey, state: WsConnectionStateEnum): boolean { return this.getConnectionState(key) === state; } /* subscribed topics */ - getTopics(key: string): WsTopicList { + getTopics(key: WsKey): WsTopicList { return this.get(key, true).subscribedTopics; } getTopicsByKey(): Record { const result = {}; for (const refKey in this.wsState) { - result[refKey] = this.getTopics(refKey); + result[refKey] = this.getTopics(refKey as WsKey); } return result; } - addTopic(key: string, topic: WsTopic) { + addTopic(key: WsKey, topic: WsTopic) { return this.getTopics(key).add(topic); } - deleteTopic(key: string, topic: WsTopic) { + deleteTopic(key: WsKey, topic: WsTopic) { return this.getTopics(key).delete(topic); } } diff --git a/src/util/websocket-util.ts b/src/util/websocket-util.ts index 0b639af..5dafc45 100644 --- a/src/util/websocket-util.ts +++ b/src/util/websocket-util.ts @@ -1,38 +1,84 @@ import { WsKey } from '../types'; -export const wsKeyInverse = 'inverse'; -export const wsKeyLinearPrivate = 'linearPrivate'; -export const wsKeyLinearPublic = 'linearPublic'; -export const wsKeySpotPrivate = 'spotPrivate'; -export const wsKeySpotPublic = 'spotPublic'; +interface NetworkMapV3 { + livenet: string; + livenet2?: string; + testnet: string; + testnet2?: string; +} -export const WS_KEY_MAP = { - inverse: wsKeyInverse, - linearPrivate: wsKeyLinearPrivate, - linearPublic: wsKeyLinearPublic, - spotPrivate: wsKeySpotPrivate, - spotPublic: wsKeySpotPublic, +type PublicPrivateNetwork = 'public' | 'private'; + +export const WS_BASE_URL_MAP: Record< + string, + Record +> = { + inverse: { + private: { + livenet: 'wss://stream.bybit.com/realtime', + testnet: 'wss://stream-testnet.bybit.com/realtime', + }, + public: { + livenet: 'wss://stream.bybit.com/realtime', + testnet: 'wss://stream-testnet.bybit.com/realtime', + }, + }, + linear: { + private: { + livenet: 'wss://stream.bybit.com/realtime_private', + livenet2: 'wss://stream.bytick.com/realtime_private', + testnet: 'wss://stream-testnet.bybit.com/realtime_private', + }, + public: { + livenet: 'wss://stream.bybit.com/realtime_public', + livenet2: 'wss://stream.bytick.com/realtime_public', + testnet: 'wss://stream-testnet.bybit.com/realtime_public', + }, + }, + spot: { + private: { + livenet: 'wss://stream.bybit.com/spot/ws', + testnet: 'wss://stream-testnet.bybit.com/spot/ws', + }, + public: { + livenet: 'wss://stream.bybit.com/spot/quote/ws/v1', + livenet2: 'wss://stream.bybit.com/spot/quote/ws/v2', + testnet: 'wss://stream-testnet.bybit.com/spot/quote/ws/v1', + testnet2: 'wss://stream-testnet.bybit.com/spot/quote/ws/v2', + }, + }, }; -export const PUBLIC_WS_KEYS = [WS_KEY_MAP.linearPublic, WS_KEY_MAP.spotPublic]; +export const WS_KEY_MAP = { + inverse: 'inverse', + linearPrivate: 'linearPrivate', + linearPublic: 'linearPublic', + spotPrivate: 'spotPrivate', + spotPublic: 'spotPublic', +} as const; + +export const PUBLIC_WS_KEYS = [ + WS_KEY_MAP.linearPublic, + WS_KEY_MAP.spotPublic, +] as string[]; export function getLinearWsKeyForTopic(topic: string): WsKey { - const privateLinearTopics = [ + const privateTopics = [ 'position', 'execution', 'order', 'stop_order', 'wallet', ]; - if (privateLinearTopics.includes(topic)) { - return wsKeyLinearPrivate; + if (privateTopics.includes(topic)) { + return WS_KEY_MAP.linearPrivate; } - return wsKeyLinearPublic; + return WS_KEY_MAP.linearPublic; } export function getSpotWsKeyForTopic(topic: string): WsKey { - const privateLinearTopics = [ + const privateTopics = [ 'position', 'execution', 'order', @@ -42,9 +88,8 @@ export function getSpotWsKeyForTopic(topic: string): WsKey { 'ticketInfo', ]; - if (privateLinearTopics.includes(topic)) { - return wsKeySpotPrivate; + if (privateTopics.includes(topic)) { + return WS_KEY_MAP.spotPrivate; } - - return wsKeySpotPublic; + return WS_KEY_MAP.spotPublic; } diff --git a/src/websocket-client.ts b/src/websocket-client.ts index 759a4a3..2b966c9 100644 --- a/src/websocket-client.ts +++ b/src/websocket-client.ts @@ -6,7 +6,9 @@ import { LinearClient } from './linear-client'; import { SpotClientV3 } from './spot-client-v3'; import { SpotClient } from './spot-client'; -import { DefaultLogger } from './util/logger'; +import { signMessage } from './util/node-support'; +import WsStore from './util/WsStore'; + import { APIMarket, KlineInterval, @@ -17,82 +19,22 @@ import { WsTopic, } from './types'; -import { signMessage } from './util/node-support'; - -import WsStore from './util/WsStore'; import { serializeParams, isWsPong, getLinearWsKeyForTopic, getSpotWsKeyForTopic, - wsKeyInverse, - wsKeyLinearPrivate, - wsKeyLinearPublic, - wsKeySpotPrivate, - wsKeySpotPublic, WsConnectionStateEnum, PUBLIC_WS_KEYS, + WS_KEY_MAP, + DefaultLogger, + WS_BASE_URL_MAP, } from './util'; -const inverseEndpoints = { - livenet: 'wss://stream.bybit.com/realtime', - testnet: 'wss://stream-testnet.bybit.com/realtime', -}; - -interface NetworkMapV3 { - livenet: string; - livenet2?: string; - testnet: string; - testnet2?: string; -} - -type NetworkType = 'public' | 'private'; - function neverGuard(x: never, msg: string): Error { return new Error(`Unhandled value exception "x", ${msg}`); } -const WS_BASE_URL_MAP: Record> = { - linear: { - private: { - livenet: 'wss://stream.bybit.com/realtime_private', - livenet2: 'wss://stream.bytick.com/realtime_private', - testnet: 'wss://stream-testnet.bybit.com/realtime_private', - }, - public: { - livenet: 'wss://stream.bybit.com/realtime_public', - livenet2: 'wss://stream.bytick.com/realtime_public', - testnet: 'wss://stream-testnet.bybit.com/realtime_public', - }, - }, -}; - -const linearEndpoints: Record = { - private: { - livenet: 'wss://stream.bybit.com/realtime_private', - livenet2: 'wss://stream.bytick.com/realtime_private', - testnet: 'wss://stream-testnet.bybit.com/realtime_private', - }, - public: { - livenet: 'wss://stream.bybit.com/realtime_public', - livenet2: 'wss://stream.bytick.com/realtime_public', - testnet: 'wss://stream-testnet.bybit.com/realtime_public', - }, -}; - -const spotEndpoints: Record = { - private: { - livenet: 'wss://stream.bybit.com/spot/ws', - testnet: 'wss://stream-testnet.bybit.com/spot/ws', - }, - public: { - livenet: 'wss://stream.bybit.com/spot/quote/ws/v1', - livenet2: 'wss://stream.bybit.com/spot/quote/ws/v2', - testnet: 'wss://stream-testnet.bybit.com/spot/quote/ws/v1', - testnet2: 'wss://stream-testnet.bybit.com/spot/quote/ws/v2', - }, -}; - const loggerCategory = { category: 'bybit-ws' }; export declare interface WebsocketClient { @@ -280,16 +222,19 @@ export class WebsocketClient extends EventEmitter { public connectAll(): Promise[] { switch (this.options.market) { case 'inverse': { - return [this.connect(wsKeyInverse)]; + return [this.connect(WS_KEY_MAP.inverse)]; } case 'linear': { return [ - this.connect(wsKeyLinearPublic), - this.connect(wsKeyLinearPrivate), + this.connect(WS_KEY_MAP.linearPublic), + this.connect(WS_KEY_MAP.linearPrivate), ]; } case 'spot': { - return [this.connect(wsKeySpotPublic), this.connect(wsKeySpotPrivate)]; + return [ + this.connect(WS_KEY_MAP.spotPublic), + this.connect(WS_KEY_MAP.spotPrivate), + ]; } default: { throw neverGuard(this.options.market, `connectAll(): Unhandled market`); @@ -300,13 +245,13 @@ export class WebsocketClient extends EventEmitter { public connectPublic(): Promise { switch (this.options.market) { case 'inverse': { - return this.connect(wsKeyInverse); + return this.connect(WS_KEY_MAP.inverse); } case 'linear': { - return this.connect(wsKeyLinearPublic); + return this.connect(WS_KEY_MAP.linearPublic); } case 'spot': { - return this.connect(wsKeySpotPublic); + return this.connect(WS_KEY_MAP.spotPublic); } default: { throw neverGuard( @@ -320,13 +265,13 @@ export class WebsocketClient extends EventEmitter { public connectPrivate(): Promise | undefined { switch (this.options.market) { case 'inverse': { - return this.connect(wsKeyInverse); + return this.connect(WS_KEY_MAP.inverse); } case 'linear': { - return this.connect(wsKeyLinearPrivate); + return this.connect(WS_KEY_MAP.linearPrivate); } case 'spot': { - return this.connect(wsKeySpotPrivate); + return this.connect(WS_KEY_MAP.spotPrivate); } default: { throw neverGuard( @@ -672,7 +617,7 @@ export class WebsocketClient extends EventEmitter { } } - private getWs(wsKey: string) { + private getWs(wsKey: WsKey) { return this.wsStore.getWs(wsKey); } @@ -688,20 +633,21 @@ export class WebsocketClient extends EventEmitter { const networkKey = this.isLivenet() ? 'livenet' : 'testnet'; switch (wsKey) { - case wsKeyLinearPublic: { - return linearEndpoints.public[networkKey]; + case WS_KEY_MAP.linearPublic: { + return WS_BASE_URL_MAP.linear.public[networkKey]; } - case wsKeyLinearPrivate: { - return linearEndpoints.private[networkKey]; + case WS_KEY_MAP.linearPrivate: { + return WS_BASE_URL_MAP.linear.private[networkKey]; } - case wsKeySpotPublic: { - return spotEndpoints.public[networkKey]; + case WS_KEY_MAP.spotPublic: { + return WS_BASE_URL_MAP.spot.public[networkKey]; } - case wsKeySpotPrivate: { - return spotEndpoints.private[networkKey]; + case WS_KEY_MAP.spotPrivate: { + return WS_BASE_URL_MAP.linear.private[networkKey]; } - case wsKeyInverse: { - return inverseEndpoints[networkKey]; + case WS_KEY_MAP.inverse: { + // private and public are on the same WS connection + return WS_BASE_URL_MAP.inverse.public[networkKey]; } default: { this.logger.error('getWsUrl(): Unhandled wsKey: ', { @@ -713,14 +659,24 @@ export class WebsocketClient extends EventEmitter { } } - private getWsKeyForTopic(topic: string) { - if (this.isInverse()) { - return wsKeyInverse; + private getWsKeyForTopic(topic: string): WsKey { + switch (this.options.market) { + case 'inverse': { + return WS_KEY_MAP.inverse; + } + case 'linear': { + return getLinearWsKeyForTopic(topic); + } + case 'spot': { + return getSpotWsKeyForTopic(topic); + } + default: { + throw neverGuard( + this.options.market, + `connectPublic(): Unhandled market` + ); + } } - if (this.isLinear()) { - return getLinearWsKeyForTopic(topic); - } - return getSpotWsKeyForTopic(topic); } private wrongMarketError(market: APIMarket) { @@ -736,7 +692,7 @@ export class WebsocketClient extends EventEmitter { } return this.tryWsSend( - wsKeySpotPublic, + WS_KEY_MAP.spotPublic, JSON.stringify({ topic: 'trade', event: 'sub', @@ -754,7 +710,7 @@ export class WebsocketClient extends EventEmitter { } return this.tryWsSend( - wsKeySpotPublic, + WS_KEY_MAP.spotPublic, JSON.stringify({ symbol, topic: 'realtimes', @@ -776,7 +732,7 @@ export class WebsocketClient extends EventEmitter { } return this.tryWsSend( - wsKeySpotPublic, + WS_KEY_MAP.spotPublic, JSON.stringify({ symbol, topic: 'kline_' + candleSize, @@ -791,6 +747,7 @@ export class WebsocketClient extends EventEmitter { //ws.send('{"symbol":"BTCUSDT","topic":"depth","event":"sub","params":{"binary":false}}'); //ws.send('{"symbol":"BTCUSDT","topic":"mergedDepth","event":"sub","params":{"binary":false,"dumpScale":1}}'); //ws.send('{"symbol":"BTCUSDT","topic":"diffDepth","event":"sub","params":{"binary":false}}'); + public subscribePublicSpotOrderbook( symbol: string, depth: 'full' | 'merge' | 'delta', @@ -831,6 +788,6 @@ export class WebsocketClient extends EventEmitter { if (dumpScale) { msg.params.dumpScale = dumpScale; } - return this.tryWsSend(wsKeySpotPublic, JSON.stringify(msg)); + return this.tryWsSend(WS_KEY_MAP.spotPublic, JSON.stringify(msg)); } }