From 304e1e9bd0ff4b0877c7201b2c7b6d78634034a3 Mon Sep 17 00:00:00 2001 From: Jack Waller Date: Wed, 22 May 2024 14:06:54 +1000 Subject: [PATCH] chore: update channel prefixes for pubsub --- src/dlob-subscriber/DLOBSubscriberIO.ts | 3 ++- src/index.ts | 17 +++++++---------- src/publishers/priorityFeesPublisher.ts | 8 ++++---- src/publishers/tradesPublisher.ts | 6 +++--- src/serverLite.ts | 5 +---- src/utils/utils.ts | 8 +++++--- src/wsConnectionManager.ts | 19 +++++++++++-------- 7 files changed, 33 insertions(+), 33 deletions(-) diff --git a/src/dlob-subscriber/DLOBSubscriberIO.ts b/src/dlob-subscriber/DLOBSubscriberIO.ts index b3e2ddd..e09fcf1 100644 --- a/src/dlob-subscriber/DLOBSubscriberIO.ts +++ b/src/dlob-subscriber/DLOBSubscriberIO.ts @@ -137,6 +137,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber { } getL2AndSendMsg(marketArgs: wsMarketArgs): void { + const clientPrefix = this.redisClient.forceGetClient().options.keyPrefix; const grouping = marketArgs.grouping; const { marketName, ...l2FuncArgs } = marketArgs; const l2 = this.getL2(l2FuncArgs); @@ -263,7 +264,7 @@ export class DLOBSubscriberIO extends DLOBSubscriber { }); this.redisClient.publish( - `orderbook_${marketType}_${marketArgs.marketIndex}`, + `${clientPrefix}orderbook_${marketType}_${marketArgs.marketIndex}`, l2Formatted ); this.redisClient.set( diff --git a/src/index.ts b/src/index.ts index 65bd714..60ed13c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -51,10 +51,7 @@ import { getDLOBProviderFromOrderSubscriber } from './dlobProvider'; require('dotenv').config(); // Reading in Redis env vars -const REDIS_CLIENTS = process.env.REDIS_CLIENTS?.replace(/^\[|\]$/g, '') - .split(',') - .map((clients) => clients.trim()) || ['DLOB']; - +const REDIS_CLIENTS = [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS]; console.log('Redis Clients:', REDIS_CLIENTS); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; @@ -294,8 +291,7 @@ const main = async (): Promise => { if (useRedis) { logger.info('Connecting to redis'); for (let i = 0; i < REDIS_CLIENTS.length; i++) { - const prefix = RedisClientPrefix[REDIS_CLIENTS[i]]; - redisClients.push(new RedisClient({ prefix })); + redisClients.push(new RedisClient({ prefix: REDIS_CLIENTS[i] })); } for (let i = 0; i < sdkConfig.SPOT_MARKETS.length; i++) { @@ -402,11 +398,12 @@ const main = async (): Promise => { const { marketIndex, marketType } = req.query; const fees = await redisClients - .find( - (client) => + .find((client) => { + return ( client.forceGetClient().options.keyPrefix === RedisClientPrefix.DLOB_HELIUS - ) + ); + }) .getRaw(`priorityFees_${marketType}_${marketIndex}`); if (fees) { @@ -521,7 +518,7 @@ const main = async (): Promise => { let accountFlag = false; if (includeAccounts) { - accountFlag = includeAccounts === 'true'; + accountFlag = (includeAccounts as string).toLowerCase() === 'true'; } let topMakers: string[]; diff --git a/src/publishers/priorityFeesPublisher.ts b/src/publishers/priorityFeesPublisher.ts index 149bc66..241210d 100644 --- a/src/publishers/priorityFeesPublisher.ts +++ b/src/publishers/priorityFeesPublisher.ts @@ -21,7 +21,7 @@ require('dotenv').config(); const stateCommitment: Commitment = 'confirmed'; const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; - +const redisClientPrefix = RedisClientPrefix.DLOB_HELIUS; // Set up express for health checks const app = express(); @@ -135,7 +135,7 @@ class PriorityFeeSubscriber { dataPerp.forEach((result: any) => { const marketIndex = parseInt(result['id']); this.redisClient.publish( - `priorityFees_perp_${marketIndex}`, + `${redisClientPrefix}priorityFees_perp_${marketIndex}`, result.result['priorityFeeLevels'] ); this.redisClient.set( @@ -147,7 +147,7 @@ class PriorityFeeSubscriber { dataSpot.forEach((result: any) => { const marketIndex = parseInt(result['id']) - 100; this.redisClient.publish( - `priorityFees_spot_${marketIndex}`, + `${redisClientPrefix}priorityFees_spot_${marketIndex}`, result.result['priorityFeeLevels'] ); this.redisClient.set( @@ -165,7 +165,7 @@ const main = async () => { }); const redisClient = new RedisClient({ - prefix: RedisClientPrefix.DLOB_HELIUS, + prefix: redisClientPrefix, }); await redisClient.connect(); diff --git a/src/publishers/tradesPublisher.ts b/src/publishers/tradesPublisher.ts index 71b24fd..186f8bb 100644 --- a/src/publishers/tradesPublisher.ts +++ b/src/publishers/tradesPublisher.ts @@ -29,7 +29,7 @@ import { fromEvent, filter, map } from 'rxjs'; require('dotenv').config(); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; const commitHash = process.env.COMMIT; - +const redisClientPrefix = RedisClientPrefix.DLOB; //@ts-ignore const sdkConfig = initialize({ env: process.env.ENV }); @@ -67,7 +67,7 @@ const main = async () => { env: driftEnv, }); - const redisClient = new RedisClient({ prefix: RedisClientPrefix.DLOB }); + const redisClient = new RedisClient({ prefix: redisClientPrefix }); await redisClient.connect(); const slotSubscriber = new SlotSubscriber(connection, {}); @@ -180,7 +180,7 @@ const main = async () => { ) .subscribe((fillEvent) => { redisClient.publish( - `trades_${fillEvent.marketType}_${fillEvent.marketIndex}`, + `${redisClientPrefix}trades_${fillEvent.marketType}_${fillEvent.marketIndex}`, fillEvent ); }); diff --git a/src/serverLite.ts b/src/serverLite.ts index ca36ca0..7d606d2 100644 --- a/src/serverLite.ts +++ b/src/serverLite.ts @@ -30,10 +30,7 @@ import { RedisClient, RedisClientPrefix } from '@drift/common'; require('dotenv').config(); // Reading in Redis env vars -const REDIS_CLIENTS = process.env.REDIS_CLIENTS?.replace(/^\[|\]$/g, '') - .split(',') - .map((clients) => clients.trim()) || ['DLOB']; - +const REDIS_CLIENTS = [RedisClientPrefix.DLOB, RedisClientPrefix.DLOB_HELIUS]; console.log('Redis Clients:', REDIS_CLIENTS); const driftEnv = (process.env.ENV || 'devnet') as DriftEnv; diff --git a/src/utils/utils.ts b/src/utils/utils.ts index 70c839f..e5dd5db 100644 --- a/src/utils/utils.ts +++ b/src/utils/utils.ts @@ -347,11 +347,13 @@ export const getAccountFromId = async ( ) => { return Promise.all( topMakers.map(async (accountId) => { - const userMap = await userMapClient.getRaw(accountId); - if (userMap) { + const userAccountEncoded = await userMapClient.getRaw(accountId); + if (userAccountEncoded) { return { accountId, - account: decodeUser(Buffer.from(userMap.split('::')[1], 'base64')), + account: decodeUser( + Buffer.from(userAccountEncoded.split('::')[1], 'base64') + ), }; } return { diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index df192a7..2bfb2c6 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -6,7 +6,7 @@ import { WebSocket, WebSocketServer } from 'ws'; import { sleep } from './utils/utils'; import { register, Gauge } from 'prom-client'; import { DriftEnv, PerpMarkets, SpotMarkets } from '@drift-labs/sdk'; -import { RedisClient } from '@drift/common'; +import { RedisClient, RedisClientPrefix } from '@drift/common'; // Set up env constants require('dotenv').config(); @@ -34,6 +34,7 @@ const WS_PORT = process.env.WS_PORT || '3000'; console.log(`WS LISTENER PORT : ${WS_PORT}`); const MAX_BUFFERED_AMOUNT = 300000; +const CHANNEL_PREFIX = RedisClientPrefix.DLOB; const safeGetRawChannelFromMessage = (message: any): string => { return message?.channel; @@ -64,11 +65,11 @@ const getRedisChannelFromMessage = (message: any): string => { switch (channel.toLowerCase()) { case 'trades': - return `trades_${marketType}_${marketIndex}`; + return `${CHANNEL_PREFIX}trades_${marketType}_${marketIndex}`; case 'orderbook': - return `orderbook_${marketType}_${marketIndex}`; + return `${CHANNEL_PREFIX}orderbook_${marketType}_${marketIndex}`; case 'priorityfees': - return `priorityFees_${marketType}_${marketIndex}`; + return `${CHANNEL_PREFIX}priorityFees_${marketType}_${marketIndex}`; case undefined: default: throw new Error('Bad channel specified'); @@ -77,7 +78,7 @@ const getRedisChannelFromMessage = (message: any): string => { async function main() { const redisClient = new RedisClient({}); - const lastMessageRetriever = new RedisClient({}); + const lastMessageRetriever = new RedisClient({ prefix: CHANNEL_PREFIX }); await redisClient.connect(); await lastMessageRetriever.connect(); @@ -183,11 +184,13 @@ async function main() { ); // Fetch and send last message if (redisChannel.includes('orderbook')) { - const lastUpdateChannel = `last_update_${redisChannel}`; - const lastMessage = await lastMessageRetriever.get( + const lastUpdateChannel = `last_update_${redisChannel}`.replace( + CHANNEL_PREFIX, + '' + ); + const lastMessage = await lastMessageRetriever.getRaw( lastUpdateChannel ); - if (lastMessage) { ws.send( JSON.stringify({