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 7f5b1c235..9edc05938 100644
--- a/client/homebrew/pages/sharePage/sharePage.jsx
+++ b/client/homebrew/pages/sharePage/sharePage.jsx
@@ -46,21 +46,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();
};
}, []);
console.log('brew: ', currentBrew);
diff --git a/client/homebrew/pages/vaultPage/vaultPage.jsx b/client/homebrew/pages/vaultPage/vaultPage.jsx
index e15443db0..d935a0841 100644
--- a/client/homebrew/pages/vaultPage/vaultPage.jsx
+++ b/client/homebrew/pages/vaultPage/vaultPage.jsx
@@ -68,7 +68,7 @@ const VaultPage = (props)=>{
updateStateWithBrews([], 1);
});
- if(response.ok)
+ if(response?.ok)
updateStateWithBrews(response.body.brews, page);
};
@@ -342,13 +342,23 @@ const VaultPage = (props)=>{
}
if(error) {
- const errorText = ErrorIndex()[error.HBErrorCode.toString()] || '';
+ if(error.HBErrorCode === undefined) {
+ return (
+
+
{error.toString()}
+
+ )
+ } else {
- return (
-
+ const errorText = ErrorIndex()[error.HBErrorCode.toString()] || '';
+
+ return (
+
Error: {errorText}
-
- );
+
+ );
+ }
+
}
if(!brewCollection) {
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