mirror of
https://github.com/clusterio/clusterio.git
synced 2026-11-12 07:04:52 +09:00
Danielv123
1ed46dff9c
- Saves of a datastore run one at a time, and changes made while saving are saved by the next save. Overlapping saves from approving and revoking hosts could write an older snapshot after a newer one. - Map subscribers fetch everything again when a new session starts, as the controller only sends values that exist. A host revoked while a browser was disconnected stayed in its list. - The host token check accepts the same pre-rename tokens as the WebSocket, so upgrading a host with such a token does not make it ask to join again.
1650 lines
57 KiB
JavaScript
1650 lines
57 KiB
JavaScript
import events from "node:events";
|
|
import assert from "node:assert/strict";
|
|
import * as lib from "@clusterio/lib";
|
|
import { MockController, MockConnector, MockControl } from "../mock.js";
|
|
import { testMatrix, testRoundTripJsonSerialisable } from "../common.js";
|
|
import "../setup_logging.js";
|
|
|
|
const addr = lib.Address.fromShorthand;
|
|
|
|
describe("lib/subscriptions", function() {
|
|
class UnregisteredEvent {
|
|
static type = "event";
|
|
static src = ["controller", "control"];
|
|
static dst = ["controller", "control"];
|
|
static permission = null;
|
|
}
|
|
|
|
class RegisteredEvent {
|
|
static type = "event";
|
|
static src = ["controller", "control"];
|
|
static dst = ["controller", "control"];
|
|
static permission = null;
|
|
constructor(updates = []) {
|
|
this.updates = updates;
|
|
}
|
|
|
|
get filters() {
|
|
return "Filters";
|
|
}
|
|
}
|
|
if (!lib.Link._eventsByName.has(RegisteredEvent.name)) {
|
|
lib.Link.register(RegisteredEvent);
|
|
}
|
|
|
|
class StringPermissionEvent {
|
|
static type = "event";
|
|
static src = ["controller", "control"];
|
|
static dst = ["controller", "control"];
|
|
static permission = "StringPermission";
|
|
}
|
|
if (!lib.Link._eventsByName.has(StringPermissionEvent.name)) {
|
|
lib.Link.register(StringPermissionEvent);
|
|
}
|
|
|
|
class FunctionPermissionEvent {
|
|
static type = "event";
|
|
static src = ["controller", "control"];
|
|
static dst = ["controller", "control"];
|
|
static permission(user, message) {
|
|
user.checkPermission("FunctionPermission");
|
|
};
|
|
}
|
|
if (!lib.Link._eventsByName.has(FunctionPermissionEvent.name)) {
|
|
lib.Link.register(FunctionPermissionEvent);
|
|
}
|
|
|
|
class MockUser {
|
|
checkPermission(permission) {
|
|
this.lastPermissionCheck = permission;
|
|
}
|
|
}
|
|
|
|
describe("class SubscriptionFilters", function() {
|
|
it("should be round trip serialisable", function() {
|
|
const f = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]);
|
|
const reconstructed = lib.SubscriptionFilters.fromJSON(JSON.parse(JSON.stringify(f)));
|
|
assert.deepEqual(new Set(reconstructed.toJSON()), new Set(["foo", "bar"]));
|
|
});
|
|
|
|
it("should not preserve 'all' through JSON", function() {
|
|
// This is because we use 'undefined' to indicate all
|
|
const f = lib.SubscriptionFilters.all();
|
|
const reconstructed = lib.SubscriptionFilters.fromJSON(JSON.parse(JSON.stringify(f)));
|
|
assert.equal(reconstructed.isAll(), false);
|
|
assert.equal(reconstructed.isEmpty(), true);
|
|
});
|
|
|
|
describe("construction", function() {
|
|
it("should create an 'all' filter", function() {
|
|
const f = lib.SubscriptionFilters.all();
|
|
assert.equal(f.isAll(), true);
|
|
assert.equal(f.isEmpty(), false);
|
|
});
|
|
|
|
it("should create an 'empty' filter", function() {
|
|
const f = lib.SubscriptionFilters.empty();
|
|
assert.equal(f.isAll(), false);
|
|
assert.equal(f.isEmpty(), true);
|
|
});
|
|
|
|
it("should create from shorthand undefined as all", function() {
|
|
const f = lib.SubscriptionFilters.fromShorthand(undefined);
|
|
assert.equal(f.isAll(), true);
|
|
});
|
|
|
|
it("should create from shorthand string", function() {
|
|
const f = lib.SubscriptionFilters.fromShorthand("foo");
|
|
assert.equal(f.isAll(), false);
|
|
assert.deepEqual(new Set(f.toJSON()), new Set(["foo"]));
|
|
});
|
|
|
|
it("should create from shorthand array", function() {
|
|
const f = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]);
|
|
assert.deepEqual(new Set(f.toJSON()), new Set(["foo", "bar"]));
|
|
});
|
|
});
|
|
|
|
describe(".toString()", function() {
|
|
it("should return 'All' for all filters", function() {
|
|
const filters = lib.SubscriptionFilters.all();
|
|
assert.equal(filters.toString(), "[SubscriptionFilters All]");
|
|
});
|
|
|
|
it("should return empty set representation for empty filters", function() {
|
|
const filters = lib.SubscriptionFilters.empty();
|
|
assert.equal(filters.toString(), "[SubscriptionFilters Empty]");
|
|
});
|
|
|
|
it("should include single filter value", function() {
|
|
const filters = lib.SubscriptionFilters.fromShorthand("foo");
|
|
assert.equal(filters.toString(), "[SubscriptionFilters Set<1>]");
|
|
});
|
|
|
|
it("should include multiple filter values", function() {
|
|
const filters = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]);
|
|
assert.equal(filters.toString(), "[SubscriptionFilters Set<2>]");
|
|
});
|
|
});
|
|
|
|
describe(".accepts()", function() {
|
|
it("should accept any value when all", function() {
|
|
const f = lib.SubscriptionFilters.all();
|
|
assert.equal(f.accepts("foo"), true);
|
|
assert.equal(f.accepts("bar"), true);
|
|
});
|
|
|
|
it("should reject all values when empty", function() {
|
|
const f = lib.SubscriptionFilters.empty();
|
|
assert.equal(f.accepts("foo"), false);
|
|
assert.equal(f.accepts("bar"), false);
|
|
});
|
|
|
|
it("should match exact values", function() {
|
|
const f = lib.SubscriptionFilters.fromShorthand(["foo"]);
|
|
assert.equal(f.accepts("foo"), true);
|
|
assert.equal(f.accepts("bar"), false);
|
|
});
|
|
});
|
|
|
|
describe(".extends()", function() {
|
|
it("all should extend any filter", function() {
|
|
const all = lib.SubscriptionFilters.all();
|
|
const foo = lib.SubscriptionFilters.fromShorthand("foo");
|
|
assert.equal(all.extends(foo), true);
|
|
});
|
|
|
|
it("should return true when superset", function() {
|
|
const a = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]);
|
|
const b = lib.SubscriptionFilters.fromShorthand(["foo"]);
|
|
assert.equal(a.extends(b), true);
|
|
});
|
|
|
|
it("should return false when not superset", function() {
|
|
const a = lib.SubscriptionFilters.fromShorthand(["foo"]);
|
|
const b = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]);
|
|
assert.equal(a.extends(b), false);
|
|
});
|
|
});
|
|
|
|
describe(".intersects()", function() {
|
|
it("should intersect when sharing values", function() {
|
|
const a = lib.SubscriptionFilters.fromShorthand(["foo"]);
|
|
const b = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]);
|
|
assert.equal(a.intersects(b), true);
|
|
assert.equal(b.intersects(a), true);
|
|
});
|
|
|
|
it("should not intersect when disjoint", function() {
|
|
const a = lib.SubscriptionFilters.fromShorthand(["foo"]);
|
|
const b = lib.SubscriptionFilters.fromShorthand(["bar"]);
|
|
assert.equal(a.intersects(b), false);
|
|
assert.equal(b.intersects(a), false);
|
|
});
|
|
|
|
it("all should intersect non-empty", function() {
|
|
const all = lib.SubscriptionFilters.all();
|
|
const foo = lib.SubscriptionFilters.fromShorthand("foo");
|
|
assert.equal(all.intersects(foo), true);
|
|
assert.equal(foo.intersects(all), true);
|
|
});
|
|
|
|
it("all should not intersect empty", function() {
|
|
const all = lib.SubscriptionFilters.all();
|
|
const empty = lib.SubscriptionFilters.empty();
|
|
assert.equal(all.intersects(empty), false);
|
|
assert.equal(empty.intersects(all), false);
|
|
});
|
|
});
|
|
|
|
describe(".union()", function() {
|
|
it("should merge filters", function() {
|
|
const a = lib.SubscriptionFilters.fromShorthand(["foo"]);
|
|
const b = lib.SubscriptionFilters.fromShorthand(["bar"]);
|
|
a.union(b);
|
|
assert.deepEqual(new Set(a.toJSON()), new Set(["foo", "bar"]));
|
|
});
|
|
|
|
it("should become all if union with all", function() {
|
|
const a = lib.SubscriptionFilters.fromShorthand(["foo"]);
|
|
const all = lib.SubscriptionFilters.all();
|
|
a.union(all);
|
|
assert.equal(a.isAll(), true);
|
|
});
|
|
|
|
it("should remain all if already all", function() {
|
|
const a = lib.SubscriptionFilters.all();
|
|
const b = lib.SubscriptionFilters.fromShorthand(["foo"]);
|
|
a.union(b);
|
|
assert.equal(a.isAll(), true);
|
|
});
|
|
});
|
|
|
|
describe(".subtract()", function() {
|
|
it("should remove matching filters", function() {
|
|
const a = lib.SubscriptionFilters.fromShorthand(["foo", "bar"]);
|
|
const b = lib.SubscriptionFilters.fromShorthand("foo");
|
|
a.subtract(b);
|
|
assert.deepEqual(new Set(a.toJSON()), new Set(["bar"]));
|
|
});
|
|
|
|
it("should become empty when subtracting all", function() {
|
|
const a = lib.SubscriptionFilters.fromShorthand(["foo"]);
|
|
const all = lib.SubscriptionFilters.all();
|
|
a.subtract(all);
|
|
assert.equal(a.isEmpty(), true);
|
|
|
|
const b = lib.SubscriptionFilters.all();
|
|
b.subtract(all);
|
|
assert.equal(b.isEmpty(), true);
|
|
});
|
|
|
|
it("should do nothing when subtracting from all", function() {
|
|
const a = lib.SubscriptionFilters.all();
|
|
const b = lib.SubscriptionFilters.fromShorthand("foo");
|
|
a.subtract(b);
|
|
assert.equal(a.isAll(), true);
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("class SubscriptionRequest", function() {
|
|
it("should be round trip json serialisable", function() {
|
|
testRoundTripJsonSerialisable(lib.SubscriptionRequest, testMatrix(
|
|
[RegisteredEvent.name],
|
|
["subscribe", "unsubscribe", "replace"],
|
|
[0, 123],
|
|
[
|
|
undefined,
|
|
lib.SubscriptionFilters.all(),
|
|
lib.SubscriptionFilters.empty(),
|
|
lib.SubscriptionFilters.fromShorthand("foo"),
|
|
lib.SubscriptionFilters.fromShorthand(["foo", "bar"]),
|
|
]
|
|
));
|
|
});
|
|
|
|
it("should throw when constructed via fromJSON with unregistered event", function() {
|
|
assert.throws(
|
|
() => lib.SubscriptionRequest.fromJSON([UnregisteredEvent.name,, "subscribe", 0]),
|
|
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
|
|
);
|
|
});
|
|
|
|
describe("static permission()", function() {
|
|
function newSubscriptionRequestMessage(request) {
|
|
return new lib.MessageRequest(
|
|
0,
|
|
addr({ controlId: 0 }),
|
|
addr("controller"),
|
|
lib.SubscriptionRequest.name,
|
|
request.toJSON()
|
|
);
|
|
}
|
|
|
|
it("should do nothing when the event has no permission property", function() {
|
|
const mockUser = new MockUser();
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
|
|
lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request));
|
|
assert.equal(mockUser.lastPermissionCheck, undefined);
|
|
});
|
|
|
|
it("should check user permission when the permission property is a string", function() {
|
|
const mockUser = new MockUser();
|
|
const request = new lib.SubscriptionRequest(StringPermissionEvent.name, "subscribe");
|
|
lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request));
|
|
assert.equal(mockUser.lastPermissionCheck, "StringPermission");
|
|
});
|
|
|
|
it("should call the permission property when it is a function", function() {
|
|
const mockUser = new MockUser();
|
|
const request = new lib.SubscriptionRequest(FunctionPermissionEvent.name, "subscribe");
|
|
lib.SubscriptionRequest.permission(mockUser, newSubscriptionRequestMessage(request));
|
|
assert.equal(mockUser.lastPermissionCheck, "FunctionPermission");
|
|
});
|
|
});
|
|
|
|
describe("construction", function() {
|
|
describe("compatibility 'action'", function() {
|
|
it("should map true to subscribe", function() {
|
|
const req = new lib.SubscriptionRequest(RegisteredEvent.name, true);
|
|
assert.equal(req.action, "subscribe");
|
|
});
|
|
|
|
it("should map false to unsubscribe", function() {
|
|
const req = new lib.SubscriptionRequest(RegisteredEvent.name, false);
|
|
assert.equal(req.action, "unsubscribe");
|
|
});
|
|
});
|
|
|
|
describe("compatibility 'filters'", function() {
|
|
it("should convert string filters to SubscriptionFilters", function() {
|
|
const req = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo");
|
|
assert(req.filters instanceof lib.SubscriptionFilters);
|
|
assert.equal(req.filters.accepts("foo"), true);
|
|
});
|
|
|
|
it("should convert array filters to SubscriptionFilters", function() {
|
|
const req = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]);
|
|
assert.equal(req.filters.accepts("foo"), true);
|
|
assert.equal(req.filters.accepts("bar"), true);
|
|
});
|
|
|
|
it("should default to 'all' when filters undefined", function() {
|
|
const req = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0);
|
|
assert.equal(req.filters.isAll(), true);
|
|
});
|
|
|
|
it("should accept SubscriptionFilters directly", function() {
|
|
const filters = lib.SubscriptionFilters.fromShorthand("foo");
|
|
const req = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, filters);
|
|
assert.equal(req.filters, filters);
|
|
});
|
|
});
|
|
|
|
it("should throw when given an unregistered event", function() {
|
|
assert.throws(
|
|
() => new lib.SubscriptionRequest(UnregisteredEvent.name, "subscribe"),
|
|
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
|
|
);
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("class SubscriptionController", function() {
|
|
let mockController, subscriptions;
|
|
|
|
const connectorSetupDate = [{
|
|
id: 0,
|
|
dst: addr({ controlId: 0 }),
|
|
src: addr("controller"),
|
|
}, {
|
|
id: 1,
|
|
dst: addr({ controlId: 1 }),
|
|
src: addr("controller"),
|
|
}];
|
|
|
|
function getLink(id) {
|
|
return mockController.wsServer.controlConnections.get(id);
|
|
}
|
|
|
|
function onceConnectorSend(id) {
|
|
return events.once(getLink(id).connector, "send");
|
|
}
|
|
|
|
function assertLastEvent(connectorId, Event) {
|
|
const link = getLink(connectorId);
|
|
const lastMessage = link.connector.sentMessages.at(-1);
|
|
const name = Event.plugin ? `${Event.plugin}:${Event.name}` : Event.name;
|
|
assert.equal(lastMessage.name, name);
|
|
}
|
|
|
|
function assertNoEvent(connectorId) {
|
|
const link = getLink(connectorId);
|
|
assert.deepEqual(link.connector.sentMessages, []);
|
|
}
|
|
|
|
beforeEach(function() {
|
|
mockController = new MockController();
|
|
mockController.wsServer = {
|
|
controlConnections: new Map(connectorSetupDate.map(function (connectorData) {
|
|
return [connectorData.id, new lib.Link(new MockConnector(connectorData.src, connectorData.dst))];
|
|
})),
|
|
};
|
|
subscriptions = new lib.SubscriptionController();
|
|
});
|
|
|
|
describe(".handle()", function() {
|
|
it("should accept registered events", function() {
|
|
subscriptions.handle(RegisteredEvent);
|
|
});
|
|
|
|
it("should not accept unregistered events", function() {
|
|
assert.throws(
|
|
() => subscriptions.handle(UnregisteredEvent),
|
|
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
|
|
);
|
|
});
|
|
|
|
it("should not accept events already handled by the class", function() {
|
|
subscriptions.handle(RegisteredEvent);
|
|
assert.throws(
|
|
() => subscriptions.handle(RegisteredEvent),
|
|
new Error(`Event ${RegisteredEvent.name} is already registered`)
|
|
);
|
|
});
|
|
});
|
|
|
|
describe(".broadcast()", function() {
|
|
beforeEach(function() {
|
|
subscriptions.handle(RegisteredEvent);
|
|
for (let connectorData of connectorSetupDate) {
|
|
const link = getLink(connectorData.id);
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
|
|
subscriptions.handleRequest(link, request, connectorData.dst, connectorData.src);
|
|
link.connector.sentMessages.pop();
|
|
}
|
|
});
|
|
|
|
it("should not accept unregistered events", function() {
|
|
assert.throws(
|
|
() => subscriptions.broadcast(new UnregisteredEvent()),
|
|
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
|
|
);
|
|
});
|
|
|
|
it("should not accept events not handled by the class", function() {
|
|
assert.throws(
|
|
() => subscriptions.broadcast(new StringPermissionEvent()),
|
|
new Error(`Event ${StringPermissionEvent.name} is not a registered as subscribable`)
|
|
);
|
|
});
|
|
|
|
it("should notify all links who subscribed", async function() {
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await Promise.all([onceConnectorSend(0), onceConnectorSend(1)]);
|
|
assertLastEvent(0, RegisteredEvent); // RegisteredEvent: All
|
|
assertLastEvent(1, RegisteredEvent); // RegisteredEvent: All
|
|
});
|
|
|
|
it("should respect filters of subscribers", async function() {
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "foo");
|
|
await subscriptions.handleRequest(
|
|
getLink(0), request0, connectorSetupDate[0].dst, connectorSetupDate[0].src
|
|
);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "bar");
|
|
await subscriptions.handleRequest(
|
|
getLink(1), request1, connectorSetupDate[1].dst, connectorSetupDate[1].src
|
|
);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "foo");
|
|
await onceConnectorSend(0);
|
|
assertNoEvent(1);
|
|
getLink(0).connector.sentMessages.pop();
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), ["bar", "baz"]);
|
|
await onceConnectorSend(1);
|
|
assertNoEvent(0);
|
|
});
|
|
|
|
it("should always notify subscribers without filters", async function() {
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "replace");
|
|
await subscriptions.handleRequest(
|
|
getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
|
|
);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "foo");
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, RegisteredEvent);
|
|
});
|
|
|
|
it("should use event filters property if present", async function() {
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "Filters");
|
|
await subscriptions.handleRequest(
|
|
getLink(0), request0, connectorSetupDate[0].dst, connectorSetupDate[0].src
|
|
);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "bar");
|
|
await subscriptions.handleRequest(
|
|
getLink(1), request1, connectorSetupDate[1].dst, connectorSetupDate[1].src
|
|
);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await onceConnectorSend(0);
|
|
assertNoEvent(1);
|
|
});
|
|
|
|
it("should not notify a link who unsubscribed", async function() {
|
|
const connectorData = connectorSetupDate[0];
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe");
|
|
await subscriptions.handleRequest(
|
|
getLink(connectorData.id), request, connectorData.dst, connectorData.src
|
|
);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await onceConnectorSend(1);
|
|
assertNoEvent(0);
|
|
assertLastEvent(1, RegisteredEvent);
|
|
});
|
|
|
|
it("should not notify links which are closed or closing", async function() {
|
|
const link = getLink(0);
|
|
link.connector.closing = true;
|
|
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await onceConnectorSend(1);
|
|
|
|
assertNoEvent(0);
|
|
assertLastEvent(1, RegisteredEvent);
|
|
});
|
|
});
|
|
|
|
describe(".unsubscribeLink() / .unsubscribeAddress()", function() {
|
|
beforeEach(function() {
|
|
subscriptions.handle(RegisteredEvent);
|
|
for (let connectorData of connectorSetupDate) {
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
|
|
subscriptions.handleRequest(
|
|
getLink(connectorData.id), request, connectorData.dst, connectorData.src
|
|
);
|
|
}
|
|
});
|
|
|
|
it("should remove an active subscription link", async function() {
|
|
const link = getLink(0);
|
|
subscriptions.unsubscribeLink(link);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await onceConnectorSend(1);
|
|
assertNoEvent(0);
|
|
assertLastEvent(1, RegisteredEvent);
|
|
});
|
|
|
|
it("should remove an active subscription address", async function() {
|
|
const link = getLink(0);
|
|
subscriptions.unsubscribeAddress(connectorSetupDate[0].dst);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await onceConnectorSend(1);
|
|
assertNoEvent(0);
|
|
assertLastEvent(1, RegisteredEvent);
|
|
});
|
|
});
|
|
|
|
describe(".checkLinkPermissions()", function() {
|
|
class DenyingUser {
|
|
constructor(denied) {
|
|
this.denied = denied;
|
|
}
|
|
|
|
checkPermission(permission) {
|
|
if (permission === this.denied) {
|
|
throw new lib.PermissionError("Permission denied");
|
|
}
|
|
}
|
|
}
|
|
beforeEach(async function() {
|
|
subscriptions.handle(StringPermissionEvent);
|
|
subscriptions.handle(FunctionPermissionEvent);
|
|
for (const connectorData of connectorSetupDate) {
|
|
for (const Event of [StringPermissionEvent, FunctionPermissionEvent]) {
|
|
const request = new lib.SubscriptionRequest(Event.name, "subscribe");
|
|
await subscriptions.handleRequest(getLink(connectorData.id), request, connectorData.dst);
|
|
}
|
|
}
|
|
});
|
|
|
|
it("should remove subscriptions with a string permission the user lost", async function() {
|
|
subscriptions.checkLinkPermissions(getLink(0), new DenyingUser("StringPermission"));
|
|
subscriptions.broadcast(new StringPermissionEvent());
|
|
await onceConnectorSend(1);
|
|
assertNoEvent(0);
|
|
subscriptions.broadcast(new FunctionPermissionEvent());
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, FunctionPermissionEvent);
|
|
});
|
|
|
|
it("should remove subscriptions with a function permission the user lost", async function() {
|
|
subscriptions.checkLinkPermissions(getLink(0), new DenyingUser("FunctionPermission"));
|
|
subscriptions.broadcast(new FunctionPermissionEvent());
|
|
await onceConnectorSend(1);
|
|
assertNoEvent(0);
|
|
subscriptions.broadcast(new StringPermissionEvent());
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, StringPermissionEvent);
|
|
});
|
|
|
|
it("should keep subscriptions the user still has permission for", async function() {
|
|
subscriptions.checkLinkPermissions(getLink(0), new DenyingUser(null));
|
|
subscriptions.broadcast(new StringPermissionEvent());
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, StringPermissionEvent);
|
|
});
|
|
});
|
|
|
|
describe(".handleRequest()", function() {
|
|
beforeEach(function() {
|
|
subscriptions.handle(RegisteredEvent);
|
|
});
|
|
|
|
it("should not accept subscriptions to unregistered events", async function() {
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
|
|
request.eventName = UnregisteredEvent.name;
|
|
await assert.rejects(
|
|
subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")),
|
|
new Error(`Event ${UnregisteredEvent.name} is not a registered event`)
|
|
);
|
|
});
|
|
|
|
it("should not accept subscriptions to events not handled by the class", async function() {
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
|
|
request.eventName = StringPermissionEvent.name;
|
|
await assert.rejects(
|
|
subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")),
|
|
new Error(`Event ${StringPermissionEvent.name} is not a registered as subscribable`)
|
|
);
|
|
});
|
|
|
|
it("should accept a response with an event replay when returned by the handler", async function() {
|
|
subscriptions.handle(StringPermissionEvent, async function() {
|
|
return new StringPermissionEvent();
|
|
});
|
|
|
|
const link = getLink(0);
|
|
const request = new lib.SubscriptionRequest(StringPermissionEvent.name, "subscribe");
|
|
await subscriptions.handleRequest(
|
|
link, request, connectorSetupDate[0].dst, connectorSetupDate[0].src
|
|
);
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, StringPermissionEvent);
|
|
|
|
const lastMessage = link.connector.sentMessages.pop();
|
|
const eventEntry = lib.Link._eventsByClass.get(StringPermissionEvent);
|
|
const response = eventEntry.eventFromJSON(lastMessage.data);
|
|
assert.deepEqual(response, new StringPermissionEvent());
|
|
});
|
|
|
|
it("should accept a response with null when returned by the handler", async function() {
|
|
subscriptions.handle(StringPermissionEvent, async function() {
|
|
return null;
|
|
});
|
|
|
|
const request = new lib.SubscriptionRequest(StringPermissionEvent.name, "subscribe");
|
|
await subscriptions.handleRequest(
|
|
getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
|
|
);
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoEvent(0);
|
|
});
|
|
|
|
it("should throw if action is not supported", async function() {
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "invalid action");
|
|
await assert.rejects(subscriptions.handleRequest(
|
|
getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
|
|
), /unreachable case: invalid action/);
|
|
});
|
|
|
|
describe("addressing", function() {
|
|
// A subscribe arrives as a request, so the source address it is
|
|
// handled with carries the request id which should not be persisted
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
function requestSource(requestId) {
|
|
return new lib.Address(connectorData.dst.type, connectorData.dst.id, requestId);
|
|
}
|
|
|
|
function lastSentMessage() {
|
|
return getLink(connectorData.id).connector.sentMessages.at(-1);
|
|
}
|
|
|
|
it("should strip the request id from broadcast events", async function() {
|
|
const ownSubscriptions = new lib.SubscriptionController();
|
|
ownSubscriptions.handle(RegisteredEvent);
|
|
await ownSubscriptions.handleRequest(
|
|
getLink(connectorData.id),
|
|
new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"),
|
|
requestSource(213), connectorData.src
|
|
);
|
|
|
|
ownSubscriptions.broadcast(new RegisteredEvent());
|
|
await onceConnectorSend(connectorData.id);
|
|
|
|
const message = lastSentMessage();
|
|
assert.equal(message.name, RegisteredEvent.name);
|
|
assert.equal(message.dst.requestId, undefined, "event addressed to a request id");
|
|
assert.deepEqual(message.dst.toJSON(), connectorData.dst.toJSON());
|
|
});
|
|
|
|
it("should strip the request id from the replayed event", async function() {
|
|
const ownSubscriptions = new lib.SubscriptionController();
|
|
ownSubscriptions.handle(RegisteredEvent, async () => new RegisteredEvent());
|
|
await ownSubscriptions.handleRequest(
|
|
getLink(connectorData.id),
|
|
new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"),
|
|
requestSource(417), connectorData.src
|
|
);
|
|
|
|
const message = lastSentMessage();
|
|
assert.equal(message.name, RegisteredEvent.name);
|
|
assert.equal(message.dst.requestId, undefined, "replay addressed to a request id");
|
|
assert.deepEqual(message.dst.toJSON(), connectorData.dst.toJSON());
|
|
});
|
|
|
|
it("should strip the request id when replacing filters creates the subscription", async function() {
|
|
const ownSubscriptions = new lib.SubscriptionController();
|
|
ownSubscriptions.handle(RegisteredEvent);
|
|
await ownSubscriptions.handleRequest(
|
|
getLink(connectorData.id),
|
|
new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "foo"),
|
|
requestSource(918), connectorData.src
|
|
);
|
|
|
|
ownSubscriptions.broadcast(new RegisteredEvent(), "foo");
|
|
await onceConnectorSend(connectorData.id);
|
|
|
|
const message = lastSentMessage();
|
|
assert.equal(message.name, RegisteredEvent.name);
|
|
assert.equal(message.dst.requestId, undefined, "event addressed to a request id");
|
|
assert.deepEqual(message.dst.toJSON(), connectorData.dst.toJSON());
|
|
});
|
|
});
|
|
|
|
describe("subscribe", function() {
|
|
it("should create a new subscription", async function() {
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
|
|
await subscriptions.handleRequest(
|
|
getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
|
|
);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, RegisteredEvent);
|
|
});
|
|
|
|
it("should union filters for existing subscriber", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo");
|
|
await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "bar");
|
|
await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "foo");
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, RegisteredEvent);
|
|
link.connector.sentMessages.pop();
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "bar");
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, RegisteredEvent);
|
|
link.connector.sentMessages.pop();
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "baz");
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoEvent(0);
|
|
});
|
|
|
|
it("should union filters with all when not specified", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo");
|
|
await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
|
|
await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "baz");
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, RegisteredEvent);
|
|
});
|
|
|
|
it("should reject filters over the limit", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
const ownSubscriptions = new lib.SubscriptionController(2);
|
|
ownSubscriptions.handle(RegisteredEvent);
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]);
|
|
await ownSubscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "baz");
|
|
await assert.rejects(
|
|
ownSubscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src),
|
|
lib.ValidationError,
|
|
);
|
|
|
|
ownSubscriptions.broadcast(new RegisteredEvent(), "baz");
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoEvent(0);
|
|
});
|
|
|
|
it("should not count existing filters towards the limit", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
const ownSubscriptions = new lib.SubscriptionController(2);
|
|
ownSubscriptions.handle(RegisteredEvent);
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]);
|
|
await ownSubscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo");
|
|
await ownSubscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
});
|
|
|
|
it("should not limit filters when subscribed to all", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
const ownSubscriptions = new lib.SubscriptionController(2);
|
|
ownSubscriptions.handle(RegisteredEvent);
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
|
|
await ownSubscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "subscribe", 0, ["foo", "bar", "baz"]
|
|
);
|
|
await ownSubscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
});
|
|
|
|
it("should not limit filters when the limit is 0", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request = new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "subscribe", 0, ["foo", "bar", "baz"]
|
|
);
|
|
await subscriptions.handleRequest(link, request, connectorData.dst, connectorData.src);
|
|
});
|
|
});
|
|
|
|
describe("unsubscribe", function() {
|
|
it("should remove a subscription", async function() {
|
|
const request = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe");
|
|
await subscriptions.handleRequest(
|
|
getLink(0), request, connectorSetupDate[0].dst, connectorSetupDate[0].src
|
|
);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoEvent(0);
|
|
});
|
|
|
|
it("should subtract filters", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, ["foo", "bar"]);
|
|
await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe", 0, "foo");
|
|
await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "foo");
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoEvent(0);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "bar");
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, RegisteredEvent);
|
|
});
|
|
|
|
it("should remove subscription when empty", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo");
|
|
await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe", 0, "foo");
|
|
await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "foo");
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoEvent(0);
|
|
});
|
|
|
|
it("should remain subscribed to all if subtracting a filter", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");
|
|
await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "unsubscribe", 0, "foo");
|
|
await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, RegisteredEvent);
|
|
});
|
|
});
|
|
|
|
describe("replace", function() {
|
|
it("should replace existing filters", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo");
|
|
await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "bar");
|
|
await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "foo");
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoEvent(0);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "bar");
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, RegisteredEvent);
|
|
});
|
|
|
|
it("should subscribe when no existing subscriber exists", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "foo");
|
|
await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "foo");
|
|
await onceConnectorSend(0);
|
|
assertLastEvent(0, RegisteredEvent);
|
|
});
|
|
|
|
it("should remove subscription when replacing with empty", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request0 = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe", 0, "foo");
|
|
await subscriptions.handleRequest(link, request0, connectorData.dst, connectorData.src);
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, []);
|
|
await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent(), "foo");
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoEvent(0);
|
|
});
|
|
|
|
it("should do nothing when replacing with empty if no subscriber exists", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
|
|
const request1 = new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, []);
|
|
await subscriptions.handleRequest(link, request1, connectorData.dst, connectorData.src);
|
|
|
|
subscriptions.broadcast(new RegisteredEvent());
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoEvent(0);
|
|
});
|
|
|
|
it("should reject filters over the limit", async function() {
|
|
const link = getLink(0);
|
|
const connectorData = connectorSetupDate[0];
|
|
const ownSubscriptions = new lib.SubscriptionController(2);
|
|
ownSubscriptions.handle(RegisteredEvent);
|
|
|
|
const request = new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", 0, ["foo", "bar", "baz"]
|
|
);
|
|
await assert.rejects(
|
|
ownSubscriptions.handleRequest(link, request, connectorData.dst, connectorData.src),
|
|
lib.ValidationError,
|
|
);
|
|
});
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("class EventSubscriber", function() {
|
|
/** @type {MockControl} */
|
|
let mockControl;
|
|
/** @type {lib.EventSubscriber} */
|
|
let eventSubscriber;
|
|
|
|
beforeEach(function() {
|
|
mockControl = new MockControl(new MockConnector(
|
|
addr({ controlId: 0 }),
|
|
addr("controller"),
|
|
));
|
|
eventSubscriber = new lib.EventSubscriber(RegisteredEvent, mockControl);
|
|
});
|
|
|
|
function onceConnectorSend() {
|
|
return events.once(mockControl.connector, "send");
|
|
}
|
|
|
|
function assertLastMessage(message) {
|
|
const Message = message.constructor;
|
|
const lastMessage = mockControl.connector.sentMessages.at(-1);
|
|
const name = Message.plugin ? `${Message.plugin}:${Message.name}` : Message.name;
|
|
assert.equal(lastMessage.name, name);
|
|
assert.deepEqual(lastMessage.data, message);
|
|
}
|
|
|
|
function assertNoMessageSent() {
|
|
assert.deepEqual(mockControl.connector.sentMessages, []);
|
|
}
|
|
|
|
describe("constructor()", function() {
|
|
it("should not accept unregistered events", function() {
|
|
assert.throws(
|
|
() => new lib.EventSubscriber(UnregisteredEvent, mockControl),
|
|
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
|
|
);
|
|
});
|
|
|
|
it("should handle the provided event", function() {
|
|
assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true);
|
|
});
|
|
});
|
|
|
|
describe("subscribe()", function() {
|
|
it("should send subscribe(all) on first subscription", async function() {
|
|
eventSubscriber.subscribe(() => {});
|
|
await onceConnectorSend();
|
|
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "subscribe", -1, lib.SubscriptionFilters.all()
|
|
));
|
|
});
|
|
|
|
it("should send replace(filters) when last callback unsubscribes", async function() {
|
|
const unsubscribe = eventSubscriber.subscribe(() => {});
|
|
await onceConnectorSend();
|
|
|
|
unsubscribe();
|
|
await onceConnectorSend();
|
|
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.empty()
|
|
));
|
|
});
|
|
|
|
it("should unsubscribe without setImmediate, which browsers lack", async function() {
|
|
const unsubscribe = eventSubscriber.subscribe(() => {});
|
|
await onceConnectorSend();
|
|
|
|
const setImmediate = globalThis.setImmediate;
|
|
delete globalThis.setImmediate;
|
|
try {
|
|
unsubscribe();
|
|
} finally {
|
|
globalThis.setImmediate = setImmediate;
|
|
}
|
|
await onceConnectorSend();
|
|
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.empty()
|
|
));
|
|
});
|
|
|
|
it("should notify all callbacks", async function() {
|
|
let called1, called2;
|
|
const event = new RegisteredEvent();
|
|
|
|
const unsubscribe = eventSubscriber.subscribe((e, s) => { called1 = [e, s]; });
|
|
eventSubscriber.subscribe((e, s) => { called2 = [e, s]; });
|
|
|
|
await eventSubscriber._handleEvent(event);
|
|
|
|
assert.deepEqual(called1, [event, true]);
|
|
assert.deepEqual(called2, [event, true]);
|
|
});
|
|
|
|
it("should not notify unsubscribed callbacks", async function() {
|
|
let called1, called2;
|
|
const event = new RegisteredEvent();
|
|
|
|
const unsubscribe = eventSubscriber.subscribe((e, s) => { called1 = [e, s]; });
|
|
eventSubscriber.subscribe((e, s) => { called2 = [e, s]; });
|
|
|
|
unsubscribe();
|
|
await new Promise(r => setTimeout(r));
|
|
await eventSubscriber._handleEvent(event);
|
|
|
|
assert.deepEqual(called1, undefined);
|
|
assert.deepEqual(called2, [event, true]);
|
|
});
|
|
|
|
it("should not throw when unsubscribing multiple times", async function() {
|
|
const unsubscribe = eventSubscriber.subscribe(() => {});
|
|
unsubscribe();
|
|
await onceConnectorSend();
|
|
unsubscribe();
|
|
});
|
|
});
|
|
|
|
describe("filters", function() {
|
|
it("should add filters and send subscribe", async function() {
|
|
eventSubscriber.addFilters("foo");
|
|
await onceConnectorSend();
|
|
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "subscribe", -1, lib.SubscriptionFilters.fromShorthand("foo")
|
|
));
|
|
});
|
|
|
|
it("should remove filters and send unsubscribe", async function() {
|
|
eventSubscriber.addFilters("foo");
|
|
await onceConnectorSend();
|
|
|
|
eventSubscriber.removeFilters("foo");
|
|
await onceConnectorSend();
|
|
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "unsubscribe", -1, lib.SubscriptionFilters.fromShorthand("foo")
|
|
));
|
|
});
|
|
|
|
it("should clear filters and unsubscribe when no callbacks", async function() {
|
|
eventSubscriber.addFilters("foo");
|
|
await onceConnectorSend();
|
|
|
|
eventSubscriber.clearFilters();
|
|
await onceConnectorSend();
|
|
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "unsubscribe", -1, lib.SubscriptionFilters.all()
|
|
));
|
|
});
|
|
|
|
it("should not unsubscribe on clearFilters if callbacks exist", async function() {
|
|
eventSubscriber.subscribe(() => {});
|
|
await onceConnectorSend();
|
|
|
|
eventSubscriber.clearFilters();
|
|
await new Promise(r => setImmediate(r));
|
|
|
|
// No new message
|
|
assert.equal(mockControl.connector.sentMessages.length, 1);
|
|
});
|
|
|
|
it("should correctly report hasFilters", function() {
|
|
eventSubscriber.addFilters(["foo", "bar"]);
|
|
assert.equal(eventSubscriber.hasFilters("foo"), true);
|
|
assert.equal(eventSubscriber.hasFilters("baz"), false);
|
|
});
|
|
|
|
it("should only fire when filters are not empty", async function() {
|
|
let called = false;
|
|
eventSubscriber.filteredHandler = () => { called = true; };
|
|
|
|
await eventSubscriber._handleEvent(new RegisteredEvent());
|
|
assert.equal(called, false);
|
|
|
|
eventSubscriber.addFilters("foo");
|
|
await onceConnectorSend();
|
|
|
|
await eventSubscriber._handleEvent(new RegisteredEvent());
|
|
assert.equal(called, true);
|
|
});
|
|
});
|
|
|
|
describe("getSnapshot()", function() {
|
|
it("should cache snapshot until state changes", async function() {
|
|
const snap1 = eventSubscriber.getSnapshot();
|
|
const snap2 = eventSubscriber.getSnapshot();
|
|
assert.strictEqual(snap1, snap2);
|
|
|
|
await eventSubscriber._handleEvent(new RegisteredEvent());
|
|
const snap3 = eventSubscriber.getSnapshot();
|
|
assert.notStrictEqual(snap1, snap3);
|
|
|
|
eventSubscriber.synced = !eventSubscriber.synced;
|
|
const snap4 = eventSubscriber.getSnapshot();
|
|
assert.notStrictEqual(snap3, snap4);
|
|
});
|
|
});
|
|
|
|
describe("handleConnectionEvent", function() {
|
|
it("should send replace(all) on reconnect if subscribed", async function() {
|
|
eventSubscriber.subscribe(() => {});
|
|
await onceConnectorSend();
|
|
|
|
eventSubscriber.control.connector.emit("connect");
|
|
await onceConnectorSend();
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all()
|
|
));
|
|
|
|
mockControl.connector.sentMessages.pop();
|
|
eventSubscriber.handleConnectionEvent("resume");
|
|
await onceConnectorSend();
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all()
|
|
));
|
|
});
|
|
|
|
it("should send replace(filters) when only filters exist", async function() {
|
|
eventSubscriber.addFilters("foo");
|
|
await onceConnectorSend();
|
|
|
|
eventSubscriber.control.connector.emit("connect");
|
|
await onceConnectorSend();
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.fromShorthand("foo")
|
|
));
|
|
|
|
mockControl.connector.sentMessages.pop();
|
|
eventSubscriber.handleConnectionEvent("resume");
|
|
await onceConnectorSend();
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.fromShorthand("foo")
|
|
));
|
|
});
|
|
|
|
it("should set synced false on close", async function() {
|
|
eventSubscriber.synced = true;
|
|
eventSubscriber.control.connector.emit("close");
|
|
assert.equal(eventSubscriber.synced, false);
|
|
|
|
eventSubscriber.synced = true;
|
|
eventSubscriber.handleConnectionEvent("drop");
|
|
assert.equal(eventSubscriber.synced, false);
|
|
|
|
eventSubscriber.synced = true;
|
|
eventSubscriber.handleConnectionEvent("close");
|
|
assert.equal(eventSubscriber.synced, false);
|
|
});
|
|
|
|
it("should invoke callbacks on close", async function() {
|
|
let calledWith;
|
|
eventSubscriber.subscribe((v, s) => { calledWith = [v, s]; });
|
|
const [msg] = await onceConnectorSend();
|
|
|
|
mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
|
|
lib.SubscriptionRequest.Response.fromJSON(false)
|
|
));
|
|
await new Promise(r => setImmediate(r));
|
|
assert.equal(eventSubscriber.synced, true);
|
|
calledWith = undefined;
|
|
|
|
eventSubscriber.control.connector.emit("close");
|
|
assert.equal(eventSubscriber.synced, false);
|
|
assert.deepEqual(calledWith, [null, false]);
|
|
});
|
|
});
|
|
|
|
describe(".handleAccountEvent()", function() {
|
|
it("should send replace(all) if subscribed", async function() {
|
|
eventSubscriber.subscribe(() => {});
|
|
await onceConnectorSend();
|
|
|
|
eventSubscriber.handleAccountEvent();
|
|
await onceConnectorSend();
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all()
|
|
));
|
|
});
|
|
|
|
it("should send replace(filters) when only filters exist", async function() {
|
|
eventSubscriber.addFilters("foo");
|
|
await onceConnectorSend();
|
|
|
|
eventSubscriber.handleAccountEvent();
|
|
await onceConnectorSend();
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.fromShorthand("foo")
|
|
));
|
|
});
|
|
|
|
it("should send nothing if not subscribed", async function() {
|
|
eventSubscriber.handleAccountEvent();
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoMessageSent();
|
|
});
|
|
|
|
it("should send once when connecting and the account is reported", async function() {
|
|
eventSubscriber.subscribe(() => {});
|
|
await onceConnectorSend();
|
|
mockControl.connector.sentMessages.pop();
|
|
|
|
eventSubscriber.handleAccountEvent();
|
|
mockControl.connector.emit("connect");
|
|
await new Promise(r => setImmediate(r));
|
|
assert.equal(mockControl.connector.sentMessages.length, 1);
|
|
});
|
|
|
|
it("should be called on account updates from the control", async function() {
|
|
const handlers = [];
|
|
mockControl = new MockControl(new MockConnector(
|
|
addr({ controlId: 0 }),
|
|
addr("controller"),
|
|
));
|
|
mockControl.onAccountUpdate = handler => { handlers.push(handler); };
|
|
const subscriber = new lib.EventSubscriber(RegisteredEvent, mockControl);
|
|
assert.equal(handlers.length, 1);
|
|
|
|
subscriber.subscribe(() => {});
|
|
await onceConnectorSend();
|
|
handlers[0]({ name: "test", roles: [] });
|
|
await onceConnectorSend();
|
|
assertLastMessage(new lib.SubscriptionRequest(
|
|
RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all()
|
|
));
|
|
});
|
|
});
|
|
|
|
describe("._sendRequest()", function() {
|
|
it("should do nothing if connection is not valid", async function() {
|
|
mockControl.connector.valid = false;
|
|
eventSubscriber.subscribe(() => {});
|
|
await new Promise(r => setImmediate(r));
|
|
assertNoMessageSent();
|
|
});
|
|
|
|
it("should set synced when no updates returned", async function() {
|
|
let calledWith;
|
|
eventSubscriber.subscribe((e, s) => { calledWith = [e, s]; });
|
|
const [msg] = await onceConnectorSend();
|
|
|
|
mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
|
|
lib.SubscriptionRequest.Response.fromJSON(false)
|
|
));
|
|
|
|
await new Promise(r => setImmediate(r));
|
|
assert.deepEqual(calledWith, [null, true]);
|
|
assert.equal(eventSubscriber.synced, true);
|
|
});
|
|
|
|
it("should set synced when updates returned via _handleEvent", async function() {
|
|
let calledWith;
|
|
eventSubscriber.subscribe((e, s) => { calledWith = [e, s]; });
|
|
const [msg] = await onceConnectorSend();
|
|
|
|
mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
|
|
lib.SubscriptionRequest.Response.fromJSON(true)
|
|
));
|
|
|
|
await new Promise(r => setImmediate(r));
|
|
assert.equal(calledWith, undefined);
|
|
assert.equal(eventSubscriber.synced, false);
|
|
|
|
const eventUpdate = new RegisteredEvent();
|
|
eventSubscriber._handleEvent(eventUpdate);
|
|
|
|
assert.deepEqual(calledWith, [eventUpdate, true]);
|
|
assert.equal(eventSubscriber.synced, true);
|
|
});
|
|
|
|
it("should log error when sendTo throws", async function() {
|
|
let logged;
|
|
const originalError = lib.logger.error;
|
|
lib.logger.error = msg => { logged = msg; };
|
|
|
|
// Mock sendTo to throw generic error
|
|
mockControl.sendTo = async function() {
|
|
throw new Error("boom");
|
|
};
|
|
|
|
try {
|
|
eventSubscriber.subscribe(() => {});
|
|
await new Promise(r => setImmediate(r));
|
|
} finally {
|
|
lib.logger.error = originalError;
|
|
}
|
|
assert.notEqual(logged, undefined);
|
|
});
|
|
|
|
it("should not log error when sendTo throws SessionLost", async function() {
|
|
let logged;
|
|
const originalError = lib.logger.error;
|
|
lib.logger.error = msg => { logged = msg; };
|
|
|
|
mockControl.sendTo = async function() {
|
|
throw new lib.SessionLost("Session lost");
|
|
};
|
|
|
|
try {
|
|
eventSubscriber.subscribe(() => {});
|
|
await new Promise(r => setImmediate(r));
|
|
} finally {
|
|
lib.logger.error = originalError;
|
|
}
|
|
assert.equal(logged, undefined);
|
|
});
|
|
|
|
it("should notify callbacks as not synced when the request fails", async function() {
|
|
let calledWith;
|
|
eventSubscriber.subscribe((e, s) => { calledWith = [e, s]; });
|
|
const [msg] = await onceConnectorSend();
|
|
mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
|
|
lib.SubscriptionRequest.Response.fromJSON(false)
|
|
));
|
|
await new Promise(r => setImmediate(r));
|
|
assert.equal(eventSubscriber.synced, true);
|
|
|
|
mockControl.sendTo = async function() {
|
|
throw new lib.PermissionError("Permission denied");
|
|
};
|
|
eventSubscriber.handleAccountEvent();
|
|
await new Promise(r => setImmediate(r));
|
|
assert.equal(eventSubscriber.synced, false);
|
|
assert.deepEqual(calledWith, [null, false]);
|
|
assert.ok(eventSubscriber.error instanceof lib.PermissionError);
|
|
});
|
|
|
|
it("should expose errors through the snapshot", async function() {
|
|
eventSubscriber.control.sendTo = function() {
|
|
throw new lib.PermissionError("Permission denied");
|
|
};
|
|
|
|
const unsubscribe = eventSubscriber.subscribe(() => {});
|
|
await new Promise(r => setImmediate(r));
|
|
const [,, error1] = eventSubscriber.getSnapshot();
|
|
assert.ok(error1 instanceof lib.PermissionError);
|
|
assert.equal(error1.message, "Permission denied");
|
|
assert.equal(lib.Link.errorName(error1), "PermissionError");
|
|
|
|
eventSubscriber.control.sendTo = function() {};
|
|
unsubscribe();
|
|
await new Promise(r => setTimeout(r));
|
|
const [,, error2] = eventSubscriber.getSnapshot();
|
|
assert.deepEqual(error2, null);
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("class ValueSubscriber", function() {
|
|
let mockControl, subscriber;
|
|
|
|
class ValueEvent {
|
|
static type = "event";
|
|
static src = ["controller", "control"];
|
|
static dst = ["controller", "control"];
|
|
static permission = null;
|
|
constructor(value) {
|
|
this.value = value;
|
|
}
|
|
}
|
|
if (!lib.Link._eventsByName.has(ValueEvent.name)) {
|
|
lib.Link.register(ValueEvent);
|
|
}
|
|
|
|
beforeEach(function() {
|
|
mockControl = new MockControl(new MockConnector(
|
|
addr({ controlId: 0 }),
|
|
addr("controller"),
|
|
));
|
|
subscriber = new lib.ValueSubscriber(ValueEvent, mockControl);
|
|
});
|
|
|
|
it("should update value on event", async function() {
|
|
const value = { id: 1, updatedAtMs: 100, isDeleted: false };
|
|
await subscriber._handleEvent(new ValueEvent(value));
|
|
|
|
assert.deepEqual(subscriber.value, value);
|
|
});
|
|
|
|
it("should update lastUpdatedMs from value", async function() {
|
|
const value = { id: 1, updatedAtMs: 123, isDeleted: false };
|
|
await subscriber._handleEvent(new ValueEvent(value));
|
|
|
|
assert.equal(subscriber.lastUpdatedMs, 123);
|
|
});
|
|
|
|
it("should return value in snapshot", async function() {
|
|
const value = { id: 1, updatedAtMs: 50, isDeleted: false };
|
|
await subscriber._handleEvent(new ValueEvent(value));
|
|
|
|
const [snap, synced] = subscriber.getSnapshot();
|
|
assert.deepEqual(snap, value);
|
|
assert.equal(synced, false); // no active subscription
|
|
});
|
|
|
|
it("should replace value on newer event", async function() {
|
|
await subscriber._handleEvent(new ValueEvent({
|
|
id: 1, updatedAtMs: 10, isDeleted: false,
|
|
}));
|
|
|
|
await subscriber._handleEvent(new ValueEvent({
|
|
id: 1, updatedAtMs: 20, isDeleted: false,
|
|
}));
|
|
|
|
assert.equal(subscriber.value.updatedAtMs, 20);
|
|
});
|
|
});
|
|
|
|
describe("class MapSubscriber", function() {
|
|
let mockControl, subscriber;
|
|
|
|
class MapEvent {
|
|
static type = "event";
|
|
static src = ["controller", "control"];
|
|
static dst = ["controller", "control"];
|
|
static permission = null;
|
|
constructor(updates) {
|
|
this.updates = updates;
|
|
}
|
|
}
|
|
if (!lib.Link._eventsByName.has(MapEvent.name)) {
|
|
lib.Link.register(MapEvent);
|
|
}
|
|
|
|
beforeEach(function() {
|
|
mockControl = new MockControl(new MockConnector(
|
|
addr({ controlId: 0 }),
|
|
addr("controller"),
|
|
));
|
|
subscriber = new lib.MapSubscriber(MapEvent, mockControl);
|
|
});
|
|
|
|
it("should add values from updates", async function() {
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false },
|
|
]));
|
|
|
|
assert.equal(subscriber.values.size, 1);
|
|
assert.equal(subscriber.values.get(1).updatedAtMs, 10);
|
|
});
|
|
|
|
it("should update existing values", async function() {
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false },
|
|
]));
|
|
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 20, isDeleted: false },
|
|
]));
|
|
|
|
assert.equal(subscriber.values.get(1).updatedAtMs, 20);
|
|
});
|
|
|
|
it("should delete values when isDeleted is true", async function() {
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false },
|
|
]));
|
|
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 20, isDeleted: true },
|
|
]));
|
|
|
|
assert.equal(subscriber.values.has(1), false);
|
|
});
|
|
|
|
it("should compute lastUpdatedMs as max of updates", async function() {
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false },
|
|
{ id: 2, updatedAtMs: 50, isDeleted: false },
|
|
]));
|
|
|
|
assert.equal(subscriber.lastUpdatedMs, 50);
|
|
});
|
|
|
|
it("should return snapshot as a copy", async function() {
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false },
|
|
]));
|
|
|
|
const [snap] = subscriber.getSnapshot();
|
|
|
|
assert.deepEqual(snap, subscriber.values);
|
|
assert.notStrictEqual(snap, subscriber.values); // must be new Map
|
|
});
|
|
|
|
it("should not mutate previous snapshots", async function() {
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false },
|
|
]));
|
|
|
|
const [snap1] = subscriber.getSnapshot();
|
|
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 2, updatedAtMs: 20, isDeleted: false },
|
|
]));
|
|
|
|
const [snap2] = subscriber.getSnapshot();
|
|
|
|
assert.equal(snap1.has(2), false);
|
|
assert.equal(snap2.has(2), true);
|
|
});
|
|
|
|
it("should fetch everything again when a new session starts", async function() {
|
|
subscriber.subscribe(() => {});
|
|
await events.once(mockControl.connector, "send");
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false },
|
|
]));
|
|
const [before] = subscriber.getSnapshot();
|
|
mockControl.connector.sentMessages.length = 0;
|
|
|
|
mockControl.connector.emit("connect");
|
|
assert.equal(subscriber.values.size, 0);
|
|
assert.notStrictEqual(subscriber.getSnapshot()[0], before);
|
|
await new Promise(resolve => setImmediate(resolve));
|
|
const requests = mockControl.connector.sentMessages.filter(m => m.name === "SubscriptionRequest");
|
|
assert.equal(requests.length, 1);
|
|
assert.equal(requests[0].data.lastRequestTimeMs, -1);
|
|
});
|
|
|
|
it("should keep values when a session is resumed", async function() {
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false },
|
|
]));
|
|
subscriber.handleConnectionEvent("drop");
|
|
subscriber.handleConnectionEvent("resume");
|
|
assert.equal(subscriber.values.size, 1);
|
|
assert.equal(subscriber.lastUpdatedMs, 10);
|
|
});
|
|
|
|
it("should handle multiple updates correctly", async function() {
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false },
|
|
{ id: 2, updatedAtMs: 20, isDeleted: false },
|
|
]));
|
|
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 30, isDeleted: false },
|
|
{ id: 2, updatedAtMs: 40, isDeleted: true },
|
|
]));
|
|
|
|
assert.equal(subscriber.values.get(1).updatedAtMs, 30);
|
|
assert.equal(subscriber.values.has(2), false);
|
|
});
|
|
|
|
it("should warn on same timestamp but different content", async function() {
|
|
let logged = false;
|
|
const originalWarn = lib.logger.warn;
|
|
lib.logger.warn = msg => { logged = msg; };
|
|
|
|
try {
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false, foo: "a" },
|
|
]));
|
|
|
|
await subscriber._handleEvent(new MapEvent([
|
|
{ id: 1, updatedAtMs: 10, isDeleted: false, foo: "b" },
|
|
]));
|
|
} finally {
|
|
lib.logger.warn = originalWarn;
|
|
}
|
|
assert.notEqual(logged, undefined);
|
|
});
|
|
});
|
|
});
|