mirror of
https://github.com/clusterio/clusterio.git
synced 2026-11-11 23:04:52 +09:00
Convert inventory_sync to entrypoint functions
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
b2705b8788
commit
1ca0c658b9
4 files changed
+215
-225
No files matched your search
@@ -1,4 +1,4 @@
|
||||
import { BaseControllerPlugin, type InstanceRecord } from "@clusterio/controller";
|
||||
import type { ControllerPluginContext, InstanceRecord } from "@clusterio/controller";
|
||||
import type { IpcPlayerData } from "./messages.js";
|
||||
|
||||
import fs from "node:fs/promises";
|
||||
@@ -35,35 +35,40 @@ async function saveDatabase(
|
||||
}
|
||||
}
|
||||
|
||||
export class ControllerPlugin extends BaseControllerPlugin {
|
||||
acquiredPlayers!: Map<string, { instanceId: number, expiresMs?: number }>;
|
||||
playerDatastore!: Map<string, IpcPlayerData>;
|
||||
playerDatastoreDirty = false;
|
||||
export default async function(context: ControllerPluginContext) {
|
||||
const { controller, logger, plugin } = context;
|
||||
const acquiredPlayers = new Map<string, { instanceId: number, expiresMs?: number }>();
|
||||
const playerDatastore: Map<string, IpcPlayerData> = await loadDatabase(controller.config, logger);
|
||||
let playerDatastoreDirty = false;
|
||||
|
||||
async init() {
|
||||
this.acquiredPlayers = new Map();
|
||||
this.playerDatastore = await loadDatabase(this.controller.config, this.logger);
|
||||
function acquire(instanceId: number, playerName: string): boolean {
|
||||
let acquisitionRecord = acquiredPlayers.get(playerName);
|
||||
if (
|
||||
!acquisitionRecord
|
||||
|| acquisitionRecord.instanceId === instanceId
|
||||
|| !controller.instances.has(acquisitionRecord.instanceId)
|
||||
|| acquisitionRecord.expiresMs && acquisitionRecord.expiresMs < Date.now()
|
||||
) {
|
||||
acquiredPlayers.set(playerName, { instanceId });
|
||||
return true;
|
||||
}
|
||||
|
||||
this.controller.handle(msg.AcquireRequest, this.handleAcquireRequest.bind(this));
|
||||
this.controller.handle(msg.ReleaseRequest, this.handleReleaseRequest.bind(this));
|
||||
this.controller.handle(msg.UploadRequest, this.handleUploadRequest.bind(this));
|
||||
this.controller.handle(msg.DownloadRequest, this.handleDownloadRequest.bind(this));
|
||||
this.controller.handle(msg.DatabaseStatsRequest, this.handleDatabaseStatsRequest.bind(this));
|
||||
return false;
|
||||
}
|
||||
|
||||
async onInstanceStatusChanged(instance: InstanceRecord) {
|
||||
controller.hooks.instanceStatusChanged.attach(plugin.name, async (instance: InstanceRecord) => {
|
||||
let instanceId = instance.id;
|
||||
if (["unassigned", "deleted"].includes(instance.status)) {
|
||||
for (let [playerName, acquisitionRecord] of this.acquiredPlayers) {
|
||||
for (let [playerName, acquisitionRecord] of acquiredPlayers) {
|
||||
if (acquisitionRecord.instanceId === instanceId) {
|
||||
this.acquiredPlayers.delete(playerName);
|
||||
acquiredPlayers.delete(playerName);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (["unknown", "stopped"].includes(instance.status)) {
|
||||
let timeoutMs = this.controller.config.get("inventory_sync.player_lock_timeout") * 1000;
|
||||
for (let acquisitonRecord of this.acquiredPlayers.values()) {
|
||||
let timeoutMs = controller.config.get("inventory_sync.player_lock_timeout") * 1000;
|
||||
for (let acquisitonRecord of acquiredPlayers.values()) {
|
||||
if (acquisitonRecord.instanceId === instanceId && !acquisitonRecord.expiresMs) {
|
||||
acquisitonRecord.expiresMs = Date.now() + timeoutMs;
|
||||
}
|
||||
@@ -71,82 +76,74 @@ export class ControllerPlugin extends BaseControllerPlugin {
|
||||
}
|
||||
|
||||
if (instance.status === "running") {
|
||||
for (let acquisitonRecord of this.acquiredPlayers.values()) {
|
||||
for (let acquisitonRecord of acquiredPlayers.values()) {
|
||||
if (acquisitonRecord.instanceId === instanceId && acquisitonRecord.expiresMs) {
|
||||
delete acquisitonRecord.expiresMs;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
acquire(instanceId: number, playerName: string): boolean {
|
||||
let acquisitionRecord = this.acquiredPlayers.get(playerName);
|
||||
if (
|
||||
!acquisitionRecord
|
||||
|| acquisitionRecord.instanceId === instanceId
|
||||
|| !this.controller.instances.has(acquisitionRecord.instanceId)
|
||||
|| acquisitionRecord.expiresMs && acquisitionRecord.expiresMs < Date.now()
|
||||
) {
|
||||
this.acquiredPlayers.set(playerName, { instanceId });
|
||||
return true;
|
||||
controller.hooks.save.attach(plugin.name, async () => {
|
||||
if (playerDatastoreDirty) {
|
||||
playerDatastoreDirty = false;
|
||||
await saveDatabase(controller.config, playerDatastore, logger);
|
||||
}
|
||||
});
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
async handleAcquireRequest(request: msg.AcquireRequest) {
|
||||
controller.handle(msg.AcquireRequest, async (request: msg.AcquireRequest) => {
|
||||
let { instanceId, playerName } = request;
|
||||
if (!this.acquire(instanceId, playerName)) {
|
||||
let acquisitionRecord = this.acquiredPlayers.get(playerName);
|
||||
let instance = this.controller.instances.get(acquisitionRecord!.instanceId)!;
|
||||
if (!acquire(instanceId, playerName)) {
|
||||
let acquisitionRecord = acquiredPlayers.get(playerName);
|
||||
let instance = controller.instances.get(acquisitionRecord!.instanceId)!;
|
||||
return {
|
||||
status: "busy",
|
||||
message: instance.config.get("instance.name"),
|
||||
};
|
||||
}
|
||||
|
||||
let playerData = this.playerDatastore.get(playerName);
|
||||
let playerData = playerDatastore.get(playerName);
|
||||
return new msg.AcquireRequest.Response(
|
||||
"acquired",
|
||||
playerData ? playerData.generation : 0,
|
||||
Boolean(playerData),
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
async handleReleaseRequest(request: msg.ReleaseRequest) {
|
||||
controller.handle(msg.ReleaseRequest, async (request: msg.ReleaseRequest) => {
|
||||
let { instanceId, playerName } = request;
|
||||
let acquisitionRecord = this.acquiredPlayers.get(playerName);
|
||||
let acquisitionRecord = acquiredPlayers.get(playerName);
|
||||
if (!acquisitionRecord) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (acquisitionRecord.instanceId === instanceId) {
|
||||
this.acquiredPlayers.delete(playerName);
|
||||
acquiredPlayers.delete(playerName);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
async handleUploadRequest(request: msg.UploadRequest) {
|
||||
controller.handle(msg.UploadRequest, async (request: msg.UploadRequest) => {
|
||||
let { instanceId, playerName, playerData } = request;
|
||||
let instanceName = this.controller.instances.get(instanceId)!.config.get("instance.name");
|
||||
let instanceName = controller.instances.get(instanceId)!.config.get("instance.name");
|
||||
let store = true;
|
||||
let acquisitionRecord = this.acquiredPlayers.get(playerName);
|
||||
let acquisitionRecord = acquiredPlayers.get(playerName);
|
||||
if (!acquisitionRecord) {
|
||||
this.logger.warn(`${instanceName} uploaded ${playerName} without an acquisition`);
|
||||
logger.warn(`${instanceName} uploaded ${playerName} without an acquisition`);
|
||||
// Allow upload in this case as it might come from a crashed instance that restarted and is now
|
||||
// uploading the player data for all the players that were online during the last autosave.
|
||||
|
||||
} else if (acquisitionRecord.instanceId !== instanceId) {
|
||||
this.logger.warn(`${instanceName} uploaded ${playerName} while another instance has acquired it`);
|
||||
logger.warn(`${instanceName} uploaded ${playerName} while another instance has acquired it`);
|
||||
store = false;
|
||||
|
||||
} else {
|
||||
this.acquiredPlayers.delete(playerName);
|
||||
acquiredPlayers.delete(playerName);
|
||||
}
|
||||
|
||||
this.acquiredPlayers.delete(playerName);
|
||||
let oldPlayerData = this.playerDatastore.get(playerName);
|
||||
acquiredPlayers.delete(playerName);
|
||||
let oldPlayerData = playerDatastore.get(playerName);
|
||||
if (store && oldPlayerData && oldPlayerData.generation >= playerData.generation) {
|
||||
this.logger.warn(
|
||||
logger.warn(
|
||||
`${instanceName} uploaded generation ${playerData.generation} while the stored` +
|
||||
`generation is ${oldPlayerData.generation} for ${playerName}`
|
||||
);
|
||||
@@ -154,48 +151,41 @@ export class ControllerPlugin extends BaseControllerPlugin {
|
||||
}
|
||||
|
||||
if (store) {
|
||||
this.logger.verbose(`Received player data for ${playerName} from ${instanceName}`);
|
||||
this.playerDatastore.set(playerName, playerData);
|
||||
this.playerDatastoreDirty = true;
|
||||
logger.verbose(`Received player data for ${playerName} from ${instanceName}`);
|
||||
playerDatastore.set(playerName, playerData);
|
||||
playerDatastoreDirty = true;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
async handleDownloadRequest(request: msg.DownloadRequest) {
|
||||
controller.handle(msg.DownloadRequest, async (request: msg.DownloadRequest) => {
|
||||
let { instanceId, playerName } = request;
|
||||
let instanceName = this.controller.instances.get(instanceId)!.config.get("instance.name");
|
||||
let instanceName = controller.instances.get(instanceId)!.config.get("instance.name");
|
||||
|
||||
let acquisitionRecord = this.acquiredPlayers.get(playerName);
|
||||
let acquisitionRecord = acquiredPlayers.get(playerName);
|
||||
if (!acquisitionRecord) {
|
||||
this.logger.warn(`${instanceName} downloaded ${playerName} without an acquisition`);
|
||||
logger.warn(`${instanceName} downloaded ${playerName} without an acquisition`);
|
||||
} else if (acquisitionRecord.instanceId !== instanceId) {
|
||||
this.logger.warn(`${instanceName} downloaded ${playerName} while another instance has acquired it`);
|
||||
logger.warn(`${instanceName} downloaded ${playerName} while another instance has acquired it`);
|
||||
}
|
||||
|
||||
this.logger.verbose(`Sending player data for ${playerName} to ${instanceName}`);
|
||||
return new msg.DownloadRequest.Response(this.playerDatastore.get(playerName) || null);
|
||||
}
|
||||
logger.verbose(`Sending player data for ${playerName} to ${instanceName}`);
|
||||
return new msg.DownloadRequest.Response(playerDatastore.get(playerName) || null);
|
||||
});
|
||||
|
||||
async onSaveData() {
|
||||
if (this.playerDatastoreDirty) {
|
||||
this.playerDatastoreDirty = false;
|
||||
await saveDatabase(this.controller.config, this.playerDatastore, this.logger);
|
||||
}
|
||||
}
|
||||
|
||||
async handleDatabaseStatsRequest() {
|
||||
let playerDatastore = Array.from(this.playerDatastore.keys())
|
||||
controller.handle(msg.DatabaseStatsRequest, async () => {
|
||||
let entries = Array.from(playerDatastore.keys())
|
||||
.map(name => ({
|
||||
name,
|
||||
length: JSON.stringify(this.playerDatastore.get(name)).length,
|
||||
length: JSON.stringify(playerDatastore.get(name)).length,
|
||||
}))
|
||||
.sort((a, b) => b.length - a.length);
|
||||
return new msg.DatabaseStatsRequest.Response(
|
||||
playerDatastore.map(x => x.length).reduce((acc, val) => acc + val, 0),
|
||||
playerDatastore.length,
|
||||
entries.map(x => x.length).reduce((acc, val) => acc + val, 0),
|
||||
entries.length,
|
||||
{
|
||||
name: playerDatastore[0] && playerDatastore[0].name || "-",
|
||||
size: playerDatastore[0] && playerDatastore[0].length || 0,
|
||||
name: entries[0] && entries[0].name || "-",
|
||||
size: entries[0] && entries[0].length || 0,
|
||||
},
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
+110
-109
@@ -1,7 +1,7 @@
|
||||
import util from "util";
|
||||
import zlib from "zlib";
|
||||
import * as lib from "@clusterio/lib";
|
||||
import { BaseInstancePlugin } from "@clusterio/host";
|
||||
import type { InstancePluginContext } from "@clusterio/host";
|
||||
import {
|
||||
AcquireRequest, AcquireResponse, ReleaseRequest, UploadRequest, DownloadRequest, DownloadResponse, IpcPlayerData,
|
||||
} from "./messages.js";
|
||||
@@ -76,67 +76,36 @@ export function recipeNotificationDelta(stored: string[], current: string[]): Re
|
||||
return delta;
|
||||
}
|
||||
|
||||
export class InstancePlugin extends BaseInstancePlugin {
|
||||
playersToRelease!: Set<string>;
|
||||
disconnecting!: boolean;
|
||||
|
||||
async init() {
|
||||
this.playersToRelease = new Set();
|
||||
this.disconnecting = false;
|
||||
|
||||
// Handle IPC from scenario script
|
||||
this.instance.server.on(
|
||||
"ipc-inventory_sync_acquire",
|
||||
(request: IpcPlayerName) => this.handleAcquire(request).catch(
|
||||
err => this.logger.error(`Error handling ipc-inventory_sync_acquire:\n${err.stack}`)
|
||||
),
|
||||
);
|
||||
this.instance.server.on(
|
||||
"ipc-inventory_sync_release",
|
||||
(request: IpcPlayerName) => this.handleRelease(request).catch(
|
||||
err => this.logger.error(`Error handling ipc-inventory_sync_release:\n${err.stack}`)
|
||||
),
|
||||
);
|
||||
this.instance.server.on(
|
||||
"ipc-inventory_sync_upload",
|
||||
(player_data: IpcPlayerData) => this.handleUpload(player_data).catch(
|
||||
err => this.logger.error(`Error handling ipc-inventory_sync_upload:\n${err.stack}`)
|
||||
),
|
||||
);
|
||||
this.instance.server.on(
|
||||
"ipc-inventory_sync_download",
|
||||
(request: IpcDownloadRequest) => this.handleDownload(request).catch(
|
||||
err => this.logger.error(`Error handling ipc-inventory_sync_download:\n${err.stack}`)
|
||||
),
|
||||
);
|
||||
/**
|
||||
* Replace the stored recipe notifications with the difference from the
|
||||
* current state on the instance to reduce the amount of data sent to Lua.
|
||||
*/
|
||||
export async function applyRecipeNotificationDelta(
|
||||
playerData: IpcPlayerData,
|
||||
current: string | undefined,
|
||||
logger: lib.Logger,
|
||||
) {
|
||||
if (!playerData.recipe_notifications) {
|
||||
return;
|
||||
}
|
||||
|
||||
async onPrepareControllerDisconnect() {
|
||||
this.disconnecting = true;
|
||||
try {
|
||||
const delta = recipeNotificationDelta(
|
||||
JSON.parse(await decodeLuaString(playerData.recipe_notifications)),
|
||||
current ? JSON.parse(await decodeLuaString(current)) : [],
|
||||
);
|
||||
playerData.recipe_notifications = await encodeLuaString(JSON.stringify(delta));
|
||||
} catch (err: any) {
|
||||
logger.warn(`Dropping invalid recipe notifications for ${playerData.name}:\n${err.stack}`);
|
||||
delete playerData.recipe_notifications;
|
||||
}
|
||||
}
|
||||
|
||||
export default async function(context: InstancePluginContext) {
|
||||
const { instance, host, logger, plugin } = context;
|
||||
const playersToRelease = new Set<string>();
|
||||
let disconnecting = false;
|
||||
|
||||
onControllerConnectionEvent(event: "connect" | "drop" | "resume" | "close") {
|
||||
if (event === "connect") {
|
||||
this.disconnecting = false;
|
||||
(async () => {
|
||||
for (let player_name of this.playersToRelease) {
|
||||
if (!this.host.connector.connected || this.disconnecting) {
|
||||
return;
|
||||
}
|
||||
this.playersToRelease.delete(player_name);
|
||||
await this.instance.sendTo(
|
||||
"controller",
|
||||
new ReleaseRequest(this.instance.id, player_name)
|
||||
);
|
||||
}
|
||||
})().catch(
|
||||
err => this.logger.error(`Unpexpected error releasing queued up players:\n${err.stack}`)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async handleAcquire(request: IpcPlayerName) {
|
||||
async function handleAcquire(request: IpcPlayerName) {
|
||||
let response: IpcAcquireResponse = {
|
||||
player_name: request.player_name,
|
||||
status: "error",
|
||||
@@ -145,11 +114,11 @@ export class InstancePlugin extends BaseInstancePlugin {
|
||||
generation: undefined,
|
||||
};
|
||||
|
||||
if (this.host.connector.connected && !this.disconnecting) {
|
||||
if (host.connector.connected && !disconnecting) {
|
||||
try {
|
||||
let acquireResponse: AcquireResponse = await this.instance.sendTo(
|
||||
let acquireResponse: AcquireResponse = await instance.sendTo(
|
||||
"controller",
|
||||
new AcquireRequest(this.instance.id, request.player_name),
|
||||
new AcquireRequest(instance.id, request.player_name),
|
||||
);
|
||||
response = {
|
||||
player_name: request.player_name,
|
||||
@@ -160,105 +129,137 @@ export class InstancePlugin extends BaseInstancePlugin {
|
||||
};
|
||||
} catch (err: any) {
|
||||
if (!(err instanceof lib.SessionLost)) {
|
||||
this.logger.error(`Unexpected error sending aquire request:\n${err.stack}`);
|
||||
logger.error(`Unexpected error sending aquire request:\n${err.stack}`);
|
||||
response.message = err.message;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let json = lib.escapeString(JSON.stringify(response));
|
||||
await this.sendRcon(`/sc inventory_sync.acquire_response("${json}")`, true);
|
||||
await instance.sendRcon(`/sc inventory_sync.acquire_response("${json}")`, true, plugin.name);
|
||||
}
|
||||
|
||||
async handleRelease(request: IpcPlayerName) {
|
||||
if (!this.host.connector.connected) {
|
||||
this.playersToRelease.add(request.player_name);
|
||||
async function handleRelease(request: IpcPlayerName) {
|
||||
if (!host.connector.connected) {
|
||||
playersToRelease.add(request.player_name);
|
||||
}
|
||||
|
||||
try {
|
||||
await this.instance.sendTo(
|
||||
await instance.sendTo(
|
||||
"controller",
|
||||
new ReleaseRequest(this.instance.id, request.player_name)
|
||||
new ReleaseRequest(instance.id, request.player_name)
|
||||
);
|
||||
} catch (err: any) {
|
||||
if (err instanceof lib.SessionLost) {
|
||||
this.playersToRelease.add(request.player_name);
|
||||
playersToRelease.add(request.player_name);
|
||||
} else {
|
||||
this.logger.error(`Unexpected error releasing player ${request.player_name}:\n${err.stack}`);
|
||||
logger.error(`Unexpected error releasing player ${request.player_name}:\n${err.stack}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async handleUpload(player_data: IpcPlayerData) {
|
||||
if (!this.host.connector.connected || this.disconnecting) {
|
||||
async function handleUpload(player_data: IpcPlayerData) {
|
||||
if (!host.connector.connected || disconnecting) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.verbose(`Uploading ${player_data.name} (${JSON.stringify(player_data).length / 1000}kB)`);
|
||||
logger.verbose(`Uploading ${player_data.name} (${JSON.stringify(player_data).length / 1000}kB)`);
|
||||
try {
|
||||
await this.instance.sendTo(
|
||||
await instance.sendTo(
|
||||
"controller",
|
||||
new UploadRequest(this.instance.id, player_data.name, player_data),
|
||||
new UploadRequest(instance.id, player_data.name, player_data),
|
||||
);
|
||||
|
||||
} catch (err: any) {
|
||||
if (!(err instanceof lib.SessionLost)) {
|
||||
this.logger.error(`Unexpected error uploading inventory for ${player_data.name}:\n${err.stack}`);
|
||||
logger.error(`Unexpected error uploading inventory for ${player_data.name}:\n${err.stack}`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
await this.sendRcon(
|
||||
`/sc inventory_sync.confirm_upload("${player_data.name}", ${player_data.generation})`, true
|
||||
await instance.sendRcon(
|
||||
`/sc inventory_sync.confirm_upload("${player_data.name}", ${player_data.generation})`, true, plugin.name
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Replace the stored recipe notifications with the difference from the
|
||||
* current state on the instance to reduce the amount of data sent to Lua.
|
||||
*/
|
||||
async applyRecipeNotificationDelta(playerData: IpcPlayerData, current?: string) {
|
||||
if (!playerData.recipe_notifications) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const delta = recipeNotificationDelta(
|
||||
JSON.parse(await decodeLuaString(playerData.recipe_notifications)),
|
||||
current ? JSON.parse(await decodeLuaString(current)) : [],
|
||||
);
|
||||
playerData.recipe_notifications = await encodeLuaString(JSON.stringify(delta));
|
||||
} catch (err: any) {
|
||||
this.logger.warn(`Dropping invalid recipe notifications for ${playerData.name}:\n${err.stack}`);
|
||||
delete playerData.recipe_notifications;
|
||||
}
|
||||
}
|
||||
|
||||
async handleDownload(request: IpcDownloadRequest) {
|
||||
async function handleDownload(request: IpcDownloadRequest) {
|
||||
const playerName = request.player_name;
|
||||
this.logger.verbose(`Downloading ${playerName}`);
|
||||
logger.verbose(`Downloading ${playerName}`);
|
||||
|
||||
let response: DownloadResponse = await this.instance.sendTo(
|
||||
let response: DownloadResponse = await instance.sendTo(
|
||||
"controller",
|
||||
new DownloadRequest(this.instance.id, playerName)
|
||||
new DownloadRequest(instance.id, playerName)
|
||||
);
|
||||
|
||||
if (!response.playerData) {
|
||||
await this.sendRcon(`/sc inventory_sync.download_inventory('${playerName}',nil,0,0)`, true);
|
||||
await instance.sendRcon(
|
||||
`/sc inventory_sync.download_inventory('${playerName}',nil,0,0)`, true, plugin.name
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
await this.applyRecipeNotificationDelta(response.playerData, request.recipe_notifications);
|
||||
await applyRecipeNotificationDelta(response.playerData, request.recipe_notifications, logger);
|
||||
|
||||
const chunkSize = this.instance.config.get("inventory_sync.rcon_chunk_size");
|
||||
const chunkSize = instance.config.get("inventory_sync.rcon_chunk_size");
|
||||
const chunks = chunkify(chunkSize, JSON.stringify(response.playerData));
|
||||
this.logger.verbose(`Sending inventory for ${playerName} in ${chunks.length} chunks`);
|
||||
logger.verbose(`Sending inventory for ${playerName} in ${chunks.length} chunks`);
|
||||
for (let i = 0; i < chunks.length; i++) {
|
||||
// this.logger.verbose(`Sending chunk ${i+1} of ${chunks.length}`)
|
||||
const chunk = lib.escapeString(chunks[i]);
|
||||
await this.sendRcon(
|
||||
await instance.sendRcon(
|
||||
`/sc inventory_sync.download_inventory('${playerName}','${chunk}',${i + 1},${chunks.length})`,
|
||||
true
|
||||
true,
|
||||
plugin.name,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Handle IPC from scenario script
|
||||
instance.server.on(
|
||||
"ipc-inventory_sync_acquire",
|
||||
(request: IpcPlayerName) => handleAcquire(request).catch(
|
||||
err => logger.error(`Error handling ipc-inventory_sync_acquire:\n${err.stack}`)
|
||||
),
|
||||
);
|
||||
instance.server.on(
|
||||
"ipc-inventory_sync_release",
|
||||
(request: IpcPlayerName) => handleRelease(request).catch(
|
||||
err => logger.error(`Error handling ipc-inventory_sync_release:\n${err.stack}`)
|
||||
),
|
||||
);
|
||||
instance.server.on(
|
||||
"ipc-inventory_sync_upload",
|
||||
(player_data: IpcPlayerData) => handleUpload(player_data).catch(
|
||||
err => logger.error(`Error handling ipc-inventory_sync_upload:\n${err.stack}`)
|
||||
),
|
||||
);
|
||||
instance.server.on(
|
||||
"ipc-inventory_sync_download",
|
||||
(request: IpcDownloadRequest) => handleDownload(request).catch(
|
||||
err => logger.error(`Error handling ipc-inventory_sync_download:\n${err.stack}`)
|
||||
),
|
||||
);
|
||||
|
||||
instance.hooks.prepareControllerDisconnect.attach(plugin.name, async () => {
|
||||
disconnecting = true;
|
||||
});
|
||||
|
||||
instance.hooks.controllerConnectionEvent.attach(plugin.name, async (event) => {
|
||||
if (event === "connect") {
|
||||
disconnecting = false;
|
||||
(async () => {
|
||||
for (let player_name of playersToRelease) {
|
||||
if (!host.connector.connected || disconnecting) {
|
||||
return;
|
||||
}
|
||||
playersToRelease.delete(player_name);
|
||||
await instance.sendTo(
|
||||
"controller",
|
||||
new ReleaseRequest(instance.id, player_name)
|
||||
);
|
||||
}
|
||||
})().catch(
|
||||
err => logger.error(`Unpexpected error releasing queued up players:\n${err.stack}`)
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -4,7 +4,7 @@ import zlib from "node:zlib";
|
||||
import * as lib from "@clusterio/lib";
|
||||
|
||||
import * as mock from "../../../test/mock.js";
|
||||
import { recipeNotificationDelta, InstancePlugin } from "../dist/node/instance.js";
|
||||
import loadInstancePlugin, { recipeNotificationDelta, applyRecipeNotificationDelta } from "../dist/node/instance.js";
|
||||
import { DownloadResponse } from "../dist/node/messages.js";
|
||||
import { plugin as info } from "../dist/node/index.js";
|
||||
|
||||
@@ -38,10 +38,10 @@ describe("inventory_sync", function() {
|
||||
});
|
||||
});
|
||||
|
||||
describe("InstancePlugin.applyRecipeNotificationDelta()", function() {
|
||||
describe("applyRecipeNotificationDelta()", function() {
|
||||
const warnings = [];
|
||||
const plugin = { logger: { warn: msg => warnings.push(msg) } };
|
||||
const apply = InstancePlugin.prototype.applyRecipeNotificationDelta.bind(plugin);
|
||||
const logger = { warn: msg => warnings.push(msg) };
|
||||
const apply = (playerData, current) => applyRecipeNotificationDelta(playerData, current, logger);
|
||||
|
||||
it("should replace the stored list with an encoded delta", async function() {
|
||||
const playerData = { name: "test", recipe_notifications: await encode(["a", "b"]) };
|
||||
@@ -71,28 +71,35 @@ describe("inventory_sync", function() {
|
||||
});
|
||||
});
|
||||
|
||||
describe("InstancePlugin.handleDownload()", function() {
|
||||
let instancePlugin;
|
||||
describe("ipc-inventory_sync_download", function() {
|
||||
let instance;
|
||||
before(async function() {
|
||||
try {
|
||||
lib.registerPluginMessages([info]);
|
||||
} catch (err) {
|
||||
// Already registered by the full test suite
|
||||
}
|
||||
instancePlugin = await mock.createInstancePlugin(InstancePlugin, info);
|
||||
instancePlugin.instance.mockConfigEntries.set("inventory_sync.rcon_chunk_size", 100000);
|
||||
({ instance } = await mock.loadInstancePlugin(loadInstancePlugin, info));
|
||||
instance.mockConfigEntries.set("inventory_sync.rcon_chunk_size", 100000);
|
||||
});
|
||||
beforeEach(function() {
|
||||
instancePlugin.instance.server.reset();
|
||||
instance.server.reset();
|
||||
});
|
||||
|
||||
function respondWith(playerData) {
|
||||
instancePlugin.instance.connector.once("send", message => {
|
||||
instancePlugin.instance.connector.emit("message", new lib.MessageResponse(
|
||||
instance.connector.once("send", message => {
|
||||
instance.connector.emit("message", new lib.MessageResponse(
|
||||
1, message.dst, message.src, new DownloadResponse(playerData)
|
||||
));
|
||||
});
|
||||
}
|
||||
async function download(request) {
|
||||
instance.server.emit("ipc-inventory_sync_download", request);
|
||||
// zlib runs on the threadpool, so poll on timers rather than setImmediate
|
||||
for (let i = 0; i < 1000 && !instance.server.rconCommands.length; i++) {
|
||||
await lib.wait(1);
|
||||
}
|
||||
}
|
||||
function downloadCommand(playerData) {
|
||||
const json = lib.escapeString(JSON.stringify(playerData));
|
||||
return `/sc inventory_sync.download_inventory('test','${json}',1,1)`;
|
||||
@@ -100,31 +107,23 @@ describe("inventory_sync", function() {
|
||||
|
||||
it("should send an empty download when the controller has no data", async function() {
|
||||
respondWith(undefined);
|
||||
await instancePlugin.handleDownload({ player_name: "test" });
|
||||
await download({ player_name: "test" });
|
||||
assert.deepEqual(
|
||||
instancePlugin.instance.server.rconCommands,
|
||||
instance.server.rconCommands,
|
||||
["/sc inventory_sync.download_inventory('test',nil,0,0)"]
|
||||
);
|
||||
});
|
||||
it("should send everything as a delta when no snapshot was given", async function() {
|
||||
respondWith({ generation: 1, name: "test", recipe_notifications: await encode(["a"]) });
|
||||
await instancePlugin.handleDownload({ player_name: "test" });
|
||||
await download({ player_name: "test" });
|
||||
const expected = { generation: 1, name: "test", recipe_notifications: await encode({ add: ["a"] }) };
|
||||
assert.deepEqual(instancePlugin.instance.server.rconCommands, [downloadCommand(expected)]);
|
||||
assert.deepEqual(instance.server.rconCommands, [downloadCommand(expected)]);
|
||||
});
|
||||
it("should send a delta when a snapshot was given", async function() {
|
||||
respondWith({ generation: 1, name: "test", recipe_notifications: await encode(["a", "b"]) });
|
||||
await instancePlugin.handleDownload({ player_name: "test", recipe_notifications: await encode(["b"]) });
|
||||
await download({ player_name: "test", recipe_notifications: await encode(["b"]) });
|
||||
const expected = { generation: 1, name: "test", recipe_notifications: await encode({ add: ["a"] }) };
|
||||
assert.deepEqual(instancePlugin.instance.server.rconCommands, [downloadCommand(expected)]);
|
||||
});
|
||||
it("should be invoked by the download ipc", async function() {
|
||||
respondWith(undefined);
|
||||
instancePlugin.instance.server.emit("ipc-inventory_sync_download", { player_name: "test" });
|
||||
for (let i = 0; i < 100 && !instancePlugin.instance.server.rconCommands.length; i++) {
|
||||
await new Promise(resolve => setImmediate(resolve));
|
||||
}
|
||||
assert.equal(instancePlugin.instance.server.rconCommands.length, 1);
|
||||
assert.deepEqual(instance.server.rconCommands, [downloadCommand(expected)]);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -1,6 +1,6 @@
|
||||
import React, { useContext, useEffect, useState } from "react";
|
||||
|
||||
import { BaseWebPlugin, PageLayout, PageHeader, ControlContext } from "@clusterio/web_ui";
|
||||
import { PageLayout, PageHeader, ControlContext, type WebPluginContext } from "@clusterio/web_ui";
|
||||
import { DatabaseStatsRequest, DatabaseStatsResponse } from "../messages.js";
|
||||
|
||||
import "./style.css";
|
||||
@@ -27,15 +27,15 @@ function InventoryPage() {
|
||||
</PageLayout>;
|
||||
}
|
||||
|
||||
export class WebPlugin extends BaseWebPlugin {
|
||||
async init() {
|
||||
this.pages = [
|
||||
{
|
||||
path: "/inventory",
|
||||
sidebarName: "Inventory sync",
|
||||
permission: "inventory_sync.inventory.view",
|
||||
content: <InventoryPage />,
|
||||
},
|
||||
];
|
||||
}
|
||||
export default async function(context: WebPluginContext) {
|
||||
const { control, plugin } = context;
|
||||
|
||||
control.hooks.pages.attach(plugin.name, () => [
|
||||
{
|
||||
path: "/inventory",
|
||||
sidebarName: "Inventory sync",
|
||||
permission: "inventory_sync.inventory.view",
|
||||
content: <InventoryPage />,
|
||||
},
|
||||
]);
|
||||
}
|
||||
Reference in new issue
Block a user