jan/openreplay
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(assist): small ref

Browse source
This commit is contained in:
Alexander 2025-03-26 16:59:09 +01:00
parent fb58f3f155
commit 0db952e764
4 changed files with 12 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
ee/assist/servers/websocket-cluster.js
View file

@ -21,7 +21,9 @@ const {createAdapter} = require("@socket.io/redis-adapter");
const {createClient} = require("redis");
const REDIS_URL = (process.env.REDIS_URL || "localhost:6379").replace(/((^\w+:|^)\/\/|^)/, 'redis://');
const pubClient = createClient({url: REDIS_URL});
pubClient.on("error", (error) => logger.error(`Pub redis client error : ${error}`));
const subClient = pubClient.duplicate();
subClient.on("error", (error) => logger.error(`Sub redis client error : ${error}`));
logger.info(`Using Redis: ${REDIS_URL}`);
const wsRouter = express.Router();

11
ee/assist/utils/httpHandlers.js
View file

@ -42,17 +42,6 @@ const respond = function (req, res, data) {
}
const getParticularSession = async function (roomId, filters, all=false) {
// let connected_sockets = await fetchSockets(roomId, all);
// if (connected_sockets.length === 0) {
// return null;
// }
// let sessInfo;
// for (let item of connected_sockets) {
// if (item.handshake.query.identity === IDENTITIES.session && item.handshake.query.sessionInfo) {
// sessInfo = item.handshake.query.sessionInfo;
// break;
// }
// }
let sessInfo = await getSessionFromCache(roomId);
if (!sessInfo) {
return null;

3
ee/assist/utils/socketHandlers.js
View file

@ -189,6 +189,9 @@ async function onUpdateEvent(socket, ...args) {
args[0] = updateSessionData(socket, args[0])
socket.handshake.query.sessionInfo = deepMerge(socket.handshake.query.sessionInfo, args[0]?.data, {tabId: args[0]?.meta?.tabId});
// update session cache
await addSessionToCache(socket.handshake.query.sessId, socket.handshake.query.sessionInfo);
// Update sessionInfo for all agents in the room
const connected_sockets = await fetchSockets(socket.handshake.query.roomId);
for (let item of connected_sockets) {

14
ee/assist/utils/wsServer.js
View file

@ -15,16 +15,16 @@ if (useRedis) {
const {createClient} = require("redis");
const REDIS_URL = (process.env.REDIS_URL || "localhost:6379").replace(/((^\w+:|^)\/\/|^)/, 'redis://');
redisClient = createClient({url: REDIS_URL});
redisClient.on("error", (error) => logger.error(`Redis error : ${error}`));
redisClient.on("error", (error) => logger.error(`Redis cache error : ${error}`));
void redisClient.connect();
}
const addSessionToCache = async function (sessionID, sessionData) {
try {
await redisClient.set(`active_sessions:${sessionID}`, JSON.stringify(sessionData), 'EX', 3600); // 60 minutes
console.log(`Session ${sessionID} stored in Redis`);
logger.debug(`Session ${sessionID} stored in Redis`);
} catch (error) {
console.log(error);
logger.error(error);
}
}
@ -32,12 +32,12 @@ const getSessionFromCache = async function (sessionID) {
try {
const sessionData = await redisClient.get(`active_sessions:${sessionID}`);
if (sessionData) {
console.log(`Session ${sessionID} retrieved from Redis`);
logger.debug(`Session ${sessionID} retrieved from Redis`);
return JSON.parse(sessionData);
}
return null;
} catch (error) {
console.log(error);
logger.error(error);
return null;
}
}
@ -45,9 +45,9 @@ const getSessionFromCache = async function (sessionID) {
const removeSessionFromCache = async function (sessionID) {
try {
await redisClient.del(`active_sessions:${sessionID}`);
console.log(`Session ${sessionID} removed from Redis`);
logger.debug(`Session ${sessionID} removed from Redis`);
} catch (error) {
console.log(error);
logger.error(error);
}
}