Merge pull request #534 from Hornwitser/refactor-subscriptions

Refactor subscriptions
This commit is contained in:
Hornwitser authored and GitHub committed 2024-01-07 02:54:20 +01:00
commit 943c0d9cb4
52 files changed
+1257 -1496

No files matched your search

+3 -2
View File
@@ -350,8 +350,9 @@ module.exports = {
"node/exports-style": "off",
"node/file-extension-in-import": "off",
"node/global-require": "off",
"node/handle-callback-err": "error",
"node/no-callback-literal": "error",
// Node.js callback passing style of (err, value) is not used in this project
"node/handle-callback-err": "off",
"node/no-callback-literal": "off",
"node/no-deprecated-api": "error",
"node/no-exports-assign": "error",
"node/no-extraneous-import": "off",
@@ -33,6 +33,11 @@ export default class BaseConnection extends lib.Link {
this.handle(lib.ModPackGetRequest, this.handleModPackGetRequest.bind(this));
this.handle(lib.ModPackGetDefaultRequest, this.handleModPackGetDefaultRequest.bind(this));
this.handle(lib.ModDownloadRequest, this.handleModDownloadRequest.bind(this));
this.handle(lib.SubscriptionRequest, this.handleSubscriptionRequest.bind(this));
this.connector.on("close", () => {
this._controller.subscriptions.unsubscribe(this);
});
}
async disconnect(code: number, reason: string) {
@@ -98,4 +103,8 @@ export default class BaseConnection extends lib.Link {
return stream.id;
}
async handleSubscriptionRequest(request: lib.SubscriptionRequest, src: lib.Address, dst: lib.Address) {
return await this._controller.subscriptions.handleRequest(this, request, src, dst);
}
}
+11 -11
View File
@@ -229,29 +229,29 @@ export default class BaseControllerPlugin {
async onPrepareHostDisconnect(connection: HostConnection) { }
/**
* Called when a mod pack is updated
* Called when one or more mod packs are updated
*
* Invoked when a mod pack has been added, updated or deleted in the
* cluster.
* Invoked when one or more mod pack have been added, updated or deleted
* in the cluster.
*
* If the mod pack has been deleted its `.isDeleted` property will be
* If a mod pack has been deleted its `.isDeleted` property will be
* true.
*
* @param modPack - Mod pack that updated.
* @param modPacks - Mod packs that updated.
*/
async onModPackUpdated(modPack: ModPack) { }
async onModPacksUpdated(modPacks: ModPack[]) { }
/**
* Called when a mod stored on the controller is updated
* Called when one or more mod stored on the controller are updated
*
* Invoked when a mod has been added, updated or deleted from the pool
* of shared mods stored on the cluster.
* Invoked when one or more mods have been added, updated or deleted
* from the pool of shared mods stored on the cluster.
*
* If a mod has been deleted its `.isDeleted` property will be true.
*
* @param mod - Mod that updated.
* @param mods - Mods that updated.
*/
async onModUpdated(mod: ModInfo) { }
async onModsUpdated(mods: ModInfo[]) { }
/**
* Called when a player joins or leaves an instance
+20 -66
View File
@@ -1,5 +1,4 @@
import type WsServerConnector from "./WsServerConnector";
import type { HostInfo } from "./HostConnection";
import events from "events";
@@ -86,6 +85,7 @@ export default class ControlConnection extends BaseConnection {
this.handle(lib.InstanceConfigSetFieldRequest, this.handleInstanceConfigSetFieldRequest.bind(this));
this.handle(lib.InstanceConfigSetPropRequest, this.handleInstanceConfigSetPropRequest.bind(this));
this.handle(lib.InstanceAssignRequest, this.handleInstanceAssignRequest.bind(this));
this.handle(lib.InstanceSaveDetailsListRequest, this.handleInstanceSaveDetailsListRequest.bind(this));
this.handle(lib.InstanceRenameSaveRequest, controller.sendRequestToHostByInstanceId.bind(controller));
this.handle(lib.InstanceCopySaveRequest, controller.sendRequestToHostByInstanceId.bind(controller));
this.handle(lib.InstanceDeleteSaveRequest, controller.sendRequestToHostByInstanceId.bind(controller));
@@ -188,30 +188,18 @@ export default class ControlConnection extends BaseConnection {
throw new Error(`Unknown host id (${request.hostId})`);
}
host.token_valid_after = Math.floor(Date.now() / 1000);
host.tokenValidAfter = Math.floor(Date.now() / 1000);
const hostConnection = this._controller.wsServer.hostConnections.get(request.hostId);
if (hostConnection) {
hostConnection.connector.terminate();
}
this._controller.hostUpdated(host as HostInfo);
this._controller.hostsUpdated([host]);
}
async handleHostListRequest(): Promise<lib.HostDetails[]> {
let list = [];
for (let host of this._controller.hosts.values()) {
list.push(new lib.HostDetails(
host.agent,
host.version,
host.name,
host.id,
this._controller.wsServer.hostConnections.has(host.id),
host.public_address,
host.token_valid_after,
));
}
return list;
return [...this._controller.hosts.values()].map(host => host.toHostDetails());
}
async handleHostGenerateTokenRequest(message: lib.HostGenerateTokenRequest) {
@@ -241,49 +229,11 @@ export default class ControlConnection extends BaseConnection {
}
async handleInstanceDetailsGetRequest(request: lib.InstanceDetailsGetRequest) {
let instance = this._controller.getRequestInstance(request.instanceId);
let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host");
if (assigned_host === null) {
assigned_host = undefined;
}
let game_port: number|null|undefined = instance.game_port;
if (game_port === null) {
game_port = undefined;
}
return new lib.InstanceDetails(
instance.config.get("instance.name"),
instance.id,
assigned_host,
game_port,
instance.status,
);
return this._controller.getRequestInstance(request.instanceId).toInstanceDetails();
}
async handleInstanceDetailsListRequest() {
let list = [];
for (let instance of this._controller.instances.values()) {
let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host");
if (assigned_host === null) {
assigned_host = undefined;
}
let game_port: number|null|undefined = instance.game_port;
if (game_port === null) {
game_port = undefined;
}
list.push(new lib.InstanceDetails(
instance.config.get("instance.name"),
instance.id,
assigned_host,
game_port,
instance.status,
));
}
return list;
return [...this._controller.instances.values()].map(instance => instance.toInstanceDetails());
}
// XXX should probably add a hook for host reuqests?
@@ -328,6 +278,10 @@ export default class ControlConnection extends BaseConnection {
await this._controller.instanceAssign(request.instanceId, request.hostId);
}
async handleInstanceSaveDetailsListRequest() {
return [...this._controller.saves.values()];
}
async handleInstanceDownloadSaveRequest(request: lib.InstanceDownloadSaveRequest) {
let { instanceId, name } = request;
let stream = await routes.createProxyStream(this._controller.app);
@@ -434,7 +388,7 @@ export default class ControlConnection extends BaseConnection {
throw new lib.RequestError(`Mod pack with ID ${modPack.id} already exist`);
}
this._controller.modPacks.set(modPack.id, modPack);
this._controller.modPackUpdated(modPack);
this._controller.modPacksUpdated([modPack]);
}
async handleModPackUpdateRequest(request: lib.ModPackUpdateRequest) {
@@ -443,7 +397,7 @@ export default class ControlConnection extends BaseConnection {
throw new lib.RequestError(`Mod pack with ID ${modPack.id} does not exist`);
}
this._controller.modPacks.set(modPack.id, modPack);
this._controller.modPackUpdated(modPack);
this._controller.modPacksUpdated([modPack]);
}
async handleModPackDeleteRequest(request: lib.ModPackDeleteRequest) {
@@ -454,7 +408,7 @@ export default class ControlConnection extends BaseConnection {
}
modPack.isDeleted = true;
this._controller.modPacks.delete(id);
this._controller.modPackUpdated(modPack);
this._controller.modPacksUpdated([modPack]);
}
async handleModGetRequest(request: lib.ModGetRequest) {
@@ -680,7 +634,7 @@ export default class ControlConnection extends BaseConnection {
async handleUserCreateRequest(request: lib.UserCreateRequest) {
let user = this._controller.userManager.createUser(request.name);
this._controller.userUpdated(user);
this._controller.usersUpdated([user]);
}
async handleUserRevokeTokenRequest(request: lib.UserRevokeTokenRequest) {
@@ -698,7 +652,7 @@ export default class ControlConnection extends BaseConnection {
controlConnection.connector.terminate();
}
}
this._controller.userUpdated(user);
this._controller.usersUpdated([user]);
}
async handleUserUpdateRolesRequest(request: lib.UserUpdateRolesRequest) {
@@ -715,7 +669,7 @@ export default class ControlConnection extends BaseConnection {
user.roleIds = new Set(request.roles);
this._controller.userPermissionsUpdated(user);
this._controller.userUpdated(user);
this._controller.usersUpdated([user]);
}
async handleUserSetAdminRequest(request: lib.UserSetAdminRequest) {
@@ -731,7 +685,7 @@ export default class ControlConnection extends BaseConnection {
}
user.isAdmin = admin;
this._controller.userUpdated(user);
this._controller.usersUpdated([user]);
this._controller.sendTo("allInstances", new lib.InstanceAdminlistUpdateEvent(name, admin));
}
@@ -749,7 +703,7 @@ export default class ControlConnection extends BaseConnection {
user.isBanned = banned;
user.banReason = reason;
this._controller.userUpdated(user);
this._controller.usersUpdated([user]);
this._controller.sendTo("allInstances", new lib.InstanceBanlistUpdateEvent(name, banned, reason));
}
@@ -766,7 +720,7 @@ export default class ControlConnection extends BaseConnection {
}
user.isWhitelisted = whitelisted;
this._controller.userUpdated(user);
this._controller.usersUpdated([user]);
this._controller.sendTo("allInstances", new lib.InstanceWhitelistUpdateEvent(name, whitelisted));
}
@@ -779,7 +733,7 @@ export default class ControlConnection extends BaseConnection {
user.isDeleted = true;
this._controller.userManager.users.delete(name);
this._controller.userUpdated(user);
this._controller.usersUpdated([user]);
if (user.isAdmin) {
this._controller.sendTo("allInstances", new lib.InstanceAdminlistUpdateEvent(name, false));
+174 -75
View File
@@ -2,6 +2,7 @@ import type winston from "winston";
import type { AddressInfo } from "net";
import type { ControllerArgs } from "../controller";
import express, { type Request, type Response, type NextFunction, type Application } from "express";
import type { Static } from "@sinclair/typebox";
import compression from "compression";
import events, { EventEmitter } from "events";
@@ -16,13 +17,14 @@ import * as lib from "@clusterio/lib";
const { logger, Summary, Gauge } = lib;
import HttpCloser from "./HttpCloser";
import InstanceInfo, { InstanceStatus } from "./InstanceInfo";
import InstanceInfo from "./InstanceInfo";
import * as metrics from "./metrics";
import * as routes from "./routes";
import ControllerUser from "./ControllerUser";
import UserManager from "./UserManager";
import WsServer from "./WsServer";
import HostConnection, { type HostInfo } from "./HostConnection";
import HostConnection from "./HostConnection";
import HostInfo from "./HostInfo";
import BaseControllerPlugin from "./BaseControllerPlugin";
const endpointDurationSummary = new Summary(
@@ -55,6 +57,8 @@ export default class Controller {
hostsDirty = false;
/** True if {@link Controller.instances} has changed since last time it was saved */
instancesDirty = false;
/** True if {@link Controller.saves} has changed since last time it was saved */
savesDirty = false;
/** True if {@link Controller.modPacks} has changed since last time it was saved */
modPacksDirty = false;
@@ -72,7 +76,7 @@ export default class Controller {
private _events: events.EventEmitter = new events.EventEmitter();
/** Event subscription controller */
subscriptions: lib.SubscriptionController;
subscriptions = new lib.SubscriptionController();
// Possible states are new, starting, running, stopping, stopped
private _state: string = "new";
@@ -96,6 +100,7 @@ export default class Controller {
const hosts = await Controller.loadHosts(path.join(databaseDirectory, "hosts.json"));
const instances = await Controller.loadInstances(path.join(databaseDirectory, "instances.json"));
const saves = await Controller.loadSaves(path.join(databaseDirectory, "saves.json"));
const modPacks = await Controller.loadModPacks(path.join(databaseDirectory, "mod-packs.json"));
const userManager = new UserManager(config);
await userManager.load(path.join(databaseDirectory, "users.json"));
@@ -107,6 +112,7 @@ export default class Controller {
return [
hosts,
instances,
saves,
modPacks,
modStore,
userManager,
@@ -123,6 +129,8 @@ export default class Controller {
public hosts = new Map<number, HostInfo>(),
/** Mapping of instance id to instance info */
public instances = new Map<number, InstanceInfo>(),
/** Mapping of save id to save details */
public saves = new Map<string, lib.SaveDetails>(),
/** Mapping of mod pack id to mod pack */
public modPacks = new Map<number, lib.ModPack>(),
/** Mods stored on the controller */
@@ -140,19 +148,20 @@ export default class Controller {
this.app.locals.streams = new Map();
this.wsServer = new WsServer(this);
this.subscriptions = new lib.SubscriptionController(this);
this.modStore.on("change", mod => {
this.modUpdated(mod);
this.modsUpdated([mod]);
});
// Handle subscriptions for all internal properties
this.subscriptions.handle(lib.HostUpdateEvent);
this.subscriptions.handle(lib.InstanceDetailsUpdateEvent);
this.subscriptions.handle(lib.InstanceSaveListUpdateEvent);
this.subscriptions.handle(lib.ModPackUpdateEvent);
this.subscriptions.handle(lib.ModUpdateEvent);
this.subscriptions.handle(lib.UserUpdateEvent);
this.subscriptions.handle(lib.HostUpdatesEvent, this.handleHostSubscription.bind(this));
this.subscriptions.handle(lib.InstanceDetailsUpdatesEvent, this.handleInstanceDetailsSubscription.bind(this));
this.subscriptions.handle(
lib.InstanceSaveDetailsUpdatesEvent, this.handleInstanceSaveDetailsSubscription.bind(this)
);
this.subscriptions.handle(lib.ModPackUpdatesEvent, this.handleModPackSubscription.bind(this));
this.subscriptions.handle(lib.ModUpdatesEvent, this.handleModSubscription.bind(this));
this.subscriptions.handle(lib.UserUpdatesEvent, this.handleUserSubscription.bind(this));
}
async start(args: ControllerArgs) {
@@ -426,6 +435,11 @@ export default class Controller {
await Controller.saveInstances(path.join(databaseDirectory, "instances.json"), this.instances);
}
if (this.savesDirty) {
this.savesDirty = false;
await Controller.saveSaves(path.join(databaseDirectory, "saves.json"), this.saves);
}
if (this.modPacksDirty) {
this.modPacksDirty = false;
await Controller.saveModPacks(path.join(databaseDirectory, "mod-packs.json"), this.modPacks);
@@ -439,7 +453,7 @@ export default class Controller {
}
static async loadHosts(filePath: string): Promise<Map<number, HostInfo>> {
let serialized: [number, HostInfo][];
let serialized: [number, Static<typeof HostInfo.jsonSchema>][];
try {
serialized = JSON.parse(await fs.readFile(filePath, { encoding: "utf8" }));
@@ -450,12 +464,21 @@ export default class Controller {
return new Map();
}
// TODO: Remove after release.
// TODO: migrate remove after release.
if (serialized.length && !(serialized[0] instanceof Array)) {
return new Map(); // Discard old format.
}
return new Map(serialized);
const hosts = new Map();
for (let [id, json] of serialized) {
const host = HostInfo.fromJSON(json);
if (host.connected) {
host.connected = false;
host.updatedAt = Date.now();
}
hosts.set(id, host);
}
return hosts;
}
static async saveHosts(filePath: string, hosts: Map<number, HostInfo>) {
@@ -467,16 +490,22 @@ export default class Controller {
let instances = new Map();
try {
let serialized = JSON.parse(await fs.readFile(filePath, { encoding: "utf8" }));
for (let serializedConfig of serialized) {
let serialized = JSON.parse(
await fs.readFile(filePath, { encoding: "utf8" })
) as Static<typeof InstanceInfo.jsonSchema>[];
for (let json of serialized) {
if (!json.config) { // migrate: from pre Alpha 14 format.
json = { config: json, status: "running" }; // Use running to force updatedAt
}
let instanceConfig = new lib.InstanceConfig("controller");
await instanceConfig.load(serializedConfig);
let status = instanceConfig.get("instance.assigned_host") === null ? "unassigned" : "unknown";
let instance = new InstanceInfo({
config: instanceConfig,
status: status as InstanceStatus,
});
instances.set(instanceConfig.get("instance.id"), instance);
await instanceConfig.load(json.config as lib.SerializedConfig);
let instance = InstanceInfo.fromJSON(json, instanceConfig);
const status = instance.config.get("instance.assigned_host") === null ? "unassigned" : "unknown";
if (instance.status !== status) {
instance.status = status;
instance.updatedAt = Date.now();
}
instances.set(instance.id, instance);
}
} catch (err: any) {
@@ -490,14 +519,28 @@ export default class Controller {
}
static async saveInstances(filePath: string, instances: Map<number, InstanceInfo>) {
let serialized = [];
for (const instance of instances.values()) {
serialized.push(instance.config.serialize());
}
let serialized = [...instances.values()].map(instance => instance.toJSON());
await lib.safeOutputFile(filePath, JSON.stringify(serialized, null, "\t"));
}
static async loadSaves(filePath: string): Promise<Map<string, lib.SaveDetails>> {
let json: Static<typeof lib.SaveDetails.jsonSchema>[];
try {
json = JSON.parse(await fs.readFile(filePath, { encoding: "utf8" }));
} catch (err: any) {
if (err.code !== "ENOENT") {
throw err;
}
return new Map();
}
return new Map(json.map(s => lib.SaveDetails.fromJSON(s)).map(s => [s.id, s]));
}
static async saveSaves(filePath: string, saves: Map<string, lib.SaveDetails>) {
await lib.safeOutputFile(filePath, JSON.stringify([...saves.values()], null, "\t"));
}
static async loadModPacks(filePath: string): Promise<Map<number, lib.ModPack>> {
let json;
try {
@@ -607,18 +650,21 @@ export default class Controller {
);
}
hostUpdated(host: HostInfo) {
let update = new lib.HostDetails(
host.agent,
host.version,
host.name,
host.id,
this.wsServer.hostConnections.has(host.id),
host.public_address,
host.token_valid_after,
);
hostsUpdated(hosts: HostInfo[]) {
const now = Date.now();
for (const host of hosts) {
host.updatedAt = now;
}
this.hostsDirty = true;
let updates = hosts.map(host => host.toHostDetails());
this.subscriptions.broadcast(new lib.HostUpdatesEvent(updates));
}
this.subscriptions.broadcast(new lib.HostUpdateEvent(update));
async handleHostSubscription(request: lib.SubscriptionRequest) {
const hosts = [...this.hosts.values()].filter(
host => host.updatedAt > request.lastRequestTime,
).map(host => host.toHostDetails());
return hosts.length ? new lib.HostUpdatesEvent(hosts) : null;
}
/**
@@ -685,15 +731,34 @@ export default class Controller {
};
instanceConfig.set("factorio.settings", settings);
let instance = new InstanceInfo({ config: instanceConfig, status: "unassigned" });
let instance = new InstanceInfo(instanceConfig, "unassigned", undefined, Date.now());
this.instances.set(instanceId, instance);
this.instancesDirty = true;
await lib.invokeHook(this.plugins, "onInstanceStatusChanged", instance);
this.addInstanceHooks(instance);
this.instanceDetailsUpdated(instance);
this.instanceDetailsUpdated([instance]);
return instance;
}
/**
* Removes all saves currently stored for the given instance, if any
* @param instanceId - Id of Instance to clear saves for.
* @internal
*/
clearSavesOfInstance(instanceId: number) {
const updates = [];
for (const [id, save] of this.saves) {
if (save.instanceId === instanceId) {
save.isDeleted = true;
updates.push(save);
this.saves.delete(id);
}
}
if (updates.length) {
this.savesUpdated(updates);
}
}
/**
* Change assigned host of an instance
*
@@ -733,16 +798,19 @@ export default class Controller {
}
}
// Remove saves recorded from currently assigned host if any
this.clearSavesOfInstance(instanceId);
// Assign to target
instance.config.set("instance.assigned_host", hostId);
this.instancesDirty = true;
// "fieldChanged" event handler will set this.instancesDirty
if (hostId !== undefined && newHostConnection) {
await newHostConnection.send(
new lib.InstanceAssignInternalRequest(instanceId, instance.config.serialize("host"))
);
} else {
instance.status = "unassigned";
this.instanceDetailsUpdated(instance);
this.instanceDetailsUpdated([instance]);
}
}
@@ -763,9 +831,12 @@ export default class Controller {
this.instances.delete(instanceId);
this.instancesDirty = true;
this.clearSavesOfInstance(instanceId);
let prev = instance.status;
instance.status = "deleted";
this.instanceDetailsUpdated(instance);
instance.updatedAt = Date.now();
this.instanceDetailsUpdated([instance]);
await lib.invokeHook(this.plugins, "onInstanceStatusChanged", instance, prev);
}
@@ -792,8 +863,9 @@ export default class Controller {
addInstanceHooks(instance: InstanceInfo) {
instance.config.on("fieldChanged", (group: lib.ConfigGroup, field: string, prev: any) => {
instance.updatedAt = Date.now();
if (group.name === "instance" && field === "name") {
this.instanceDetailsUpdated(instance);
this.instanceDetailsUpdated([instance]);
}
this.instancesDirty = true;
@@ -801,46 +873,73 @@ export default class Controller {
});
}
instanceDetailsUpdated(instance: InstanceInfo) {
let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host");
if (assigned_host === null) {
assigned_host = undefined;
}
let game_port: number|null|undefined = instance.game_port;
if (game_port === null) {
game_port = undefined;
}
this.subscriptions.broadcast(new lib.InstanceDetailsUpdateEvent(
new lib.InstanceDetails(
instance.config.get("instance.name"),
instance.id,
assigned_host,
game_port,
instance.status,
)
));
instanceDetailsUpdated(instances: InstanceInfo[]) {
const updates = instances.map(instance => instance.toInstanceDetails());
this.subscriptions.broadcast(new lib.InstanceDetailsUpdatesEvent(updates));
}
saveListUpdate(instanceId: number, saves: lib.SaveDetails[]) {
this.subscriptions.broadcast(new lib.InstanceSaveListUpdateEvent(instanceId, saves));
async handleInstanceDetailsSubscription(request: lib.SubscriptionRequest) {
const instances = [...this.instances.values()].filter(
instance => instance.updatedAt > request.lastRequestTime,
).map(instance => instance.toInstanceDetails());
return instances.length ? new lib.InstanceDetailsUpdatesEvent(instances) : null;
}
modPackUpdated(modPack: lib.ModPack) {
savesUpdated(saves: lib.SaveDetails[]) {
this.savesDirty = true;
this.subscriptions.broadcast(new lib.InstanceSaveDetailsUpdatesEvent(saves));
}
async handleInstanceSaveDetailsSubscription(request: lib.SubscriptionRequest) {
const saves = [...this.saves.values()].filter(save => save.updatedAt > request.lastRequestTime);
return saves.length ? new lib.InstanceSaveDetailsUpdatesEvent(saves) : null;
}
modPacksUpdated(modPacks: lib.ModPack[]) {
const now = Date.now();
for (const modPack of modPacks) {
modPack.updatedAt = now;
}
this.modPacksDirty = true;
this.subscriptions.broadcast(new lib.ModPackUpdateEvent(modPack));
lib.invokeHook(this.plugins, "onModPackUpdated", modPack);
this.subscriptions.broadcast(new lib.ModPackUpdatesEvent(modPacks));
lib.invokeHook(this.plugins, "onModPacksUpdated", modPacks);
}
modUpdated(mod: lib.ModInfo) {
this.subscriptions.broadcast(new lib.ModUpdateEvent(mod));
lib.invokeHook(this.plugins, "onModUpdated", mod);
async handleModPackSubscription(request: lib.SubscriptionRequest) {
const modPacks = [...this.modPacks.values()].filter(
modPack => modPack.updatedAt > request.lastRequestTime,
);
return modPacks.length ? new lib.ModPackUpdatesEvent(modPacks) : null;
}
userUpdated(user: ControllerUser) {
modsUpdated(mods: lib.ModInfo[]) {
// ModStore sets updatedAt for mods
this.subscriptions.broadcast(new lib.ModUpdatesEvent(mods));
lib.invokeHook(this.plugins, "onModsUpdated", mods);
}
async handleModSubscription(request: lib.SubscriptionRequest) {
const mods = [...this.modStore.files.values()].filter(
mod => mod.updatedAt > request.lastRequestTime,
);
return mods.length ? new lib.ModUpdatesEvent(mods) : null;
}
usersUpdated(users: ControllerUser[]) {
const now = Date.now();
for (const user of users) {
user.updatedAt = now;
}
this.userManager.dirty = true;
this.subscriptions.broadcast(new lib.UserUpdateEvent(user));
this.subscriptions.broadcast(new lib.UserUpdatesEvent(users));
}
async handleUserSubscription(request: lib.SubscriptionRequest) {
const users = [...this.userManager.users.values()].filter(
user => user.updatedAt > request.lastRequestTime,
);
return users.length ? new lib.UserUpdatesEvent(users) : null;
}
/**
@@ -35,6 +35,7 @@ export default class ControllerUser extends User implements IControllerUser {
json.is_banned,
json.is_whitelisted,
json.ban_reason,
json.updated_at,
json.is_deleted,
playerStats,
instanceStats,
+75 -39
View File
@@ -5,19 +5,10 @@ import * as lib from "@clusterio/lib";
const { logger } = lib;
import BaseConnection from "./BaseConnection";
import InstanceInfo, { InstanceStatus } from "./InstanceInfo";
import InstanceInfo from "./InstanceInfo";
import HostInfo from "./HostInfo";
export interface HostInfo {
agent: string;
id: number;
name: string;
version: string;
plugins: Record<string, string>;
public_address: string | undefined;
token_valid_after?: number;
}
/**
* Represents the connection to a host
*
@@ -25,10 +16,10 @@ export interface HostInfo {
* @alias module:controller/src/HostConnection
*/
export default class HostConnection extends BaseConnection {
private _agent: any;
private _id: any;
private _name: any;
private _version: any;
private _agent: string;
private _id: number;
private _name: string;
private _version: string;
plugins: Map<string, string>;
constructor(
@@ -45,18 +36,21 @@ export default class HostConnection extends BaseConnection {
this.plugins = new Map(Object.entries(registerData.plugins));
this._checkPluginVersions();
let currentHostInfo = this._controller.hosts.get(this._id);
this._controller.hosts.set(this._id, {
agent: this._agent,
id: this._id,
name: this._name,
version: this._version,
public_address: registerData.publicAddress,
plugins: registerData.plugins,
token_valid_after: currentHostInfo?.token_valid_after,
});
this._controller.hostsDirty = true;
const previousHostInfo = this._controller.hosts.get(this._id);
const currentHostInfo = new HostInfo(
this._agent,
this._id,
this._name,
this._version,
this.plugins,
true,
registerData.publicAddress ?? "",
previousHostInfo?.tokenValidAfter,
Date.now(),
false,
);
this._controller.hosts.set(this._id, currentHostInfo);
this._controller.hostsUpdated([currentHostInfo]);
for (let event of ["connect", "drop", "resume", "close"] as const) {
// eslint-disable-next-line no-loop-func
@@ -69,21 +63,26 @@ export default class HostConnection extends BaseConnection {
this.connector.on("close", () => {
// Update status to unknown for instances on this host.
const instances: InstanceInfo[] = [];
for (let instance of this._controller.instances.values()) {
if (instance.config.get("instance.assigned_host") !== this._id) {
continue;
}
instances.push(instance);
let prev = instance.status;
instance.status = "unknown";
this._controller.instanceDetailsUpdated(instance);
lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev);
}
this._controller.instanceDetailsUpdated(instances);
currentHostInfo.connected = false;
this._controller.hostsUpdated([currentHostInfo]);
});
this.handle(lib.InstanceStatusChangedEvent, this.handleInstanceStatusChangedEvent.bind(this));
this.handle(lib.InstancesUpdateRequest, this.handleInstancesUpdateRequest.bind(this));
this.handle(lib.InstanceSaveListUpdateEvent, this.handleInstanceSaveListUpdateEvent.bind(this));
this.handle(lib.InstanceSaveDetailsUpdatesEvent, this.handleInstanceSaveDetailsUpdatesEvent.bind(this));
this.handle(lib.LogMessageEvent, this.handleLogMessageEvent.bind(this));
this.handle(lib.InstancePlayerUpdateEvent, this.handleInstancePlayerUpdateEvent.bind(this));
}
@@ -177,10 +176,11 @@ export default class HostConnection extends BaseConnection {
}
let prev = instance.status;
instance.status = request.status as lib.InstanceStatus;
instance.game_port = request.gamePort || null;
instance.status = request.status;
instance.gamePort = request.gamePort;
instance.updatedAt = Date.now();
logger.verbose(`Instance ${instance.config.get("instance.name")} State: ${instance.status}`);
this._controller.instanceDetailsUpdated(instance);
this._controller.instanceDetailsUpdated([instance]);
await lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev);
}
@@ -195,6 +195,7 @@ export default class HostConnection extends BaseConnection {
}
// Assign instances the host has but controller does not
const instanceUpdates: InstanceInfo[] = [];
for (let instanceData of request.instances) {
let instanceConfig = new lib.InstanceConfig("controller");
await instanceConfig.load(instanceData.config as lib.SerializedConfig, "host");
@@ -212,9 +213,11 @@ export default class HostConnection extends BaseConnection {
// Already have this instance, update state instead
if (controllerInstance.status !== instanceData.status) {
let prev = controllerInstance.status;
controllerInstance.status = instanceData.status as InstanceStatus;
controllerInstance.status = instanceData.status;
controllerInstance.gamePort = instanceData.gamePort;
controllerInstance.updatedAt = Date.now();
logger.verbose(`Instance ${instanceConfig.get("instance.name")} State: ${instanceData.status}`);
this._controller.instanceDetailsUpdated(controllerInstance);
instanceUpdates.push(controllerInstance);
await lib.invokeHook(
this._controller.plugins, "onInstanceStatusChanged", controllerInstance, prev
);
@@ -224,7 +227,10 @@ export default class HostConnection extends BaseConnection {
instanceConfig.set("instance.assigned_host", this._id);
let newInstance = new InstanceInfo(
{ config: instanceConfig, status: instanceData.status as InstanceStatus }
instanceConfig,
instanceData.status,
instanceData.gamePort,
Date.now(),
);
this._controller.instances.set(instanceConfig.get("instance.id"), newInstance);
this._controller.instancesDirty = true;
@@ -234,10 +240,14 @@ export default class HostConnection extends BaseConnection {
instanceConfig.get("instance.id"), instanceConfig.serialize("host")
)
);
this._controller.instanceDetailsUpdated(newInstance);
instanceUpdates.push(newInstance);
await lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", newInstance);
}
if (instanceUpdates.length) {
this._controller.instanceDetailsUpdated(instanceUpdates);
}
// Push lists to make sure they are in sync.
let adminlist: Set<string> = new Set();
let banlist: Map<string, string> = new Map();
@@ -258,8 +268,34 @@ export default class HostConnection extends BaseConnection {
await this.send(new lib.SyncUserListsEvent(adminlist, banlist, whitelist));
}
async handleInstanceSaveListUpdateEvent(event: lib.InstanceSaveListUpdateEvent) {
this._controller.saveListUpdate(event.instanceId, event.saves);
async handleInstanceSaveDetailsUpdatesEvent(event: lib.InstanceSaveDetailsUpdatesEvent) {
const now = Date.now();
const updates: lib.SaveDetails[] = [];
for (const save of event.updates) {
const existingSave = this._controller.saves.get(save.id);
if (existingSave && save.equals(existingSave)) {
continue;
}
save.updatedAt = now;
this._controller.saves.set(save.id, save);
updates.push(save);
}
if (event.instanceId !== undefined) {
const updatedSaves = new Set(event.updates.map(s => s.id));
for (const [id, save] of this._controller.saves) {
if (save.instanceId === event.instanceId && !updatedSaves.has(id)) {
save.isDeleted = true;
save.updatedAt = now;
updates.push(save);
this._controller.saves.delete(id);
}
}
}
// Hosts eagerly send updates, which means we may get an update
// where nothing actualy changed.
if (updates.length) {
this._controller.savesUpdated(updates);
}
}
async handleLogMessageEvent(event: lib.LogMessageEvent) {
@@ -286,7 +322,7 @@ export default class HostConnection extends BaseConnection {
user.instanceStats.set(instanceId, event.stats);
user.recalculatePlayerStats();
this._controller.userUpdated(user);
this._controller.usersUpdated([user]);
let instance = this._controller.instances.get(instanceId)!;
await lib.invokeHook(this._controller.plugins, "onPlayerEvent", instance, {
+88
View File
@@ -0,0 +1,88 @@
import { Type, Static } from "@sinclair/typebox";
import { HostDetails } from "@clusterio/lib";
/**
* Info about a host known to the controller.
*/
export default class HostInfo {
constructor(
/** Agent this host last connected with */
public agent: string,
/** Id of this host */
public id: number,
/** Version this host last connected with */
public name: string,
/** Version this host last connected with */
public version: string,
/** Plugins this host last connected with */
public plugins: Map<string, string>,
/** True if this host is currently connected to controller */
public connected: boolean = false,
/** Value of host.public_address configured for this host */
public publicAddress: string = "",
/** Unix timestamp in seconds host token must be issued afterto be valid */
public tokenValidAfter: number = 0,
/** Millisecond Unix timestamp this entry was last updated at */
public updatedAt: number = 0,
/** True if this host has been deleted */
public isDeleted: boolean = false,
) { }
static jsonSchema = Type.Object({
"agent": Type.String(),
"id": Type.Number(),
"name": Type.String(),
"version": Type.String(),
"plugins": Type.Record(Type.String(), Type.String()),
"connected": Type.Boolean(),
"public_address": Type.String(),
"token_valid_after": Type.Number(),
"updated_at": Type.Number(),
"is_deleted": Type.Boolean(),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(
json.agent,
json.id,
json.name,
json.version,
new Map(Object.entries(json.plugins)),
json.connected,
json.public_address,
json.token_valid_after,
json.updated_at,
json.is_deleted,
);
}
toJSON(): Static<typeof HostInfo.jsonSchema> {
return {
agent: this.agent,
id: this.id,
name: this.name,
version: this.version,
plugins: Object.fromEntries(this.plugins),
connected: this.connected,
public_address: this.publicAddress,
token_valid_after: this.tokenValidAfter,
updated_at: this.updatedAt,
is_deleted: this.isDeleted,
};
}
toHostDetails() {
return new HostDetails(
this.agent,
this.version,
this.name,
this.id,
this.connected,
this.publicAddress,
this.tokenValidAfter,
this.updatedAt,
this.isDeleted,
);
}
}
+40 -26
View File
@@ -1,22 +1,5 @@
import type * as lib from "@clusterio/lib";
/**
* Current status of the instance. One of:
* - `unknown`: Instance is assigned to a host but this host is currently
* not connected to the contreller.
* - `unassigned`: Instance is not assigned to a a host and exists only on
* the controller.
* - `stopped`: Instance is stopped.
* - `starting`: Instance is in the process of starting up.
* - `running`: Instance is running normally.
* - `stopping`: Instance is in the process of stopping.
* - `creating_save`: Instance is in the process of creating a save.
* - `exporting_data`: Instance is in the process of exporting game data.
* - `deleted`: Instance has been deleted.
*/
export type InstanceStatus =
"unknown" | "unassigned" | "stopped" | "starting" | "running" |
"stopping" | "creating_save" | "exporting_data" | "deleted";
import * as lib from "@clusterio/lib";
import { Static, Type } from "@sinclair/typebox";
/**
@@ -24,25 +7,56 @@ export type InstanceStatus =
* @alias module:controller/src/InstanceInfo
*/
export default class InstanceInfo {
config: lib.InstanceConfig;
status: InstanceStatus;
game_port: number | null = null;
constructor(
{ config, status }:
{ config: lib.InstanceConfig, status: InstanceStatus }
public config: lib.InstanceConfig,
public status: lib.InstanceStatus,
public gamePort?: number,
public updatedAt = 0,
) {
this.config = config;
this.status = status;
}
toJSON() {
static jsonSchema = Type.Object({
"config": Type.Object({}),
"status": lib.InstanceStatus,
"gamePort": Type.Optional(Type.Number()),
"updatedAt": Type.Optional(Type.Number()),
});
static fromJSON(json: Static<typeof this.jsonSchema>, config: lib.InstanceConfig) {
return new this(
config,
json.status,
json.gamePort,
json.updatedAt,
);
}
toJSON(): Static<typeof InstanceInfo.jsonSchema> {
return {
config: this.config.serialize(),
status: this.status,
gamePort: this.gamePort,
updatedAt: this.updatedAt,
};
}
toInstanceDetails() {
return new lib.InstanceDetails(
this.config.get("instance.name"),
this.id,
this.config.get("instance.assigned_host") ?? undefined,
this.gamePort,
this.status,
this.updatedAt,
);
}
get isDeleted() {
return this.status === "deleted";
}
/** Shorthand for `instance.config.get("instance.id")` */
get id():number {
return this.config.get("instance.id");
+2 -4
View File
@@ -9,7 +9,7 @@ import * as lib from "@clusterio/lib";
const { logger } = lib;
import ControlConnection from "./ControlConnection";
import HostConnection, { HostInfo } from "./HostConnection";
import HostConnection from "./HostConnection";
import WsServerConnector from "./WsServerConnector";
import { version as packageVersion } from "../package.json";
@@ -243,7 +243,7 @@ ${err.stack}`
}
let host = this.controller.hosts.get(data.id);
if (!tokenPayload.iat || tokenPayload.iat < (host?.token_valid_after??0)) {
if (!tokenPayload.iat || tokenPayload.iat < (host?.tokenValidAfter ?? 0)) {
throw new Error("invalid token");
}
} catch (err: any) {
@@ -274,13 +274,11 @@ ${err.stack}`
connector.on("close", () => {
if (this.hostConnections.get(data.id) === connection) {
this.hostConnections.delete(data.id);
this.controller.hostUpdated(this.controller.hosts.get(data.id) as HostInfo);
} else {
logger.warn("Unlisted HostConnection closed");
}
});
this.hostConnections.set(data.id, connection);
this.controller.hostUpdated(this.controller.hosts.get(data.id) as HostInfo);
let src = new lib.Address(lib.Address.host, data.id);
connector.ready(socket, src, sessionToken, undefined);
}
+2 -2
View File
@@ -152,7 +152,7 @@ function validateHostToken(req: Request, res: Response, next: any) {
if (!host) {
throw new Error("invalid host");
}
if (!tokenPayload.iat || tokenPayload.iat < (host.token_valid_after??0)) {
if (!tokenPayload.iat || tokenPayload.iat < (host.tokenValidAfter ?? 0)) {
throw new Error("invalid token");
}
} catch (err) {
@@ -256,7 +256,7 @@ async function uploadExport(req: Request, res: Response) {
modPack.exportManifest = new lib.ExportManifest(assets);
modPack.fillDefaultSettings(settingPrototypes, logger);
res.app.locals.controller.modPackUpdated(modPack);
res.app.locals.controller.modPacksUpdated([modPack]);
res.sendStatus(200);
}
+3 -2
View File
@@ -532,7 +532,8 @@ instanceSaveCommands.add(new lib.Command({
}],
handler: async function(args: { instance: string }, control: Control) {
let instanceId = await lib.resolveInstance(control, args.instance);
let saves = await control.sendTo({ instanceId }, new lib.InstanceListSavesRequest());
let saves = await control.sendTo("controller", new lib.InstanceSaveDetailsListRequest());
saves = saves.filter(save => save.instanceId === instanceId);
print(asTable(saves.map(
({ mtimeMs, ...rest }) => ({ ...rest, mtime: new Date(mtimeMs).toLocaleString() })
)));
@@ -866,7 +867,7 @@ modPackCommands.add(new lib.Command({
}],
handler: async function(args: { modPack: string }, control: Control) {
let mods = new Map(
(await control.send(new lib.ModListRequest())).map(m => [`${m.name}_${m.version}`, m])
(await control.send(new lib.ModListRequest())).map(m => [m.id, m])
);
let modPack = await control.send(
new lib.ModPackGetRequest(await lib.resolveModPack(control, args.modPack))
+16 -6
View File
@@ -347,7 +347,9 @@ export default class Host extends lib.Link {
this.handle(lib.InstanceAssignInternalRequest, this.handleInstanceAssignInternalRequest.bind(this));
this.handle(lib.InstanceUnassignInternalRequest, this.handleInstanceUnassignInternalRequest.bind(this));
this.handle(lib.HostMetricsRequest, this.handleHostMetricsRequest.bind(this));
this.fallbackRequest(lib.InstanceListSavesRequest, this.fallbackInstanceListSavesRequest.bind(this));
this.fallbackRequest(
lib.InstanceSaveDetailsListRequest, this.fallbackInstanceSaveDetailsListRequest.bind(this),
);
this.handle(lib.InstanceRenameSaveRequest, this.handleInstanceRenameSaveRequest.bind(this));
this.handle(lib.InstanceCopySaveRequest, this.handleInstanceCopySaveRequest.bind(this));
this.handle(lib.InstanceTransferSaveRequest, this.handleInstanceTransferSaveRequest.bind(this));
@@ -600,6 +602,9 @@ export default class Host extends lib.Link {
)
);
// Send the new list of saves for this assigned instance to the controller.
await this.sendSaveListUpdate(instanceId, path.join(instanceInfo.path, "saves"));
// save a copy of the instance config
let warnedOutput = {
_warning: "Changes to this file will be overwritten by the controller's copy.",
@@ -700,9 +705,13 @@ export default class Host extends lib.Link {
return { results };
}
async fallbackInstanceListSavesRequest(request: lib.InstanceListSavesRequest, src: lib.Address, dst: lib.Address) {
async fallbackInstanceSaveDetailsListRequest(
request: lib.InstanceSaveDetailsListRequest,
src: lib.Address,
dst: lib.Address
) {
let instanceInfo = this.getRequestInstanceInfo(dst.id);
return await Instance.listSaves(path.join(instanceInfo.path, "saves"), null);
return await Instance.listSaves(dst.id, path.join(instanceInfo.path, "saves"), null);
}
async handleInstanceRenameSaveRequest(request: lib.InstanceRenameSaveRequest) {
@@ -788,12 +797,12 @@ export default class Host extends lib.Link {
let instanceConnection = this.instanceConnections.get(instanceId);
let saveList: lib.SaveDetails[];
if (instanceConnection) {
saveList = await instanceConnection.send(new lib.InstanceListSavesRequest());
saveList = await instanceConnection.send(new lib.InstanceSaveDetailsListRequest());
} else {
saveList = await Instance.listSaves(savesDir, null);
saveList = await Instance.listSaves(instanceId, savesDir, null);
}
this.send(new lib.InstanceSaveListUpdateEvent(instanceId, saveList));
this.send(new lib.InstanceSaveDetailsUpdatesEvent(saveList, instanceId));
}
async handleInstanceDeleteSaveRequest(request: lib.InstanceDeleteSaveRequest) {
@@ -913,6 +922,7 @@ export default class Host extends lib.Link {
list.push(new lib.RawInstanceInfo(
instanceInfo.config.serialize("controller"),
instanceConnection ? instanceConnection.instance.status : "stopped",
instanceConnection ? instanceConnection.instance.server.gamePort : undefined,
));
}
await this.send(new lib.InstancesUpdateRequest(list));
+11 -8
View File
@@ -108,7 +108,7 @@ export default class Instance extends lib.Link {
/**
* Mod pack currently running on this instance
*/
activeModPack = new lib.ModPack();
activeModPack!: lib.ModPack; // This is set in syncMods
/**
* Per player statistics recorded by this instance.
*/
@@ -228,7 +228,7 @@ export default class Instance extends lib.Link {
this.handle(lib.InstanceMetricsRequest, this.handleInstanceMetricsRequest.bind(this));
this.handle(lib.InstanceStartRequest, this.handleInstanceStartRequest.bind(this));
this.handle(lib.InstanceLoadScenarioRequest, this.handleInstanceLoadScenarioRequest.bind(this));
this.handle(lib.InstanceListSavesRequest, this.handleInstanceListSavesRequest.bind(this));
this.handle(lib.InstanceSaveDetailsListRequest, this.handleInstanceSaveDetailsListRequest.bind(this));
this.handle(lib.InstanceCreateSaveRequest, this.handleInstanceCreateSaveRequest.bind(this));
this.handle(lib.InstanceExportDataRequest, this.handleInstanceExportDataRequest.bind(this));
this.handle(lib.InstanceStopRequest, this.handleInstanceStopRequest.bind(this));
@@ -374,7 +374,7 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
}
}
static async listSaves(savesDir: string, loadedSave: string | null) {
static async listSaves(instanceId: number, savesDir: string, loadedSave: string | null) {
let defaultSave = null;
if (loadedSave === null) {
defaultSave = await lib.getNewestFile(
@@ -395,12 +395,15 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
}
list.push(new lib.SaveDetails(
instanceId,
type,
name,
stat.size,
stat.mtimeMs,
name === loadedSave,
name === defaultSave,
0, // Set by controller
false,
));
}
@@ -410,9 +413,9 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
async sendSaveListUpdate() {
this.sendTo(
"controller",
new lib.InstanceSaveListUpdateEvent(
new lib.InstanceSaveDetailsUpdatesEvent(
await Instance.listSaves(this.id, this.path("saves"), this._loadedSave),
this.id,
await Instance.listSaves(this.path("saves"), this._loadedSave),
),
);
}
@@ -446,7 +449,7 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
new lib.InstanceStatusChangedEvent(
this.id,
status,
this.server.gamePort || this.config.get("factorio.game_port") || undefined,
status !== "stopped" ? this.server.gamePort : undefined,
),
);
}
@@ -1034,8 +1037,8 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
}
}
async handleInstanceListSavesRequest() {
return await Instance.listSaves(this.path("saves"), this._loadedSave);
async handleInstanceSaveDetailsListRequest() {
return await Instance.listSaves(this.id, this.path("saves"), this._loadedSave);
}
async handleInstanceCreateSaveRequest(request: lib.InstanceCreateSaveRequest) {
+1
View File
@@ -66,6 +66,7 @@ export default class ModStore extends TypedEventEmitter<keyof ModStoreEvents, Mo
await fs.unlink(path.join(this.modsDirectory, file));
this.files.delete(file);
modInfo.isDeleted = true;
modInfo.updatedAt = Date.now();
this.emit("change", modInfo);
}
+17 -1
View File
@@ -16,6 +16,14 @@ import { integerModVersion, integerFactorioVersion, modVersionRegExp } from "./v
* details of the invidiual fields sourced from info.json.
*/
export default class ModInfo {
/**
* String containing {mod.name}_{mod.version}, uniquely identifies this
* mod.
*/
get id() {
return `${this.name}_${this.version}`;
}
/**
* Internal name of this mod.
* This is the name of the zip file as well as the name that appears in
@@ -113,6 +121,9 @@ export default class ModInfo {
*/
sha1?: string;
/** Millisecond Unix timestamp this entry was last updated at */
updatedAt = 0;
/**
* True if this mod has been deleted
*/
@@ -137,6 +148,7 @@ export default class ModInfo {
...this.infoJsonSchema.properties,
"size": Type.Optional(Type.Integer()),
"sha1": Type.Optional(Type.String()),
"updated_at": Type.Optional(Type.Number()),
"is_deleted": Type.Optional(Type.Boolean()),
});
@@ -159,6 +171,7 @@ export default class ModInfo {
// Additional data
if (json.size) { modInfo.size = json.size; }
if (json.sha1) { modInfo.sha1 = json.sha1; }
if (json.updated_at) { modInfo.updatedAt = json.updated_at; }
if (json.is_deleted) { modInfo.isDeleted = json.is_deleted; }
return modInfo;
@@ -180,6 +193,7 @@ export default class ModInfo {
}
if (this.size) { json.size = this.size; }
if (this.sha1) { json.sha1 = this.sha1; }
if (this.updatedAt) { json.updated_at = this.updatedAt; }
if (this.isDeleted) { json.is_deleted = this.isDeleted; }
return json;
}
@@ -209,11 +223,13 @@ export default class ModInfo {
throw new Error(`Mod's version (${modInfo.version}) is invalid`);
}
const stat = await fs.stat(modPath);
return this.fromJSON({
...modInfo,
size: (await fs.stat(modPath)).size,
size: stat.size,
sha1: await libHash.hashFile(modPath),
updated_at: stat.mtimeMs,
is_deleted: false,
});
}
+10 -1
View File
@@ -58,7 +58,7 @@ export default class ModPack {
/**
* Id of this mod pack
*/
id?: number;
id: number = Math.random() * 2**31 | 0;
/**
* Name of this mod pack.
@@ -101,6 +101,9 @@ export default class ModPack {
*/
exportManifest?: ExportManifest;
/** Millisecond Unix timestamp this entry was last updated at */
updatedAt = 0;
/**
* True if this mod pack has been deleted from the list of mod packs.
*/
@@ -115,6 +118,7 @@ export default class ModPack {
clone.mods = this.mods;
clone.settings = this.settings;
clone.exportManifest = this.exportManifest;
clone.updatedAt = this.updatedAt;
clone.isDeleted = this.isDeleted;
return clone;
}
@@ -131,6 +135,7 @@ export default class ModPack {
"runtime-per-user": ModSettingsJsonSchema,
}),
"export_manifest": Type.Optional(ExportManifest.jsonSchema),
"updated_at": Type.Optional(Type.Number()),
"is_deleted": Type.Optional(Type.Boolean()),
});
@@ -156,6 +161,7 @@ export default class ModPack {
};
}
if (json.export_manifest) { modPack.exportManifest = ExportManifest.fromJSON(json.export_manifest); }
if (json.updated_at) { modPack.updatedAt = json.updated_at; }
if (json.is_deleted) { modPack.isDeleted = json.is_deleted; }
if (!modPack.mods.has("base")) {
@@ -179,6 +185,7 @@ export default class ModPack {
},
};
if (this.exportManifest) { json.export_manifest = this.exportManifest; }
if (this.updatedAt) { json.updated_at = this.updatedAt; }
if (this.isDeleted) { json.is_deleted = this.isDeleted; }
return json;
}
@@ -187,6 +194,8 @@ export default class ModPack {
const json = this.toJSON();
delete json.id;
delete json.export_manifest;
delete json.updated_at;
delete json.is_deleted;
// eslint-disable-next-line node/no-sync
let buf = zlib.deflateSync(JSON.stringify(json));
+13
View File
@@ -23,6 +23,8 @@ export default class User {
public isWhitelisted = false,
/** Reason for being banned. Ignored if isBanned is false. */
public banReason = "",
/** Millisecond Unix timestamp this entry was last updated at */
public updatedAt = 0,
/** True if this user object has been removed from the cluster. */
public isDeleted = false,
/** Combined statistics for the player this user account is tied to. */
@@ -40,6 +42,7 @@ export default class User {
is_banned: Type.Optional(Type.Boolean()),
is_whitelisted: Type.Optional(Type.Boolean()),
ban_reason: Type.Optional(Type.String()),
updated_at: Type.Optional(Type.Number()),
is_deleted: Type.Optional(Type.Boolean()),
instance_stats: Type.Optional(
Type.Array(
@@ -64,6 +67,7 @@ export default class User {
json.is_banned,
json.is_whitelisted,
json.ban_reason,
json.updated_at,
json.is_deleted,
playerStats,
instanceStats,
@@ -99,6 +103,10 @@ export default class User {
json.ban_reason = this.banReason;
}
if (this.updatedAt) {
json.updated_at = this.updatedAt;
}
if (this.isDeleted) {
json.is_deleted = this.isDeleted;
}
@@ -135,6 +143,11 @@ export default class User {
}
return playerStats;
}
/** Name of the user, used by event subscriptions */
get id() {
return this.name;
}
}
export interface IControllerUser extends User {
+12 -9
View File
@@ -11,6 +11,9 @@ export class HostDetails {
public connected: boolean,
public publicAddress?: string,
public tokenValidAfter?: number,
/** Millisecond Unix timestamp this entry was last updated at */
public updatedAt = 0,
public isDeleted = false,
) { }
static jsonSchema = Type.Object({
@@ -21,6 +24,8 @@ export class HostDetails {
"connected": Type.Boolean(),
"publicAddress": Type.Optional(Type.String()),
"tokenValidAfter": Type.Optional(Type.Number()),
"updatedAt": Type.Optional(Type.Number()),
"isDeleted": Type.Optional(Type.Boolean()),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
@@ -32,6 +37,8 @@ export class HostDetails {
json.connected,
json.publicAddress,
json.tokenValidAfter,
json.updatedAt,
json.isDeleted,
);
}
}
@@ -45,27 +52,23 @@ export class HostListRequest {
static Response = jsonArray(HostDetails);
}
export class HostUpdateEvent {
declare ["constructor"]: typeof HostUpdateEvent;
export class HostUpdatesEvent {
declare ["constructor"]: typeof HostUpdatesEvent;
static type = "event" as const;
static src = "controller" as const;
static dst = "control" as const;
static permission = "core.host.subscribe" as const;
constructor(
public update: HostDetails,
public updates: HostDetails[],
) { }
static jsonSchema = Type.Object({
"update": HostDetails.jsonSchema,
"updates": Type.Array(HostDetails.jsonSchema),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(HostDetails.fromJSON(json.update));
}
get subscriptionChannel() {
return this.update.id;
return new this(json.updates.map(update => HostDetails.fromJSON(update)));
}
}
+117 -39
View File
@@ -7,10 +7,34 @@ import type { MessageRequest } from "./messages_core";
import { CollectorResultSerialized } from "../prometheus";
export type InstanceStatus =
"unknown" | "unassigned" | "stopped" | "starting" | "running"
| "stopping" | "creating_save" | "exporting_data" | "deleted"
;
export const InstanceStatus = StringEnum([
"unknown",
"unassigned",
"stopped",
"starting",
"running",
"stopping",
"creating_save",
"exporting_data",
"deleted",
]);
/**
* Current status of the instance. One of:
* - `unknown`: Instance is assigned to a host but this host is currently
* not connected to the contreller.
* - `unassigned`: Instance is not assigned to a a host and exists only on
* the controller.
* - `stopped`: Instance is stopped.
* - `starting`: Instance is in the process of starting up.
* - `running`: Instance is running normally.
* - `stopping`: Instance is in the process of stopping.
* - `creating_save`: Instance is in the process of creating a save.
* - `exporting_data`: Instance is in the process of exporting game data.
* - `deleted`: Instance has been deleted.
*/
export type InstanceStatus = Static<typeof InstanceStatus>;
export class InstanceDetails {
constructor(
public name: string,
@@ -18,6 +42,8 @@ export class InstanceDetails {
public assignedHost: number | undefined,
public gamePort: number | undefined,
public status: InstanceStatus,
/** Millisecond Unix timestamp this entry was last updated at */
public updatedAt = 0,
) { }
static jsonSchema = Type.Object({
@@ -29,10 +55,22 @@ export class InstanceDetails {
"unknown", "unassigned", "stopped", "starting", "running", "stopping",
"creating_save", "exporting_data", "deleted",
]),
"updatedAt": Type.Optional(Type.Number()),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(json.name, json.id, json.assignedHost, json.gamePort, json.status);
return new this(
json.name,
json.id,
json.assignedHost,
json.gamePort,
json.status,
json.updatedAt,
);
}
get isDeleted() {
return this.status === "deleted";
}
}
@@ -67,29 +105,23 @@ export class InstanceDetailsListRequest {
static Response = jsonArray(InstanceDetails);
};
export class InstanceDetailsUpdateEvent {
declare ["constructor"]: typeof InstanceDetailsUpdateEvent;
export class InstanceDetailsUpdatesEvent {
declare ["constructor"]: typeof InstanceDetailsUpdatesEvent;
static type = "event" as const;
static src = "controller" as const;
static dst = "control" as const;
static permission = "core.instance.subscribe" as const;
constructor(
public details: InstanceDetails,
public updates: InstanceDetails[],
) { }
static jsonSchema = InstanceDetails.jsonSchema;
toJSON() {
return this.details;
}
static jsonSchema = Type.Object({
"updates": Type.Array(InstanceDetails.jsonSchema),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(InstanceDetails.fromJSON(json));
}
get subscriptionChannel() {
return this.details.id;
return new this(json.updates.map(update => InstanceDetails.fromJSON(update)));
}
};
@@ -251,60 +283,104 @@ export class InstanceStartRequest {
export class SaveDetails {
constructor(
public type: string,
public instanceId: number,
public type: "file" | "directory" | "special",
public name: string,
public size: number,
public mtimeMs: number,
public loaded: boolean,
public loadByDefault: boolean,
/** Millisecond Unix timestamp this entry was last updated at */
public updatedAt: number,
public isDeleted: boolean,
) { }
static jsonSchema = Type.Object({
"instanceId": Type.Integer(),
"type": StringEnum(["file", "directory", "special"]),
"name": Type.String(),
"size": Type.Integer(),
"mtimeMs": Type.Number(),
"loaded": Type.Boolean(),
"loadByDefault": Type.Boolean(),
"updatedAt": Type.Number(),
"isDeleted": Type.Boolean(),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(json.type, json.name, json.size, json.mtimeMs, json.loaded, json.loadByDefault);
return new this(
json.instanceId,
json.type,
json.name,
json.size,
json.mtimeMs,
json.loaded,
json.loadByDefault,
json.updatedAt,
json.isDeleted,
);
}
/** Concatenation of instance id and name forming a unique string for this save */
get id() {
return `${this.instanceId}/${this.name}`;
}
/**
* Compare this save with another save.
* @param other - save to compare with.
* @returns true if this save is identical to the provided save
*/
equals(other: SaveDetails) {
// Note that updatedAt is not included in the equality check because
// instances send the whole list of saves with updatedAt set to zero
// to the controller whenever any save changes and this equals check
// is used to filter whether a save was updated.
return (
this.instanceId === other.instanceId
&& this.type === other.type
&& this.name === other.name
&& this.size === other.size
&& this.mtimeMs === other.mtimeMs
&& this.loaded === other.loaded
&& this.loadByDefault === other.loadByDefault
&& this.isDeleted === other.isDeleted
);
}
}
export class InstanceListSavesRequest {
declare ["constructor"]: typeof InstanceListSavesRequest;
export class InstanceSaveDetailsListRequest {
declare ["constructor"]: typeof InstanceSaveDetailsListRequest;
static type = "request" as const;
static src = "control" as const;
static dst = "instance" as const;
static src = ["control", "host"] as const;
static dst = ["controller", "instance"] as const;
static permission = "core.instance.save.list" as const;
static Response = jsonArray(SaveDetails);
}
export class InstanceSaveListUpdateEvent {
declare ["constructor"]: typeof InstanceSaveListUpdateEvent;
export class InstanceSaveDetailsUpdatesEvent {
declare ["constructor"]: typeof InstanceSaveDetailsUpdatesEvent;
static type = "event" as const;
static src = ["instance", "host", "controller"] as const;
static dst = ["controller", "control"] as const;
static permission = "core.instance.save.list_subscribe" as const;
static permission = "core.instance.save.list.subscribe" as const;
constructor(
public instanceId: number,
public saves: SaveDetails[],
public updates: SaveDetails[],
/**
* Present if this update was sent by a host and updates contains all
* saves of the given instance.
*/
public instanceId?: number,
) { }
static jsonSchema = Type.Object({
"instanceId": Type.Integer(),
"saves": Type.Array(SaveDetails.jsonSchema),
"updates": Type.Array(SaveDetails.jsonSchema),
"instanceId": Type.Optional(Type.Integer()),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(json.instanceId, json.saves.map(i => SaveDetails.fromJSON(i)));
}
get subscriptionChannel() {
return this.instanceId;
return new this(json.updates.map(i => SaveDetails.fromJSON(i)), json.instanceId);
}
}
@@ -643,7 +719,8 @@ export class RawInstanceInfo {
constructor(
public config: object,
public status: string,
public status: InstanceStatus,
public gamePort: undefined | number,
) { }
static jsonSchema = Type.Object({
@@ -651,10 +728,11 @@ export class RawInstanceInfo {
"status": StringEnum([
"stopped", "starting", "running", "stopping", "creating_save", "exporting_data",
]),
"gamePort": Type.Optional(Type.Number()),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(json.config, json.status);
return new this(json.config, json.status, json.gamePort);
}
}
@@ -747,7 +825,7 @@ export class InstanceStatusChangedEvent {
constructor(
public instanceId: number,
public status: string,
public status: InstanceStatus,
public gamePort?: number,
) { }
+10 -18
View File
@@ -245,50 +245,42 @@ export class ModDeleteRequest {
}
}
export class ModPackUpdateEvent {
declare ["constructor"]: typeof ModPackUpdateEvent;
export class ModPackUpdatesEvent {
declare ["constructor"]: typeof ModPackUpdatesEvent;
static type = "event" as const;
static src = "controller" as const;
static dst = "control" as const;
static permission = "core.mod_pack.subscribe" as const;
constructor(
public modPack: ModPack,
public updates: ModPack[],
) { }
static jsonSchema = Type.Object({
"modPack": ModPack.jsonSchema,
"updates": Type.Array(ModPack.jsonSchema),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(ModPack.fromJSON(json.modPack));
}
get subscriptionChannel() {
return this.modPack.id;
return new this(json.updates.map(update => ModPack.fromJSON(update)));
}
}
export class ModUpdateEvent {
declare ["constructor"]: typeof ModUpdateEvent;
export class ModUpdatesEvent {
declare ["constructor"]: typeof ModUpdatesEvent;
static type = "event" as const;
static src = "controller" as const;
static dst = "control" as const;
static permission = "core.mod.subscribe" as const;
constructor(
public mod: ModInfo,
public updates: ModInfo[],
) { }
static jsonSchema = Type.Object({
"mod": ModInfo.jsonSchema,
"updates": Type.Array(ModInfo.jsonSchema),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(ModInfo.fromJSON(json.mod));
}
get subscriptionChannel() {
return this.mod.name;
return new this(json.updates.map(update => ModInfo.fromJSON(update)));
}
}
+5 -9
View File
@@ -301,26 +301,22 @@ export class UserDeleteRequest {
}
}
export class UserUpdateEvent {
declare ["constructor"]: typeof UserUpdateEvent;
export class UserUpdatesEvent {
declare ["constructor"]: typeof UserUpdatesEvent;
static type = "event" as const;
static src = "controller" as const;
static dst = "control" as const;
static permission = "core.user.subscribe" as const;
constructor(
public user: User,
public updates: User[],
) { }
static jsonSchema = Type.Object({
"user": User.jsonSchema,
"updates": Type.Array(User.jsonSchema),
});
static fromJSON(json: Static<typeof this.jsonSchema>) {
return new this(User.fromJSON(json.user));
}
get subscriptionChannel() {
return this.user.name;
return new this(json.updates.map(update => User.fromJSON(update)));
}
}
+7 -7
View File
@@ -31,7 +31,7 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
subscriptions.SubscriptionRequest,
host.HostListRequest,
host.HostUpdateEvent,
host.HostUpdatesEvent,
host.HostMetricsRequest,
host.ControllerConnectionEvent,
host.PrepareControllerDisconnectRequest,
@@ -40,7 +40,7 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
instance.InstanceDetailsGetRequest,
instance.InstanceDetailsListRequest,
instance.InstanceDetailsUpdateEvent,
instance.InstanceDetailsUpdatesEvent,
instance.InstanceCreateRequest,
instance.InstanceConfigGetRequest,
instance.InstanceConfigSetFieldRequest,
@@ -48,8 +48,8 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
instance.InstanceAssignRequest,
instance.InstanceMetricsRequest,
instance.InstanceStartRequest,
instance.InstanceListSavesRequest,
instance.InstanceSaveListUpdateEvent,
instance.InstanceSaveDetailsListRequest,
instance.InstanceSaveDetailsUpdatesEvent,
instance.InstanceCreateSaveRequest,
instance.InstanceRenameSaveRequest,
instance.InstanceCopySaveRequest,
@@ -88,8 +88,8 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
mod.ModSearchRequest,
mod.ModDownloadRequest,
mod.ModDeleteRequest,
mod.ModPackUpdateEvent,
mod.ModUpdateEvent,
mod.ModPackUpdatesEvent,
mod.ModUpdatesEvent,
user.PermissionListRequest,
user.RoleListRequest,
@@ -106,5 +106,5 @@ export const dataClasses: (RequestClass<unknown, unknown> | EventClass<unknown>)
user.UserSetWhitelistedRequest,
user.UserSetBannedRequest,
user.UserDeleteRequest,
user.UserUpdateEvent,
user.UserUpdatesEvent,
];
+1 -1
View File
@@ -201,7 +201,7 @@ definePermission({
description: "List the saves currently on the instance.",
});
definePermission({
name: "core.instance.save.list_subscribe",
name: "core.instance.save.list.subscribe",
title: "Subscribe to save list updatess",
description: "Subscribe to be notifed on updates to the list of saves of instances.",
});
+149 -226
View File
@@ -1,73 +1,21 @@
import { Type, Static } from "@sinclair/typebox";
import { Link, Event, EventClass, RequestHandler, WebSocketClientConnector, WebSocketBaseConnector } from "./link";
import { logger } from "./logging";
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
* 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();
}
}
}
export type EventSubscriberCallback<T> = (updates: T[], synced: boolean) => void
/**
* 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;
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>;
@@ -84,8 +32,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 +43,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 +57,7 @@ export class SubscriptionRequest {
type EventData = {
subscriptionUpdate?: SubscriptionRequestHandler<unknown>,
subscriptions: Map<Link, { all: boolean, channels: Array<string | number>, onceClose: () => void }>,
subscriptions: Set<Link>,
};
/**
@@ -121,18 +67,12 @@ type EventData = {
export class SubscriptionController {
_events = new Map<string, EventData>();
constructor(
private controller: any, // Controller | Link
) {
this.controller.handle(SubscriptionRequest, this._handleRequest.bind(this));
}
/**
* Allow clients to subscribe to an event by telling the subscription controller to accept them
* Has an optional subscription update handler which is called when any client updates their subscription
* Has an optional subscription update handler which is called when a client subscribes
* @param Event - Event class which is sent out as updates.
* @param subscriptionUpdate -
* Optional handler called when a client updates its subscriptions.
* Optional handler called when a client subscribes.
*/
handle<T>(Event: EventClass<T>, subscriptionUpdate?: SubscriptionRequestHandler<T>) {
const entry = Link._eventsByClass.get(Event);
@@ -144,15 +84,15 @@ export class SubscriptionController {
}
this._events.set(entry.name, {
subscriptionUpdate: subscriptionUpdate,
subscriptions: new Map(),
subscriptions: new Set(),
});
}
/**
* 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,25 +101,34 @@ 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) {
for (let link of eventData.subscriptions) {
if ((link.connector as WebSocketBaseConnector).closing) {
eventData.subscriptions.delete(link);
} else if ((subscription.all || (channel && subscription.channels.includes(channel)))) {
} else {
link.send(event);
}
}
}
/**
* Handle incoming event subscription requests
* Unsubscribe from all events of a given link.
* Used when a link is closed to stop all active subscriptions.
* @param link - Link to stop sending events to.
*/
unsubscribe(link: Link) {
for (let eventData of this._events.values()) {
eventData.subscriptions.delete(link);
}
}
/**
* Handle incoming subscription requests on a link
* @param link - Link message was received on
* @param event - incomming event.
* @param src - Source address of incomming request.
* @param dst - destination address of incomming request.
* @returns Response to subscription request.
* @internal
*/
async _handleRequest(event: SubscriptionRequest, src: Address, dst: Address) {
async handleRequest(link: Link, event: SubscriptionRequest, src: Address, dst: Address) {
if (!Link._eventsByName.has(event.eventName)) {
throw new Error(`Event ${event.eventName} is not a registered event`);
}
@@ -187,184 +136,158 @@ export class SubscriptionController {
if (!eventData) {
throw new Error(`Event ${event.eventName} is not a registered as subscribable`);
}
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) {
let onceClose = eventData.subscriptions.get(link)?.onceClose;
if (onceClose) {
link.connector.off("close", onceClose);
eventData.subscriptions.delete(link);
}
if (event.subscribe === false) {
eventData.subscriptions.delete(link);
} else {
let onceClose = eventData.subscriptions.get(link)?.onceClose;
if (!onceClose) {
onceClose = () => eventData.subscriptions.delete(link);
link.connector.once("close", onceClose);
}
eventData.subscriptions.set(
link, { all: event.allChannels, channels: event.channels, onceClose: onceClose },
);
}
return new SubscriptionResponse(eventReplay);
}
}
/**
* 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
*/
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 prehandler?: (e: T) => V,
control?: Link
) {
const entry = Link._eventsByClass.get(this.event);
if (!entry) {
throw new Error(`Unregistered Event class ${this.event.name}`);
}
if (control) {
this.connectControl(control);
}
}
/**
* Handle incoming events and distribute it to the correct callbacks
* @param response - event from subscribed resource
* @internal
*/
async _handle(response: Event<T> | ChannelEvent<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);
eventData.subscriptions.add(link);
if (eventData.subscriptionUpdate) {
const eventReplay = await eventData.subscriptionUpdate(event, src, dst);
if (eventReplay) {
link.send(eventReplay);
}
}
}
}
}
export interface SubscribableValue {
id: number | string,
updatedAt: number,
isDeleted: boolean,
}
export interface EventSubscribable<T, V extends SubscribableValue> extends Event<T> {
updates: V[],
}
/**
* Component for subscribing to and tracking updates of a remote resource
* Multiple handlers can be subscribed at the same time
*/
export class EventSubscriber<
T extends EventSubscribable<T, V>,
K extends string | number = T["updates"][number]["id"],
V extends SubscribableValue = T["updates"][number],
> {
_callbacks = new Array<EventSubscriberCallback<V>>();
/** Values of the subscribed resource */
values = new Map<K, V>();
/** True if this subscriber is currently synced with the source */
synced = false;
_snapshot?: readonly [ReadonlyMap<K, Readonly<V>>, boolean];
_snapshotLastUpdated = 0;
lastResponseTime = -1;
constructor(
private Event: EventClass<T>,
public control: Link,
) {
control.handle(Event, this._handle.bind(this));
control.connector.on("connect", () => {
this._updateSubscription();
});
control.connector.on("close", () => {
if (this.synced) {
this.synced = false;
for (let callback of this._callbacks) {
callback([], this.synced);
}
}
});
}
/**
* If a control link was not connected at creation, it can be connected
* here (normally during onControllerConnectionEvent)
* @param control - Control link to associate with.
* Handle incoming events and distribute it to the correct callbacks
* @param event - event from subscribed resource
* @internal
*/
async connectControl(control: Link) {
if (this.control === control) {
return;
async _handle(event: EventSubscribable<T, V>) {
for (const value of event.updates) {
this.lastResponseTime = Math.max(this.lastResponseTime, value.updatedAt);
if (value.isDeleted) {
this.values.delete(value.id as K);
} else {
this.values.set(value.id as K, value);
}
}
for (let callback of this._callbacks) {
callback(event.updates, this.synced);
}
this.control = control;
this.control.handle(this.event, this._handle.bind(this));
await this._updateSubscription();
}
/**
* Subscribe to receive all event notifications
* @param handler -
* callback invoked whenever the subscribed resource changes.
* callback invoked whenever the subscribed resource changes or the
* synced property changes, in which case the updates will be empty.
* @returns function that will unsubscribe from notifications
*/
async subscribe(handler: EventSubscriberCallback<V>) {
subscribe(handler: EventSubscriberCallback<V>) {
this._callbacks.push(handler);
await this._updateSubscription();
if (this._callbacks.length === 1) {
this._updateSubscription();
}
return () => {
// During a page transition the components currently rendered
// are unmounted and then the components for the new page is
// mounted. This means that if a resource is used by both pages
// it is first unsubscribed by the unmounted component causing
// the callbacks count to go to zero and a subscription update
// to be sent, and then subscribed by the mounted component
// causing another subscription update to be sent.
// By delaying the unsubscription here the subscription happens
// before the unsubscription, thus preventing the redundant
// updates from being sent out.
setImmediate(() => {
let index = this._callbacks.lastIndexOf(handler);
if (index === -1) {
return;
}
this._callbacks.splice(index, 1);
if (this._callbacks.length === 0) {
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.
* Obtain a snapshot of the current state of the tracked resource
* @returns tuple of values map snapshot and synced property.
*/
async subscribeToChannel(channel: string | number, handler: EventSubscriberCallback<V>) {
if (!this._channelCallbacks.get(channel)) {
this._channelCallbacks.set(channel, []);
getSnapshot() {
if (this._snapshotLastUpdated !== this.lastResponseTime) {
this._snapshotLastUpdated = this.lastResponseTime;
this._snapshot = [new Map(this.values), this.synced];
}
this._channelCallbacks.get(channel)!.push(handler);
await this._updateSubscription();
}
/**
* Unsubscribe from receiving all event notifications, does not effect
* the subscriptions of other handlers.
*
* @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
*/
async unsubscribe(handler: EventSubscriberCallback<V>) {
let index = this._callbacks.lastIndexOf(handler);
if (index === -1) {
throw new Error("handler is not registered");
}
this._callbacks.splice(index, 1);
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();
return this._snapshot!;
}
/**
* Update the subscription with the controller based on current handler counts
*/
async _updateSubscription() {
if (!this.control || !(this.control.connector as WebSocketClientConnector).connected) {
if (!(this.control.connector as WebSocketClientConnector).connected) {
return;
}
const entry = Link._eventsByClass.get(this.event)!;
const entry = Link._eventsByClass.get(this.Event)!;
const response = await this.control.send(new SubscriptionRequest(
entry.name,
this._callbacks.length > 0,
[...this._channelCallbacks.keys()],
this.lastResponseTime
));
if (response.eventReplay) {
await this._handle(response.eventReplay as any);
try {
await this.control.send(new SubscriptionRequest(
entry.name,
this._callbacks.length > 0,
this.lastResponseTime
));
this.synced = this._callbacks.length > 0;
this._snapshotLastUpdated = 0;
for (let callback of this._callbacks) {
callback([], this.synced);
}
} catch (err: any) {
logger.error(
`Unexpected error updating ${entry.name} subscription:\n${err.stack}`
);
}
}
}
@@ -6,7 +6,7 @@ import * as lib from "@clusterio/lib";
import ControlContext from "./ControlContext";
import { notifyErrorHandler } from "../util/notify";
import { useHostList } from "../model/host";
import { useHosts } from "../model/host";
const { Paragraph } = Typography;
@@ -20,7 +20,7 @@ type AssignInstanceModalProps = {
};
export default function AssignInstanceModal(props: AssignInstanceModalProps) {
let [open, setOpen] = useState(false);
let [hostList] = useHostList();
let [hosts] = useHosts();
let [applying, setApplying] = useState(false);
let [form] = Form.useForm();
let control = useContext(ControlContext);
@@ -80,7 +80,7 @@ export default function AssignInstanceModal(props: AssignInstanceModalProps) {
<Select.Option value={"null"}>
<Typography.Text italic>Unassigned</Typography.Text>
</Select.Option>
{hostList.map((host) => <Select.Option
{[...hosts.values()].map((host) => <Select.Option
key={host["id"]}
value={host["id"]}
disabled={!host["connected"]}
+11 -13
View File
@@ -4,13 +4,12 @@ import { useParams } from "react-router-dom";
import * as lib from "@clusterio/lib";
import notify, { notifyErrorHandler } from "../util/notify";
import type { Control } from "../util/websocket";
import ControlContext from "./ControlContext";
import InstanceList from "./InstanceList";
import LogConsole from "./LogConsole";
import { useAccount } from "../model/account";
import { useInstanceList } from "../model/instance";
import { HostState, useHost } from "../model/host";
import { useInstances } from "../model/instance";
import { useHost } from "../model/host";
import PageHeader from "./PageHeader";
import PageLayout from "./PageLayout";
import PluginExtra from "./PluginExtra";
@@ -23,17 +22,16 @@ export default function HostViewPage() {
let hostId = Number(params.id);
let control = useContext(ControlContext);
let account = useAccount();
let [instanceList] = useInstanceList();
let [host] = useHost(hostId);
let [instances] = useInstances();
const [host, synced] = useHost(hostId);
const hostInstances = new Map([...instances].filter(([_id, instance]) => instance.assignedHost === hostId));
instanceList = instanceList.filter(instance => instance.assignedHost === hostId);
let nav = [{ name: "Hosts", path: "/hosts" }, { name: host?.name ?? String(hostId) }];
if (!host) {
if (!synced) {
return <PageLayout nav={nav}><Spin size="large" /></PageLayout>;
}
let nav = [{ name: "Hosts", path: "/hosts" }, { name: host.name || String(hostId) }];
if (host.loading) {
return <PageLayout nav={nav}><Spin size="large" /></PageLayout>;
}
if (host.missing) {
return <PageLayout nav={nav}>
<PageHeader title={String(hostId)} />
<p>Host with id {hostId} was not found on the controller.</p>
@@ -83,7 +81,7 @@ export default function HostViewPage() {
</Descriptions>
{account.hasPermission("core.instance.list") && <>
<Title level={5} style={{ marginTop: 16 }}>Instances</Title>
<InstanceList instances={instanceList} size="small" hideAssignedHost />
<InstanceList instances={hostInstances} size="small" hideAssignedHost />
</>}
{account.hasPermission("core.log.follow") && <>
<Title level={5} style={{ marginTop: 16 }}>Console</Title>
+3 -3
View File
@@ -10,7 +10,7 @@ import ControlContext from "./ControlContext";
import PageHeader from "./PageHeader";
import PageLayout from "./PageLayout";
import PluginExtra from "./PluginExtra";
import { useHostList } from "../model/host";
import { useHosts } from "../model/host";
import notify, { notifyErrorHandler } from "../util/notify";
const strcmp = new Intl.Collator(undefined, { numeric: true, sensitivity: "base" }).compare;
@@ -184,7 +184,7 @@ function CopyButton({ text, message }: { text:string, message:string }) {
export default function HostsPage() {
let account = useAccount();
let navigate = useNavigate();
let [hostList] = useHostList();
let [hosts] = useHosts();
return <PageLayout nav={[{ name: "Hosts" }]}>
<PageHeader
@@ -226,7 +226,7 @@ export default function HostsPage() {
sorter: (a, b) => Number(a.connected) - Number(b.connected),
},
]}
dataSource={hostList}
dataSource={[...hosts.values()]}
rowKey={host => host.id}
pagination={false}
onRow={(record, rowIndex) => ({
+17 -18
View File
@@ -6,7 +6,7 @@ import type { SizeType } from "antd/es/config-provider/SizeContext";
import type { ColumnsType } from "antd/es/table";
import { useAccount } from "../model/account";
import { useHostList } from "../model/host";
import { useHosts } from "../model/host";
import InstanceStatusTag from "./InstanceStatusTag";
import StartStopInstanceButton from "./StartStopInstanceButton";
import { InstanceDetails } from "@clusterio/lib";
@@ -14,7 +14,7 @@ import { InstanceDetails } from "@clusterio/lib";
const strcmp = new Intl.Collator(undefined, { numeric: true, sensitivity: "base" }).compare;
type InstanceListProps = {
instances: InstanceDetails[];
instances: ReadonlyMap<number, Readonly<InstanceDetails>>;
size?: SizeType;
hideAssignedHost?: boolean;
};
@@ -22,28 +22,27 @@ type InstanceListProps = {
export default function InstanceList(props: InstanceListProps) {
let account = useAccount();
let navigate = useNavigate();
let [hostList] = useHostList();
let [hosts] = useHosts();
function hostName(hostId?: number|null) {
if (hostId === null ||hostId === undefined) {
function hostName(hostId?: number) {
if (hostId === undefined) {
return "";
}
let host = hostList.find(s => s.id === hostId);
if (host) {
return host.name;
}
return String(hostId);
return hosts.get(hostId)?.name ?? String(hostId);
}
function instancePublicAddress(instance: InstanceDetails) {
let host = hostList.find(s => s.id === instance.assignedHost);
if (host && host.publicAddress) {
if (instance.gamePort) {
return `${host.publicAddress}:${instance.gamePort}`;
}
if (instance.assignedHost === undefined) {
return "";
}
let host = hosts.get(instance.assignedHost);
if (!host || !host.publicAddress) {
return "";
}
if (instance.gamePort === undefined) {
return host.publicAddress;
}
return null;
return `${host.publicAddress}:${instance.gamePort}`;
}
let columns: ColumnsType<InstanceDetails> = [
@@ -78,7 +77,7 @@ export default function InstanceList(props: InstanceListProps) {
/>
</> : "";
},
sorter: (a, b) => strcmp(instancePublicAddress(a)||"", instancePublicAddress(b)||""),
sorter: (a, b) => strcmp(instancePublicAddress(a), instancePublicAddress(b)),
responsive: ["lg"],
},
{
@@ -109,7 +108,7 @@ export default function InstanceList(props: InstanceListProps) {
return <Table
size={props.size || "large"}
columns={columns}
dataSource={props.instances}
dataSource={[...props.instances.values()]}
rowKey={instance => instance["id"]}
pagination={false}
onRow={(record, rowIndex) => ({
@@ -20,15 +20,15 @@ import StartStopInstanceButton from "./StartStopInstanceButton";
import LoadScenarioModal from "./LoadScenarioModal";
import SavesList from "./SavesList";
import { notifyErrorHandler } from "../util/notify";
import { InstanceState, useInstance } from "../model/instance";
import { HostState, useHost } from "../model/host";
import { useInstance } from "../model/instance";
import { useHost } from "../model/host";
import InstanceStatusTag from "./InstanceStatusTag";
const { Title } = Typography;
type InstanceDescriptionProps = {
host: HostState;
instance: InstanceState;
host?: Readonly<lib.HostDetails>;
instance: Readonly<lib.InstanceDetails>;
};
function InstanceDescription(props: InstanceDescriptionProps) {
let account = useAccount();
@@ -42,7 +42,7 @@ function InstanceDescription(props: InstanceDescriptionProps) {
<Descriptions.Item label="Host">
{!assigned
? <em>Unassigned</em>
: host.name || instance.assignedHost
: host?.name ?? instance.assignedHost
}
{account.hasPermission("core.instance.assign") && <AssignInstanceModal
id={instance.id}
@@ -60,7 +60,7 @@ function InstanceDescription(props: InstanceDescriptionProps) {
</Descriptions>;
}
function InstanceButtons(props: { instance: InstanceState }) {
function InstanceButtons(props: { instance: lib.InstanceDetails }) {
let account = useAccount();
let control = useContext(ControlContext);
let navigate = useNavigate();
@@ -169,18 +169,18 @@ export default function InstanceViewPage() {
let navigate = useNavigate();
let account = useAccount();
let [instance] = useInstance(instanceId);
let [host] = useHost(Number(instance.assignedHost));
let [instance, synced] = useInstance(instanceId);
let [host] = useHost(instance?.assignedHost);
let nav = [{ name: "Instances", path: "/instances" }, { name: instance.name || "Unknown" }];
if (instance.loading) {
return <PageLayout nav={nav}><Spin size="large" /></PageLayout>;
}
let nav = [{ name: "Instances", path: "/instances" }, { name: instance?.name ?? String(instanceId) }];
if (!instance) {
if (!synced) {
return <PageLayout nav={nav}><Spin size="large" /></PageLayout>;
}
if (instance.missing || instance.status === "deleted") {
return <PageLayout nav={nav}>
<Alert
message={instance.status === "deleted" ? "Instance has been deleted" : "Instance not found" }
message={"Instance not found" }
showIcon
description={<>Instance with id {instanceId} was not found on the controller.</>}
type="warning"
@@ -9,7 +9,7 @@ import ControlContext from "./ControlContext";
import PageHeader from "./PageHeader";
import PageLayout from "./PageLayout";
import PluginExtra from "./PluginExtra";
import { useInstanceList } from "../model/instance";
import { useInstances } from "../model/instance";
import InstanceList from "./InstanceList";
import { notifyErrorHandler } from "../util/notify";
@@ -62,7 +62,7 @@ function CreateInstanceButton() {
export default function InstancesPage() {
let control = useContext(ControlContext);
let account = useAccount();
let [instanceList] = useInstanceList();
let [instances] = useInstances();
return <PageLayout nav={[{ name: "Instances" }]}>
<PageHeader
@@ -70,7 +70,7 @@ export default function InstancesPage() {
extra={<>
{account.hasPermission("core.instance.create") && <CreateInstanceButton />}
{account.hasPermission("core.instance.start")
&& <Button onClick={e => instanceList.forEach(instance => {
&& <Button onClick={e => instances.forEach(instance => {
if (instance.status === "stopped") {
control.sendTo(
{ instanceId: instance.id },
@@ -82,7 +82,7 @@ export default function InstancesPage() {
Start all
</Button>}
{account.hasPermission("core.instance.stop")
&& <Button onClick={e => instanceList.forEach(instance => {
&& <Button onClick={e => instances.forEach(instance => {
if (["starting", "running"].includes(instance.status)) {
control.sendTo(
{ instanceId: instance.id },
@@ -96,7 +96,7 @@ export default function InstancesPage() {
</>}
/>
<InstanceList instances={instanceList} />
<InstanceList instances={instances} />
<PluginExtra component="InstancesPage" />
</PageLayout>;
}
@@ -5,10 +5,9 @@ import * as lib from "@clusterio/lib";
import ControlContext from "./ControlContext";
import { notifyErrorHandler } from "../util/notify";
import { InstanceState } from "../model/instance";
type LoadScenarioModalProps = {
instance: InstanceState;
instance: lib.InstanceDetails;
};
export default function LoadScenarioModal(props: LoadScenarioModalProps) {
let [open, setOpen] = useState(false);
@@ -55,7 +54,7 @@ export default function LoadScenarioModal(props: LoadScenarioModalProps) {
setLoadingScenario(true);
control.sendTo(
{ instanceId: props.instance.id! },
{ instanceId: props.instance.id },
new lib.InstanceLoadScenarioRequest(scenario, seed, mapGenSettings, mapSettings),
).then(() => {
setOpen(false);
@@ -19,7 +19,7 @@ import * as lib from "@clusterio/lib";
import ControlContext from "./ControlContext";
import { useAccount } from "../model/account";
import { useModPack } from "../model/mod_pack";
import { useModList } from "../model/mods";
import { useMods } from "../model/mods";
import { useClipboard } from "../util/clipboard";
import notify, { notifyErrorHandler } from "../util/notify";
import PageHeader from "./PageHeader";
@@ -245,7 +245,7 @@ type ModsTableProps = {
};
function ModsTable(props: ModsTableProps) {
const [showAddMods, setShowAddMods] = useState(false);
const [modList] = useModList();
const [modsMap] = useMods();
const deletedMods: Map<string, lib.ModRecord|lib.ModInfo> = new Map();
const changedMods: Map<string, lib.ModRecord|lib.ModInfo> = new Map();
@@ -261,10 +261,9 @@ function ModsTable(props: ModsTableProps) {
}
}
const modListMap = new Map(modList.map(mod => [`${mod.name}_${mod.version}`, mod]));
const mods = [...props.modPack.mods.values(), ...deletedMods.values()].map(
(mod: lib.ModRecord|lib.ModInfo): lib.ModInfo|lib.ModRecord => {
const candidate = modListMap.get(`${mod.name}_${mod.version}`);
const candidate = modsMap.get(`${mod.name}_${mod.version}`);
if (!candidate) {
return {
...mod,
@@ -478,11 +477,11 @@ type SettingsTableProps = {
onRevert: (change: ModChange) => void;
};
function SettingsTable(props: SettingsTableProps) {
const [modList] = useModList();
const [mods] = useMods();
const modsInPack = new Set([...props.modPack.mods.values()].map(mod => `${mod.name}_${mod.version}`));
const modTitles = new Map(modList
.filter(mod => modsInPack.has(`${mod.name}_${mod.version}`))
.map(mod => [`${mod.name}`, mod.title])
const modTitles = new Map([...mods.values()]
.filter(mod => modsInPack.has(mod.id))
.map(mod => [mod.name, mod.title])
);
const types = ["bool-setting", "int-setting", "double-setting", "string-setting"];
@@ -609,7 +608,7 @@ function ExportButton(props: { modPack: lib.ModPack }) {
</>;
}
function useExportedAsset(modPack: lib.ModPack, asset: "settings"|"locale"): Map<any, any> | any {
function useExportedAsset(modPack: lib.ModPack | undefined, asset: "settings"|"locale"): Map<any, any> | any {
let [assetData, setAssetData] = useState<Map<any, any>|any>(asset === "locale" ? new Map() : {});
let assetFilename: string | undefined;
if (modPack instanceof lib.ModPack && modPack.exportManifest?.assets[asset]) {
@@ -689,20 +688,13 @@ export default function ModPackViewPage() {
let modPackId = Number(params.id);
const [form] = Form.useForm();
let [modPack] = useModPack(modPackId) as lib.ModPack[];
const [modPack, synced] = useModPack(modPackId);
let prototypes: any = useExportedAsset(modPack, "settings");
let locale: Map<any, any> = useExportedAsset(modPack, "locale");
let [changes, setChanges] = useState<ModChange[]>([]);
let syncTimeout = useRef<ReturnType<typeof setTimeout>|undefined>();
let modifiedModPack: any;
if (!(modPack instanceof lib.ModPack)) {
// Loading or invalid id
modifiedModPack = modPack;
} else {
modifiedModPack = applyModPackChanges(modPack, changes);
}
const modifiedModPack = modPack ? applyModPackChanges(modPack, changes) : undefined;
useEffect(() => () => { clearTimeout(syncTimeout.current); });
@@ -744,34 +736,35 @@ export default function ModPackViewPage() {
continue;
}
if (["name", "description", "factorioVersion"].includes(field.name[0])) {
if (modifiedModPack[field.name[0]] !== field.value) {
pushChange({ type: field.name[0], value: field.value });
const name = field.name[0] as "name" | "description" | "factorioVersion";
if (modifiedModPack![name] !== field.value) {
pushChange({ type: name, value: field.value });
}
} else if (["startup", "runtime-global", "runtime-per-user"].includes(field.name[0])) {
let [scope, settingName]: ["startup"|"runtime-global"|"runtime-per-user", string] = field.name;
let value = field.value;
if (typeof modPack.settings[scope].get(settingName)?.value === "number") {
if (typeof modPack!.settings[scope].get(settingName)?.value === "number") {
value = Number(value);
if (Number.isNaN(value)) {
continue;
}
}
if (modifiedModPack.settings[scope].get(settingName).value !== value) {
if (modifiedModPack!.settings[scope].get(settingName)!.value !== value) {
pushChange({ type: "settings.set", scope, name: settingName, value: { value }});
}
}
}
}
let nav = [{ name: "Mods", path: "/mods" }, { name: "Mod Packs" }, { name: modPack.name || String(modPackId) }];
if (modifiedModPack.loading) {
return <PageLayout nav={nav}>
<PageHeader title={String(modPackId)} />
<Spin size="large" />
</PageLayout>;
}
let nav = [{ name: "Mods", path: "/mods" }, { name: "Mod Packs" }, { name: modPack?.name ?? String(modPackId) }];
if (!modifiedModPack || !modPack) {
if (!synced) {
return <PageLayout nav={nav}>
<PageHeader title={String(modPackId)} />
<Spin size="large" />
</PageLayout>;
}
if (modifiedModPack.missing) {
return <PageLayout nav={nav}>
<PageHeader title="Mod Pack Not Found" />
<p>Mod pack with id {modPackId} was not found on the controller.</p>
+7 -7
View File
@@ -8,8 +8,8 @@ import UploadOutlined from "@ant-design/icons/UploadOutlined";
import * as lib from "@clusterio/lib";
import { useAccount } from "../model/account";
import { useModList } from "../model/mods";
import { useModPackList } from "../model/mod_pack";
import { useMods } from "../model/mods";
import { useModPacks } from "../model/mod_pack";
import { notifyErrorHandler } from "../util/notify";
import ControlContext from "./ControlContext";
import PageHeader from "./PageHeader";
@@ -129,8 +129,8 @@ export default function ModsPage() {
let account = useAccount();
let control = useContext(ControlContext);
let navigate = useNavigate();
let [modList] = useModList();
let [modPackList] = useModPackList();
let [mods] = useMods();
let [modPacks] = useModPacks();
function actions(mod: lib.ModInfo) {
return <Space>
@@ -209,7 +209,7 @@ export default function ModsPage() {
render: modPack => modPack.mods.size,
},
]}
dataSource={modPackList}
dataSource={[...modPacks.values()]}
pagination={false}
rowKey={modPack => Number(modPack.id)}
onRow={(modPack, rowIndex) => ({
@@ -263,9 +263,9 @@ export default function ModsPage() {
expandedRowRender: (mod: lib.ModInfo) => <ModDetails mod={mod} actions={actions} />,
expandedRowClassName: () => "no-expanded-padding",
}}
dataSource={modList}
dataSource={[...mods.values()]}
pagination={false}
rowKey={mod => `${mod.name}_${mod.version}`}
rowKey={mod => mod.id}
/>
<PluginExtra component="ModsPage" />
</PageLayout>;
@@ -4,18 +4,14 @@ import * as lib from "@clusterio/lib";
import ControlContext from "./ControlContext";
import BaseWebPlugin from "../BaseWebPlugin";
import { RawUserState } from "../model/user";
import { HostState } from "../model/host";
import { InstanceState } from "../model/instance";
import { RawRoleState } from "./RoleViewPage";
type PluginExtraProps = {
component: string;
instance?: InstanceState;
user?: RawUserState;
host?: HostState;
role?: RawRoleState;
instance?: lib.InstanceDetails;
user?: lib.User;
host?: lib.HostDetails;
role?: lib.Role;
};
export default function PluginExtra(props: PluginExtraProps) {
let control = useContext(ControlContext);
+18 -26
View File
@@ -13,27 +13,19 @@ import PluginExtra from "./PluginExtra";
import { notifyErrorHandler } from "../util/notify";
export type RawRoleState = Partial<lib.Role> & {
loading?: boolean;
missing?: boolean;
present?: boolean;
};
function useRole(id: number): [RawRoleState, () => void] {
function useRole(id: number): [Readonly<lib.Role> | undefined, boolean, () => void] {
let control = useContext(ControlContext);
let [role, setRole] = useState<RawRoleState>({ loading: true });
let [role, setRole] = useState<lib.Role>();
let [synced, setSynced] = useState(false);
function updateRole() {
// XXX optimize by requesting only the role in question
control.send(new lib.RoleListRequest()).then(roles => {
let match = roles.find(u => u.id === id);
if (!match) {
setRole({ missing: true });
} else {
setRole({
...match,
present: true,
});
if (match) {
setRole(match);
}
setSynced(true);
});
}
@@ -41,7 +33,7 @@ function useRole(id: number): [RawRoleState, () => void] {
updateRole();
}, [id]);
return [role, updateRole];
return [role, synced, updateRole];
}
export default function RoleViewPage() {
@@ -52,19 +44,19 @@ export default function RoleViewPage() {
let account = useAccount();
let control = useContext(ControlContext);
let [role, updateRole] = useRole(roleId);
const [role, synced, updateRole] = useRole(roleId);
let [edited, setEdited] = useState(false);
let nav = [{ name: "Roles", path: "/roles" }, { name: role.name || String(roleId) }];
if (role.loading) {
return <PageLayout nav={nav}>
<PageHeader title={String(roleId)} />
<Spin size="large" />
</PageLayout>;
}
let nav = [{ name: "Roles", path: "/roles" }, { name: role?.name ?? String(roleId) }];
if (!role) {
if (!synced) {
return <PageLayout nav={nav}>
<PageHeader title={String(roleId)} />
<Spin size="large" />
</PageLayout>;
}
if (role.missing) {
return <PageLayout nav={nav}>
<PageHeader title="Role not found" />
<p>Role with id {roleId} was not found on the controller.</p>
@@ -76,7 +68,7 @@ export default function RoleViewPage() {
description: role["description"],
permissions: {
...Object.fromEntries([...lib.permissions.values()].map(perm => [
perm.name, [...role.permissions!].includes(perm.name),
perm.name, [...role.permissions].includes(perm.name),
])),
},
};
@@ -101,7 +93,7 @@ export default function RoleViewPage() {
}}
>
<PageHeader
title={role.name!}
title={role.name}
extra={<>
{canUpdate && <Button type={edited ? "primary" : "default"} htmlType="submit">Apply</Button>}
{account.hasPermission("core.role.delete") && <Popconfirm
+14 -14
View File
@@ -12,8 +12,8 @@ import { useAccount } from "../model/account";
import ControlContext from "./ControlContext";
import CreateSaveModal from "./CreateSaveModal";
import SectionHeader from "./SectionHeader";
import { InstanceState, useInstanceList } from "../model/instance";
import { useSaves } from "../model/saves";
import { useInstances } from "../model/instance";
import { useSavesOfInstance } from "../model/saves";
import { notifyErrorHandler } from "../util/notify";
@@ -107,7 +107,7 @@ function TransferModal(props: ModalProps) {
let control = useContext(ControlContext);
let [open, setOpen] = useState(false);
let [form] = Form.useForm();
let [instanceList] = useInstanceList();
let [instances] = useInstances();
return <>
<Button disabled={props.disabled} onClick={() => setOpen(true)}>Transfer</Button>
@@ -153,17 +153,17 @@ function TransferModal(props: ModalProps) {
(option?.title.toLowerCase().indexOf(input.toLowerCase())??-1) >= 0
)}
>
{instanceList.filter(
{[...instances.values()].filter(
instance => instance["id"] !== props.instanceId
).map((instance) => <Select.Option
key={instance["id"]}
value={instance["id"]}
title={instance["name"]}
disabled={["unassigned", "unknown"].includes(instance["status"])}
key={instance.id}
value={instance.id}
title={instance.name}
disabled={["unassigned", "unknown"].includes(instance.status)}
>
{instance["name"]}
{instance["status"] === "unassigned" && " (unassigned)"}
{instance["status"] === "unknown" && " (offline)"}
{instance.name}
{instance.status === "unassigned" && " (unassigned)"}
{instance.status === "unknown" && " (offline)"}
</Select.Option>)}
</Select>
</Form.Item>
@@ -198,10 +198,10 @@ type File = {
status: string;
}
export default function SavesList(props: { instance: InstanceState }) {
export default function SavesList(props: { instance: lib.InstanceDetails }) {
let account = useAccount();
let control = useContext(ControlContext);
let saves = useSaves(props.instance.id);
let [saves] = useSavesOfInstance(props.instance.id);
let [starting, setStarting] = useState(false);
let [uploadingFiles, setUploadingFiles] = useState<File[]>([]);
@@ -234,7 +234,7 @@ export default function SavesList(props: { instance: InstanceState }) {
defaultSortOrder: "descend",
},
]}
dataSource={saves}
dataSource={[...saves.values()]}
rowKey={save => save.name}
pagination={false}
expandable={{
@@ -5,11 +5,10 @@ import * as lib from "@clusterio/lib";
import ControlContext from "./ControlContext";
import { notifyErrorHandler } from "../util/notify";
import type { InstanceState } from "../model/instance";
import { ButtonProps } from "antd/es/button";
type StartStopInstanceButtonProps = {
instance: InstanceState;
instance: lib.InstanceDetails;
onFinish?: () => void;
buttonProps?: ButtonProps;
};
@@ -23,15 +22,15 @@ export default function StartStopInstanceButton(props: StartStopInstanceButtonPr
let action;
if (props.instance.status === "stopped") {
action = control.sendTo(
{ instanceId: props.instance.id! },
{ instanceId: props.instance.id },
new lib.InstanceStartRequest(undefined),
).catch(
notifyErrorHandler("Error starting instance")
);
} else if (["starting", "running"].includes(props.instance.status!)) {
} else if (["starting", "running"].includes(props.instance.status)) {
action = control.sendTo(
{ instanceId: props.instance.id! },
{ instanceId: props.instance.id },
new lib.InstanceStopRequest(),
).catch(
notifyErrorHandler("Error stopping instance")
@@ -54,7 +53,7 @@ export default function StartStopInstanceButton(props: StartStopInstanceButtonPr
{...(props.buttonProps || {})}
loading={switching}
type="primary"
disabled={!["starting", "running", "stopped"].includes(props.instance.status!)}
disabled={!["starting", "running", "stopped"].includes(props.instance.status)}
onClick={onClick}
>
{props.instance.status === "stopped" ? "Start" : "Stop"}
+14 -14
View File
@@ -9,7 +9,7 @@ import DeleteOutlined from "@ant-design/icons/DeleteOutlined";
import * as lib from "@clusterio/lib";
import { useAccount } from "../model/account";
import { useInstanceList } from "../model/instance";
import { useInstances } from "../model/instance";
import ControlContext from "./ControlContext";
import PageHeader from "./PageHeader";
import PageLayout from "./PageLayout";
@@ -33,8 +33,8 @@ export default function UserViewPage() {
let account = useAccount();
let control = useContext(ControlContext);
let [instanceList] = useInstanceList();
let [user] = useUser(userName);
let [instances] = useInstances();
const [user, synced] = useUser(userName);
let [roles, setRoles] = useState<Map<number, lib.Role>>(new Map());
let [form] = Form.useForm();
let [rolesDirty, setRolesDirty] = useState<boolean>(false);
@@ -54,6 +54,7 @@ export default function UserViewPage() {
if (
account.hasPermission("core.user.set_banned")
&& !banReasonDirty
&& user
&& user.banReason !== form.getFieldValue("banReason")
) {
form.setFieldsValue({ "banReason": user.banReason });
@@ -61,14 +62,14 @@ export default function UserViewPage() {
}, [account, form, user, banReasonDirty]);
let nav = [{ name: "Users", path: "/users" }, { name: userName }];
if (user.loading) {
return <PageLayout nav={nav}>
<PageHeader title={userName} />
<Spin size="large" />
</PageLayout>;
}
if (!user) {
if (!synced) {
return <PageLayout nav={nav}>
<PageHeader title={userName} />
<Spin size="large" />
</PageLayout>;
}
if (user.missing) {
return <PageLayout nav={nav}>
<PageHeader title="User not found" />
<p>User with name {userName} was not found on the controller.</p>
@@ -78,7 +79,7 @@ export default function UserViewPage() {
let roleSelector = <Form.Item
noStyle
name="roles"
initialValue={user.roleIds}
initialValue={[...user.roleIds]}
rules={[
{ validator: () => (rolesError ? Promise.reject(rolesError) : Promise.resolve()) },
]}
@@ -100,9 +101,8 @@ export default function UserViewPage() {
</Select>
</Form.Item>;
function instanceName(id: number): string {
let instance = instanceList.find(i => i.id === id);
return instance ? instance.name : String(id);
function instanceName(id: number) {
return instances.get(id)?.name ?? String(id);
}
return <PageLayout nav={nav}>
+3 -3
View File
@@ -11,7 +11,7 @@ import { formatDuration } from "../util/time_format";
import PageHeader from "./PageHeader";
import PageLayout from "./PageLayout";
import PluginExtra from "./PluginExtra";
import { formatLastSeen, sortLastSeen, useUserList } from "../model/user";
import { formatLastSeen, sortLastSeen, useUsers } from "../model/user";
const strcmp = new Intl.Collator(undefined, { numeric: true, sensitivity: "base" }).compare;
@@ -60,7 +60,7 @@ export default function UsersPage() {
let account = useAccount();
let control = useContext(ControlContext);
let navigate = useNavigate();
let [userList] = useUserList();
let [users] = useUsers();
let [roles, setRoles] = useState(new Map());
@@ -117,7 +117,7 @@ export default function UsersPage() {
responsive: ["lg"],
},
]}
dataSource={userList}
dataSource={[...users.values()]}
pagination={false}
rowKey={user => user.name}
onRow={(user, rowIndex) => ({
+8 -82
View File
@@ -1,87 +1,13 @@
import { useEffect, useContext, useState } from "react";
import { useCallback, useContext, useSyncExternalStore } from "react";
import ControlContext from "../components/ControlContext";
import * as lib from "@clusterio/lib";
export type HostState = Partial<lib.HostDetails> & {
loading?: boolean,
present?: boolean,
missing?: boolean,
export function useHost(id?: number) {
const [hosts, synced] = useHosts();
return [id !== undefined ? hosts.get(id) : undefined, synced] as const;
}
export function useHost(id: number): [ HostState, () => void ] {
let control = useContext(ControlContext);
let [host, setHost] = useState<HostState>({ loading: true });
function updateHost() {
// XXX optimize by requesting only the host in question
control.send(
new lib.HostListRequest()
).then((hosts: lib.HostDetails[]) => {
let match = hosts.find(i => i.id === id);
if (!match) {
setHost({ missing: true });
} else {
setHost({ ...match, present: true });
}
});
}
useEffect(() => {
if (!Number.isInteger(id)) {
setHost({ missing: true });
return undefined;
}
updateHost();
function updateHandler(newHost: lib.HostDetails) {
setHost({ ...newHost, loading: false, missing: false, present: true });
}
control.hostUpdate.subscribeToChannel(id, updateHandler);
return () => {
control.hostUpdate.unsubscribeFromChannel(id, updateHandler);
};
}, [id]);
return [host, updateHost];
}
export function useHostList() {
let control = useContext(ControlContext);
let [hostList, setHostList] = useState<lib.HostDetails[]>([]);
function updateHostList() {
control.send(
new lib.HostListRequest()
).then((hosts: lib.HostDetails[]) => {
setHostList(hosts);
});
}
useEffect(() => {
updateHostList();
function updateHandler(newHost: lib.HostDetails) {
setHostList(oldList => {
let newList = oldList.concat();
let index = newList.findIndex(s => s.id === newHost.id);
if (index !== -1) {
newList[index] = newHost;
} else {
newList.push(newHost);
}
return newList;
});
}
control.hostUpdate.subscribe(updateHandler);
return () => {
control.hostUpdate.unsubscribe(updateHandler);
};
}, []);
return [hostList];
export function useHosts() {
const control = useContext(ControlContext);
const subscribe = useCallback((callback: () => void) => control.hosts.subscribe(callback), [control]);
return useSyncExternalStore(subscribe, () => control.hosts.getSnapshot());
}
+8 -81
View File
@@ -1,86 +1,13 @@
import { useEffect, useContext, useState } from "react";
import { useCallback, useContext, useSyncExternalStore } from "react";
import ControlContext from "../components/ControlContext";
import * as lib from "@clusterio/lib";
const { logger } = lib;
export type InstanceState = Partial<lib.InstanceDetails> & {
loading?: boolean,
present?: boolean,
missing?: boolean,
export function useInstance(id?: number) {
const [instances, synced] = useInstances();
return [id !== undefined ? instances.get(id) : undefined, synced] as const;
}
export function useInstance(id: number): [InstanceState, ()=>void] {
let control = useContext(ControlContext);
let [instance, setInstance] = useState<InstanceState>({ loading: true });
function updateInstance() {
control.send(new lib.InstanceDetailsGetRequest(id)).then(updatedInstance => {
setInstance({ ...updatedInstance, present: true });
}).catch((err: any) => {
logger.error(`Failed to get instance: ${err}`);
setInstance({ missing: true });
});
}
useEffect(() => {
if (!Number.isInteger(id)) {
setInstance({ missing: true });
return undefined;
}
updateInstance();
function updateHandler(newInstance: lib.InstanceDetails) {
setInstance({ ...newInstance, present: true });
}
control.instanceUpdate.subscribeToChannel(id, updateHandler);
return () => {
control.instanceUpdate.unsubscribeFromChannel(id, updateHandler);
};
}, [id]);
return [instance, updateInstance];
}
export function useInstanceList() {
let control = useContext(ControlContext);
let [instanceList, setInstanceList] = useState<lib.InstanceDetails[]>([]);
function updateInstanceList() {
control.send(new lib.InstanceDetailsListRequest()).then(instances => {
setInstanceList(instances);
}).catch(err => {
logger.error(`Failed to list instances:\n${err}`);
});
}
useEffect(() => {
updateInstanceList();
function updateHandler(newInstance: lib.InstanceDetails) {
setInstanceList(oldList => {
let newList = oldList.concat();
let index = newList.findIndex(s => s.id === newInstance.id);
if (newInstance.status !== "deleted") {
if (index !== -1) {
newList[index] = newInstance;
} else {
newList.push(newInstance);
}
} else if (index !== -1) {
newList.splice(index, 1);
}
return newList;
});
}
control.instanceUpdate.subscribe(updateHandler);
return () => {
control.instanceUpdate.unsubscribe(updateHandler);
};
}, []);
return [instanceList];
export function useInstances() {
const control = useContext(ControlContext);
const subscribe = useCallback((callback: () => void) => control.instances.subscribe(callback), [control]);
return useSyncExternalStore(subscribe, () => control.instances.getSnapshot());
}
+8 -80
View File
@@ -1,85 +1,13 @@
import { useContext, useEffect, useState } from "react";
import * as lib from "@clusterio/lib";
import { useCallback, useContext, useSyncExternalStore } from "react";
import ControlContext from "../components/ControlContext";
const { logger } = lib;
export type StateLoadStatus = {
loading?: boolean;
present?: boolean;
missing?: boolean;
};
export function useModPack(id: number) {
let control = useContext(ControlContext);
let [modPack, setModPack] = useState<lib.ModPack|StateLoadStatus>({ loading: true });
function updateModPack() {
control.send(new lib.ModPackGetRequest(id)).then(updatedModPack => {
setModPack(updatedModPack);
}).catch(err => {
logger.error(`Failed to get mod pack: ${err}`);
setModPack({ missing: true });
});
}
useEffect(() => {
if (typeof id !== "number") {
setModPack({ missing: true });
return undefined;
}
updateModPack();
control.modPackUpdate.subscribeToChannel(id, setModPack);
return () => {
control.modPackUpdate.unsubscribeFromChannel(id, setModPack);
};
}, [id]);
return [modPack];
export function useModPack(id?: number) {
const [modPacks, synced] = useModPacks();
return [id !== undefined ? modPacks.get(id) : undefined, synced] as const;
}
export function useModPackList() {
let control = useContext(ControlContext);
let [modPackList, setModPackList] = useState<lib.ModPack[]>([]);
function updateModPackList() {
control.send(new lib.ModPackListRequest()).then(modPacks => {
setModPackList(modPacks);
}).catch(err => {
logger.error(`Failed to list mod packs:\n${err}`);
});
}
useEffect(() => {
updateModPackList();
function updateHandler(newModPack: lib.ModPack) {
setModPackList(oldList => {
let newList = oldList.concat();
let index = newList.findIndex(u => u.id === newModPack.id);
if (!newModPack.isDeleted) {
if (index !== -1) {
newList[index] = newModPack;
} else {
newList.push(newModPack);
}
} else if (index !== -1) {
newList.splice(index, 1);
}
return newList;
});
}
control.modPackUpdate.subscribe(updateHandler);
return () => {
control.modPackUpdate.unsubscribe(updateHandler);
};
}, []);
return [modPackList];
export function useModPacks() {
const control = useContext(ControlContext);
const subscribe = useCallback((callback: () => void) => control.modPacks.subscribe(callback), [control]);
return useSyncExternalStore(subscribe, () => control.modPacks.getSnapshot());
}
+9 -45
View File
@@ -1,49 +1,13 @@
import { useContext, useEffect, useState } from "react";
import * as lib from "@clusterio/lib";
import { useCallback, useContext, useSyncExternalStore } from "react";
import ControlContext from "../components/ControlContext";
const { logger } = lib;
export function useModList() {
let control = useContext(ControlContext);
let [modList, setModList] = useState<lib.ModInfo[]>([]);
function updateModList() {
control.send(new lib.ModListRequest()).then(mods => {
setModList(mods);
}).catch(err => {
logger.error(`Failed to list mods:\n${err}`);
});
}
useEffect(() => {
updateModList();
function updateHandler(newMod: lib.ModInfo) {
setModList(oldList => {
let newList = oldList.concat();
let index = newList.findIndex(mod => mod.name === newMod.name && mod.version === newMod.version);
if (!newMod.isDeleted) {
if (index !== -1) {
newList[index] = newMod;
} else {
newList.push(newMod);
}
} else if (index !== -1) {
newList.splice(index, 1);
}
return newList;
});
}
control.modUpdate.subscribe(updateHandler);
return () => {
control.modUpdate.unsubscribe(updateHandler);
};
}, []);
return [modList];
export function useMod(id?: string) {
const [mods, synced] = useMods();
return [id !== undefined ? mods.get(id) : undefined, synced] as const;
}
export function useMods() {
const control = useContext(ControlContext);
const subscribe = useCallback((callback: () => void) => control.mods.subscribe(callback), [control]);
return useSyncExternalStore(subscribe, () => control.mods.getSnapshot());
}
+11 -36
View File
@@ -1,39 +1,14 @@
import { useEffect, useContext, useState } from "react";
import { useCallback, useContext, useSyncExternalStore } from "react";
import ControlContext from "../components/ControlContext";
import * as lib from "@clusterio/lib";
const { logger } = lib;
export function useSaves(instanceId?: number): lib.SaveDetails[] {
let control = useContext(ControlContext);
let [saves, setSaves] = useState<lib.SaveDetails[]>([]);
function updateSaves() {
control.sendTo({ instanceId: instanceId! }, new lib.InstanceListSavesRequest()).then(updatedSaves => {
setSaves(updatedSaves);
}).catch(err => {
logger.error(`Failed to list instance saves: ${err}`);
setSaves([]);
});
}
useEffect(() => {
if (!Number.isInteger(instanceId)) {
setSaves([]);
return undefined;
}
updateSaves();
function updateHandler(data: lib.InstanceSaveListUpdateEvent) {
setSaves(data.saves);
}
control.saveListUpdate.subscribeToChannel(instanceId!, updateHandler);
return () => {
control.saveListUpdate.unsubscribeFromChannel(instanceId!, updateHandler);
};
}, [instanceId]);
return saves;
export function useSavesOfInstance(instanceId?: number) {
const [saves, synced] = useSaves();
const savesOfInstance = new Map([...saves].filter(([_id, save]) => save.instanceId === instanceId));
return [savesOfInstance, synced] as const;
}
export function useSaves() {
const control = useContext(ControlContext);
const subscribe = useCallback((callback: () => void) => control.saves.subscribe(callback), [control]);
return useSyncExternalStore(subscribe, () => control.saves.getSnapshot());
}
+16 -87
View File
@@ -1,31 +1,29 @@
import React, { useEffect, useContext, useState } from "react";
import React, { useCallback, useContext, useSyncExternalStore } from "react";
import { Tag } from "antd";
import ControlContext from "../components/ControlContext";
import * as lib from "@clusterio/lib";
const { logger } = lib;
function calculateLastSeen(user: RawUserState, instanceId?: number) {
function calculateLastSeen(user: lib.User, instanceId?: number) {
let stats;
if (instanceId === undefined) {
stats = user.playerStats;
} else {
stats = (user.instanceStats || new Map()).get(instanceId);
stats = user.instanceStats.get(instanceId);
if (!stats) {
return undefined;
}
}
if (stats.lastLeaveAt > stats.lastJoinAt) {
return stats.lastLeaveAt;
if ((stats.lastLeaveAt?.getTime() ?? 0) > (stats.lastJoinAt?.getTime() ?? 0)) {
return stats.lastLeaveAt?.getTime();
}
if (stats.lastJoinAt) {
return stats.lastLeaveAt;
return stats.lastLeaveAt?.getTime();
}
return undefined;
}
export function formatLastSeen(user: RawUserState, instanceId?: number) {
export function formatLastSeen(user: lib.User, instanceId?: number) {
if (user.instances && [...user.instances].some(id => instanceId === undefined || id === instanceId)) {
return <Tag color="green">Online</Tag>;
}
@@ -36,8 +34,8 @@ export function formatLastSeen(user: RawUserState, instanceId?: number) {
return new Date(lastSeen).toLocaleString();
}
export function sortLastSeen(userA:RawUserState, userB:RawUserState, instanceIdA?: number, instanceIdB?: number) {
function epoch(user:RawUserState, instanceId?: number) {
export function sortLastSeen(userA: lib.User, userB: lib.User, instanceIdA?: number, instanceIdB?: number) {
function epoch(user: lib.User, instanceId?: number) {
return user.instances && [...user.instances].some(id => instanceId === undefined || id === instanceId);
}
@@ -52,82 +50,13 @@ export function sortLastSeen(userA:RawUserState, userB:RawUserState, instanceIdA
return lastSeenA - lastSeenB;
}
export type RawUserState = Partial<lib.User> & {
loading?: boolean;
present?: boolean;
missing?: boolean;
};
export function useUser(name: string): [RawUserState, () => void] {
let control = useContext(ControlContext);
let [user, setUser] = useState<RawUserState>({ loading: true });
function updateUser() {
control.send(new lib.UserGetRequest(name)).then(updatedUser => {
setUser({ ...updatedUser, present: true });
}).catch(err => {
logger.error(`Failed to get user: ${err}`);
setUser({ missing: true });
});
}
useEffect(() => {
if (typeof name !== "string") {
setUser({ missing: true });
return undefined;
}
updateUser();
function updateHandler(newUser: lib.User) {
setUser({ ...newUser, present: true });
}
control.userUpdate.subscribeToChannel(name, updateHandler);
return () => {
control.userUpdate.unsubscribeFromChannel(name, updateHandler);
};
}, [name]);
return [user, updateUser];
export function useUser(name?: string) {
const [users, synced] = useUsers();
return [name !== undefined ? users.get(name) : undefined, synced] as const;
}
export function useUserList() {
let control = useContext(ControlContext);
let [userList, setUserList] = useState<lib.User[]>([]);
function updateUserList() {
control.send(new lib.UserListRequest()).then(users => {
setUserList(users);
}).catch(err => {
logger.error(`Failed to list users:\n${err}`);
});
}
useEffect(() => {
updateUserList();
function updateHandler(newUser: lib.User) {
setUserList(oldList => {
let newList = oldList.concat();
let index = newList.findIndex(u => u.name === newUser.name);
if (!newUser.isDeleted) {
if (index !== -1) {
newList[index] = newUser;
} else {
newList.push(newUser);
}
} else if (index !== -1) {
newList.splice(index, 1);
}
return newList;
});
}
control.userUpdate.subscribe(updateHandler);
return () => {
control.userUpdate.unsubscribe(updateHandler);
};
}, []);
return [userList];
export function useUsers() {
const control = useContext(ControlContext);
const subscribe = useCallback((callback: () => void) => control.users.subscribe(callback), [control]);
return useSyncExternalStore(subscribe, () => control.users.getSnapshot());
}
+7 -42
View File
@@ -7,7 +7,7 @@ const { logFilter, logger } = lib;
type accountHandler = (account: lib.AccountDetails) => void;
type hostHandler = (hostDetails: lib.HostDetails) => void;
type instanceHandler = (instanceDetails: lib.InstanceDetails) => void;
type saveListHandler = (saveListEvent: lib.InstanceSaveListUpdateEvent) => void;
type saveListHandler = (saveListEvent: lib.SaveDetails) => void;
type modPackHandler = (modPack: lib.ModPack) => void;
type modInfoHandler = (modInfo: lib.ModInfo) => void;
type userHandler = (rawUser: lib.User) => void;
@@ -60,29 +60,12 @@ export class Control extends lib.Link {
logHandlers: Map<lib.LogFilter, logHandler[]> = new Map();
/** Updates handled by the subscription service */
hostUpdate = new lib.EventSubscriber<lib.HostUpdateEvent, lib.HostDetails>(
lib.HostUpdateEvent, event => event.update,
);
instanceUpdate = new lib.EventSubscriber<lib.InstanceDetailsUpdateEvent, lib.InstanceDetails>(
lib.InstanceDetailsUpdateEvent, event => event.details,
);
saveListUpdate = new lib.EventSubscriber<lib.InstanceSaveListUpdateEvent>(
lib.InstanceSaveListUpdateEvent,
);
modPackUpdate = new lib.EventSubscriber<lib.ModPackUpdateEvent, lib.ModPack>(
lib.ModPackUpdateEvent, event => event.modPack,
);
modUpdate = new lib.EventSubscriber<lib.ModUpdateEvent, lib.ModInfo>(
lib.ModUpdateEvent, event => event.mod,
);
userUpdate = new lib.EventSubscriber<lib.UserUpdateEvent, lib.User>(
lib.UserUpdateEvent, event => event.user,
);
hosts = new lib.EventSubscriber(lib.HostUpdatesEvent, this);
instances = new lib.EventSubscriber(lib.InstanceDetailsUpdatesEvent, this);
saves = new lib.EventSubscriber(lib.InstanceSaveDetailsUpdatesEvent, this);
modPacks = new lib.EventSubscriber(lib.ModPackUpdatesEvent, this);
mods = new lib.EventSubscriber(lib.ModUpdatesEvent, this);
users = new lib.EventSubscriber(lib.UserUpdatesEvent, this);
declare connector: ControlConnector;
@@ -95,24 +78,6 @@ export class Control extends lib.Link {
this.accountName = data.account.name;
this.accountRoles = data.account.roles;
this.emitAccountUpdate();
this.hostUpdate.connectControl(this).catch(err => logger.error(
`Unexpected error updating host subscriptions:\n${err.stack}`
));
this.instanceUpdate.connectControl(this).catch(err => logger.error(
`Unexpected error updating instance subscriptions:\n${err.stack}`
));
this.saveListUpdate.connectControl(this).catch(err => logger.error(
`Unexpected error updating save list subscriptions:\n${err.stack}`
));
this.modPackUpdate.connectControl(this).catch(err => logger.error(
`Unexpected error updating mod pack subscriptions:\n${err.stack}`
));
this.modUpdate.connectControl(this).catch(err => logger.error(
`Unexpected error updating mod subscriptions:\n${err.stack}`
));
this.userUpdate.connectControl(this).catch(err => logger.error(
`Unexpected error updating user subscriptions:\n${err.stack}`
));
this.updateLogSubscriptions().catch(err => logger.error(
`Unexpected error updating log subscriptions:\n${err.stack}`
));
+40
View File
@@ -0,0 +1,40 @@
"use strict";
const assert = require("assert").strict;
const { InstanceInfo } = require("@clusterio/controller");
const lib = require("@clusterio/lib");
describe("controller/InstanceInfo", function() {
describe("class InstanceInfo", function() {
it("should round trip serialize", async function() {
const validate = lib.compile(InstanceInfo.jsonSchema);
function check(info) {
const json = JSON.parse(JSON.stringify(info));
if (!validate(json)) {
throw validate.errors;
}
assert.deepEqual(InstanceInfo.fromJSON(json, info.config), info);
}
if (!lib.InstanceConfig._finalized) {
lib.InstanceConfig.finalize();
}
const config = new lib.InstanceConfig("controller");
await config.init();
check(InstanceInfo.fromJSON({ config, status: "stopped" }, config));
check(InstanceInfo.fromJSON({ config, status: "running" }, config));
check(InstanceInfo.fromJSON({ config, status: "running", gamePort: 34197 }, config));
check(InstanceInfo.fromJSON({ config, status: "running", updatedAt: Date.now() }, config));
// All at once
check(InstanceInfo.fromJSON({
config,
status: "unknown",
gamePort: 10000,
updateAt: 20000,
}, config));
});
});
});
+16 -8
View File
@@ -315,16 +315,16 @@ describe("Integration of Clusterio", function() {
describe("instance save create", function() {
it("creates a save", async function() {
slowTest(this);
getControl().saveListUpdates = [];
getControl().saveUpdates = [];
await execCtl("instance save create test");
await checkInstanceStatus(44, "stopped");
});
});
describe("saveListUpdateEventHandler()", function() {
describe("InstanceSaveDetailsUpdatesEvent", function() {
it("should have triggered for the created save", function() {
slowTest(this);
assert.equal(getControl().saveListUpdates.slice(-1)[0].saves[0].name, "world.zip");
assert.equal(getControl().saveUpdates.slice(-1)[0].updates[0].name, "world.zip");
});
});
@@ -391,7 +391,7 @@ describe("Integration of Clusterio", function() {
let savesDir = path.join("temp", "test", "instances", "test", "saves");
await fs.copy(path.join(savesDir, "world.zip"), path.join(savesDir, "_autosave1.zip"));
await execCtl("instance start test");
let saves = await getControl().sendTo({ instanceId: 44 }, new lib.InstanceListSavesRequest());
let saves = await getControl().sendTo({ instanceId: 44 }, new lib.InstanceSaveDetailsListRequest());
let running = saves.find(s => s.loaded);
assert(running.name !== "_autosave1.zip");
});
@@ -407,19 +407,19 @@ describe("Integration of Clusterio", function() {
assert(log.some(info => /technobabble/.test(info.message)), "Command was not sent");
});
it("should trigger saveListUpdate on save", async function() {
it("should trigger InstanceSaveDetailsUpdatesEvent on save", async function() {
slowTest(this);
getControl().saveListUpdates = [];
getControl().saveUpdates = [];
await execCtl("instance send-rcon test /server-save");
let received = false;
for (let x = 0; x < 10; x++) {
if (getControl().saveListUpdates.length) {
if (getControl().saveUpdates.length) {
received = true;
break;
}
await wait(100);
}
assert(received, "saveListUpdate not sent");
assert(received, "InstanceSaveDetailsUpdatesEvent not sent");
});
});
@@ -827,10 +827,16 @@ describe("Integration of Clusterio", function() {
describe("instance delete", function() {
it("deletes the instance", async function() {
slowTest(this);
getControl().saveUpdates = [];
await execCtl("instance delete test");
assert(!await fs.exists(path.join(instancesDir, "test")), "Instance files was not deleted");
let instances = await getInstances();
assert(!instances.has(44), "instance was not deleted from controller");
const saveUpdates = getControl().saveUpdates;
assert.equal(saveUpdates.length, 1, "Expected one update sent on saves");
assert(saveUpdates[0].updates.length > 0, "Expected saves update to contain at least one save updated");
assert(saveUpdates[0].updates.every(update => update.isDeleted), "Expected all saves to be deleted");
});
});
@@ -885,6 +891,7 @@ describe("Integration of Clusterio", function() {
reference.settings["runtime-global"].set("MyInt", { value: 1235 });
reference.settings["runtime-global"].set("MyDouble", { value: 12.25 });
reference.settings["runtime-per-user"].set("MyString", { value: "a-string" });
reference.updatedAt = modPack.updatedAt;
assert.deepEqual(modPack, reference);
});
});
@@ -989,6 +996,7 @@ describe("Integration of Clusterio", function() {
"dependencies:\n" +
`size: ${stat.size}\n` +
`sha1: ${hash}\n` +
`updatedAt: ${stat.mtimeMs}\n` +
"isDeleted: false\n",
);
});
+26 -26
View File
@@ -26,7 +26,7 @@ class TestControl extends lib.Link {
super(connector);
this.hostUpdates = [];
this.instanceUpdates = [];
this.saveListUpdates = [];
this.saveUpdates = [];
this.modUpdates = [];
this.modPackUpdates = [];
this.userUpdates = [];
@@ -36,56 +36,56 @@ class TestControl extends lib.Link {
return;
}
this.send(
new lib.SubscriptionRequest(lib.HostUpdateEvent.name, true)
new lib.SubscriptionRequest(lib.HostUpdatesEvent.name, true)
).catch(err => logger.error(`Error setting host subscriptions:\n${err.stack}`));
this.send(
new lib.SubscriptionRequest(lib.InstanceDetailsUpdateEvent.name, true)
new lib.SubscriptionRequest(lib.InstanceDetailsUpdatesEvent.name, true)
).catch(err => logger.error(`Error setting instance subscriptions:\n${err.stack}`));
this.send(
new lib.SubscriptionRequest(lib.InstanceSaveListUpdateEvent.name, true)
).catch(err => logger.error(`Error setting save list subscriptions:\n${err.stack}`));
new lib.SubscriptionRequest(lib.InstanceSaveDetailsUpdatesEvent.name, true)
).catch(err => logger.error(`Error setting save subscriptions:\n${err.stack}`));
this.send(
new lib.SubscriptionRequest(lib.ModUpdateEvent.name, true)
new lib.SubscriptionRequest(lib.ModUpdatesEvent.name, true)
).catch(err => logger.error(`Error setting mod subscriptions:\n${err.stack}`));
this.send(
new lib.SubscriptionRequest(lib.ModPackUpdateEvent.name, true)
new lib.SubscriptionRequest(lib.ModPackUpdatesEvent.name, true)
).catch(err => logger.error(`Error setting mod pack subscriptions:\n${err.stack}`));
this.send(
new lib.SubscriptionRequest(lib.UserUpdateEvent.name, true)
new lib.SubscriptionRequest(lib.UserUpdatesEvent.name, true)
).catch(err => logger.error(`Error setting user subscriptions:\n${err.stack}`));
});
this.handle(lib.AccountUpdateEvent);
this.handle(lib.HostUpdateEvent, this.handleHostUpdateEvent.bind(this));
this.handle(lib.InstanceDetailsUpdateEvent, this.handleInstanceDetailsUpdateEvent.bind(this));
this.handle(lib.InstanceSaveListUpdateEvent, this.handleInstanceSaveListUpdateEvent.bind(this));
this.handle(lib.ModUpdateEvent, this.handleModUpdateEvent.bind(this));
this.handle(lib.ModPackUpdateEvent, this.handleModPackUpdateEvent.bind(this));
this.handle(lib.UserUpdateEvent, this.handleUserUpdateEvent.bind(this));
this.handle(lib.HostUpdatesEvent, this.handleHostUpdatesEvent.bind(this));
this.handle(lib.InstanceDetailsUpdatesEvent, this.handleInstanceDetailsUpdatesEvent.bind(this));
this.handle(lib.InstanceSaveDetailsUpdatesEvent, this.handleInstanceSaveDetailsUpdatesEvent.bind(this));
this.handle(lib.ModUpdatesEvent, this.handleModUpdatesEvent.bind(this));
this.handle(lib.ModPackUpdatesEvent, this.handleModPackUpdatesEvent.bind(this));
this.handle(lib.UserUpdatesEvent, this.handleUserUpdatesEvent.bind(this));
}
async handleHostUpdateEvent(event) {
this.hostUpdates.push(event.update);
async handleHostUpdatesEvent(event) {
this.hostUpdates.push(...event.updates);
}
async handleInstanceDetailsUpdateEvent(event) {
this.instanceUpdates.push(event.details);
async handleInstanceDetailsUpdatesEvent(event) {
this.instanceUpdates.push(...event.updates);
}
async handleInstanceSaveListUpdateEvent(event) {
this.saveListUpdates.push(event);
async handleInstanceSaveDetailsUpdatesEvent(event) {
this.saveUpdates.push(event);
}
async handleModUpdateEvent(event) {
this.modUpdates.push(event.mod);
async handleModUpdatesEvent(event) {
this.modUpdates.push(...event.updates);
}
async handleModPackUpdateEvent(event) {
this.modPackUpdates.push(event.modPack);
async handleModPackUpdatesEvent(event) {
this.modPackUpdates.push(...event.updates);
}
async handleUserUpdateEvent(event) {
this.userUpdates.push(event.user);
async handleUserUpdatesEvent(event) {
this.userUpdates.push(...event.updates);
}
}
+1
View File
@@ -92,6 +92,7 @@ describe("lib/data/ModInfo", function() {
dependencies: [],
filename: "empty_mod_1.0.0.zip",
size: stat.size,
updated_at: stat.mtimeMs,
sha1: hash,
})
);
+167 -290
View File
@@ -19,20 +19,11 @@ describe("lib/subscriptions", function() {
static src = ["controller", "control"];
static dst = ["controller", "control"];
static permission = null;
}
lib.Link.register(RegisteredEvent);
class ChannelEvent {
static type = "event";
static src = ["controller", "control"];
static dst = ["controller", "control"];
static permission = null;
get subscriptionChannel() {
return "channelOne";
constructor(updates = []) {
this.updates = updates;
}
}
lib.Link.register(ChannelEvent);
lib.Link.register(RegisteredEvent);
class StringPermissionEvent {
static type = "event";
@@ -58,46 +49,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 a null event replay", function() {
const response = new lib.SubscriptionResponse(null);
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) {
@@ -130,14 +81,14 @@ 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));
assert.deepEqual(reconstructed, request);
});
it("should be throw when given an unregistered event", function() {
it("should throw when given an unregistered event", function() {
assert.throws(
() => new lib.SubscriptionRequest(UnregisteredEvent.name, true),
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
@@ -157,37 +108,24 @@ 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 getLink(id) {
return mockController.wsServer.controlConnections.get(id);
}
function onceConnectorSend(id) {
const link = mockController.wsServer.controlConnections.get(id);
return events.once(link.connector, "send");
return events.once(getLink(id).connector, "send");
}
function assertLastEvent(connectorId, Event) {
const link = mockController.wsServer.controlConnections.get(connectorId);
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 = mockController.wsServer.controlConnections.get(connectorId);
const link = getLink(connectorId);
assert.equal(link.connector.sentMessages.length, 0);
}
@@ -201,10 +139,6 @@ describe("lib/subscriptions", function() {
subscriptions = new lib.SubscriptionController(mockController);
});
it("should handle the SubscriptionRequest event", function() {
assert.equal(mockController.handles.has(lib.SubscriptionRequest), true);
});
describe("handle()", function() {
it("should accept registered events", function() {
subscriptions.handle(RegisteredEvent);
@@ -227,9 +161,10 @@ 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);
subscriptions.handleRequest(
getLink(connectorData.id), connectorData.request, connectorData.src, connectorData.dst
);
}
});
@@ -245,74 +180,63 @@ 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);
mockController.wsServer.controlConnections.get(0).connector.sentMessages.pop(); // Response to request
await subscriptions.handleRequest(
getLink(connectorData.id), request, connectorData.src, connectorData.dst,
);
subscriptions.broadcast(new RegisteredEvent());
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);
const link = getLink(0);
link.connector.closing = true;
subscriptions.broadcast(new RegisteredEvent());
await onceConnectorSend(1);
assertNoEvent(0);
assertLastEvent(1, RegisteredEvent);
assertNoEvent(2);
assertNoEvent(3);
assertNoEvent(4);
});
});
describe("_handleRequest()", function() {
describe("unsubscribe()", function() {
beforeEach(function() {
subscriptions.handle(RegisteredEvent);
for (let connectorData of connectorSetupDate) {
subscriptions.handleRequest(
getLink(connectorData.id), connectorData.request, connectorData.src, connectorData.dst
);
}
});
it("should remove an active subscription", async function() {
const link = getLink(0);
subscriptions.unsubscribe(link);
subscriptions.broadcast(new RegisteredEvent());
await onceConnectorSend(1);
assertNoEvent(0);
assertLastEvent(1, RegisteredEvent);
});
});
describe("handleRequest()", function() {
beforeEach(function() {
subscriptions.handle(RegisteredEvent);
subscriptions.handle(ChannelEvent);
});
it("should not accept subscriptions to unregistered events", function() {
const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
request.eventName = UnregisteredEvent.name;
assert.rejects(
subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")),
subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")),
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
);
});
@@ -320,83 +244,79 @@ describe("lib/subscriptions", function() {
const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
request.eventName = StringPermissionEvent.name;
assert.rejects(
subscriptions._handleRequest(request, addr({ controlId: 0 }), addr("controller")),
subscriptions.handleRequest(getLink(0), request, addr({ controlId: 0 }), addr("controller")),
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);
await subscriptions.handleRequest(
getLink(0), 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 an unsubscription", async function() {
const request = new lib.SubscriptionRequest(RegisteredEvent.name, true);
await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
await subscriptions.handleRequest(
getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst
);
subscriptions.broadcast(new RegisteredEvent());
await onceConnectorSend(0);
assertLastEvent(0, RegisteredEvent);
const link = mockController.wsServer.controlConnections.get(0);
const link = getLink(0);
link.connector.sentMessages.pop();
const unsubRequest = new lib.SubscriptionRequest(RegisteredEvent.name, false);
await subscriptions._handleRequest(unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst);
await subscriptions.handleRequest(
getLink(0), unsubRequest, connectorSetupDate[0].src, connectorSetupDate[0].dst
);
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() {
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 request = new lib.SubscriptionRequest(StringPermissionEvent.name, true);
await subscriptions._handleRequest(request, connectorSetupDate[0].src, connectorSetupDate[0].dst);
subscriptions.broadcast(new StringPermissionEvent());
await subscriptions.handleRequest(
getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst
);
await onceConnectorSend(0);
assertLastEvent(0, StringPermissionEvent);
const link = mockController.wsServer.controlConnections.get(0);
const link = getLink(0);
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, true);
await subscriptions.handleRequest(
getLink(0), request, connectorSetupDate[0].src, connectorSetupDate[0].dst
);
await new Promise(resolve => setImmediate(resolve));
assertNoEvent(0);
});
});
});
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);
registeredEvent = new lib.EventSubscriber(RegisteredEvent, mockControl);
});
function assertLastRequest(request) {
@@ -407,178 +327,135 @@ describe("lib/subscriptions", function() {
assert.deepEqual(lastMessage.data, request);
}
function assertNoMessageSent() {
assert.equal(mockControl.connector.sentMessages.length, 0, "messages were sent");
}
describe("constructor()", function() {
it("should not accept unregistered events", function() {
assert.throws(
() => new lib.EventSubscriber(UnregisteredEvent),
() => new lib.EventSubscriber(UnregisteredEvent, mockControl),
new Error(`Unregistered Event class ${UnregisteredEvent.name}`)
);
});
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;
function prehandler() { return "PreHandlerReturn"; }
const eventSubscriber = new lib.EventSubscriber(StringPermissionEvent, prehandler, mockControl);
eventSubscriber.subscribe(function(value) { calledWith = value; });
eventSubscriber._handle(new StringPermissionEvent());
assert.equal(calledWith, "PreHandlerReturn");
});
});
describe("connectControl()", function() {
it("should handle the provided event", function() {
const control = new MockControl(new MockConnector(addr({ controlId: 0 }), addr("controller")));
registeredEvent.connectControl(control);
assert.equal(control._eventHandlers.has(RegisteredEvent), true);
});
it("should do nothing if the control is already connected", function() {
assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true);
registeredEvent.connectControl(mockControl);
assert.equal(mockControl._eventHandlers.has(RegisteredEvent), true);
});
});
describe("subscribe()", function() {
it("should allow subscriptions to an event", function() {
it("should allow subscriptions to an event", async function() {
let calledWith = null;
let calledWithTwo = null;
const event = new RegisteredEvent();
registeredEvent.subscribe(function(value) { calledWith = value; });
registeredEvent.subscribe(function(value) { calledWithTwo = value; });
registeredEvent._handle(event);
assert.deepEqual(calledWith, event);
assert.deepEqual(calledWithTwo, event);
const item = { id: 1, updatedAt: 1, isDeleted: false };
const event = new RegisteredEvent([item]);
registeredEvent.subscribe(function(updates, synced) { calledWith = [updates, synced]; });
registeredEvent.subscribe(function(updates, synced) { calledWithTwo = [updates, synced]; });
await registeredEvent._handle(event);
assert.deepEqual(calledWith, [[item], false]);
assert.deepEqual(calledWithTwo, [[item], false]);
});
});
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;
it("should allow unsubscribing from an event", async function() {
let called = false;
let event = new RegisteredEvent();
function callback(value) { calledWith = value; }
registeredEvent.subscribe(callback);
registeredEvent._handle(event);
assert.deepEqual(calledWith, event);
function callback() { called = true; }
const unsubscribe = registeredEvent.subscribe(callback);
await registeredEvent._handle(event);
assert.deepEqual(called, true);
calledWith = null;
called = false;
event = new RegisteredEvent();
registeredEvent.unsubscribe(callback);
registeredEvent._handle(event);
assert.equal(calledWith, null);
unsubscribe();
// Wait for unsubscribe to happen.
await new Promise(resolve => setImmediate(resolve));
await registeredEvent._handle(event);
assert.equal(called, false);
});
it("should throw an error if the handler was not subscribed", function() {
function callback(value) { calledWith = value; }
assert.rejects(
() => registeredEvent.unsubscribe(callback),
new Error("handler is not registered")
);
it("should not throw error on incorrectly unsubscribing multiple times", async function() {
const unsubscribe = registeredEvent.subscribe(() => {});
unsubscribe();
unsubscribe();
// Wait for unsubscribe to happen.
await new Promise(resolve => setImmediate(resolve));
});
});
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("getSnapshot()", function() {
it("should give an immutable copy of the tracked values", async function() {
const update1 = [{ id: 1, updatedAt: 1, isDeleted: false }];
const update2 = [{ id: 1, updatedAt: 2, isDeleted: false }, { id: 2, updatedAt: 3, isDeleted: false }];
const update3 = [{ id: 1, updatedAt: 4, isDeleted: true }];
const snap0 = registeredEvent.getSnapshot();
assert.deepEqual(snap0, [new Map(), false]);
await registeredEvent._handle(new RegisteredEvent(update1));
const snap1 = registeredEvent.getSnapshot();
assert.deepEqual(snap1, [new Map([[1, update1[0]]]), false]);
await registeredEvent._handle(new RegisteredEvent(update2));
const snap2 = registeredEvent.getSnapshot();
assert.deepEqual(snap2, [new Map([[1, update2[0]], [2, update2[1]]]), false]);
await registeredEvent._handle(new RegisteredEvent(update3));
const snap3 = registeredEvent.getSnapshot();
assert.deepEqual(snap3, [new Map([[2, update2[1]]]), false]);
// validate previous snapshots were not modified
assert.deepEqual(snap2, [new Map([[1, update2[0]], [2, update2[1]]]), false]);
assert.deepEqual(snap1, [new Map([[1, update1[0]]]), false]);
assert.deepEqual(snap0, [new Map(), false]);
});
});
describe("_updateSubscription()", function() {
it("should correctly request a subscription for all channels", async function() {
const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true);
it("should do nothing if not connected", async function() {
registeredEvent.control.connector.connected = false;
registeredEvent.subscribe(() => true);
await new Promise(resolve => setImmediate(resolve));
assertNoMessageSent();
});
it("should correctly request a subscription", async function() {
const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true, -1);
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() {
const expected = new lib.SubscriptionRequest(RegisteredEvent.name, false);
it("should correctly request no subscriptions", async function() {
const expected = new lib.SubscriptionRequest(RegisteredEvent.name, false, -1);
function callback() {}
registeredEvent.subscribe(callback);
const unsubscribe = registeredEvent.subscribe(callback);
await events.once(mockControl.connector, "send");
registeredEvent.unsubscribe(callback);
unsubscribe();
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; });
it("should set synced and invoke callbacks when request is responded to", async function() {
let calledWith;
registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; });
const [msg] = await events.once(mockControl.connector, "send");
assert(!registeredEvent.synced, "synced was set before response");
mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src));
await new Promise(resolve => setImmediate(resolve));
assert(calledWith !== undefined, "callback not called");
assert.deepEqual(calledWith, [[], true]);
assert(registeredEvent.synced, "synced was not set");
});
it("should be called on connector connect event", async function() {
const expected = new lib.SubscriptionRequest(RegisteredEvent.name, true, -1);
registeredEvent._callbacks.push(() => {});
registeredEvent.control.connector.emit("connect");
await events.once(mockControl.connector, "send");
assertLastRequest(expected);
});
it("should revert synced and invoke callbacks on connector close event", async function() {
let calledWith;
registeredEvent.subscribe((values, synced) => { calledWith = [values, synced]; });
const [msg] = await events.once(mockControl.connector, "send");
mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src));
await new Promise(resolve => setImmediate(resolve));
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);
assert(registeredEvent.synced, "synced was not set before close");
calledWith = undefined;
registeredEvent.control.connector.emit("close");
assert(calledWith !== undefined, "callback not called");
assert.deepEqual(calledWith, [[], false]);
assert(!registeredEvent.synced, "synced was not reset");
});
});
});