From a51231535c276300af78a818835c6d6b1bb2cd2e Mon Sep 17 00:00:00 2001 From: "G.Ambatte" Date: Sat, 10 Oct 2026 15:54:59 +1300 Subject: [PATCH] Incremental commit --- server/app.js | 2 +- server/eventStreamSource.js | 38 +++++++++++++++++++++++++------------ 2 files changed, 27 insertions(+), 13 deletions(-) diff --git a/server/app.js b/server/app.js index 9556b8a49..c834c1616 100644 --- a/server/app.js +++ b/server/app.js @@ -203,7 +203,7 @@ export default async function createApp(vite) { 'Content-Encoding' : 'none' }); ; - Stream.subscribe(res); + Stream.subscribe(req.params.id, res); }); diff --git a/server/eventStreamSource.js b/server/eventStreamSource.js index 9d0574d40..2f4acd312 100644 --- a/server/eventStreamSource.js +++ b/server/eventStreamSource.js @@ -12,21 +12,27 @@ let subscribers = []; const debounceUnsub = _.debounce((id)=>{ unsubscribe(id); }, UNSUB_DELAY, { leading: false, trailing: true }); // Listener functions -const subscribe = function(res){ - const id = nanoid(); - subscribers.push({ id, 'stream': res }); +const subscribe = function(shareId, res){ + const subscriber = { + id : nanoid(), // NanoID is assumed globally unique + shareId, + stream : res, + time : new Date + }; - res.write(`event: subscribe\ndata: { id: ${id} }\n\n`); + subscribers.push(subscriber); - debounceUnsub(id); - return id; + Stream.emit('sendUpdate', 'subscribe', { id: subscriber.id, shareId: subscriber.shareId, time: subscriber.time }); + + debounceUnsub(subscriber.id); + return subscriber.id; }; const unsubscribe = function(id){ subscribers .filter((sub)=>{return sub.id == id;}) .forEach((sub)=>{ - sub.stream.write(`event: unsubscribe\ndata: { id: ${id} }\n\n`); + Stream.emit('sendUpdate', 'unsubscribe', { id: sub.id, shareId: sub.shareId, time: new Date }); }); subscribers = subscribers.filter((sub)=>{ return sub.id != id; }); @@ -35,14 +41,22 @@ const unsubscribe = function(id){ // Create global sendUpdate listener Stream.on('sendUpdate', (event, data)=>{ console.log('Event:', event, '\nData:', data); - subscribers.forEach((sub)=>{ - sub?.stream?.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); + subscribers + .filter((sub)=>{return data.shareId == sub.shareId; }) + .forEach((sub)=>{ + sub?.stream?.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); - // invoke debounced unsub to delay execution - debounceUnsub(sub.id); - }); + // invoke debounced unsub to delay execution + debounceUnsub(sub.id); + }); }); +// DEBUG: Report subscriber count periodically +const REPORT_DELAY = 1000; //ms +const reportIntervalId = setInterval(()=>{ + console.log('Subscriber count:', subscribers.length); +}, REPORT_DELAY); + export default { emit : function(event) {return Stream.emit(event, ...([...arguments].slice(1)));}, // Arguments doesn't work for arrow functions on : (event, listener)=>{return Stream.on(event, listener);},