mirror of
https://github.com/clusterio/clusterio.git
synced 2026-11-11 15:04:52 +09:00
Convert subspace_storage to entrypoint functions
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
1ca0c658b9
commit
60dbb461e7
5 files changed
+363
-277
No files matched your search
@@ -1,4 +1,4 @@
|
||||
import { BaseControllerPlugin, type ControlConnection } from "@clusterio/controller";
|
||||
import type { ControlConnection, ControllerPluginContext } from "@clusterio/controller";
|
||||
|
||||
import fs from "node:fs/promises";
|
||||
import path from "path";
|
||||
@@ -70,58 +70,21 @@ async function saveDatabase(
|
||||
}
|
||||
}
|
||||
|
||||
export class ControllerPlugin extends BaseControllerPlugin {
|
||||
items!: lib.ItemDatabase;
|
||||
itemUpdateRateLimiter!: lib.RateLimiter;
|
||||
itemsLastUpdate: Map<string, lib.ItemCountWithQuality> = new Map();
|
||||
subscribedControlLinks!: Set<ControlConnection>;
|
||||
doleMagicId!: ReturnType<typeof setInterval>;
|
||||
neuralDole!: dole.NeuralDole;
|
||||
storageDirty = false;
|
||||
export default async function(context: ControllerPluginContext) {
|
||||
const { controller, logger, plugin } = context;
|
||||
|
||||
async init() {
|
||||
this.items = await loadDatabase(this.controller.config, this.logger);
|
||||
this.itemUpdateRateLimiter = new lib.RateLimiter({
|
||||
maxRate: 1,
|
||||
action: () => {
|
||||
try {
|
||||
this.broadcastStorage();
|
||||
} catch (err: any) {
|
||||
this.logger.error(`Unexpected error sending storage update:\n${err.stack}`);
|
||||
}
|
||||
},
|
||||
});
|
||||
this.itemsLastUpdate = new Map();
|
||||
for (const [name, qualities] of this.items.getEntries()) {
|
||||
this.itemsLastUpdate.set(name, { ...qualities });
|
||||
}
|
||||
|
||||
this.neuralDole = new dole.NeuralDole({ items: this.items });
|
||||
this.doleMagicId = setInterval(() => {
|
||||
if (this.controller.config.get("subspace_storage.division_method") === "neural_dole") {
|
||||
this.neuralDole.doMagic();
|
||||
}
|
||||
}, 1000);
|
||||
|
||||
this.subscribedControlLinks = new Set();
|
||||
|
||||
routes.addApiRoutes(this.controller.app, this.items);
|
||||
|
||||
this.controller.handle(GetStorageRequest, this.handleGetStorageRequest.bind(this));
|
||||
this.controller.handle(PlaceEvent, this.handlePlaceEvent.bind(this));
|
||||
this.controller.handle(RemoveRequest, this.handleRemoveRequest.bind(this));
|
||||
this.controller.handle(SetStorageSubscriptionRequest, this.handleSetStorageSubscriptionRequest.bind(this));
|
||||
const items = await loadDatabase(controller.config, logger);
|
||||
let storageDirty = false;
|
||||
let itemsLastUpdate = new Map<string, lib.ItemCountWithQuality>();
|
||||
for (const [name, qualities] of items.getEntries()) {
|
||||
itemsLastUpdate.set(name, { ...qualities });
|
||||
}
|
||||
const subscribedControlLinks = new Set<ControlConnection>();
|
||||
|
||||
updateStorage() {
|
||||
this.itemUpdateRateLimiter.activate();
|
||||
this.storageDirty = true;
|
||||
}
|
||||
|
||||
broadcastStorage() {
|
||||
function broadcastStorage() {
|
||||
let itemsToUpdate: Item[] = [];
|
||||
for (const [name, qualities] of this.items.getEntries()) {
|
||||
const lastQualities = this.itemsLastUpdate.get(name);
|
||||
for (const [name, qualities] of items.getEntries()) {
|
||||
const lastQualities = itemsLastUpdate.get(name);
|
||||
for (const [quality, count] of Object.entries(qualities)) {
|
||||
if (!lastQualities || lastQualities[quality] !== count) {
|
||||
itemsToUpdate.push(new Item(name, count, quality));
|
||||
@@ -134,68 +97,93 @@ export class ControllerPlugin extends BaseControllerPlugin {
|
||||
}
|
||||
|
||||
let update = new UpdateStorageEvent(itemsToUpdate);
|
||||
this.controller.sendTo("allInstances", update);
|
||||
for (let link of this.subscribedControlLinks) {
|
||||
controller.sendTo("allInstances", update);
|
||||
for (let link of subscribedControlLinks) {
|
||||
link.send(update);
|
||||
}
|
||||
|
||||
this.itemsLastUpdate = new Map();
|
||||
for (const [name, qualities] of this.items.getEntries()) {
|
||||
this.itemsLastUpdate.set(name, { ...qualities });
|
||||
itemsLastUpdate = new Map();
|
||||
for (const [name, qualities] of items.getEntries()) {
|
||||
itemsLastUpdate.set(name, { ...qualities });
|
||||
}
|
||||
}
|
||||
|
||||
async handleGetStorageRequest() {
|
||||
const itemUpdateRateLimiter = new lib.RateLimiter({
|
||||
maxRate: 1,
|
||||
action: () => {
|
||||
try {
|
||||
broadcastStorage();
|
||||
} catch (err: any) {
|
||||
logger.error(`Unexpected error sending storage update:\n${err.stack}`);
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
function updateStorage() {
|
||||
itemUpdateRateLimiter.activate();
|
||||
storageDirty = true;
|
||||
}
|
||||
|
||||
const neuralDole = new dole.NeuralDole({ items });
|
||||
const doleMagicId = setInterval(() => {
|
||||
if (controller.config.get("subspace_storage.division_method") === "neural_dole") {
|
||||
neuralDole.doMagic();
|
||||
}
|
||||
}, 1000);
|
||||
|
||||
routes.addApiRoutes(controller.app, items);
|
||||
|
||||
controller.handle(GetStorageRequest, async () => {
|
||||
const result: Item[] = [];
|
||||
for (const [name, qualities] of this.items.getEntries()) {
|
||||
for (const [name, qualities] of items.getEntries()) {
|
||||
for (const [quality, count] of Object.entries(qualities)) {
|
||||
result.push(new Item(name, count, quality));
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
});
|
||||
|
||||
async handlePlaceEvent(request: PlaceEvent, src: lib.Address) {
|
||||
controller.handle(PlaceEvent, async (request: PlaceEvent, src: lib.Address) => {
|
||||
let instanceId = src.id;
|
||||
|
||||
for (let item of request.items) {
|
||||
this.items.addItem(item.name, item.count, item.quality || "normal");
|
||||
items.addItem(item.name, item.count, item.quality || "normal");
|
||||
exportCounter.labels(String(instanceId), item.name, item.quality || "normal").inc(item.count);
|
||||
}
|
||||
|
||||
this.updateStorage();
|
||||
updateStorage();
|
||||
|
||||
if (this.controller.config.get("subspace_storage.log_item_transfers")) {
|
||||
this.logger.verbose(
|
||||
if (controller.config.get("subspace_storage.log_item_transfers")) {
|
||||
logger.verbose(
|
||||
`Imported the following from ${instanceId}:\n${JSON.stringify(request.items)}`
|
||||
);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
async handleRemoveRequest(request: RemoveRequest, src: lib.Address) {
|
||||
let method = this.controller.config.get("subspace_storage.division_method");
|
||||
controller.handle(RemoveRequest, async (request: RemoveRequest, src: lib.Address) => {
|
||||
let method = controller.config.get("subspace_storage.division_method");
|
||||
let instanceId = src.id;
|
||||
|
||||
let itemsRemoved = [];
|
||||
if (method === "simple") {
|
||||
for (let item of request.items) {
|
||||
const quality = item.quality || "normal";
|
||||
let count = this.items.getItemCount(item.name, quality);
|
||||
let count = items.getItemCount(item.name, quality);
|
||||
let toRemove = Math.min(count, item.count);
|
||||
if (toRemove > 0) {
|
||||
this.items.removeItem(item.name, toRemove, quality);
|
||||
items.removeItem(item.name, toRemove, quality);
|
||||
itemsRemoved.push(new Item(item.name, toRemove, quality));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let instance = this.controller.instances.get(instanceId);
|
||||
let instance = controller.instances.get(instanceId);
|
||||
let instanceName = instance ? instance.config.get("instance.name") : "unknown";
|
||||
|
||||
// use fancy neural net to calculate a "fair" dole division rate.
|
||||
if (method === "neural_dole") {
|
||||
for (let item of request.items) {
|
||||
const quality = item.quality || "normal";
|
||||
let count = this.neuralDole.divider(
|
||||
let count = neuralDole.divider(
|
||||
{ name: item.name, quality, count: item.count, instanceId, instanceName }
|
||||
);
|
||||
if (count > 0) {
|
||||
@@ -209,9 +197,9 @@ export class ControllerPlugin extends BaseControllerPlugin {
|
||||
const quality = item.quality || "normal";
|
||||
let count = dole.doleDivider({
|
||||
object: { name: item.name, quality, count: item.count, instanceId, instanceName },
|
||||
items: this.items,
|
||||
logItemTransfers: this.controller.config.get("subspace_storage.log_item_transfers"),
|
||||
logger: this.logger,
|
||||
items,
|
||||
logItemTransfers: controller.config.get("subspace_storage.log_item_transfers"),
|
||||
logger,
|
||||
});
|
||||
if (count > 0) {
|
||||
itemsRemoved.push(new Item(item.name, count, quality));
|
||||
@@ -229,50 +217,51 @@ export class ControllerPlugin extends BaseControllerPlugin {
|
||||
importCounter.labels(String(instanceId), item.name, item.quality || "normal").inc(item.count);
|
||||
}
|
||||
|
||||
this.updateStorage();
|
||||
updateStorage();
|
||||
|
||||
if (itemsRemoved.length && this.controller.config.get("subspace_storage.log_item_transfers")) {
|
||||
this.logger.verbose(`Exported the following to ${instanceId}:\n${JSON.stringify(itemsRemoved)}`);
|
||||
if (itemsRemoved.length && controller.config.get("subspace_storage.log_item_transfers")) {
|
||||
logger.verbose(`Exported the following to ${instanceId}:\n${JSON.stringify(itemsRemoved)}`);
|
||||
}
|
||||
}
|
||||
|
||||
return itemsRemoved;
|
||||
}
|
||||
});
|
||||
|
||||
async handleSetStorageSubscriptionRequest(request: SetStorageSubscriptionRequest, src: lib.Address) {
|
||||
let link = this.controller.wsServer.controlConnections.get(src.id)!;
|
||||
controller.handle(SetStorageSubscriptionRequest, async (
|
||||
request: SetStorageSubscriptionRequest,
|
||||
src: lib.Address,
|
||||
) => {
|
||||
let link = controller.wsServer.controlConnections.get(src.id)!;
|
||||
if (request.storage) {
|
||||
this.subscribedControlLinks.add(link);
|
||||
subscribedControlLinks.add(link);
|
||||
} else {
|
||||
this.subscribedControlLinks.delete(link);
|
||||
subscribedControlLinks.delete(link);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
onControlConnectionEvent(connection: ControlConnection, event: string) {
|
||||
controller.hooks.controlConnectionEvent.attach(plugin.name, (connection, event) => {
|
||||
if (event === "close") {
|
||||
this.subscribedControlLinks.delete(connection);
|
||||
subscribedControlLinks.delete(connection);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
async onMetrics() {
|
||||
if (this.items) {
|
||||
for (const [name, qualities] of this.items.getEntries()) {
|
||||
for (const [quality, count] of Object.entries(qualities)) {
|
||||
controllerInventoryGauge.labels(name, quality).set(Number(count) || 0);
|
||||
}
|
||||
controller.hooks.metrics.attach(plugin.name, async () => {
|
||||
for (const [name, qualities] of items.getEntries()) {
|
||||
for (const [quality, count] of Object.entries(qualities)) {
|
||||
controllerInventoryGauge.labels(name, quality).set(Number(count) || 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
async onShutdown() {
|
||||
this.itemUpdateRateLimiter.cancel();
|
||||
clearInterval(this.doleMagicId);
|
||||
}
|
||||
controller.hooks.shutdown.attach(plugin.name, async () => {
|
||||
itemUpdateRateLimiter.cancel();
|
||||
clearInterval(doleMagicId);
|
||||
});
|
||||
|
||||
async onSaveData() {
|
||||
if (this.storageDirty) {
|
||||
this.storageDirty = false;
|
||||
await saveDatabase(this.controller.config, this.items, this.logger);
|
||||
controller.hooks.save.attach(plugin.name, async () => {
|
||||
if (storageDirty) {
|
||||
storageDirty = false;
|
||||
await saveDatabase(controller.config, items, logger);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import * as lib from "@clusterio/lib";
|
||||
import { BaseInstancePlugin } from "@clusterio/host";
|
||||
import type { InstancePluginContext } from "@clusterio/host";
|
||||
|
||||
import {
|
||||
PlaceEvent,
|
||||
@@ -11,127 +11,124 @@ import {
|
||||
|
||||
type IpcItems = [name: string, count: number, quality: string][];
|
||||
|
||||
export class InstancePlugin extends BaseInstancePlugin {
|
||||
pendingTasks!: Set<any>;
|
||||
pingId?: ReturnType<typeof setTimeout>;
|
||||
timeUpdateId?: ReturnType<typeof setInterval>;
|
||||
cachedInventoryItems: Item[] = [];
|
||||
export default async function(context: InstancePluginContext) {
|
||||
const { instance, host, logger, plugin } = context;
|
||||
const pendingTasks = new Set<Promise<unknown>>();
|
||||
let pingId: ReturnType<typeof setInterval> | undefined;
|
||||
let timeUpdateId: ReturnType<typeof setInterval> | undefined;
|
||||
let cachedInventoryItems: Item[] = [];
|
||||
|
||||
unexpectedError(err: Error) {
|
||||
this.logger.error(`Unexpected error:\n${err.stack}`);
|
||||
}
|
||||
|
||||
async init() {
|
||||
this.pendingTasks = new Set();
|
||||
this.instance.server.on("ipc-subspace_storage:output", (output: IpcItems) => {
|
||||
this.logger.info("Received output items:");
|
||||
this.logger.info(JSON.stringify(output));
|
||||
this.provideItems(output).catch(err => this.unexpectedError(err));
|
||||
});
|
||||
this.instance.server.on("ipc-subspace_storage:orders", (orders: IpcItems) => {
|
||||
if (this.instance.status !== "running" || !this.host.connected) {
|
||||
return;
|
||||
}
|
||||
|
||||
let task = this.requestItems(orders).catch(err => this.unexpectedError(err));
|
||||
this.pendingTasks.add(task);
|
||||
task.finally(() => { this.pendingTasks.delete(task); });
|
||||
});
|
||||
|
||||
this.instance.handle(UpdateStorageEvent, this.handleUpdateStorageEvent.bind(this));
|
||||
}
|
||||
|
||||
async onStart() {
|
||||
this.pingId = setInterval(() => {
|
||||
if (!this.host.connected) {
|
||||
return; // Only ping if we are actually connected to the controller.
|
||||
}
|
||||
this.sendRcon(
|
||||
"/sc __subspace_storage__ global.ticksSinceMasterPinged = 0", true
|
||||
).catch(err => this.unexpectedError(err));
|
||||
}, 5000);
|
||||
|
||||
let items = await this.instance.sendTo("controller", new GetStorageRequest());
|
||||
|
||||
// Cache items for periodic time updates
|
||||
this.cachedInventoryItems = items;
|
||||
|
||||
// Ensure a payload with updated time is sent every second
|
||||
this.timeUpdateId = setInterval(() => {
|
||||
if (this.instance.status !== "running") {
|
||||
return;
|
||||
}
|
||||
|
||||
const payloadItems = [
|
||||
...this.cachedInventoryItems,
|
||||
new Item("signal-unixtime", Math.floor(Date.now() / 1000), "normal"),
|
||||
];
|
||||
const payloadJson = lib.escapeString(JSON.stringify(payloadItems));
|
||||
const task = this.sendRcon(`/sc __subspace_storage__ UpdateInvData("${payloadJson}")`, true)
|
||||
.catch(err => this.unexpectedError(err));
|
||||
this.pendingTasks.add(task);
|
||||
task.finally(() => { this.pendingTasks.delete(task); });
|
||||
}, 1000);
|
||||
}
|
||||
|
||||
async onStop() {
|
||||
clearInterval(this.pingId);
|
||||
clearInterval(this.timeUpdateId);
|
||||
await Promise.all(this.pendingTasks);
|
||||
}
|
||||
|
||||
onExit() {
|
||||
clearInterval(this.pingId);
|
||||
clearInterval(this.timeUpdateId);
|
||||
function unexpectedError(err: Error) {
|
||||
logger.error(`Unexpected error:\n${err.stack}`);
|
||||
}
|
||||
|
||||
// provide items --------------------------------------------------------------
|
||||
async provideItems(items: IpcItems) {
|
||||
if (!this.host.connector.hasSession) {
|
||||
async function provideItems(items: IpcItems) {
|
||||
if (!host.connector.hasSession) {
|
||||
// For now the items are voided if the controller connection is
|
||||
// down, which is no different from the previous behaviour.
|
||||
if (this.instance.config.get("subspace_storage.log_item_transfers")) {
|
||||
this.logger.verbose("Voided the following items:");
|
||||
this.logger.verbose(JSON.stringify(items));
|
||||
if (instance.config.get("subspace_storage.log_item_transfers")) {
|
||||
logger.verbose("Voided the following items:");
|
||||
logger.verbose(JSON.stringify(items));
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const fromIpcItems = items.map(item => new Item(item[0], item[1], item[2]));
|
||||
this.instance.sendTo("controller", new PlaceEvent(fromIpcItems));
|
||||
instance.sendTo("controller", new PlaceEvent(fromIpcItems));
|
||||
|
||||
if (this.instance.config.get("subspace_storage.log_item_transfers")) {
|
||||
this.logger.verbose("Exported the following to controller:");
|
||||
this.logger.verbose(JSON.stringify(items));
|
||||
if (instance.config.get("subspace_storage.log_item_transfers")) {
|
||||
logger.verbose("Exported the following to controller:");
|
||||
logger.verbose(JSON.stringify(items));
|
||||
}
|
||||
}
|
||||
|
||||
// request items --------------------------------------------------------------
|
||||
async requestItems(requestItems: IpcItems) {
|
||||
this.logger.info(`Requesting items: ${JSON.stringify(requestItems)}`);
|
||||
async function requestItems(orders: IpcItems) {
|
||||
logger.info(`Requesting items: ${JSON.stringify(orders)}`);
|
||||
// Request the items all at once
|
||||
const fromIpcItems = requestItems.map(item => new Item(item[0], item[1], item[2]));
|
||||
let items = await this.instance.sendTo("controller", new RemoveRequest(fromIpcItems));
|
||||
const fromIpcItems = orders.map(item => new Item(item[0], item[1], item[2]));
|
||||
let items = await instance.sendTo("controller", new RemoveRequest(fromIpcItems));
|
||||
|
||||
if (!items.length) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.instance.config.get("subspace_storage.log_item_transfers")) {
|
||||
this.logger.verbose("Imported following from controller:");
|
||||
this.logger.verbose(JSON.stringify(items));
|
||||
if (instance.config.get("subspace_storage.log_item_transfers")) {
|
||||
logger.verbose("Imported following from controller:");
|
||||
logger.verbose(JSON.stringify(items));
|
||||
}
|
||||
|
||||
let itemsJson = lib.escapeString(JSON.stringify(items));
|
||||
await this.sendRcon(`/sc __subspace_storage__ Import("${itemsJson}")`, true);
|
||||
await instance.sendRcon(`/sc __subspace_storage__ Import("${itemsJson}")`, true, plugin.name);
|
||||
}
|
||||
|
||||
// combinator signals ---------------------------------------------------------
|
||||
async handleUpdateStorageEvent(event: UpdateStorageEvent) {
|
||||
if (this.instance.status !== "running") {
|
||||
instance.server.on("ipc-subspace_storage:output", (output: IpcItems) => {
|
||||
logger.info("Received output items:");
|
||||
logger.info(JSON.stringify(output));
|
||||
provideItems(output).catch(err => unexpectedError(err));
|
||||
});
|
||||
instance.server.on("ipc-subspace_storage:orders", (orders: IpcItems) => {
|
||||
if (instance.status !== "running" || !host.connected) {
|
||||
return;
|
||||
}
|
||||
// Cache latest inventory from controller
|
||||
this.cachedInventoryItems = event.items;
|
||||
}
|
||||
|
||||
let task = requestItems(orders).catch(err => unexpectedError(err));
|
||||
pendingTasks.add(task);
|
||||
task.finally(() => { pendingTasks.delete(task); });
|
||||
});
|
||||
|
||||
// combinator signals ---------------------------------------------------------
|
||||
instance.handle(UpdateStorageEvent, async (event: UpdateStorageEvent) => {
|
||||
if (instance.status !== "running") {
|
||||
return;
|
||||
}
|
||||
// Cache latest inventory from controller
|
||||
cachedInventoryItems = event.items;
|
||||
});
|
||||
|
||||
instance.hooks.start.attach(plugin.name, async () => {
|
||||
pingId = setInterval(() => {
|
||||
if (!host.connected) {
|
||||
return; // Only ping if we are actually connected to the controller.
|
||||
}
|
||||
instance.sendRcon(
|
||||
"/sc __subspace_storage__ global.ticksSinceMasterPinged = 0", true, plugin.name
|
||||
).catch(err => unexpectedError(err));
|
||||
}, 5000);
|
||||
|
||||
let items = await instance.sendTo("controller", new GetStorageRequest());
|
||||
|
||||
// Cache items for periodic time updates
|
||||
cachedInventoryItems = items;
|
||||
|
||||
// Ensure a payload with updated time is sent every second
|
||||
timeUpdateId = setInterval(() => {
|
||||
if (instance.status !== "running") {
|
||||
return;
|
||||
}
|
||||
|
||||
const payloadItems = [
|
||||
...cachedInventoryItems,
|
||||
new Item("signal-unixtime", Math.floor(Date.now() / 1000), "normal"),
|
||||
];
|
||||
const payloadJson = lib.escapeString(JSON.stringify(payloadItems));
|
||||
const task = instance.sendRcon(
|
||||
`/sc __subspace_storage__ UpdateInvData("${payloadJson}")`, true, plugin.name
|
||||
).catch(err => unexpectedError(err));
|
||||
pendingTasks.add(task);
|
||||
task.finally(() => { pendingTasks.delete(task); });
|
||||
}, 1000);
|
||||
});
|
||||
|
||||
instance.hooks.stop.attach(plugin.name, async () => {
|
||||
clearInterval(pingId);
|
||||
clearInterval(timeUpdateId);
|
||||
await Promise.all(pendingTasks);
|
||||
});
|
||||
|
||||
instance.hooks.exit.attach(plugin.name, () => {
|
||||
clearInterval(pingId);
|
||||
clearInterval(timeUpdateId);
|
||||
});
|
||||
}
|
||||
@@ -1,7 +1,5 @@
|
||||
// These are libraries that does not have any units test, they are
|
||||
// required here in order to track coverage for them.
|
||||
|
||||
import "../dist/node/instance.js";
|
||||
import "../dist/node/controller.js";
|
||||
import "../dist/node/dole.js";
|
||||
import "../dist/node/dole_nn_base.js";
|
||||
@@ -0,0 +1,95 @@
|
||||
import assert from "node:assert/strict";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
|
||||
import * as lib from "@clusterio/lib";
|
||||
import * as mock from "../../../test/mock.js";
|
||||
import controllerEntrypoint from "../dist/node/controller.js";
|
||||
import instanceEntrypoint from "../dist/node/instance.js";
|
||||
import { plugin as info } from "../dist/node/index.js";
|
||||
import {
|
||||
Item, PlaceEvent, RemoveRequest, GetStorageRequest, UpdateStorageEvent,
|
||||
} from "../dist/node/messages.js";
|
||||
|
||||
|
||||
describe("subspace_storage plugin", function() {
|
||||
before(function() {
|
||||
try {
|
||||
lib.registerPluginMessages([info]);
|
||||
} catch (err) {
|
||||
// Already registered by the full test suite
|
||||
}
|
||||
});
|
||||
|
||||
describe("controller entrypoint", function() {
|
||||
const databaseDir = path.join("temp", "test", "subspace_storage");
|
||||
const src = lib.Address.fromShorthand({ instanceId: 7357 });
|
||||
let controller;
|
||||
let sent;
|
||||
before(async function() {
|
||||
await fs.rm(databaseDir, { recursive: true, force: true });
|
||||
controller = await mock.loadControllerPlugin(async (context) => {
|
||||
context.controller.mockConfigEntries.set("controller.database_directory", databaseDir);
|
||||
context.controller.mockConfigEntries.set("subspace_storage.division_method", "simple");
|
||||
context.controller.mockConfigEntries.set("subspace_storage.log_item_transfers", false);
|
||||
sent = [];
|
||||
context.controller.sendTo = (dst, event) => { sent.push(event); };
|
||||
await controllerEntrypoint(context);
|
||||
}, info);
|
||||
});
|
||||
after(async function() {
|
||||
await controller.hooks.shutdown.invoke();
|
||||
});
|
||||
|
||||
it("should store placed items and broadcast the update", async function() {
|
||||
await mock.getHandler(controller, PlaceEvent)(
|
||||
new PlaceEvent([new Item("iron-plate", 20, "normal")]), src
|
||||
);
|
||||
const storage = await mock.getHandler(controller, GetStorageRequest)();
|
||||
assert.deepEqual(storage, [new Item("iron-plate", 20, "normal")]);
|
||||
assert(sent[0] instanceof UpdateStorageEvent);
|
||||
});
|
||||
it("should remove requested items up to the stored amount", async function() {
|
||||
const removed = await mock.getHandler(controller, RemoveRequest)(
|
||||
new RemoveRequest([new Item("iron-plate", 15, "normal"), new Item("copper-plate", 5, "normal")]), src
|
||||
);
|
||||
assert.deepEqual(removed, [new Item("iron-plate", 15, "normal")]);
|
||||
});
|
||||
it("should update the inventory gauge on metrics", async function() {
|
||||
for (const result of await controller.hooks.metrics.collect()) {
|
||||
for await (const _ of result) { /* drain */ }
|
||||
}
|
||||
const gauge = lib.defaultRegistry.collectors.find(
|
||||
collector => collector.metric?.name === "clusterio_subspace_storage_controller_inventory"
|
||||
);
|
||||
assert.equal(gauge.labels("iron-plate", "normal").get(), 5);
|
||||
});
|
||||
it("should save the item database", async function() {
|
||||
await controller.hooks.save.invoke();
|
||||
const content = JSON.parse(await fs.readFile(path.join(databaseDir, "items.json"), "utf8"));
|
||||
assert.deepEqual(content, { "iron-plate": { normal: 5 } });
|
||||
});
|
||||
});
|
||||
|
||||
describe("instance entrypoint", function() {
|
||||
let instance;
|
||||
before(async function() {
|
||||
({ instance } = await mock.loadInstancePlugin(instanceEntrypoint, info));
|
||||
instance.mockConfigEntries.set("subspace_storage.log_item_transfers", false);
|
||||
});
|
||||
|
||||
it("should send output items to the controller", async function() {
|
||||
instance.server.emit("ipc-subspace_storage:output", [["iron-plate", 10, "normal"]]);
|
||||
await new Promise(resolve => setImmediate(resolve));
|
||||
const message = instance.connector.sentMessages.find(m => m.data instanceof PlaceEvent);
|
||||
assert(message, "PlaceEvent was not sent");
|
||||
});
|
||||
it("should ignore orders while not running", async function() {
|
||||
instance.status = "starting";
|
||||
instance.server.emit("ipc-subspace_storage:orders", [["iron-plate", 10, "normal"]]);
|
||||
await new Promise(resolve => setImmediate(resolve));
|
||||
const message = instance.connector.sentMessages.find(m => m.data instanceof RemoveRequest);
|
||||
assert.equal(message, undefined);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -1,41 +1,99 @@
|
||||
import { useContext, useEffect, useState } from "react";
|
||||
import { useEffect, useState } from "react";
|
||||
import { Table } from "antd";
|
||||
|
||||
import {
|
||||
BaseWebPlugin, PageLayout, PageHeader, Control, ControlContext, notifyErrorHandler,
|
||||
PageLayout, PageHeader, Control, notifyErrorHandler,
|
||||
useExportLocale, useExportPrototypeMetadata, useDefaultModPack, FactorioIcon,
|
||||
useTableQueryState, useColumnSearch,
|
||||
useTableQueryState, useColumnSearch, type WebPluginContext,
|
||||
} from "@clusterio/web_ui";
|
||||
import { GetStorageRequest, Item, SetStorageSubscriptionRequest, UpdateStorageEvent } from "../messages.js";
|
||||
|
||||
import "./style.css";
|
||||
|
||||
|
||||
function useStorage(control: Control) {
|
||||
let plugin = control.plugins.get("subspace_storage") as WebPlugin;
|
||||
let [storage, setStorage] = useState([...plugin.storage]);
|
||||
class StorageSubscriber {
|
||||
storage = new Map<string, Item>();
|
||||
callbacks: (() => void)[] = [];
|
||||
|
||||
constructor(
|
||||
private control: Control,
|
||||
) { }
|
||||
|
||||
onUpdate(callback: () => void) {
|
||||
this.callbacks.push(callback);
|
||||
if (this.callbacks.length) {
|
||||
this.updateSubscription();
|
||||
}
|
||||
}
|
||||
|
||||
offUpdate(callback: () => void) {
|
||||
let index = this.callbacks.lastIndexOf(callback);
|
||||
if (index === -1) {
|
||||
throw new Error("callback is not registered");
|
||||
}
|
||||
|
||||
this.callbacks.splice(index, 1);
|
||||
if (!this.callbacks.length) {
|
||||
this.updateSubscription();
|
||||
}
|
||||
}
|
||||
|
||||
updateStorage(items: Item[]) {
|
||||
for (let item of items) {
|
||||
this.storage.set(`${item.name}:${item.quality}`, item);
|
||||
}
|
||||
for (let callback of this.callbacks) {
|
||||
callback();
|
||||
}
|
||||
}
|
||||
|
||||
updateSubscription() {
|
||||
if (!this.control.connector.connected) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.control.send(
|
||||
new SetStorageSubscriptionRequest(Boolean(this.callbacks.length))
|
||||
).catch(notifyErrorHandler("Error subscribing to storage"));
|
||||
|
||||
if (this.callbacks.length) {
|
||||
this.control.send(new GetStorageRequest()).then(
|
||||
items => {
|
||||
this.updateStorage(items);
|
||||
}
|
||||
).catch(notifyErrorHandler("Error updating storage"));
|
||||
} else {
|
||||
this.storage.clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Created when the plugin is loaded, before any page is rendered
|
||||
let storageSubscriber: StorageSubscriber;
|
||||
|
||||
function useStorage() {
|
||||
let [storage, setStorage] = useState([...storageSubscriber.storage]);
|
||||
|
||||
useEffect(() => {
|
||||
function update() {
|
||||
setStorage([...plugin.storage]);
|
||||
setStorage([...storageSubscriber.storage]);
|
||||
}
|
||||
|
||||
plugin.onUpdate(update);
|
||||
storageSubscriber.onUpdate(update);
|
||||
return () => {
|
||||
plugin.offUpdate(update);
|
||||
storageSubscriber.offUpdate(update);
|
||||
};
|
||||
}, []);
|
||||
return storage;
|
||||
}
|
||||
|
||||
function StoragePage() {
|
||||
let control = useContext(ControlContext);
|
||||
const modPack = useDefaultModPack();
|
||||
let locale = useExportLocale(modPack);
|
||||
const prototypes = useExportPrototypeMetadata(modPack);
|
||||
let itemMetadata = prototypes?.get("item");
|
||||
let fluidMetadata = prototypes?.get("fluid");
|
||||
let storage = useStorage(control);
|
||||
let storage = useStorage();
|
||||
const tableState = useTableQueryState<[string, Item]>({
|
||||
namespace: "storage", defaultSortKey: "quantity", defaultSortOrder: "descend", pagination: false,
|
||||
});
|
||||
@@ -136,77 +194,26 @@ function StoragePage() {
|
||||
</PageLayout>;
|
||||
}
|
||||
|
||||
export class WebPlugin extends BaseWebPlugin {
|
||||
storage = new Map<string, Item>();
|
||||
callbacks: (() => void)[] = [];
|
||||
export default async function(context: WebPluginContext) {
|
||||
const { control, plugin } = context;
|
||||
storageSubscriber = new StorageSubscriber(control);
|
||||
|
||||
async init() {
|
||||
this.pages = [
|
||||
{
|
||||
path: "/storage",
|
||||
sidebarName: "Storage",
|
||||
permission: "subspace_storage.storage.view",
|
||||
content: <StoragePage/>,
|
||||
},
|
||||
];
|
||||
this.control.handle(UpdateStorageEvent, this.handleUpdateStorageEvent.bind(this));
|
||||
}
|
||||
control.hooks.pages.attach(plugin.name, () => [
|
||||
{
|
||||
path: "/storage",
|
||||
sidebarName: "Storage",
|
||||
permission: "subspace_storage.storage.view",
|
||||
content: <StoragePage/>,
|
||||
},
|
||||
]);
|
||||
|
||||
onControllerConnectionEvent(event: "connect" | "drop" | "resume" | "close") {
|
||||
control.handle(UpdateStorageEvent, async (event: UpdateStorageEvent) => {
|
||||
storageSubscriber.updateStorage(event.items);
|
||||
});
|
||||
|
||||
control.hooks.controllerConnectionEvent.attach(plugin.name, (event) => {
|
||||
if (event === "connect") {
|
||||
this.updateSubscription();
|
||||
storageSubscriber.updateSubscription();
|
||||
}
|
||||
}
|
||||
|
||||
async handleUpdateStorageEvent(event: UpdateStorageEvent) {
|
||||
this.updateStorage(event.items);
|
||||
}
|
||||
|
||||
onUpdate(callback: () => void) {
|
||||
this.callbacks.push(callback);
|
||||
if (this.callbacks.length) {
|
||||
this.updateSubscription();
|
||||
}
|
||||
}
|
||||
|
||||
offUpdate(callback: () => void) {
|
||||
let index = this.callbacks.lastIndexOf(callback);
|
||||
if (index === -1) {
|
||||
throw new Error("callback is not registered");
|
||||
}
|
||||
|
||||
this.callbacks.splice(index, 1);
|
||||
if (!this.callbacks.length) {
|
||||
this.updateSubscription();
|
||||
}
|
||||
}
|
||||
|
||||
updateStorage(items: Item[]) {
|
||||
for (let item of items) {
|
||||
this.storage.set(`${item.name}:${item.quality}`, item);
|
||||
}
|
||||
for (let callback of this.callbacks) {
|
||||
callback();
|
||||
}
|
||||
}
|
||||
|
||||
updateSubscription() {
|
||||
if (!this.control.connector.connected) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.control.send(
|
||||
new SetStorageSubscriptionRequest(Boolean(this.callbacks.length))
|
||||
).catch(notifyErrorHandler("Error subscribing to storage"));
|
||||
|
||||
if (this.callbacks.length) {
|
||||
this.control!.send(new GetStorageRequest()).then(
|
||||
items => {
|
||||
this.updateStorage(items);
|
||||
}
|
||||
).catch(notifyErrorHandler("Error updating storage"));
|
||||
} else {
|
||||
this.storage.clear();
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user