mirror of
https://github.com/clusterio/clusterio.git
synced 2026-11-14 15:24:52 +09:00
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.
This commit is contained in:
1 parent
21d9fdff7b
commit
e5115df28b
4 files changed
+65
-1
No files matched your search
@@ -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)) {
|
||||
|
||||
+32
-1
@@ -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<string, 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`);
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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() {
|
||||
|
||||
Reference in new issue
Block a user