have redis working

src/dlob-subscriber/DLOBSubscriberIO.ts

@@ -11,6 +11,7 @@ import {
 import { getOracleForMarket, l2WithBNToStrings } from '../utils/utils';
 import { Server } from 'socket.io';
 import { getIOServer } from '..';
+import { RedisClient } from '../utils/redisClient';
 
 type wsMarketL2Args = {
 	marketIndex: number;
@@ -28,9 +29,11 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 	public marketL2Args: wsMarketL2Args[] = [];
 	public lastSeenL2Formatted: Map<MarketType, Map<number, any>>;
 	io: Server;
+	redisClient: RedisClient;
 
-	constructor(config: DLOBSubscriptionConfig) {
+	constructor(config: DLOBSubscriptionConfig & { redisClient: RedisClient }) {
 		super(config);
+		this.redisClient = config.redisClient;
 
 		// Set up appropriate maps
 		this.lastSeenL2Formatted = new Map();
@@ -43,7 +46,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 				marketIndex: market.marketIndex,
 				marketType: MarketType.PERP,
 				marketName: market.symbol,
-				depth: 2,
+				depth: -1,
 				includeVamm: true,
 				numVammOrders: 100,
 				updateOnChange: true,
@@ -55,7 +58,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 				marketIndex: market.marketIndex,
 				marketType: MarketType.SPOT,
 				marketName: market.symbol,
-				depth: 2,
+				depth: -1,
 				includeVamm: false,
 				updateOnChange: true,
 				fallbackL2Generators: [],
@@ -101,6 +104,9 @@ export class DLOBSubscriberIO extends DLOBSubscriber {
 			l2Args.marketType,
 			l2Args.marketIndex
 		);
-		this.io.emit('l2Update', l2Formatted);
+		this.redisClient.client.publish(
+			l2Args.marketName,
+			JSON.stringify(l2Formatted)
+		);
 	}
 }

src/index.ts

@@ -43,6 +43,7 @@ import {
 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;
@@ -130,13 +131,21 @@ const initializeAllMarketSubscribers = async (driftClient: DriftClient) => {
 		};
 
 		if (market.phoenixMarket) {
-			const phoenixSubscriber = getPhoenixSubscriber(driftClient, market);
+			const phoenixSubscriber = getPhoenixSubscriber(
+				driftClient,
+				market,
+				sdkConfig
+			);
 			await phoenixSubscriber.subscribe();
 			markets[market.marketIndex].phoenix = phoenixSubscriber;
 		}
 
 		if (market.serumMarket) {
-			const serumSubscriber = getSerumSubscriber(driftClient, market);
+			const serumSubscriber = getSerumSubscriber(
+				driftClient,
+				market,
+				sdkConfig
+			);
 			await serumSubscriber.subscribe();
 			markets[market.marketIndex].serum = serumSubscriber;
 		}
@@ -195,11 +204,15 @@ const main = async () => {
 	);
 	await userStatsMap.subscribe();
 
+	const redisClient = new RedisClient('localhost', '6379');
+	await redisClient.connect();
+
 	const dlobSubscriber = new DLOBSubscriberIO({
 		driftClient,
 		dlobSource: userMap,
 		slotSource: slotSubscriber,
 		updateFrequency: ORDERBOOK_UPDATE_INTERVAL,
+		redisClient,
 	});
 	await dlobSubscriber.subscribe();
 
@@ -385,6 +398,8 @@ const main = async () => {
 		try {
 			const { marketName, marketIndex, marketType } = req.query;
 			const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
+				driftClient,
+				driftEnv,
 				marketType as string,
 				marketIndex as string,
 				marketName as string
@@ -454,6 +469,8 @@ const main = async () => {
 			} = req.query;
 
 			const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
+				driftClient,
+				driftEnv,
 				marketType as string,
 				marketIndex as string,
 				marketName as string
@@ -561,6 +578,8 @@ const main = async () => {
 			} = req.query;
 
 			const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
+				driftClient,
+				driftEnv,
 				marketType as string,
 				marketIndex as string,
 				marketName as string
@@ -674,6 +693,8 @@ const main = async () => {
 			const l2s = normedParams.map((normedParam) => {
 				const { normedMarketType, normedMarketIndex, error } =
 					validateDlobQuery(
+						driftClient,
+						driftEnv,
 						normedParam['marketType'] as string,
 						normedParam['marketIndex'] as string,
 						normedParam['marketName'] as string
@@ -760,6 +781,8 @@ const main = async () => {
 			const { marketName, marketIndex, marketType, includeOracle } = req.query;
 
 			const { normedMarketType, normedMarketIndex, error } = validateDlobQuery(
+				driftClient,
+				driftEnv,
 				marketType as string,
 				marketIndex as string,
 				marketName as string

src/utils/redisClient.ts

@@ -0,0 +1,124 @@
+import Redis, { RedisOptions } from 'ioredis';
+import { sleep } from './utils';
+
+export const getRedisClient = (
+	host?: string,
+	port?: string,
+	password?: string,
+	db?: number,
+	opts?: RedisOptions
+): Redis => {
+	if (host && port) {
+		console.log(`Connecting to configured redis:: ${host}:${port}`);
+
+		const redisClient = new Redis({
+			host: host,
+			port: parseInt(port, 10),
+			password: password,
+			...(opts ?? {}),
+			retryStrategy: (times) => {
+				const delay = Math.min(times * 1000, 10000);
+				console.log(
+					`Reconnecting to Redis in ${delay}ms... (retries: ${times})`
+				);
+				return delay;
+			},
+			reconnectOnError: (err) => {
+				const targetError = 'ECONNREFUSED';
+				if (err.message.includes(targetError)) {
+					console.log(
+						`Redis error: ${targetError}. Attempting to reconnect...`
+					);
+					return true;
+				}
+				return false;
+			},
+			maxRetriesPerRequest: null, // unlimited retries
+		});
+
+		redisClient.on('connect', () => {
+			console.log('Connected to Redis.');
+		});
+
+		redisClient.on('error', (err) => {
+			console.error('Redis error:', err);
+		});
+
+		redisClient.on('reconnecting', () => {
+			console.log('Reconnecting to Redis...');
+		});
+
+		return redisClient;
+	}
+
+	console.log(`Using default redis`);
+	return new Redis({ db });
+};
+
+/**
+ * Wrapper around the redis client.
+ *
+ * You can hover over the underlying redis client methods for explanations of the methods, but will also include links to DOCS for some important concepts below:
+ *
+ * zRange, zRangeByScore etc.:
+ * - All of the "z" methods are methods that use sorted sets.
+ * - Sorted sets are explained here : https://redis.io/docs/data-types/sorted-sets/
+ */
+export class RedisClient {
+	public client: Redis;
+
+	connectionPromise: Promise<void>;
+
+	constructor(
+		host?: string,
+		port?: string,
+		password?: string,
+		db?: number,
+		opts?: RedisOptions
+	) {
+		this.client = getRedisClient(host, port, password, db, opts);
+	}
+
+	/**
+	 * Should avoid using this unless necessary.
+	 * @returns
+	 */
+	public forceGetClient() {
+		return this.client;
+	}
+
+	public get connected() {
+		return this?.client?.status === 'ready';
+	}
+
+	async connect() {
+		if (this.client.status === 'ready' || this.client.status === 'connect') {
+			return;
+		}
+
+		if (this.client.status === 'connecting') {
+			await sleep(100);
+			return this.connect(); // recursive call to check again
+		}
+
+		try {
+			await this.client.connect();
+		} catch (e) {
+			console.error(e);
+		}
+	}
+
+	disconnect() {
+		this.client.disconnect();
+	}
+
+	private assertConnected() {
+		if (!this.connected) {
+			throw 'Redis client not connected';
+		}
+	}
+
+	private getPrefixForMetrics(key: string) {
+		return key.split(':')?.[0];
+	}
+}

src/utils/utils.ts

@@ -1,5 +1,6 @@
 import {
 	DriftClient,
+	DriftEnv,
 	L2OrderBook,
 	MarketType,
 	PerpMarkets,
@@ -12,7 +13,6 @@ import {
 } from '@drift-labs/sdk';
 import { logger } from './logger';
 import { NextFunction, Request, Response } from 'express';
-import { driftClient, driftEnv, sdkConfig } from '..';
 
 export const l2WithBNToStrings = (l2: L2OrderBook): any => {
 	for (const key of Object.keys(l2)) {
@@ -111,7 +111,8 @@ export const normalizeBatchQueryParams = (rawParams: {
 };
 
 export const validateWsSubscribeMsg = (
-	msg: any
+	msg: any,
+	sdkConfig: any
 ): { valid: boolean; msg?: string } => {
 	const maxPerpMarketIndex = Math.max(
 		...sdkConfig.PERP_MARKETS.map((m) => m.marketIndex)
@@ -158,6 +159,8 @@ export const validateWsSubscribeMsg = (
 };
 
 export const validateDlobQuery = (
+	driftClient: DriftClient,
+	driftEnv: DriftEnv,
 	marketType?: string,
 	marketIndex?: string,
 	marketName?: string
@@ -248,7 +251,8 @@ export function errorHandler(
 
 export const getPhoenixSubscriber = (
 	driftClient: DriftClient,
-	marketConfig: SpotMarketConfig
+	marketConfig: SpotMarketConfig,
+	sdkConfig
 ): PhoenixSubscriber => {
 	return new PhoenixSubscriber({
 		connection: driftClient.connection,
@@ -262,7 +266,8 @@ export const getPhoenixSubscriber = (
 
 export const getSerumSubscriber = (
 	driftClient: DriftClient,
-	marketConfig: SpotMarketConfig
+	marketConfig: SpotMarketConfig,
+	sdkConfig
 ): SerumSubscriber => {
 	return new SerumSubscriber({
 		connection: driftClient.connection,

src/wsConnectionManager.ts

@@ -0,0 +1,49 @@
+import cors from 'cors';
+import express from 'express';
+import * as http from 'http';
+import { Server } from 'socket.io';
+import compression from 'compression';
+import { sleep } from './utils/utils';
+import { RedisClient } from './utils/redisClient';
+
+const app = express();
+app.use(cors({ origin: '*' }));
+app.use(compression());
+app.set('trust proxy', 1);
+
+const server = http.createServer(app);
+const io = new Server(server);
+
+async function main() {
+	const redisClient = new RedisClient('localhost', '6379');
+	await redisClient.connect();
+
+	io.on('connection', (socket) => {
+		socket.on('subscribe', (channel) => {
+			console.log('Subscribing to channel', channel);
+			redisClient.client.subscribe(channel);
+			redisClient.client.on('message', (subscribedChannel, message) => {
+				console.log(message);
+				if (subscribedChannel === channel) {
+					socket.emit(channel, JSON.parse(message));
+				}
+			});
+		});
+	});
+
+	server.listen('3000', () => {
+		console.log('connection manager running on 3000');
+	});
+}
+
+async function recursiveTryCatch(f: () => void) {
+	try {
+		await f();
+	} catch (e) {
+		console.error(e);
+		await sleep(15000);
+		await recursiveTryCatch(f);
+	}
+}
+
+recursiveTryCatch(() => main());