trade/trade-drift-dlob
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

send the latest message for fast retrieval upon subscribe

Browse Source
This commit is contained in:
Nour Alharithi
2023-11-07 10:22:35 -08:00
parent 3a6ff7ca45
commit 3e112f4caf
1 changed files with 13 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/wsConnectionManager.ts
View File

@@ -24,7 +24,10 @@ const REDIS_PASSWORD = process.env.REDIS_PASSWORD;
async function main() { async function main() {
const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD); const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
const lastMessageRetriever = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
await redisClient.connect(); await redisClient.connect();
await lastMessageRetriever.connect();
const channelSubscribers = new Map<string, Set<WebSocket>>(); const channelSubscribers = new Map<string, Set<WebSocket>>();
const subscribedChannels = new Set<string>(); const subscribedChannels = new Set<string>();
@@ -34,6 +37,9 @@ async function main() {
subscribers.forEach((ws) => { subscribers.forEach((ws) => {
ws.send(JSON.stringify({ channel: subscribedChannel, data: message })); ws.send(JSON.stringify({ channel: subscribedChannel, data: message }));
}); });
// Save and persist last message
lastMessageRetriever.client.set(`last_update_${subscribedChannel}`, message);
}); });
wss.on('connection', (ws: WebSocket) => { wss.on('connection', (ws: WebSocket) => {
@@ -68,6 +74,13 @@ async function main() {
channelSubscribers.set(channel, subscribers); channelSubscribers.set(channel, subscribers);
} }
channelSubscribers.get(channel).add(ws); channelSubscribers.get(channel).add(ws);
// Fetch and send last message
const lastMessage = await lastMessageRetriever.client.get(`last_update_${channel}`);
if (lastMessage !== null) {
console.log('sending last message on new subscribe');
ws.send(JSON.stringify({ channel, data: lastMessage }));
}
break; break;
} }
case 'unsubscribe': { case 'unsubscribe': {