Filter stream updates by Share ID

server/eventStreamSource.js

@@ -2,29 +2,37 @@ import { EventEmitter } from 'events';
 import _ from 'lodash';
 import { nanoid } from 'nanoid';
 
-const UNSUB_DELAY = 60000; //ms
+// DEBUG
+const DEBUG = {
+	showEvents          : true,
+	showSubscriberCount : true
+};
+
+// Delays
+const UNSUB_DELAY = 60 * 1000; //ms
+const REPORT_DELAY = 5 * 1000; //ms
+
 
 const Stream = new EventEmitter;
 
 // Create array of stream subscribers
 let subscribers = [];
 
-const debounceUnsub = _.debounce((id)=>{ unsubscribe(id); }, UNSUB_DELAY, { leading: false, trailing: true });
-
 // Listener functions
 const subscribe = function(shareId, res){
+	const id = nanoid(24); // NanoID is assumed globally unique;
 	const subscriber = {
-		id     : nanoid(), // NanoID is assumed globally unique
+		id,
 		shareId,
-		stream : res,
-		time   : new Date
+		stream     : res,
+		time       : new Date,
+		unsubTimer : setTimeout(()=>{ unsubscribe(id); }, UNSUB_DELAY)
 	};
 
 	subscribers.push(subscriber);
 
 	Stream.emit('sendUpdate', 'subscribe', { id: subscriber.id, shareId: subscriber.shareId, time: subscriber.time });
 
-	debounceUnsub(subscriber.id);
 	return subscriber.id;
 };
 
@@ -40,22 +48,26 @@ const unsubscribe = function(id){
 
 // Create global sendUpdate listener
 Stream.on('sendUpdate', (event, data)=>{
-	console.log('Event:', event, '\nData:', data);
+	if(DEBUG.showEvents) console.log('Event:', event, '\nData:', data);
 	subscribers
 		.filter((sub)=>{return data.shareId == sub.shareId; })
 		.forEach((sub)=>{
-			sub?.stream?.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
+			if(event == 'brewUpdated'){
+			// Reset the unsubscription timer
+				clearTimeout(sub.unsubTimer);
+				sub.unsubTimer = setTimeout(()=>{ unsubscribe(sub.id); }, UNSUB_DELAY);
+			}
 
-			// invoke debounced unsub to delay execution
-			debounceUnsub(sub.id);
+			sub?.stream?.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
 		});
 });
 
 // DEBUG: Report subscriber count periodically
-const REPORT_DELAY = 1000; //ms
-const reportIntervalId = setInterval(()=>{
-	console.log('Subscriber count:', subscribers.length);
-}, REPORT_DELAY);
+if(DEBUG.showSubscriberCount){
+	const reportIntervalId = setInterval(()=>{
+		console.log('Subscriber count:', subscribers.length, ' @ ', new Date);
+	}, REPORT_DELAY);
+}
 
 export default {
 	emit : function(event) {return Stream.emit(event, ...([...arguments].slice(1)));},    // Arguments doesn't work for arrow functions