Add types to event emitters and clean up connections (#772)

* Event types for controller

* Clean up ctl event emitters

* Event types for server

* Event types for lib

* Remove TypedEventEmitter

* Event types for webui

* Fix compile issues

* Small cleanup of connectors
This commit is contained in:
Cooldude2606 authored and GitHub committed 2025-06-13 22:20:50 +01:00
1 parent 5b56aa3fea
commit 331fc7d947
16 files changed
+182 -157

No files matched your search

+1 -1
View File
@@ -313,7 +313,7 @@ export default class ControlConnection extends BaseConnection {
let stream = await routes.createProxyStream(this._controller.app);
stream.filename = name;
let ready = new Promise((resolve, reject) => {
let ready = new Promise<void>((resolve, reject) => {
stream.events.on("source", resolve);
stream.events.on("timeout", () => reject(
new lib.RequestError("Timed out establishing stream from host")
+11 -3
View File
@@ -6,7 +6,7 @@ import type { Static } from "@sinclair/typebox";
import finalhandler from "finalhandler";
import compression from "compression";
import events, { EventEmitter } from "events";
import events from "events";
import fs from "fs-extra";
import http from "http";
import https from "https";
@@ -42,6 +42,14 @@ const logSizeGauge = new Gauge(
type InstanceId = { instanceId: number };
type ControllerDebugEvents = {
"message": [ { direction: "in" | "out", content: string } ],
};
type ControllerEvents = {
"stop": [],
}
/**
* Manages all controller related operations
* @alias module:controller/src/Controller
@@ -66,8 +74,8 @@ export default class Controller {
wsServer: WsServer;
router = new ControllerRouter(this);
trustedProxies: BlockList;
debugEvents: events.EventEmitter = new events.EventEmitter();
private _events: events.EventEmitter = new events.EventEmitter();
debugEvents = new events.EventEmitter<ControllerDebugEvents>();
private _events = new events.EventEmitter<ControllerEvents>();
/** Event subscription controller */
subscriptions = new lib.SubscriptionController();
+5 -8
View File
@@ -1,6 +1,7 @@
import type WebSocket from "ws";
import events from "events";
import assert from "assert/strict";
import * as lib from "@clusterio/lib";
const { logger } = lib;
@@ -113,8 +114,7 @@ export default class WsServerConnector extends lib.WebSocketBaseConnector {
_timedOut() {
logger.verbose("Connector | Connection timed out");
this._reset();
this.emit("close");
this._close();
}
_attachSocketHandlers() {
@@ -130,8 +130,7 @@ export default class WsServerConnector extends lib.WebSocketBaseConnector {
this._socket = null;
if (this._closing) {
this._reset();
this.emit("close");
this._close();
} else {
this._state = "resuming";
@@ -187,8 +186,7 @@ export default class WsServerConnector extends lib.WebSocketBaseConnector {
}
if (this._state === "resuming") {
this._reset();
this.emit("close");
this._close();
return;
}
@@ -223,8 +221,7 @@ export default class WsServerConnector extends lib.WebSocketBaseConnector {
*/
setClosing() {
if (this._state === "resuming") {
this._reset();
this.emit("close");
this._close();
return;
}
+7 -1
View File
@@ -266,6 +266,12 @@ async function uploadExport(req: Request, res: Response) {
res.sendStatus(200);
}
type ProxyStreamEvents = {
"error": [ err: any ],
"close": [],
"timeout": [],
"source": [],
}
export interface ProxyStream {
id: string;
@@ -274,7 +280,7 @@ export interface ProxyStream {
size?: string;
mime?: string;
filename: string | null;
events: events.EventEmitter;
events: events.EventEmitter<ProxyStreamEvents>;
timeout: ReturnType<typeof setTimeout>;
}
+59 -71
View File
@@ -197,28 +197,24 @@ controllerConfigCommands.add(new lib.Command({
editorSpawn.on("data", (data) => {
process.stdout.pipe(data);
});
let doneEmitter = new events.EventEmitter();
editorSpawn.on("exit", async (exit) => {
const data = await fs.readFile(tmpFile, "utf8");
const final = await configToKeyVal(data);
for (let index in final) {
if (index in final) {
try {
await control.send(new lib.ControllerConfigSetFieldRequest(
index,
final[index],
));
} catch (err) {
// eslint-disable-next-line
print(`Attempt to set ${index} to ${final[index] || String(null)} failed; set back to previous value.`);
print(err);
// If the string is empty, it's better to just print "" instead of nothing
}
await events.once(editorSpawn, "exit");
const data = await fs.readFile(tmpFile, "utf8");
const final = await configToKeyVal(data);
for (let index in final) {
if (index in final) {
try {
await control.send(new lib.ControllerConfigSetFieldRequest(
index,
final[index],
));
} catch (err) {
// eslint-disable-next-line
print(`Attempt to set ${index} to ${final[index] || String(null)} failed; set back to previous value.`);
print(err);
// If the string is empty, it's better to just print "" instead of nothing
}
}
doneEmitter.emit("dot_on_done");
});
await events.once(doneEmitter, "dot_on_done");
}
try {
await fs.unlink(tmpFile);
} catch (err) {
@@ -487,33 +483,29 @@ hostConfigCommands.add(new lib.Command({
editorSpawn.on("data", (data) => {
process.stdout.pipe(data);
});
let doneEmitter = new events.EventEmitter();
editorSpawn.on("exit", async (exit) => {
const data = await fs.readFile(tmpFile, "utf8");
const final = await configToKeyVal(data);
for (let index in final) {
if (index in final) {
try {
await control.sendTo({ hostId }, new lib.HostConfigSetFieldRequest(
index,
final[index],
));
} catch (err) {
// eslint-disable-next-line
print(`\n\n\nAttempt to set ${index} to ${final[index] || String(null)} failed; set back to previous value.`);
// If the string is empty, it's better to just print "" instead of nothing
print("This message shouldn't normally appear; if the below message does not indicate it");
print("was a user mistake, please report it to the clustorio devs.");
// added this because it could be a missed entry in disallowedList
// i've added all the vanilla clustorio configs, but there may be
// some modded ones that need to be added.
print(err);
}
await events.once(editorSpawn, "exit");
const data = await fs.readFile(tmpFile, "utf8");
const final = await configToKeyVal(data);
for (let index in final) {
if (index in final) {
try {
await control.sendTo({ hostId }, new lib.HostConfigSetFieldRequest(
index,
final[index],
));
} catch (err) {
// eslint-disable-next-line
print(`\n\n\nAttempt to set ${index} to ${final[index] || String(null)} failed; set back to previous value.`);
// If the string is empty, it's better to just print "" instead of nothing
print("This message shouldn't normally appear; if the below message does not indicate it");
print("was a user mistake, please report it to the clustorio devs.");
// added this because it could be a missed entry in disallowedList
// i've added all the vanilla clustorio configs, but there may be
// some modded ones that need to be added.
print(err);
}
}
doneEmitter.emit("dot_on_done");
});
await events.once(doneEmitter, "dot_on_done");
}
try {
await fs.unlink(tmpFile);
} catch (err) {
@@ -732,34 +724,30 @@ instanceConfigCommands.add(new lib.Command({
editorSpawn.on("data", (data) => {
process.stdout.pipe(data);
});
let doneEmitter = new events.EventEmitter();
editorSpawn.on("exit", async (exit) => {
const data = await fs.readFile(tmpFile, "utf8");
const final = await configToKeyVal(data);
for (let index in final) {
if (index in final) {
try {
await control.send(new lib.InstanceConfigSetFieldRequest(
instanceId,
index,
final[index],
));
} catch (err) {
// eslint-disable-next-line
print(`\n\n\nAttempt to set ${index} to ${final[index] || String(null)} failed; set back to previous value.`);
// If the string is empty, it's better to just print "" instead of nothing
print("This message shouldn't normally appear; if the below message does not indicate it");
print("was a user mistake, please report it to the clustorio devs.");
// added this because it could be a missed entry in disallowedList
// i've added all the vanilla clustorio configs, but there may be
// some modded ones that need to be added.
print(err);
}
await events.once(editorSpawn, "exit");
const data = await fs.readFile(tmpFile, "utf8");
const final = await configToKeyVal(data);
for (let index in final) {
if (index in final) {
try {
await control.send(new lib.InstanceConfigSetFieldRequest(
instanceId,
index,
final[index],
));
} catch (err) {
// eslint-disable-next-line
print(`\n\n\nAttempt to set ${index} to ${final[index] || String(null)} failed; set back to previous value.`);
// If the string is empty, it's better to just print "" instead of nothing
print("This message shouldn't normally appear; if the below message does not indicate it");
print("was a user mistake, please report it to the clustorio devs.");
// added this because it could be a missed entry in disallowedList
// i've added all the vanilla clustorio configs, but there may be
// some modded ones that need to be added.
print(err);
}
}
doneEmitter.emit("dot_on_done");
});
await events.once(doneEmitter, "dot_on_done");
}
try {
await fs.unlink(tmpFile);
} catch (err) {
+28 -2
View File
@@ -454,6 +454,32 @@ export interface FactorioServerOptions {
shutdownTimeoutMs?: number,
}
type FactorioServerEvents = {
// Public Events
"stdout": [ rawLine: Buffer ],
"stderr": [ rawLine: Buffer ],
"output": [ parsed: lib.ParsedFactorioOutput, line: string ]
"error": [ err: any ],
"rcon-ready": [],
"game-ready": [],
"exit": [],
"autosave-start": [ name: string ],
"autosave-finished": [ name: string ],
"save-finished": [],
"whitelist-change": [ added: string[], removed: string[] ],
// Private Events
"_autosave": [ name: string ]
"_saving": [],
"_saved": [],
// IPS events
[ipcEvent: `ipc-${string}`]: [ event: any ],
};
/**
* Factorio Server interface
*
@@ -473,7 +499,7 @@ export interface FactorioServerOptions {
* - exit - invoked when the sterver has exited
* @extends events.EventEmitter
*/
export class FactorioServer extends events.EventEmitter {
export class FactorioServer extends events.EventEmitter<FactorioServerEvents> {
/** Path to executable to invoke when starting the server */
executablePath?: string;
/** UDP port used for hosting the Factorio game server on */
@@ -1212,7 +1238,7 @@ export class FactorioServer extends events.EventEmitter {
// complete before the RCON connection can be used
if (!this._rconReady) {
// Not using events.once here to avoid throwing on error events.
await new Promise(resolve => this.once("rcon-ready", resolve));
await new Promise<void>(resolve => this.once("rcon-ready", resolve));
}
// The Factorio server may have decided to get ahead of us and
-1
View File
@@ -15,4 +15,3 @@ export * from "./src/subscriptions";
export { default as ExponentialBackoff } from "./src/ExponentialBackoff";
export { default as RateLimiter } from "./src/RateLimiter";
export { default as TypedEventEmitter } from "./src/TypedEventEmitter";
-1
View File
@@ -35,4 +35,3 @@ export * from "./src/rce_ops";
export { default as ExponentialBackoff } from "./src/ExponentialBackoff";
export { default as ModStore } from "./src/ModStore";
export { default as RateLimiter } from "./src/RateLimiter";
export { default as TypedEventEmitter } from "./src/TypedEventEmitter";
+4 -4
View File
@@ -1,16 +1,16 @@
import fs from "fs-extra";
import path from "path";
import events from "events";
import { Static } from "@sinclair/typebox";
import { logger } from "./logging";
import { integerModVersion, ModInfo, ModPack, ModRecord } from "./data";
import TypedEventEmitter from "./TypedEventEmitter";
import { ModInfo, ModPack } from "./data";
import { safeOutputFile } from "./file_ops";
import { Writable } from "stream";
import { ModPortalReleaseSchema, ModPortalDetailsSchema } from "./data/messages_mod";
export interface ModStoreEvents {
/** A stored mod was created, updated or deleted */
change: (mod: ModInfo) => void;
"change": [ mod: ModInfo ];
}
interface ModsInfoResponse {
@@ -37,7 +37,7 @@ interface ModPortalResponse {
results: Static<typeof ModPortalDetailsSchema>[];
}
export default class ModStore extends TypedEventEmitter<keyof ModStoreEvents, ModStoreEvents> {
export default class ModStore extends events.EventEmitter<ModStoreEvents> {
constructor(
public modsDirectory: string,
public files: Map<string, ModInfo>,
-27
View File
@@ -1,27 +0,0 @@
import EventEmitter from "events";
/**
* Type checked version of EventEmitter
*
* @example
* ```ts
* interface Events {
* "foo": (a: number, b: number) => void,
* "bar": (spam: string) => void,
* }
*
* class Emitter extends TypedEventEmitter<keyof Events, Events> {}
*
* const e = new Emitter();
* e.emit("foo", 1, 2) // type checked arguments
* ```
*/
export default abstract class TypedEventEmitter<
Keys extends string,
Events extends Record<Keys, (...args: [...any]) => void>,
> extends EventEmitter {
declare ["on"]: <E extends Keys>(event: E, listener: Events[E]) => this;
declare ["off"]: <E extends Keys>(event: E, listener: Events[E]) => this;
declare ["once"]: <E extends Keys>(event: E, listener: Events[E]) => this;
declare ["emit"]: <E extends Keys>(event: E, ...args: Parameters<Events[E]>) => boolean;
}
+3 -6
View File
@@ -1,8 +1,8 @@
// Configuration classes
import { Type, Static } from "@sinclair/typebox";
import * as fs from "fs-extra";
import events from "events";
import TypedEventEmitter from "../TypedEventEmitter";
import isDeepStrictEqual from "../is_deep_strict_equal";
import { basicType } from "../helpers";
import * as libSchema from "../schema";
@@ -152,7 +152,7 @@ function defaultValue(def: FieldDefinition) {
}
export type ConfigEvents = {
fieldChanged: (name: string, curr: FieldValue, prev: FieldValue) => void;
fieldChanged: [ name: string, curr: FieldValue, prev: FieldValue ],
}
export enum ConfigAccess {
@@ -178,10 +178,7 @@ export enum ConfigAccess {
*/
export class Config<
Fields extends { [Field in keyof Fields]: FieldValue },
> extends TypedEventEmitter<
keyof ConfigEvents,
ConfigEvents
> {
> extends events.EventEmitter<ConfigEvents> {
/**
* Mapping of config name to field meta data
* Note that mutating this object will lead to unexpected behaviour.
+8 -5
View File
@@ -1,11 +1,11 @@
import path from "path";
import fs from "fs-extra";
import { EventEmitter } from "events";
import { safeOutputFile } from "./file_ops";
import { SubscribableValue } from "./subscriptions";
import { logger } from "./logging";
import { EventEmitter } from "stream";
import { ControllerConfig } from "./config";
import path from "path";
type DatastoreKey = string | number;
type DatastoreValue = string | number | boolean | object;
@@ -137,7 +137,10 @@ export class JsonIdDatastoreProvider<
export abstract class Datastore<
K extends DatastoreKey,
V extends DatastoreValue,
> extends EventEmitter {
U = V,
> extends EventEmitter<{
"update": [ updates: U[] ],
}> {
private dirty = false;
constructor(
@@ -167,7 +170,7 @@ export abstract class Datastore<
}
// Set the dirty flag and call update handlers
protected touch(updates: any[]) {
protected touch(updates: U[]) {
if (updates.length) {
this.dirty = true;
this.emit("update", updates);
@@ -214,7 +217,7 @@ export abstract class Datastore<
export class KeyValueDatastore<
K extends DatastoreKey,
V extends DatastoreValue,
> extends Datastore<K, V> {
> extends Datastore<K, V, [K, V, isDeleted?: boolean]> {
// Set the value in the datastore, be careful of race conditions if you await any functions before calling set
set(key: K, value: V) {
this.data.set(key, value);
+44 -21
View File
@@ -1,8 +1,8 @@
// Connection adapters for Links
import { strict as assert } from "assert";
import events from "events";
import WebSocket from "./WebSocket";
import WebSocket from "./WebSocket";
import * as libData from "../data";
import * as libErrors from "../errors";
import { logger } from "../logging";
@@ -36,12 +36,21 @@ export enum ConnectionClosed {
RecoveryMode = 4001,
};
type BaseConnectorEvents = {
"message": [ message: libData.Message ],
"error": [ err: any ],
"invalidate": [],
"reset": [],
};
/**
* Base connector for links
*
* @extends events.EventEmitter
*/
export abstract class BaseConnector extends events.EventEmitter {
export abstract class BaseConnector<
E extends Record<string, any[]> | [never] = [never], // Should be "extends EventMap = DefaultEventMap"
> extends events.EventEmitter<E | BaseConnectorEvents> {
protected _seq = 1;
constructor(
@@ -53,10 +62,12 @@ export abstract class BaseConnector extends events.EventEmitter {
_reset() {
this._seq = 1;
this.emit("reset");
}
_invalidate() {
this._seq = 1;
this.emit("invalidate");
}
protected abstract send(message: libData.MessageRoutable): void;
@@ -131,6 +142,15 @@ export abstract class BaseConnector extends events.EventEmitter {
type ConnectorState = "closed" | "connecting" | "connected" | "resuming";
type WebSocketBaseConnectorEvents = {
"disconnectReady": [],
"disconnectPrepare": [],
"connect": [ data: any ],
"resume": [],
"close": [],
"drop": [],
};
export type WebSocketClusterio = WebSocket.WebSocket & {
clusterio_ignore_dump?: boolean;
};
@@ -140,7 +160,9 @@ export type WebSocketClusterio = WebSocket.WebSocket & {
*
* @extends module:lib.BaseConnector
*/
export abstract class WebSocketBaseConnector extends BaseConnector {
export abstract class WebSocketBaseConnector<
E extends Record<string, any[]> | [never] = [never], // Should be "extends EventMap = DefaultEventMap"
> extends BaseConnector<E | WebSocketBaseConnectorEvents> {
// One of closed, connecting (client only), connected and resuming.
_state: ConnectorState = "closed";
_closing = false;
@@ -166,7 +188,7 @@ export abstract class WebSocketBaseConnector extends BaseConnector {
_invalidate() {
this._state = "connecting";
assert(this._closing === false);
assert(this._socket);
// We do not care about the socket state
this._lastHeartbeatMs = null;
assert(this._heartbeatId === null);
this._heartbeatInterval = null;
@@ -175,6 +197,11 @@ export abstract class WebSocketBaseConnector extends BaseConnector {
super._invalidate();
}
_close() {
this._reset();
this.emit("close");
}
_check(...expectedStates: ConnectorState[]) {
if (!expectedStates.includes(this._state)) {
throw new Error(
@@ -385,12 +412,16 @@ export abstract class WebSocketBaseConnector extends BaseConnector {
}
}
type WebSocketClientConnectorEvents = {
"hello": [ data: any ],
};
/**
* Connector for controller clients
*
* @extends module:lib.WebSocketBaseConnector
*/
export abstract class WebSocketClientConnector extends WebSocketBaseConnector {
export abstract class WebSocketClientConnector extends WebSocketBaseConnector<WebSocketClientConnectorEvents> {
_reconnectId?: ReturnType<typeof setTimeout>;
_sessionToken: string | null = null;
_sessionTimeout: number | null = null;
@@ -418,8 +449,8 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector {
this._sessionToken = null;
this._sessionTimeout = null;
this._startedResumingMs = null;
super._reset();
this.src = undefined as any;
super._reset();
}
_invalidate() {
@@ -427,8 +458,8 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector {
this._sessionToken = null;
this._sessionTimeout = null;
this._startedResumingMs = null;
super._invalidate();
this.src = undefined as any;
super._invalidate();
}
/**
@@ -463,8 +494,7 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector {
await events.once(this, "close");
} else {
this._reset();
this.emit("close");
this._close();
}
}
@@ -489,8 +519,7 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector {
this._socket!.close(ConnectionClosed.NormalClosure, "Connector closing");
} else {
this._reset();
this.emit("close");
this._close();
}
}
@@ -546,13 +575,10 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector {
&& this._startedResumingMs! + this._sessionTimeout! * 1000 < Date.now() + delayMs
) {
logger.error("Connector | Session timed out trying to resume");
this._reset();
if (this._closing) {
this._reset();
this.emit("close");
this._close();
} else {
this._state = "connecting";
this.emit("invalidate");
this._invalidate();
}
}
logger.verbose(
@@ -604,8 +630,7 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector {
} else {
// eslint-disable-next-line no-lonely-if
if (this._closing) {
this._reset();
this.emit("close");
this._close();
} else {
this.reconnect();
@@ -711,11 +736,9 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector {
} else if (type === "invalidate") {
logger.warn("Connector | session invalidated by controller");
if (this._closing) {
this._reset();
this.emit("close");
this._close();
} else {
this._invalidate();
this.emit("invalidate");
this.register();
}
}
+3 -3
View File
@@ -113,7 +113,7 @@ export class Link {
// Process messages received by the connector
connector.on("message", payload => {
try {
this._processMessage(payload);
this._processMessage(payload as MessageRoutable);
} catch (err) {
if (err instanceof libErrors.InvalidMessage) {
logger.error(err.message);
@@ -128,10 +128,10 @@ export class Link {
}
});
// The events below exist only on websocket connectors, we can't early return because of test mocking
const webSocketConnector = this.connector as WebSocketBaseConnector;
connector.on("disconnectPrepare", () => {
this.prepareDisconnect().finally(() => {
// Only WebSocket connectors issue disconnection events.
const webSocketConnector = this.connector as WebSocketBaseConnector;
if (webSocketConnector.hasSession) {
webSocketConnector.sendDisconnectReady();
}
+4 -2
View File
@@ -222,10 +222,12 @@ export class EventSubscriber<
public control: Link,
) {
control.handle(Event, this._handleEvent.bind(this));
control.connector.on("connect", () => {
// The events below exist only on websocket connectors, we can't early return because of test mocking
const webSocketConnector = control.connector as WebSocketBaseConnector;
webSocketConnector.on("connect", () => {
this.handleConnectionEvent("connect");
});
control.connector.on("close", () => {
webSocketConnector.on("close", () => {
this.handleConnectionEvent("close");
});
}
+5 -1
View File
@@ -1,7 +1,11 @@
import { useEffect, useState } from "react";
import { EventEmitter } from "events";
class Permission extends EventEmitter {
type PermissionEvents = {
"change": [ status: string ]
};
class Permission extends EventEmitter<PermissionEvents> {
status!: { state: string };
constructor(name: PermissionName) {