Move to Subscription model

client/homebrew/pages/sharePage/sharePage.jsx

@@ -54,15 +54,13 @@ const SharePage = (props)=>{
 		fetchThemeBundle(undefined, setThemeBundle, currentBrew.renderer, currentBrew.theme);
 
 		// listen for changes in the brew version
-		const eventSource = new EventSource('/stream');
+		const eventSource = new EventSource(`/stream/${props.brew.shareId}`);
-		eventSource.addEventListener('message', (evt)=>{
+		eventSource.addEventListener('brewUpdated', (evt)=>{
 			const messageData = JSON.parse(evt.data);
 
-			if(messageData.eventType == 'brewUpdated'){
+			if(messageData.shareId == currentBrew.shareId && messageData.version != currentBrew.version) {
-				if(messageData.shareId == currentBrew.shareId && messageData.version != currentBrew.version) {
+				console.log('should fetch brew');
-					console.log('should fetch brew');
+				fetchUpdatedBrew();
-					fetchUpdatedBrew();
-				}
 			}
 		});
 

server/app.js

@@ -195,29 +195,17 @@ export default async function createApp(vite) {
 	});
 
 	// Create Event Stream source for pages to listen to
-	let initStreamListener = false;
+	app.get('/stream/:id', (req, res)=>{
-	app.get('/stream', (req, res)=>{
 		res.writeHead(200, {
 			'Content-Type'     : 'text/event-stream',
 			'Cache-Control'    : 'no-cache',
 			'Connection'       : 'keep-alive',
 			'Content-Encoding' : 'none'
 		});
-
+		;
-		if(!initStreamListener){
+		Stream.subscribe(res);
-			initStreamListener = true;
-			Stream.on('sendUpdate', (event, data)=>{
-				if(isLocalEnvironment) { console.log('Event:', event, '\nData:', data); }
-				res.write(`data: ${JSON.stringify({ ...data, eventType: event })}\n\n`);
-			});
-		}
 	});
 
-	// After Stream starts, send initStream event
-	setTimeout(()=>{
-		Stream.emit('sendUpdate', 'initStream', { time: new Date });
-	}, 1000);
-
 
 	// Local only
 	if(isLocalEnvironment){

server/eventStreamSource.js

@@ -1,9 +1,41 @@
 import { EventEmitter } from 'events';
+import { nanoid } from 'nanoid';
 
 const Stream = new EventEmitter;
 
+// After Stream starts, send initStream event
+Stream.on('open', ()=>{
+	Stream.emit('sendUpdate', 'initStream', { time: new Date });
+});
+
+// Create array of stream subscribers
+const subscribers = [];
+
+// Listener functions
+const subscribe = function(res){
+	const id = nanoid();
+	subscribers.push({ id, 'stream': res });
+
+	res.write(`event: subscribe\ndata: { id: ${id} }\n\n`);
+	return id;
+};
+
+const unsubscribe = function(id){
+	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)=>{
+		sub?.stream?.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
+	});
+});
+
 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);},
-	off  : (event, listener)=>{return Stream.off(event, listener);}
+	off  : (event, listener)=>{return Stream.off(event, listener);},
+	subscribe,
+	unsubscribe
 };