diff --git a/.gitignore b/.gitignore index 150d81008..57cb15ccf 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,7 @@ storage build/* config/local.* config/docker.* +config/dev.* todo.md startDB.bat diff --git a/client/homebrew/pages/sharePage/sharePage.jsx b/client/homebrew/pages/sharePage/sharePage.jsx index 5280fc850..002844c2c 100644 --- a/client/homebrew/pages/sharePage/sharePage.jsx +++ b/client/homebrew/pages/sharePage/sharePage.jsx @@ -53,21 +53,25 @@ const SharePage = (props)=>{ document.addEventListener('keydown', handleControlKeys); fetchThemeBundle(undefined, setThemeBundle, currentBrew.renderer, currentBrew.theme); - // listen for changes in the brew version - const eventSource = new EventSource('/stream'); - eventSource.addEventListener('message', (evt)=>{ - const messageData = JSON.parse(evt.data); + let eventSource; + // Start a stream if the brew has been updated in the last month + if(Date.parse(props.brew?.updatedAt) > (new Date).setDate((new Date).getDate() - 30)){ + // listen for changes in the brew version + eventSource = new EventSource(`/api/stream/${props.brew.shareId}`); - if(messageData.eventType == 'brewUpdated'){ - if(messageData.shareId == currentBrew.shareId && messageData.version != currentBrew.version) { + eventSource.addEventListener('brewUpdated', (evt)=>{ + const messageData = JSON.parse(evt.data); + + if(messageData.version != currentBrew.version) { console.log('should fetch brew'); fetchUpdatedBrew(); } - } - }); + }); + } return ()=>{ document.removeEventListener('keydown', handleControlKeys); + eventSource?.close(); }; }, []); diff --git a/server/app.js b/server/app.js index adc7c0136..dde6b6c27 100644 --- a/server/app.js +++ b/server/app.js @@ -20,6 +20,7 @@ import api from './homebrew.api.js'; const { homebrewApi, getBrew, getCSS } = api; import adminApi from './admin.api.js'; import vaultApi from './vault.api.js'; +import streamApi from './stream.api.js'; import pageRoutes from './page-routes.js'; import serveCompressedStaticAssets from './static-assets.mv.js'; @@ -32,7 +33,6 @@ import bodyParser from 'body-parser'; import cookieParser from 'cookie-parser'; import forceSSL from './forcessl.mw.js'; -import Stream from './eventStreamSource.js'; import dbCheck from './middleware/dbCheck.js'; import cors from 'cors'; @@ -115,6 +115,7 @@ export default async function createApp(vite) { app.use(homebrewApi); app.use(adminApi(vite)); app.use(vaultApi); + app.use(streamApi); String.prototype.replaceAll = function(s, r){return this.split(s).join(r);}; @@ -194,26 +195,6 @@ export default async function createApp(vite) { } }); - // Create Event Stream source for pages to listen to - app.get('/stream', (req, res)=>{ - res.writeHead(200, { - 'Content-Type' : 'text/event-stream', - 'Cache-Control' : 'no-cache', - 'Connection' : 'keep-alive', - 'Content-Encoding' : 'none' - }); - - 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){ diff --git a/server/config.js b/server/config.js index 69674595b..06f47c69a 100644 --- a/server/config.js +++ b/server/config.js @@ -4,4 +4,5 @@ export default nconf .argv() .env({ lowerCase: true }) .file('environment', { file: `config/${process.env.NODE_ENV}.json` }) + .file('development', { file: 'config/dev.json' }) .file('defaults', { file: 'config/default.json' }); \ No newline at end of file diff --git a/server/eventStreamSource.js b/server/eventStreamSource.js index 66fdb3354..52db436d1 100644 --- a/server/eventStreamSource.js +++ b/server/eventStreamSource.js @@ -1,9 +1,94 @@ import { EventEmitter } from 'events'; +import _ from 'lodash'; +import { nanoid } from 'nanoid'; + +import config from './config.js'; + +// DEBUG +const DEBUG = { + showEvents : config.get('showStreamEvents'), + showSubscriberCount : config.get('showStreamSubscriberCount') +}; + +// Delays +const REPORT_DELAY = config.get('REPORT_DELAY') || 30 * 60 * 1000; //ms +const HEARTBEAT_DELAY = config.get('HEARTBEAT_DELAY') || 20 * 1000; //ms + const Stream = new EventEmitter; +// Create array of stream subscribers +let subscribers = []; + +// Listener functions +const subscribe = function(shareId, res){ + const id = nanoid(24); // NanoID is assumed globally unique; + const subscriber = { + id, + shareId, + stream : res + }; + + subscribers.push(subscriber); + + Stream.emit('sendUpdate', 'subscribe', { id: subscriber.id, shareId: subscriber.shareId }); + + return subscriber.id; +}; + +const unsubscribe = function(id){ + subscribers + .filter((sub)=>{return sub.id == id;}) + .forEach((sub)=>{ + // Unsubscribe event - not currently used + // Stream.emit('sendUpdate', 'unsubscribe', { id: sub.id, shareId: sub.shareId }); + sub?.stream?.status(200).end(); + }); + + subscribers = subscribers.filter((sub)=>{ return sub.id != id; }); +}; + +// Create heartbeat +const heartbeatId = config.get('node_env') != 'test' ? + setInterval(()=>{ + if(DEBUG.showEvents){ console.log('STREAM HEARTBEAT @ ', new Date); }; + subscribers + .forEach((sub)=>{ + sub?.stream?.write(`:\n\n`); + }); + }, HEARTBEAT_DELAY) + : + undefined; + +Stream.on('close', ()=>{ + console.log('STREAM CLOSE'); + clearInterval(heartbeatId); +}); + +// Create global sendUpdate listener +Stream.on('sendUpdate', (event, data)=>{ + if(DEBUG.showEvents) console.log('Event:', event, '\nData:', data); + subscribers + .filter((sub)=>{ + if(data?.id) return data.id == sub.id; + return data.shareId == sub.shareId; + }) + .forEach((sub)=>{ + sub?.stream?.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); + }); +}); + +// DEBUG: Report subscriber count periodically +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 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 }; \ No newline at end of file diff --git a/server/stream.api.js b/server/stream.api.js new file mode 100644 index 000000000..79636469c --- /dev/null +++ b/server/stream.api.js @@ -0,0 +1,23 @@ +import express from 'express'; + +import Stream from './eventStreamSource.js'; + +const app = express.Router(); + +// Create Event Stream source for pages to listen to +app.get('/api/stream/:id', (req, res)=>{ + res.writeHead(200, { + 'Content-Type' : 'text/event-stream', + 'Cache-Control' : 'no-cache', + 'Connection' : 'keep-alive', + 'Content-Encoding' : 'none' + }); + + const subId = Stream.subscribe(req.params.id, res); + + res.on('close', ()=>{ + Stream.unsubscribe(subId); + }); +}); + +export default app; \ No newline at end of file