diff --git a/packages/api/package.json b/packages/api/package.json index 25cd1c693..c90456c24 100644 --- a/packages/api/package.json +++ b/packages/api/package.json @@ -46,6 +46,7 @@ "lodash": "^4.17.21", "ncp": "^2.0.0", "node-cron": "^2.0.3", + "on-finished": "^2.4.1", "portfinder": "^1.0.28", "simple-encryptor": "^4.0.0", "ssh2": "^1.11.0", diff --git a/packages/api/src/main.js b/packages/api/src/main.js index ead1c4350..eec93c02a 100644 --- a/packages/api/src/main.js +++ b/packages/api/src/main.js @@ -25,6 +25,7 @@ const plugins = require('./controllers/plugins'); const files = require('./controllers/files'); const scheduler = require('./controllers/scheduler'); const queryHistory = require('./controllers/queryHistory'); +const onFinished = require('on-finished'); const { rundir } = require('./utility/directories'); const platformInfo = require('./utility/platformInfo'); @@ -63,7 +64,10 @@ function start() { // Tell the client to retry every 10 seconds if connectivity is lost res.write('retry: 10000\n\n'); - socket.setSseResponse(res); + socket.addSseResponse(res); + onFinished(req, () => { + socket.removeSseResponse(res); + }); }); app.use(bodyParser.json({ limit: '50mb' })); diff --git a/packages/api/src/utility/socket.js b/packages/api/src/utility/socket.js index 34723ade1..5b519f107 100644 --- a/packages/api/src/utility/socket.js +++ b/packages/api/src/utility/socket.js @@ -1,36 +1,33 @@ -let sseResponse = null; +const _ = require('lodash'); + +const sseResponses = []; let electronSender = null; -let init = []; +let pingConfigured = false; module.exports = { - setSseResponse(value) { - sseResponse = value; - setInterval(() => this.emit('ping'), 29 * 1000); + ensurePing() { + if (!pingConfigured) { + setInterval(() => this.emit('ping'), 29 * 1000); + pingConfigured = true; + } + }, + addSseResponse(value) { + sseResponses.push(value); + this.ensurePing(); + }, + removeSseResponse(value) { + _.remove(sseResponses, x => x == value); }, setElectronSender(value) { electronSender = value; + this.ensurePing(); }, emit(message, data) { if (electronSender) { - if (init.length > 0) { - for (const item of init) { - electronSender.send(item.message, item.data == null ? null : item.data); - } - init = []; - } electronSender.send(message, data == null ? null : data); - } else if (sseResponse) { - if (init.length > 0) { - for (const item of init) { - sseResponse.write( - `event: ${item.message}\ndata: ${JSON.stringify(item.data == null ? null : item.data)}\n\n` - ); - } - init = []; - } - sseResponse.write(`event: ${message}\ndata: ${JSON.stringify(data == null ? null : data)}\n\n`); - } else { - init.push([{ message, data }]); + } + for (const res of sseResponses) { + res.write(`event: ${message}\ndata: ${JSON.stringify(data == null ? null : data)}\n\n`); } }, emitChanged(key) { diff --git a/yarn.lock b/yarn.lock index ef005b6ce..0bf59659a 100644 --- a/yarn.lock +++ b/yarn.lock @@ -8257,6 +8257,13 @@ object.pick@^1.3.0: dependencies: isobject "^3.0.1" +on-finished@^2.4.1: + version "2.4.1" + resolved "https://registry.yarnpkg.com/on-finished/-/on-finished-2.4.1.tgz#58c8c44116e54845ad57f14ab10b03533184ac3f" + integrity sha512-oVlzkg3ENAhCk2zdv7IJwd/QUD4z2RxRwpkcGY8psCVcCYZNq4wYnVWALHM+brtuJjePWiYF/ClmuDr8Ch5+kg== + dependencies: + ee-first "1.1.1" + on-finished@~2.3.0: version "2.3.0" resolved "https://registry.yarnpkg.com/on-finished/-/on-finished-2.3.0.tgz#20f1336481b083cd75337992a16971aa2d906947"