aperx/factorio-clusterio
1
0 Fork 0
mirror of https://github.com/clusterio/clusterio.git synced 2026-11-12 23:04:52 +09:00
Code Issues Packages Projects Releases Wiki Activity

Remove SubscriptionResponse

Browse Source 
Since subscriptions are handlede per link there's no need for the
eventReplay to be returned as a part of the SubscriptionRequest, it can
be sent and processed normally as an event.  Simplify the logic by
sending eventReplay as an event and remove SubscriptionResponse.
This commit is contained in:
Hornwitser committed 2023-12-11 21:47:15 +01:00
1 parent 4439691695
commit 6cab1fa842
3 files changed
+11 -116
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/controller/src/Controller.ts
+6 -6
View File
@@ -664,7 +664,7 @@ export default class Controller {
const hosts = [...this.hosts.values()].filter(
host => host.updatedAt > request.lastRequestTime,
).map(host => host.toHostDetails());
return new lib.HostUpdatesEvent(hosts);
return hosts.length ? new lib.HostUpdatesEvent(hosts) : null;
}
/**
@@ -882,7 +882,7 @@ export default class Controller {
const instances = [...this.instances.values()].filter(
instance => instance.updatedAt > request.lastRequestTime,
).map(instance => instance.toInstanceDetails());
return new lib.InstanceDetailsUpdatesEvent(instances);
return instances.length ? new lib.InstanceDetailsUpdatesEvent(instances) : null;
}
savesUpdated(saves: lib.SaveDetails[]) {
@@ -892,7 +892,7 @@ export default class Controller {
async handleInstanceSaveDetailsSubscription(request: lib.SubscriptionRequest) {
const saves = [...this.saves.values()].filter(save => save.updatedAt > request.lastRequestTime);
return new lib.InstanceSaveDetailsUpdatesEvent(saves);
return saves.length ? new lib.InstanceSaveDetailsUpdatesEvent(saves) : null;
}
@@ -910,7 +910,7 @@ export default class Controller {
const modPacks = [...this.modPacks.values()].filter(
modPack => modPack.updatedAt > request.lastRequestTime,
);
return new lib.ModPackUpdatesEvent(modPacks);
return modPacks.length ? new lib.ModPackUpdatesEvent(modPacks) : null;
}
modsUpdated(mods: lib.ModInfo[]) {
@@ -923,7 +923,7 @@ export default class Controller {
const mods = [...this.modStore.files.values()].filter(
mod => mod.updatedAt > request.lastRequestTime,
);
return new lib.ModUpdatesEvent(mods);
return mods.length ? new lib.ModUpdatesEvent(mods) : null;
}
usersUpdated(users: ControllerUser[]) {
@@ -939,7 +939,7 @@ export default class Controller {
const users = [...this.userManager.users.values()].filter(
user => user.updatedAt > request.lastRequestTime,
);
return new lib.UserUpdatesEvent(users);
return users.length ? new lib.UserUpdatesEvent(users) : null;
}
/**
packages/lib/src/subscriptions.ts
+5 -54
View File
@@ -5,50 +5,6 @@ import { Address, MessageRequest, IControllerUser } from "./data";
export type SubscriptionRequestHandler<T> = RequestHandler<SubscriptionRequest, Event<T> | null>;
export type EventSubscriberCallback<T> = (value: T) => void;
/**
* Response received by the subscriber after a request
* It can contain an eventReplay value if the event sender implements a subscription handler
* This replay will be send to the handlers as if an update had just occurred
*/
export class SubscriptionResponse {
constructor(
public readonly eventReplay: Event<unknown> | null = null,
) {
if (eventReplay && !Link._eventsByClass.has(eventReplay.constructor)) {
throw new Error(`Unregistered Event class ${eventReplay.constructor.name}`);
}
}
static jsonSchema = Type.Union([
Type.Tuple([
Type.String(),
Type.Unknown(),
]),
Type.Null(),
]);
toJSON() {
if (this.eventReplay) {
const entry = Link._eventsByClass.get(this.eventReplay.constructor)!;
return [entry.name, this.eventReplay];
}
return null;
}
static fromJSON(json: Static<typeof SubscriptionResponse.jsonSchema>): SubscriptionResponse {
if (json) {
const entry = Link._eventsByName.get(json[0]);
if (!entry) {
throw new Error(`Unregistered Event class ${json[0]}`);
} else {
return new SubscriptionResponse(entry.eventFromJSON(json[1]));
}
} else {
return new 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
@@ -59,7 +15,6 @@ export class SubscriptionRequest {
static type = "request" as const;
static src = ["control", "instance"] as const;
static dst = "controller" as const;
static Response = SubscriptionResponse;
static permission(user: IControllerUser, message: MessageRequest) {
if (typeof message.data === "object" && message.data !== null) {
const data = message.data as Static<typeof SubscriptionRequest.jsonSchema>;
@@ -171,7 +126,6 @@ export class SubscriptionController {
* @param event - incomming event.
* @param src - Source address of incomming request.
* @param dst - destination address of incomming request.
* @returns Response to subscription request.
*/
async handleRequest(link: Link, event: SubscriptionRequest, src: Address, dst: Address) {
if (!Link._eventsByName.has(event.eventName)) {
@@ -181,16 +135,17 @@ export class SubscriptionController {
if (!eventData) {
throw new Error(`Event ${event.eventName} is not a registered as subscribable`);
}
let eventReplay: Event<unknown> | null = null;
if (event.subscribe === false) {
eventData.subscriptions.delete(link);
} else {
eventData.subscriptions.add(link);
if (eventData.subscriptionUpdate) {
eventReplay = await eventData.subscriptionUpdate(event, src, dst);
const eventReplay = await eventData.subscriptionUpdate(event, src, dst);
if (eventReplay) {
link.send(eventReplay);
}
}
}
return new SubscriptionResponse(eventReplay);
}
}
@@ -285,14 +240,10 @@ export class EventSubscriber<T extends Event<T>, V=T> {
}
const entry = Link._eventsByClass.get(this.event)!;
const response = await this.control.send(new SubscriptionRequest(
await this.control.send(new SubscriptionRequest(
entry.name,
this._callbacks.length > 0,
this.lastResponseTime
));
if (response.eventReplay) {
await this._handle(response.eventReplay as any);
}
}
}
test/lib/subscriptions.js
-56
View File
@@ -46,46 +46,6 @@ describe("lib/subscriptions", function() {
}
}
describe("class SubscriptionResponse", function() {
it("should be round trip json serialisable without an event replay", function() {
const response = new lib.SubscriptionResponse();
const json = JSON.stringify(response);
assert.equal(typeof json, "string");
const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json));
assert.deepEqual(reconstructed, response);
});
it("should be round trip json serialisable with an event replay", function() {
const response = new lib.SubscriptionResponse(new RegisteredEvent());
const json = JSON.stringify(response);
assert.equal(typeof json, "string");
const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json));
assert.deepEqual(reconstructed, response);
});
it("should be round trip json serialisable with an undefined event replay", function() {
const response = new lib.SubscriptionResponse(undefined);
const json = JSON.stringify(response);
assert.equal(typeof json, "string");
const reconstructed = lib.SubscriptionResponse.fromJSON(JSON.parse(json));
assert.deepEqual(reconstructed, response);
});
it("should be throw when given an unregistered event in a json", function() {
assert.throws(
() => lib.SubscriptionResponse.fromJSON([UnregisteredEvent.name, {}]),
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
);
});
it("should be throw when given an unregistered event", function() {
assert.throws(
() => new lib.SubscriptionResponse(new UnregisteredEvent()),
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
);
});
});
describe("class SubscriptionRequest", function() {
describe("permission()", function() {
function newSubscriptionRequestMessage(request) {
@@ -440,22 +400,6 @@ describe("lib/subscriptions", function() {
await events.once(mockControl.connector, "send");
assertLastRequest(expected);
});
it("should call handlers when a replay event is returned", async function() {
let calledWith = null;
const event = new RegisteredEvent();
const request = registeredEvent.subscribe(function(value) { calledWith = value; });
await events.once(mockControl.connector, "send");
mockControl.connector.emit("message", new lib.MessageResponse(
0,
mockControl.connector.dst,
new lib.Address(lib.Address.control, 0, mockControl._nextRequestId - 1),
new lib.SubscriptionResponse(event).toJSON()
));
await request;
assert.deepEqual(calledWith, event);
});
});
});
});