redis pub/sub fix

This commit is contained in:
tim
2025-05-06 22:27:14 -04:00
parent 7f4c301491
commit 008b6793d1
7 changed files with 68 additions and 27 deletions

View File

@@ -1,12 +1,19 @@
import {createClient} from "redis"; import {createClient} from "redis";
export const redis = createClient({
async function createRedisClient() {
const client = createClient({
url: process.env.DEXORDER_REDIS_URL || 'redis://localhost:6379', url: process.env.DEXORDER_REDIS_URL || 'redis://localhost:6379',
returnBuffers: false, returnBuffers: false,
}) })
client.on('error', (err) => console.log('Redis Client Error', err));
await client.connect();
return client
}
redis.on('error', (err) => console.log('Redis Client Error', err));
await redis.connect(); export const redis = await createRedisClient()
export const redisSubscriber = await createRedisClient()
export class CacheSet { export class CacheSet {

View File

@@ -5,11 +5,15 @@ export const chainInfo = JSON.parse(fs.readFileSync('../contract/version.json'))
console.log('chainInfo', chainInfo) console.log('chainInfo', chainInfo)
export async function joinChain( socket, chainId ) { export async function joinChain( socket, chainId ) {
try {
if (socket.chainId) if (socket.chainId)
socket.leave(socket.chainId) socket.leave(socket.chainId)
socket.join(chainId) socket.join(chainId)
socket.chainId = chainId socket.chainId = chainId
const items = await marks.items(chainId); const items = await marks.items(chainId);
for (let [token,mark] of items) for (let [token, mark] of items)
socket.emit('mark.usd', chainId, token, mark) socket.emit('mark.usd', chainId, token, mark)
} catch (e) {
console.error('joinChain', e)
}
} }

View File

@@ -6,6 +6,16 @@ import {metadata} from "./metadata.js";
export async function gib( chainId, owner, vault, tokenAmounts ) { export async function gib( chainId, owner, vault, tokenAmounts ) {
try {
return await doGib( chainId, owner, vault, tokenAmounts )
}
catch (e) {
console.error('gib failed', e)
}
}
export async function doGib( chainId, owner, vault, tokenAmounts ) {
if (!owner || !vault) return if (!owner || !vault) return
if (chainId === 421614) { if (chainId === 421614) {
// Arbitrum-Sepolia // Arbitrum-Sepolia

11
init.js Normal file
View File

@@ -0,0 +1,11 @@
export function initLog(app) {
process.on('uncaughtException', (err) => {
console.error('Uncaught Exception:', err);
process.exit(1); // Exit with code 1
});
process.on('unhandledRejection', (reason, promise) => {
console.error('Unhandled Rejection at:', promise, 'reason:', reason);
process.exit(1); // Exit with code 1
});
}

10
io.js
View File

@@ -1,7 +1,7 @@
import {createServer} from "http"; import {createServer} from "http";
import {Server} from "socket.io"; import {Server} from "socket.io";
import { createAdapter } from "@socket.io/redis-adapter"; import { createAdapter } from "@socket.io/redis-adapter";
import {redis} from "./cache.js"; import {redis, redisSubscriber} from "./cache.js";
import {fileURLToPath} from "url"; import {fileURLToPath} from "url";
import path from "path"; import path from "path";
import express from "express"; import express from "express";
@@ -29,12 +29,14 @@ app.set('views', path.join(__dirname, 'views')); // Set the views directory
app.use(express.static(path.join(__dirname, 'public'))); app.use(express.static(path.join(__dirname, 'public')));
app.use(cors()) app.use(cors())
app.use((err, req, res, next) => {
console.error(err.stack);
res.status(500).send('Something went wrong!');
});
initSnapShare(app) initSnapShare(app)
export const httpServer = createServer(app) export const httpServer = createServer(app)
export const io = new Server(httpServer, socketIoOptions) export const io = new Server(httpServer, socketIoOptions)
const pubClient = redis.duplicate(); const adapter = createAdapter(redis, redisSubscriber, {/*key:'socket.io'*/})
await pubClient.connect()
const adapter = createAdapter(pubClient, redis, {/*key:'socket.io'*/})
io.adapter(adapter) io.adapter(adapter)

View File

@@ -1,8 +1,10 @@
import 'dotenv/config' import 'dotenv/config'
import {httpServer} from "./io.js"; import {httpServer} from "./io.js";
import {initIO} from "./route.js"; import {initIO} from "./route.js";
import {initLog} from "./init.js";
initLog();
initIO(); initIO();
const port = parseInt(process.env.DEXORDER_PORT) || 3001; const port = parseInt(process.env.DEXORDER_PORT) || 3001;

11
pool.js
View File

@@ -19,18 +19,23 @@ export function unsubPools( socket, chainId, addresses ) {
export async function subOHLCs( socket, chainId, poolPeriods) { export async function subOHLCs( socket, chainId, poolPeriods) {
try {
console.log('subOHLCs', chainId, poolPeriods) console.log('subOHLCs', chainId, poolPeriods)
for(const key of poolPeriods) { for (const key of poolPeriods) {
const room = `${chainId}|${key}`; const room = `${chainId}|${key}`;
socket.join(room) socket.join(room)
let ohlc = await ohlcs.get(chainId,key); let ohlc = await ohlcs.get(chainId, key);
console.log('got ohlc', ohlc) console.log('got ohlc', ohlc)
if (typeof(ohlc)==='string') { if (typeof (ohlc) === 'string') {
ohlc = JSON.parse(ohlc) ohlc = JSON.parse(ohlc)
} }
socket.emit('ohlc', chainId, key, ohlc) socket.emit('ohlc', chainId, key, ohlc)
console.log('joined room', room) console.log('joined room', room)
} }
}
catch (e) {
console.error('subOHLCs', e)
}
} }
export function unsubOHLCs( socket, chainId, poolPeriods ) { export function unsubOHLCs( socket, chainId, poolPeriods ) {