From 5a0b9d8aaee0ba8d08c93579d0f35298d9f36283 Mon Sep 17 00:00:00 2001 From: "G.Ambatte" Date: Fri, 9 Oct 2026 20:28:04 +1300 Subject: [PATCH] Add debounced call to unsubscribe --- server/eventStreamSource.js | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/server/eventStreamSource.js b/server/eventStreamSource.js index bfdcc8099..9d0574d40 100644 --- a/server/eventStreamSource.js +++ b/server/eventStreamSource.js @@ -1,10 +1,15 @@ import { EventEmitter } from 'events'; +import _ from 'lodash'; import { nanoid } from 'nanoid'; +const UNSUB_DELAY = 60000; //ms + const Stream = new EventEmitter; // Create array of stream subscribers -const subscribers = []; +let subscribers = []; + +const debounceUnsub = _.debounce((id)=>{ unsubscribe(id); }, UNSUB_DELAY, { leading: false, trailing: true }); // Listener functions const subscribe = function(res){ @@ -12,18 +17,29 @@ const subscribe = function(res){ subscribers.push({ id, 'stream': res }); res.write(`event: subscribe\ndata: { id: ${id} }\n\n`); + + debounceUnsub(id); return id; }; const unsubscribe = function(id){ + subscribers + .filter((sub)=>{return sub.id == id;}) + .forEach((sub)=>{ + sub.stream.write(`event: unsubscribe\ndata: { id: ${id} }\n\n`); + }); + subscribers = subscribers.filter((sub)=>{ return sub.id != id; }); }; // Create global sendUpdate listener Stream.on('sendUpdate', (event, data)=>{ console.log('Event:', event, '\nData:', data); - subscribers.forEach((sub, idx)=>{ + subscribers.forEach((sub)=>{ sub?.stream?.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); + + // invoke debounced unsub to delay execution + debounceUnsub(sub.id); }); });