Remove channels to simplify subscription

Make subscriptions either on or off to greatly simplify the logic of
subscribing to events.
This commit is contained in:
Hornwitser committed 2023-12-03 20:32:48 +01:00
1 parent a2eb98ac93
commit 58914042cc
11 files changed
+51 -279

No files matched your search

-4
View File
@@ -63,10 +63,6 @@ export class HostUpdateEvent {
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(HostDetails.fromJSON(json.update));
}
get subscriptionChannel() {
return this.update.id;
}
}
export class HostMetricsRequest {
@@ -87,10 +87,6 @@ export class InstanceDetailsUpdateEvent {
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(InstanceDetails.fromJSON(json));
}
get subscriptionChannel() {
return this.details.id;
}
};
export class InstanceCreateRequest {
@@ -302,10 +298,6 @@ export class InstanceSaveListUpdateEvent {
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(json.instanceId, json.saves.map(i => SaveDetails.fromJSON(i)));
}
get subscriptionChannel() {
return this.instanceId;
}
}
export class InstanceCreateSaveRequest {
-8
View File
@@ -263,10 +263,6 @@ export class ModPackUpdateEvent {
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(ModPack.fromJSON(json.modPack));
}
get subscriptionChannel() {
return this.modPack.id;
}
}
export class ModUpdateEvent {
@@ -287,8 +283,4 @@ export class ModUpdateEvent {
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(ModInfo.fromJSON(json.mod));
}
get subscriptionChannel() {
return this.mod.name;
}
}
-4
View File
@@ -319,8 +319,4 @@ export class UserUpdateEvent {
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(User.fromJSON(json.user));
}
get subscriptionChannel() {
return this.user.name;
}
}
+15 -83
View File
@@ -5,14 +5,6 @@ import { Address, MessageRequest, IControllerUser } from "./data";
export type SubscriptionRequestHandler<T> = RequestHandler<SubscriptionRequest, Event<T> | null>;
export type EventSubscriberCallback<T> = (value: T) => void;
export type ChannelEvent<T> = Event<T> & {
get subscriptionChannel(): number | string;
}
export type ChannelEventClass<T> = EventClass<T> & {
new(...args: any): ChannelEvent<T>
}
/**
* Response received by the subscriber after a request
* It can contain an eventReplay value if the event sender implements a subscription handler
@@ -60,7 +52,7 @@ export class SubscriptionResponse {
/**
* A subscription request sent by a subscriber, this updates what events the subscriber will be sent
* The permission for this request copies the permission from the event being subscribed to
* allChannels: false, channels: [] will unsubscribe the subscriber from all notifications
* subscribe: false will unsubscribe the subscriber from all notifications
*/
export class SubscriptionRequest {
declare ["constructor"]: typeof SubscriptionRequest;
@@ -84,8 +76,7 @@ export class SubscriptionRequest {
constructor(
public eventName: string,
public allChannels: boolean,
public channels: Array<string | number> = [],
public subscribe: boolean,
public lastRequestTime: number = 0,
) {
if (!Link._eventsByName.has(eventName)) {
@@ -96,12 +87,11 @@ export class SubscriptionRequest {
static jsonSchema = Type.Tuple([
Type.String(),
Type.Boolean(),
Type.Array(Type.Union([Type.String(), Type.Number()])),
Type.Number(),
]);
toJSON() {
return [this.eventName, this.allChannels, this.channels, this.lastRequestTime];
return [this.eventName, this.subscribe, this.lastRequestTime];
}
static fromJSON(json: Static<typeof SubscriptionRequest.jsonSchema>): SubscriptionRequest {
@@ -111,7 +101,7 @@ export class SubscriptionRequest {
type EventData = {
subscriptionUpdate?: SubscriptionRequestHandler<unknown>,
subscriptions: Map<Link, { all: boolean, channels: Array<string | number>, onceClose: () => void }>,
subscriptions: Map<Link, { all: boolean, onceClose: () => void }>,
};
/**
@@ -149,10 +139,10 @@ export class SubscriptionController {
}
/**
* Broadcast an event to all subscribers of that event, will be filtered by channel if a ChannelEvent is provided
* Broadcast an event to all subscribers of that event
* @param event - Event to broadcast.
*/
broadcast<T>(event: Event<T> | ChannelEvent<T>) {
broadcast<T>(event: Event<T>) {
const entry = Link._eventsByClass.get(event.constructor);
if (!entry) {
throw new Error(`Unregistered Event class ${event.constructor.name}`);
@@ -161,11 +151,10 @@ export class SubscriptionController {
if (!eventData) {
throw new Error(`Event ${entry.name} is not a registered as subscribable`);
}
const channel = "subscriptionChannel" in event ? event.subscriptionChannel : null;
for (let [link, subscription] of eventData.subscriptions) {
if ((link.connector as WebSocketBaseConnector).closing) {
eventData.subscriptions.delete(link);
} else if ((subscription.all || (channel && subscription.channels.includes(channel)))) {
} else if (subscription.all) {
link.send(event);
}
}
@@ -189,7 +178,7 @@ export class SubscriptionController {
}
const eventReplay = eventData.subscriptionUpdate ? await eventData.subscriptionUpdate(event, src, dst) : null;
const link: Link = this.controller.wsServer.controlConnections.get(src.id);
if (event.allChannels === false && event.channels.length === 0) {
if (event.subscribe === false) {
let onceClose = eventData.subscriptions.get(link)?.onceClose;
if (onceClose) {
link.connector.off("close", onceClose);
@@ -202,7 +191,7 @@ export class SubscriptionController {
link.connector.once("close", onceClose);
}
eventData.subscriptions.set(
link, { all: event.allChannels, channels: event.channels, onceClose: onceClose },
link, { all: event.subscribe, onceClose: onceClose },
);
}
return new SubscriptionResponse(eventReplay);
@@ -210,19 +199,17 @@ export class SubscriptionController {
}
/**
* A class component to allow subscribing and unsubscribing to/from an event and its channels
* Channels are an optional feature which can be implemented to only get notifications of reliant updates
* Multiple handlers can be subscribed at the same time, on the same or different channels
* A class component to allow subscribing and unsubscribing to/from an event
* Multiple handlers can be subscribed at the same time
*/
export class EventSubscriber<T extends Event<T>, V=T> {
_callbacks = new Array<EventSubscriberCallback<V>>();
_channelCallbacks = new Map<string | number, Array<EventSubscriberCallback<V>>>();
lastResponse: Event<T> | null = null;
lastResponseTime = 0;
control?: Link;
constructor(
private event: EventClass<T> | ChannelEventClass<T>,
private event: EventClass<T>,
private prehandler?: (e: T) => V,
control?: Link
) {
@@ -240,22 +227,13 @@ export class EventSubscriber<T extends Event<T>, V=T> {
* @param response - event from subscribed resource
* @internal
*/
async _handle(response: Event<T> | ChannelEvent<T>) {
async _handle(response: Event<T>) {
this.lastResponse = response;
this.lastResponseTime = Date.now();
const value = this.prehandler ? this.prehandler(response as T) : response as any;
for (let callback of this._callbacks) {
callback(value);
}
const channel = "subscriptionChannel" in response ? response.subscriptionChannel : null;
if (channel) {
const callbacks = this._channelCallbacks.get(channel);
if (callbacks) {
for (let callback of callbacks) {
callback(value);
}
}
}
}
/**
@@ -282,21 +260,6 @@ export class EventSubscriber<T extends Event<T>, V=T> {
await this._updateSubscription();
}
/**
* Subscribe to receive event notifications for the given channel
* @param channel - Channel to subscribe to.
* @param handler -
* callback invoked whenever the subscribed resource changes on the
* given channel.
*/
async subscribeToChannel(channel: string | number, handler: EventSubscriberCallback<V>) {
if (!this._channelCallbacks.get(channel)) {
this._channelCallbacks.set(channel, []);
}
this._channelCallbacks.get(channel)!.push(handler);
await this._updateSubscription();
}
/**
* Unsubscribe from receiving all event notifications, does not effect
* the subscriptions of other handlers.
@@ -304,8 +267,8 @@ export class EventSubscriber<T extends Event<T>, V=T> {
* @param handler - A callback previously passed to {@link subscribe}
*
* @throws {Error}
* If your handler is not subscribed to all channels, does not
* accept anonymous functions
* If your handler is not subscribed, does not accept anonymous
* functions
*/
async unsubscribe(handler: EventSubscriberCallback<V>) {
let index = this._callbacks.lastIndexOf(handler);
@@ -317,36 +280,6 @@ export class EventSubscriber<T extends Event<T>, V=T> {
await this._updateSubscription();
}
/**
* Unsubscribe from receiving event notifications for a given channel,
* does not effect the subscriptions of other handlers.
*
* @param channel - A channel previously passed to {@link subscribeToChannel}
* @param handler - A callback previously passed to {@link subscribeToChannel}
*
* @throws {Error}
* If your handler is not subscribed to the channel or does not
* accept anonymous functions
*/
async unsubscribeFromChannel(channel: string | number, handler: EventSubscriberCallback<V>) {
const channelCallbacks = this._channelCallbacks.get(channel);
if (!channelCallbacks) {
throw new Error("handler is not registered");
}
let index = channelCallbacks.lastIndexOf(handler);
if (index === -1) {
throw new Error("handler is not registered");
}
if (channelCallbacks.length === 1) {
this._channelCallbacks.delete(channel);
} else {
channelCallbacks.splice(index, 1);
}
await this._updateSubscription();
}
/**
* Update the subscription with the controller based on current handler counts
*/
@@ -359,7 +292,6 @@ export class EventSubscriber<T extends Event<T>, V=T> {
const response = await this.control.send(new SubscriptionRequest(
entry.name,
this._callbacks.length > 0,
[...this._channelCallbacks.keys()],
this.lastResponseTime
));
+5 -2
View File
@@ -36,12 +36,15 @@ export function useHost(id: number): [ HostState, () => void ] {
updateHost();
function updateHandler(newHost: lib.HostDetails) {
if (newHost.id !== id) {
return;
}
setHost({ ...newHost, loading: false, missing: false, present: true });
}
control.hostUpdate.subscribeToChannel(id, updateHandler);
control.hostUpdate.subscribe(updateHandler);
return () => {
control.hostUpdate.unsubscribeFromChannel(id, updateHandler);
control.hostUpdate.unsubscribe(updateHandler);
};
}, [id]);
+5 -2
View File
@@ -32,12 +32,15 @@ export function useInstance(id: number): [InstanceState, ()=>void] {
updateInstance();
function updateHandler(newInstance: lib.InstanceDetails) {
if (newInstance.id !== id) {
return;
}
setInstance({ ...newInstance, present: true });
}
control.instanceUpdate.subscribeToChannel(id, updateHandler);
control.instanceUpdate.subscribe(updateHandler);
return () => {
control.instanceUpdate.unsubscribeFromChannel(id, updateHandler);
control.instanceUpdate.unsubscribe(updateHandler);
};
}, [id]);
+8 -2
View File
@@ -33,9 +33,15 @@ export function useModPack(id: number) {
}
updateModPack();
control.modPackUpdate.subscribeToChannel(id, setModPack);
function updateHandler(newModPack: lib.ModPack) {
if (newModPack.id !== id) {
return;
}
setModPack(newModPack);
}
control.modPackUpdate.subscribe(updateHandler);
return () => {
control.modPackUpdate.unsubscribeFromChannel(id, setModPack);
control.modPackUpdate.unsubscribe(updateHandler);
};
}, [id]);
+5 -2
View File
@@ -26,12 +26,15 @@ export function useSaves(instanceId?: number): lib.SaveDetails[] {
updateSaves();
function updateHandler(data: lib.InstanceSaveListUpdateEvent) {
if (data.instanceId !== instanceId) {
return;
}
setSaves(data.saves);
}
control.saveListUpdate.subscribeToChannel(instanceId!, updateHandler);
control.saveListUpdate.subscribe(updateHandler);
return () => {
control.saveListUpdate.unsubscribeFromChannel(instanceId!, updateHandler);
control.saveListUpdate.unsubscribe(updateHandler);
};
}, [instanceId]);
+5 -2
View File
@@ -79,12 +79,15 @@ export function useUser(name: string): [RawUserState, () => void] {
updateUser();
function updateHandler(newUser: lib.User) {
if (newUser.name !== name) {
return;
}
setUser({ ...newUser, present: true });
}
control.userUpdate.subscribeToChannel(name, updateHandler);
control.userUpdate.subscribe(updateHandler);
return () => {
control.userUpdate.unsubscribeFromChannel(name, updateHandler);
control.userUpdate.unsubscribe(updateHandler);
};
}, [name]);
+8 -162
View File
@@ -22,18 +22,6 @@ describe("lib/subscriptions", function() {
}
lib.Link.register(RegisteredEvent);
class ChannelEvent {
static type = "event";
static src = ["controller", "control"];
static dst = ["controller", "control"];
static permission = null;
get subscriptionChannel() {
return "channelOne";
}
}
lib.Link.register(ChannelEvent);
class StringPermissionEvent {
static type = "event";
static src = ["controller", "control"];
@@ -130,7 +118,7 @@ describe("lib/subscriptions", function() {
});
it("should be round trip json serialisable", function() {
const request = new lib.SubscriptionRequest(RegisteredEvent.name, true, ["channelOne"], 123);
const request = new lib.SubscriptionRequest(RegisteredEvent.name, true, 123);
const json = JSON.stringify(request);
assert.equal(typeof json, "string");
const reconstructed = lib.SubscriptionRequest.fromJSON(JSON.parse(json));
@@ -157,21 +145,6 @@ describe("lib/subscriptions", function() {
request: new lib.SubscriptionRequest(RegisteredEvent.name, true),
src: addr({ controlId: 1 }),
dst: addr("controller"),
}, {
id: 2,
request: new lib.SubscriptionRequest(ChannelEvent.name, true),
src: addr({ controlId: 2 }),
dst: addr("controller"),
}, {
id: 3,
request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]),
src: addr({ controlId: 3 }),
dst: addr("controller"),
}, {
id: 4,
request: new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelTwo"]),
src: addr({ controlId: 4 }),
dst: addr("controller"),
}];
function onceConnectorSend(id) {
@@ -227,7 +200,6 @@ describe("lib/subscriptions", function() {
describe("broadcast()", function() {
beforeEach(function() {
subscriptions.handle(RegisteredEvent);
subscriptions.handle(ChannelEvent);
for (let connectorData of connectorSetupDate) {
subscriptions._handleRequest(connectorData.request, connectorData.src, connectorData.dst);
}
@@ -245,16 +217,13 @@ describe("lib/subscriptions", function() {
new Error(`Event ${StringPermissionEvent.name} is not a registered as subscribable`)
);
});
it("should notify all links who subscribed to all channels", async function() {
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
assertNoEvent(2); // ChannelEvent: All
assertNoEvent(3); // ChannelEvent: channelOne
assertNoEvent(4); // ChannelEvent: channelTwo
});
it("should not notify a link who unsubscribed from all all channels", async function() {
it("should not notify a link who unsubscribed", async function() {
const connectorData = connectorSetupDate[0];
const request = new lib.SubscriptionRequest(RegisteredEvent.name, false);
await subscriptions._handleRequest(request, connectorData.src, connectorData.dst);
@@ -263,31 +232,6 @@ describe("lib/subscriptions", function() {
await onceConnectorSend(1);
assertNoEvent(0);
assertLastEvent(1, RegisteredEvent);
assertNoEvent(2);
assertNoEvent(3);
assertNoEvent(4);
});
it("should notify all links who subscribed the specific channel", async function() {
subscriptions.broadcast(new ChannelEvent());
await Promise.all([onceConnectorSend(2), onceConnectorSend(3)]);
assertNoEvent(0);
assertNoEvent(1);
assertLastEvent(2, ChannelEvent);
assertLastEvent(3, ChannelEvent);
assertNoEvent(4);
});
it("should not notify a link who unsubscribed from the specific channel", async function() {
const connectorData = connectorSetupDate[3];
const request = new lib.SubscriptionRequest(ChannelEvent.name, false, []);
await subscriptions._handleRequest(request, connectorData.src, connectorData.dst);
mockController.wsServer.controlConnections.get(3).connector.sentMessages.pop(); // Response to request
subscriptions.broadcast(new ChannelEvent());
await onceConnectorSend(2);
assertNoEvent(0);
assertNoEvent(1);
assertLastEvent(2, ChannelEvent);
assertNoEvent(3);
assertNoEvent(4);
});
it("should not notify links which are closed or closing", async function() {
const link = mockController.wsServer.controlConnections.get(0);
@@ -296,16 +240,12 @@ describe("lib/subscriptions", function() {
await onceConnectorSend(1);
assertNoEvent(0);
assertLastEvent(1, RegisteredEvent);
assertNoEvent(2);
assertNoEvent(3);
assertNoEvent(4);
});
});
describe("_handleRequest()", function() {
beforeEach(function() {
subscriptions.handle(RegisteredEvent);
subscriptions.handle(ChannelEvent);
});
it("should not accept subscriptions to unregistered events", function() {
@@ -324,14 +264,14 @@ describe("lib/subscriptions", function() {
new Error(`Event ${StringPermissionEvent.eventName} is not a registered as subscribable`)
);
});
it("should accept a subscription to all channels", async function() {
it("should accept a subscription", async function() {
const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
subscriptions.broadcast(new RegisteredEvent());
await onceConnectorSend(0);
assertLastEvent(0, RegisteredEvent);
});
it("should accept a unsubscription from all channels", async function() {
it("should accept a unsubscription", async function() {
const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
subscriptions.broadcast(new RegisteredEvent());
@@ -346,28 +286,6 @@ describe("lib/subscriptions", function() {
subscriptions.broadcast(new RegisteredEvent());
assertNoEvent(0);
});
it("should accept a subscription to a specific channel", async function() {
const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]);
await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
subscriptions.broadcast(new ChannelEvent());
await onceConnectorSend(0);
assertLastEvent(0, ChannelEvent);
});
it("should accept a unsubscription from a specific channel", async function() {
const request = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]);
await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
subscriptions.broadcast(new ChannelEvent());
await onceConnectorSend(0);
assertLastEvent(0, ChannelEvent);
const link = mockController.wsServer.controlConnections.get(0);
link.connector.sentMessages.pop();
const unsubRequest = new lib.SubscriptionRequest(ChannelEvent.name, false, []);
await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst);
subscriptions.broadcast(new ChannelEvent());
assertNoEvent(0);
});
it("should accept a respond with an event replay when returned by the handler", async function() {
subscriptions.handle(StringPermissionEvent, undefined, async function() {
return new StringPermissionEvent();
@@ -389,14 +307,13 @@ describe("lib/subscriptions", function() {
});
describe("class EventSubscriber", function() {
let channelEvent, mockControl, registeredEvent;
let mockControl, registeredEvent;
beforeEach(function() {
mockControl = new MockControl(new MockConnector(
addr({ controlId: 0 }),
addr("controller"),
));
registeredEvent = new lib.EventSubscriber(RegisteredEvent, undefined, mockControl);
channelEvent = new lib.EventSubscriber(ChannelEvent, undefined, mockControl);
});
function assertLastRequest(request) {
@@ -416,7 +333,6 @@ describe("lib/subscriptions", function() {
});
it("should handle the provided event, if a control link is provided", function() {
assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true);
assert.equal(mockControl._eventHandlers.has(ChannelEvent), true);
});
it("should call and use the return of a pre-handler, if provided", function() {
let calledWith = null;
@@ -454,19 +370,6 @@ describe("lib/subscriptions", function() {
});
});
describe("subscribeToChannel()", function() {
it("should allow subscriptions to a channel for an event", function() {
let calledWith = null;
let calledWithTwo = null;
const event = new ChannelEvent();
channelEvent.subscribeToChannel("channelOne", function(value) { calledWith = value; });
channelEvent.subscribeToChannel("channelTwo", function(value) { calledWithTwo = value; });
channelEvent._handle(event);
assert.deepEqual(calledWith, event);
assert.deepEqual(calledWithTwo, null);
});
});
describe("unsubscribe()", function() {
it("should allow unsubscribing from an event", function() {
let calledWith = null;
@@ -491,71 +394,14 @@ describe("lib/subscriptions", function() {
});
});
describe("unsubscribeFromChannel()", function() {
it("should allow unsubscribing from a channel for an event", function() {
let calledWith = null;
let event = new ChannelEvent();
function callback(value) { calledWith = value; }
channelEvent.subscribeToChannel("channelOne", callback);
channelEvent._handle(event);
assert.deepEqual(calledWith, event);
calledWith = null;
event = new ChannelEvent();
channelEvent.unsubscribeFromChannel("channelOne", callback);
channelEvent._handle(event);
assert.equal(calledWith, null);
let calledWithTwo = null;
function callbackTwo(value) { calledWithTwo = value; }
channelEvent.subscribeToChannel("channelOne", callback);
channelEvent.subscribeToChannel("channelOne", callbackTwo);
channelEvent._handle(event);
assert.equal(calledWith, event);
assert.equal(calledWithTwo, event);
calledWith = null;
calledWithTwo = null;
event = new ChannelEvent();
channelEvent.unsubscribeFromChannel("channelOne", callback);
channelEvent._handle(event);
assert.equal(calledWith, null);
assert.equal(calledWithTwo, event);
});
it("should throw an error if the handler was not subscribed", function() {
function callback() {}
assert.rejects(
() => channelEvent.unsubscribeFromChannel("channelOne", callback),
new Error("handler is not registered")
);
let calledWith = null;
function callbackTwo(value) { calledWith = value; }
channelEvent.subscribeToChannel("channelOne", callbackTwo);
assert.rejects(
() => channelEvent.unsubscribeFromChannel("channelOne", callback),
new Error("handler is not registered")
);
const event = new ChannelEvent();
channelEvent._handle(event);
assert.equal(calledWith, event);
});
});
describe("_updateSubscription()", function() {
it("should correctly request a subscription for all channels", async function() {
it("should correctly request a subscription", async function() {
const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true);
registeredEvent.subscribe(() => true);
await events.once(mockControl.connector, "send");
assertLastRequest(expected);
});
it("should correctly request a subscription for some channels", async function() {
const expected = new lib.SubscriptionRequest(ChannelEvent.name, false, ["channelOne"]);
channelEvent.subscribeToChannel("channelOne", () => true);
await events.once(mockControl.connector, "send");
assertLastRequest(expected);
});
it("should correctly request a subscription for no channels", async function() {
it("should correctly request a no subscriptions", async function() {
const expected = new lib.SubscriptionRequest(RegisteredEvent.name, false);
function callback() {}
registeredEvent.subscribe(callback);