From e5115df28b1d993433e8c60b448800068967df6f Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Fri, 13 Dec 2019 21:37:27 +0100 Subject: [PATCH] Implement broadcast for events Add broadcastTo property to events that broadcasts them to all entities of the specified target that's downstream to the connection it was sent to. --- client.js | 17 +++++++++++++++++ lib/link/messages.js | 33 ++++++++++++++++++++++++++++++++- master.js | 10 ++++++++++ test/lib/link/messages.js | 6 ++++++ 4 files changed, 65 insertions(+), 1 deletion(-) diff --git a/client.js b/client.js index f7bfbe45..9bd28fb6 100644 --- a/client.js +++ b/client.js @@ -300,6 +300,17 @@ class InstanceConnection extends link.Link { async forwardEventToMaster(message, event) { event.send(this.slave, message.data); } + + async broadcastEventToInstance(message, event) { + for (let instanceConnection of this.slave.instanceConnections.values()) { + if (instanceConnection === this) { + continue; // Do not broadcast back to the source + } + + console.log("Broadcasting event"); + event.send(instanceConnection, message.data); + } + } } class SlaveConnector extends link.SocketIOClientConnector { @@ -377,6 +388,12 @@ class Slave extends link.Link { return await request.send(instanceConnection, message.data); } + async broadcastEventToInstance(message, event) { + for (let instanceConnection of this.instanceConnections.values()) { + event.send(instanceConnection, message.data); + } + } + async createInstanceRequestHandler(message) { let { id, options } = message.data; if (this.instanceInfos.has(id)) { diff --git a/lib/link/messages.js b/lib/link/messages.js index 7a597221..a385262f 100644 --- a/lib/link/messages.js +++ b/lib/link/messages.js @@ -313,19 +313,30 @@ class Event { * @param {?string} forwardTo - * Optional target to forward this request to. Only 'master' is * currently supported. + * @param {?string} broadcastTo - + * Optional target to broadcast this event to. Only 'instance' is + * currently supported. An event will only be broadcast towards + * links that are downstream, so sending an event set to instance + * broadcast that's sent to a slave will only be broadcast to that + * slave's instances. * @param {Object} eventProperties - * Mapping of property values to JSON schema specifications for the * properties that are valid in the data payload of this event. */ - constructor({ type, links, forwardTo = null, eventProperties = {} }) { + constructor({ type, links, forwardTo = null, broadcastTo = null, eventProperties = {} }) { this.type = type; this.links = links; this.forwardTo = forwardTo; + this.broadcastTo = broadcastTo; if (forwardTo !== 'master' && forwardTo !== null) { throw new Error(`Invalid forwardTo value ${forwardTo}`); } + if (broadcastTo !== 'instance' && broadcastTo !== null) { + throw new Error(`Invalid broadcastTo value ${broadcastTo}`); + } + this.eventType = type + '_event'; this._eventValidator = schema.compile({ $schema: "http://json-schema.org/draft-07/schema#", @@ -349,6 +360,9 @@ class Event { * If forwardTo was set to 'master' the `forwardEventToMaster` method on * the link is used as the default handler. * + * If broadcastTo was set to 'instance' the `broadcastEventToInstance` + * method on the link is invoked before the handler. BUT HOW? + * * Does nothing if the link is not a target for this event. * * @param {Link} link - The link to attach to. @@ -367,6 +381,23 @@ class Event { } } + if ( + this.broadcastTo === 'instance' + && [ + 'instance-slave', 'slave-master', 'control-master', 'master-slave' + ].includes(`${link.target}-${link.source}`) + ) { + if (!handler) { + handler = link.broadcastEventToInstance; + } else { + let originalHandler = handler; + handler = async (message, event) => { + await link.broadcastEventToInstance(message, event); + await originalHandler.call(link, message, event); + } + } + } + if (!handler) { throw new Error(`Missing handler for ${this.eventType} on ${link.source}-${link.target} link`); } diff --git a/master.js b/master.js index 593f7b88..244e8f73 100644 --- a/master.js +++ b/master.js @@ -267,6 +267,16 @@ class BaseConnection extends link.Link { return await request.send(connection, message.data); } + + async broadcastEventToInstance(message, event) { + for (let slaveConnection of slaveConnections.values()) { + if (slaveConnection === this) { + continue; // Do not broadcast back to the source + } + + event.send(slaveConnection, message.data); + } + } } let controlConnections = new Array(); diff --git a/test/lib/link/messages.js b/test/lib/link/messages.js index 3f9ab742..21d03629 100644 --- a/test/lib/link/messages.js +++ b/test/lib/link/messages.js @@ -115,6 +115,12 @@ describe("lib/link/messages", function() { new Error("Invalid forwardTo value invalid") ); }); + it("should throw on invalid broadcastTo", function() { + assert.throws( + () => new link.Event({ broadcastTo: 'invalid'}), + new Error("Invalid broadcastTo value invalid") + ); + }); }); describe(".attach()", function() {