publishing messages

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -3,6 +3,8 @@ import {
 	DLOBSubscriber,
 	DLOBSubscriptionConfig,
 	L2OrderBookGenerator,
+	MainnetPerpMarkets,
+	MainnetSpotMarkets,
 	MarketType,
 	groupL2,
 } from '@drift-labs/sdk';
@@ -16,16 +18,49 @@ type wsMarketL2Args = {
 	marketName: string;
 	depth: number;
 	includeVamm: boolean;
+  numVammOrders?: number;
 	grouping?: number;
 	fallbackL2Generators?: L2OrderBookGenerator[];
+  updateOnChange?: boolean;
 };
 
 export class DLOBSubscriberIO extends DLOBSubscriber {
 	public marketL2Args: wsMarketL2Args[] = [];
+  public lastSeenL2Formatted: Map<MarketType, Map<number, any>>;
 	io: Server;
 
 	constructor(config: DLOBSubscriptionConfig) {
 		super(config);
+
+    // Set up appropriate maps
+    this.lastSeenL2Formatted = new Map();
+    this.lastSeenL2Formatted.set(MarketType.SPOT, new Map());
+    this.lastSeenL2Formatted.set(MarketType.PERP, new Map());
+
+    // Add all active markets to the market L2Args 
+    for (const market of MainnetPerpMarkets) {
+      this.marketL2Args.push({
+        marketIndex: market.marketIndex,
+        marketType: MarketType.PERP,
+        marketName: market.symbol,
+        depth: 2,
+        includeVamm: true,
+        numVammOrders: 100,
+        updateOnChange: true,
+        fallbackL2Generators: []
+      });
+    }
+    for (const market of MainnetSpotMarkets) {
+      this.marketL2Args.push({
+        marketIndex: market.marketIndex,
+        marketType: MarketType.SPOT,
+        marketName: market.symbol,
+        depth: 2,
+        includeVamm: false,
+        updateOnChange: true,
+        fallbackL2Generators: []
+      });
+    }
 	}
 
 	override async updateDLOB(): Promise<void> {
@@ -38,7 +73,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 
 	getL2AndSendMsg(l2Args: wsMarketL2Args): void {
 		const grouping = l2Args.grouping;
-		const l2 = this.getL2(l2Args);
+    const l2 = this.getL2(l2Args);
 		let l2Formatted: any;
 		if (grouping) {
 			const groupingBN = new BN(grouping);
@@ -46,6 +81,11 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 		} else {
 			l2Formatted = l2WithBNToStrings(l2);
 		}
+
+    if (l2Args.updateOnChange && this.lastSeenL2Formatted.get(l2Args.marketType)?.get(l2Args.marketIndex) === JSON.stringify(l2Formatted)) {
+      return;
+    }
+    this.lastSeenL2Formatted.get(l2Args.marketType)?.set(l2Args.marketIndex, JSON.stringify(l2Formatted));
 		l2Formatted['marketName'] = l2Args.marketName;
 		l2Formatted['marketType'] = l2Args.marketType;
 		l2Formatted['marketIndex'] = l2Args.marketIndex;

src/index.ts

@@ -17,7 +17,6 @@ import {
 	DLOBOrder,
 	DLOBOrders,
 	DLOBOrdersCoder,
-	MarketType,
 	DLOBNode,
 	isVariant,
 	BN,
@@ -39,7 +38,6 @@ import {
 	errorHandler,
 	getPhoenixSubscriber,
 	getSerumSubscriber,
-	validateWsSubscribeMsg,
 	validateDlobQuery,
 } from './utils/utils';
 import { DLOBSubscriberIO } from './dlob-subscriber/DLOBSubscriberIO';
@@ -54,7 +52,7 @@ const sdkConfig = initialize({ env: process.env.ENV });
 
 const stateCommitment: Commitment = 'processed';
 const serverPort = process.env.PORT || 6969;
-const ORDERBOOK_UPDATE_INTERVAL = 100;
+const ORDERBOOK_UPDATE_INTERVAL = 1000;
 
 const rateLimitCallsPerSecond = process.env.RATE_LIMIT_CALLS_PER_SECOND
 	? parseInt(process.env.RATE_LIMIT_CALLS_PER_SECOND)
@@ -162,7 +160,6 @@ const main = async () => {
 		programID: clearingHousePublicKey,
 		accountSubscription: {
 			type: 'websocket',
-			resubTimeoutMs: 60000,
 		},
 		env: driftEnv,
 		userStats: true,
@@ -813,41 +810,6 @@ const main = async () => {
 		socket.on('disconnect', () => {
 			console.log('user disconnected');
 		});
-
-		socket.on('message', (msg: any) => {
-			const parsed = JSON.parse(msg);
-			if (parsed['type'] == 'subscribe') {
-				const valid = validateWsSubscribeMsg(parsed);
-				if (!valid.valid) {
-					socket.emit('message', {
-						type: 'error',
-						message: `Bad Request: Invalid subscribe message: ${valid.msg}`,
-					});
-					return;
-				}
-
-				dlobSubscriber.marketL2Args.push({
-					marketIndex: parseInt(parsed['marketIndex']),
-					marketType: isVariant(parsed['marketType'].toLowerCase(), 'spot')
-						? MarketType.SPOT
-						: MarketType.PERP,
-					marketName: parsed['marketName'],
-					depth: parseInt(parsed['depth']),
-					includeVamm: parsed['includeVamm'] === 'true',
-					grouping: parsed['grouping']
-						? parseInt(parsed['grouping'])
-						: undefined,
-					fallbackL2Generators: isVariant(parsed['marketType'], 'spot')
-						? [
-								`${parsed['includePhoenix']}`.toLowerCase() === 'true' &&
-									MARKET_SUBSCRIBERS[parseInt(parsed['marketIndex'])].phoenix,
-								`${parsed['includeSerum']}`.toLowerCase() === 'true' &&
-									MARKET_SUBSCRIBERS[parseInt(parsed['marketIndex'])].serum,
-						  ].filter((a) => !!a)
-						: [],
-				});
-			}
-		});
 	});
 };