separate out publisher

package.json

@@ -49,6 +49,7 @@
     "clean": "rm -rf lib",
     "start": "node lib/index.js",
     "dev": "ts-node src/index.ts",
+    "ws-publish": "ts-node src/wsPublish.ts",
     "manager": "ts-node src/wsConnectionManager.ts",
     "dev:inspect": "yarn build && node --inspect ./lib/index.js",
     "dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js",

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -9,8 +9,6 @@ import {
 	groupL2,
 } from '@drift-labs/sdk';
 import { getOracleForMarket, l2WithBNToStrings } from '../utils/utils';
-import { Server } from 'socket.io';
-import { getIOServer } from '..';
 import { RedisClient } from '../utils/redisClient';
 
 type wsMarketL2Args = {
@@ -28,7 +26,6 @@ type wsMarketL2Args = {
 export class DLOBSubscriberIO extends DLOBSubscriber {
 	public marketL2Args: wsMarketL2Args[] = [];
 	public lastSeenL2Formatted: Map<MarketType, Map<number, any>>;
-	io: Server;
 	redisClient: RedisClient;
 
 	constructor(config: DLOBSubscriptionConfig & { redisClient: RedisClient }) {
@@ -67,7 +64,6 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 	}
 
 	override async updateDLOB(): Promise<void> {
-		this.io = getIOServer();
 		await super.updateDLOB();
 		for (const l2Args of this.marketL2Args) {
 			this.getL2AndSendMsg(l2Args);
@@ -85,12 +81,13 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			l2Formatted = l2WithBNToStrings(l2);
 		}
 
-		if (
+		if (l2Args.updateOnChange) {
-			l2Args.updateOnChange
+			if (
-		) {
+				this.lastSeenL2Formatted
-			if (this.lastSeenL2Formatted
+					.get(l2Args.marketType)
-				.get(l2Args.marketType)
+					?.get(l2Args.marketIndex) === JSON.stringify(l2Formatted)
-				?.get(l2Args.marketIndex) === JSON.stringify(l2Formatted)) return;
+			)
+				return;
 		}
 		this.lastSeenL2Formatted
 			.get(l2Args.marketType)

src/index.ts

@@ -23,12 +23,12 @@ import {
 	groupL2,
 	Wallet,
 	UserStatsMap,
+	DLOBSubscriber,
 } from '@drift-labs/sdk';
 
 import { logger, setLogLevel } from './utils/logger';
 
 import * as http from 'http';
-import { Server } from 'socket.io';
 import {
 	l2WithBNToStrings,
 	sleep,
@@ -40,10 +40,8 @@ import {
 	getSerumSubscriber,
 	validateDlobQuery,
 } from './utils/utils';
-import { DLOBSubscriberIO } from './dlob-subscriber/DLOBSubscriberIO';
 import { handleResponseTime } from './core/middleware';
 import { handleHealthCheck } from './core/metrics';
-import { RedisClient } from './utils/redisClient';
 
 require('dotenv').config();
 const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
@@ -103,11 +101,6 @@ app.use((req, _res, next) => {
 
 app.use(errorHandler);
 const server = http.createServer(app);
-const io = new Server(server);
-
-export function getIOServer(): Server {
-	return io;
-}
 
 const opts = program.opts();
 setLogLevel(opts.debug ? 'debug' : 'info');
@@ -204,15 +197,11 @@ const main = async () => {
 	);
 	await userStatsMap.subscribe();
 
-	const redisClient = new RedisClient('localhost', '6379');
+	const dlobSubscriber = new DLOBSubscriber({
-	await redisClient.connect();
-
-	const dlobSubscriber = new DLOBSubscriberIO({
 		driftClient,
 		dlobSource: userMap,
 		slotSource: slotSubscriber,
 		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
-		redisClient,
 	});
 	await dlobSubscriber.subscribe();
 
@@ -826,14 +815,6 @@ const main = async () => {
 	server.listen(serverPort, () => {
 		logger.info(`DLOB server listening on port http://localhost:${serverPort}`);
 	});
-
-	io.on('connection', (socket) => {
-		console.log('a user connected');
-
-		socket.on('disconnect', () => {
-			console.log('user disconnected');
-		});
-	});
 };
 
 async function recursiveTryCatch(f: () => void) {

src/wsConnectionManager.ts

@@ -14,8 +14,11 @@ app.set('trust proxy', 1);
 const server = http.createServer(app);
 const io = new Server(server);
 
+const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
+const REDIS_PORT = process.env.REDIS_PORT || '6379';
+
 async function main() {
-	const redisClient = new RedisClient('localhost', '6379');
+	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT);
 	await redisClient.connect();
 
 	io.on('connection', (socket) => {

src/wsPublish.ts

@@ -0,0 +1,121 @@
+import { program } from 'commander';
+
+import { Connection, Commitment, PublicKey, Keypair } from '@solana/web3.js';
+
+import {
+	DriftClient,
+	initialize,
+	DriftEnv,
+	SlotSubscriber,
+	UserMap,
+	Wallet,
+	UserStatsMap,
+} from '@drift-labs/sdk';
+
+import { logger, setLogLevel } from './utils/logger';
+
+import { sleep } from './utils/utils';
+import { DLOBSubscriberIO } from './dlob-subscriber/DLOBSubscriberIO';
+import { RedisClient } from './utils/redisClient';
+
+require('dotenv').config();
+const driftEnv = (process.env.ENV || 'devnet') as DriftEnv;
+const commitHash = process.env.COMMIT;
+const REDIS_HOST = process.env.REDIS_HOST || 'localhost';
+const REDIS_PORT = process.env.REDIS_PORT || '6379';
+
+//@ts-ignore
+const sdkConfig = initialize({ env: process.env.ENV });
+
+const stateCommitment: Commitment = 'processed';
+const ORDERBOOK_UPDATE_INTERVAL = 1000;
+
+let driftClient: DriftClient;
+
+const opts = program.opts();
+setLogLevel(opts.debug ? 'debug' : 'info');
+
+const endpoint = process.env.ENDPOINT;
+const wsEndpoint = process.env.WS_ENDPOINT;
+logger.info(`RPC endpoint: ${endpoint}`);
+logger.info(`WS endpoint:  ${wsEndpoint}`);
+logger.info(`DriftEnv:     ${driftEnv}`);
+logger.info(`Commit:       ${commitHash}`);
+
+const main = async () => {
+	const wallet = new Wallet(new Keypair());
+	const clearingHousePublicKey = new PublicKey(sdkConfig.DRIFT_PROGRAM_ID);
+
+	const connection = new Connection(endpoint, {
+		wsEndpoint: wsEndpoint,
+		commitment: stateCommitment,
+	});
+
+	driftClient = new DriftClient({
+		connection,
+		wallet,
+		programID: clearingHousePublicKey,
+		accountSubscription: {
+			type: 'websocket',
+		},
+		env: driftEnv,
+		userStats: true,
+	});
+
+	const slotSubscriber = new SlotSubscriber(connection, {});
+
+	const lamportsBalance = await connection.getBalance(wallet.publicKey);
+	logger.info(
+		`DriftClient ProgramId: ${driftClient.program.programId.toBase58()}`
+	);
+	logger.info(`Wallet pubkey: ${wallet.publicKey.toBase58()}`);
+	logger.info(` . SOL balance: ${lamportsBalance / 10 ** 9}`);
+
+	await driftClient.subscribe();
+	driftClient.eventEmitter.on('error', (e) => {
+		logger.info('clearing house error');
+		logger.error(e);
+	});
+
+	await slotSubscriber.subscribe();
+
+	const userMap = new UserMap(
+		driftClient,
+		driftClient.userAccountSubscriptionConfig,
+		false
+	);
+	await userMap.subscribe();
+	const userStatsMap = new UserStatsMap(
+		driftClient,
+		driftClient.userStatsAccountSubscriptionConfig
+	);
+	await userStatsMap.subscribe();
+
+	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT);
+	await redisClient.connect();
+
+	const dlobSubscriber = new DLOBSubscriberIO({
+		driftClient,
+		dlobSource: userMap,
+		slotSource: slotSubscriber,
+		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
+		redisClient,
+	});
+	await dlobSubscriber.subscribe();
+
+	console.log('DLOBSubscriber Publishing Messages');
+};
+
+async function recursiveTryCatch(f: () => void) {
+	try {
+		await f();
+	} catch (e) {
+		console.error(e);
+		await sleep(15000);
+		await recursiveTryCatch(f);
+	}
+}
+
+recursiveTryCatch(() => main());
+
+export { sdkConfig, endpoint, wsEndpoint, driftEnv, commitHash, driftClient };