diff --git a/docs/developing-for-clusterio.md b/docs/developing-for-clusterio.md index 39ef4be1..8b11ef0c 100644 --- a/docs/developing-for-clusterio.md +++ b/docs/developing-for-clusterio.md @@ -253,17 +253,16 @@ Parameters: #### clusterio_api.send_udp(channel, data) -Send JSON data to Clusterio over the given channel using UDP instead of stdout. -Clusterio plugins receive it the same way as data sent with `send_json`. +Send data to Clusterio over the given channel using UDP instead of stdout. +Clusterio plugins receive it as `udp-channel_name` events, see the [Communicating over UDP section](writing-plugins.md#communicating-over-udp) in the Writing Plugins document. Requires `factorio.enable_lua_udp` to be enabled on the instance and Factorio 2.0.59 or later, otherwise the data is silently dropped. The port to send to is passed to the module with a script command on startup, so `factorio.enable_script_commands` must be enabled as well. -See the [Communicating over UDP section](writing-plugins.md#communicating-over-udp) in the Writing Plugins document for more information. -**Note**: Delivery is not guaranteed and payloads must be less than 64 kB. +**Note**: Delivery is not guaranteed and payloads should be kept well below 32 kB. Parameters: - `channel`: string identifying which channel to send it on. -- `data`: table that can be converted to JSON with game.table_to_json +- `data`: string with the payload to send, any bytes are allowed. ### serialize library diff --git a/docs/writing-plugins.md b/docs/writing-plugins.md index 4b8f704d..6df458dc 100644 --- a/docs/writing-plugins.md +++ b/docs/writing-plugins.md @@ -488,14 +488,26 @@ Instances can optionally exchange UDP packets with the Factorio server for low l This is disabled by default and enabled by setting `factorio.enable_lua_udp` on the instance. Sending data out of Factorio also needs `factorio.enable_script_commands`, as the port to send to is passed to the module with a script command on startup. Plugins requiring it should declare the `LuaUdp` feature. -Note that UDP does not guarantee delivery or ordering, and packets should be kept well below 64 kB. +Note that UDP does not guarantee delivery or ordering, and packets should be kept well below 32 kB. -Data out from Factorio is sent with the `send_udp` API of the Clusterio module and is delivered to plugins as the same `ipc-channel_name` events as `send_json` uses: +Data out from Factorio is sent with the `send_udp` API of the Clusterio module. +Unlike `send_json` the payload is an arbitrary string, and it is delivered to plugins as a `udp-channel_name` event with the payload as a Buffer: ```lua -clusterio_api.send_udp("my_plugin_foo", { data = 123 }) +clusterio_api.send_udp("my_plugin_foo", "123") ``` +```js +async init() { + this.instance.server.handleUdp("my_plugin_foo", async data => { + this.logger.info(`got ${data.toString()}`); + }); +} +``` + +Like `handle` for `send_json`, `handleUdp` logs errors thrown by the handler. +Use `this.instance.server.on("udp-my_plugin_foo", ...)` for custom error handling. + Data into Factorio is sent with the `sendUdp` method on the plugin object and delivered in-game as `defines.events.on_udp_packet_received` events with the packet in `event.payload`. The Clusterio module polls for received packets every tick, all modules and mods receive every packet so make sure the payload identifies your plugin. diff --git a/packages/host/lua/clusterio_lib/api.lua b/packages/host/lua/clusterio_lib/api.lua index a9242489..8aee448d 100644 --- a/packages/host/lua/clusterio_lib/api.lua +++ b/packages/host/lua/clusterio_lib/api.lua @@ -73,9 +73,7 @@ function api.send_udp(channel, data) return -- Lua UDP is not enabled for this instance end - channel = escape_channel(channel) - data = compat.table_to_json(data) - helpers.send_udp(port, "\f$ipc:" .. channel .. "?j" .. data, 0) + helpers.send_udp(port, escape_channel(channel) .. "?" .. data, 0) end diff --git a/packages/host/modules/clusterio/api.lua b/packages/host/modules/clusterio/api.lua index 8c1ad4ac..ec42588c 100644 --- a/packages/host/modules/clusterio/api.lua +++ b/packages/host/modules/clusterio/api.lua @@ -49,9 +49,7 @@ function api.send_udp(channel, data) return -- Lua UDP is not enabled for this instance end - channel = escape_channel(channel) - data = compat.table_to_json(data) - helpers.send_udp(port, "\f$ipc:" .. channel .. "?j" .. data, 0) + helpers.send_udp(port, escape_channel(channel) .. "?" .. data, 0) end diff --git a/packages/host/src/server.ts b/packages/host/src/server.ts index 000b6c40..a9aec3a7 100644 --- a/packages/host/src/server.ts +++ b/packages/host/src/server.ts @@ -290,6 +290,15 @@ async function generatePassword(length: number) { } } +function unescapeChannel(channel: Buffer) { + return channel + .toString("utf-8") + .replace(/\\x([0-9a-f]{2})/g, (match, p1) => ( + String.fromCharCode(parseInt(p1, 16)) + )) + ; +} + /** * Interpret lines of output from Factorio * @@ -412,6 +421,9 @@ const outputHeuristics: Heuristic[] = [ filter: { type: "log", message: /^Starting RCON interface/ }, action: function() { this._startRcon().catch((err) => { this.emit("error", err); }); + if (this.enableLuaUdp) { + this._startUdp().catch((err) => { this.emit("error", err); }); + } }, }, @@ -591,8 +603,9 @@ type FactorioServerEvents = { "_saving": [], "_saved": [], - // IPS events + // IPC events [ipcEvent: `ipc-${string}`]: [ event: any ], + [udpEvent: `udp-${string}`]: [ data: Buffer ], }; /** @@ -612,8 +625,8 @@ type FactorioServerEvents = { * - autosave-fnished - invoked when the autosave finished * - save-finished - invoked when the server has finished a manual save * - exit - invoked when the sterver has exited - * - ipc- - invoked with data sent by the clusterio module in-game, - * either via stdout or via UDP when Lua UDP is enabled + * - ipc- - invoked with data sent with send_json in-game + * - udp- - invoked with data sent with send_udp in-game * @extends events.EventEmitter */ export class FactorioServer extends events.EventEmitter { @@ -757,19 +770,19 @@ export class FactorioServer extends events.EventEmitter { })); } + handleUdp(channel: string, handler: (data: Buffer) => Promise) { + this.on(`udp-${channel}`, (data) => handler(data).catch((err: Error) => { + this._logger.error(`Error handling udp event:\n${err.stack ?? err.message}`); + })); + } + async _handleIpc(line: Buffer) { let channelEnd = line.indexOf("?"); if (channelEnd === -1) { throw new Error(`Malformed IPC line "${line.toString()}"`); } - let channel = line - .subarray(6, channelEnd) - .toString("utf-8") - .replace(/\\x([0-9a-f]{2})/g, (match, p1) => ( - String.fromCharCode(parseInt(p1, 16)) - )) - ; + let channel = unescapeChannel(line.subarray(6, channelEnd)); let type = line.subarray(channelEnd + 1, channelEnd + 2).toString("utf-8"); let content; @@ -815,15 +828,24 @@ export class FactorioServer extends events.EventEmitter { return; } - if (message.subarray(0, 6).equals(Buffer.from("\f$ipc:"))) { - this._handleIpc(message).catch(err => this.emit("error", err)); + let channelEnd = message.indexOf("?"); + if (channelEnd === -1) { + this._logger.warn(`Ignoring malformed UDP packet "${message.subarray(0, 100).toString("utf-8")}"`); return; } - this._logger.warn(`Ignoring UDP packet not in IPC format: ${message.subarray(0, 100).toString("utf-8")}`); + let channel = unescapeChannel(message.subarray(0, channelEnd)); + let data = message.subarray(channelEnd + 1); + if (!this.emit(`udp-${channel}`, data)) { + this._logger.warn(`Warning: Unhandled udp-${channel}`); + } } - async _bindUdpSocket() { + async _startUdp() { + if (this._udpSocket !== null) { + throw Error("UDP socket is already open"); + } + const socket = dgram.createSocket("udp4"); socket.on("message", (message, rinfo) => this._handleUdpMessage(message, rinfo)); try { @@ -834,15 +856,16 @@ export class FactorioServer extends events.EventEmitter { resolve(); }); }); - } catch (err) { + } catch (err: any) { + this._logger.error(`Failed to open UDP socket:\n${err?.stack ?? err?.message ?? err}`); socket.close(); - throw err; + return; } socket.on("error", err => this.emit("error", err)); this._udpSocket = socket; } - _closeUdpSocket() { + _stopUdp() { if (this._udpSocket) { this._udpSocket.close(); this._udpSocket = null; @@ -1130,7 +1153,7 @@ export class FactorioServer extends events.EventEmitter { this._server = null; this._rconClient = null; this._rconReady = false; - this._closeUdpSocket(); + this._udpSocket = null; this._gameReady = false; this._unexpected = []; this._killed = false; @@ -1176,6 +1199,7 @@ export class FactorioServer extends events.EventEmitter { if (this._rconClient) { this._rconClient.end().catch(() => {}); } + this._stopUdp(); if (this._whitelistWatcher) { this._whitelistWatcher.close(); @@ -1194,6 +1218,7 @@ export class FactorioServer extends events.EventEmitter { if (this._rconClient) { this._rconClient.end().catch(() => {}); } + this._stopUdp(); if (this._whitelistWatcher) { this._whitelistWatcher.close(); @@ -1292,9 +1317,6 @@ export class FactorioServer extends events.EventEmitter { */ async start(save: string) { this._check(["init"]); - if (this.enableLuaUdp) { - await this._bindUdpSocket(); - } this._state = "running"; try { @@ -1348,9 +1370,6 @@ export class FactorioServer extends events.EventEmitter { */ async startScenario(scenario: string, seed?: number, mapGenSettings?: object, mapSettings?: object) { this._check(["init"]); - if (this.enableLuaUdp) { - await this._bindUdpSocket(); - } this._state = "running"; try { @@ -1465,7 +1484,7 @@ export class FactorioServer extends events.EventEmitter { async sendUdp(message: string | Buffer) { this._check(["running", "stopping"]); if (!this._udpSocket) { - throw new Error("Lua UDP is not enabled"); + throw new Error("UDP socket is not open"); } const socket = this._udpSocket; @@ -1509,6 +1528,7 @@ export class FactorioServer extends events.EventEmitter { if (this._rconClient) { this._rconClient.end().catch(() => {}); } + this._stopUdp(); if (this._whitelistWatcher) { this._whitelistWatcher.close(); } diff --git a/test/host/server.js b/test/host/server.js index b6cc542d..18fd3628 100644 --- a/test/host/server.js +++ b/test/host/server.js @@ -346,123 +346,156 @@ describe("host/server", function() { received = []; udpServer.on("message", (msg, rinfo) => received.push([msg, rinfo])); await events.once(udpServer.bind(0, "127.0.0.1"), "listening"); + server.luaUdpPort = udpServer.address().port; }); after(function() { udpServer.close(); }); + afterEach(function() { + server._stopUdp(); + server._state = "init"; + }); + function sendFromGame(message) { + udpServer.send(Buffer.from(message), server.hostUdpPort, "127.0.0.1"); + } + function captureLogger(level) { + const logger = server._logger; + const lines = []; + server._logger = { [level]: msg => lines.push(msg) }; + return { lines, restore: () => { server._logger = logger; } }; + } it("should not have a host port when not running", function() { assert.equal(server.hostUdpPort, undefined); }); - it("should reject sendUdp when not enabled", async function() { + it("should reject sendUdp when the socket is not open", async function() { server._state = "running"; - try { - await assert.rejects(server.sendUdp("data"), new Error("Lua UDP is not enabled")); - } finally { - server._state = "init"; - } + await assert.rejects(server.sendUdp("data"), new Error("UDP socket is not open")); }); - it("should bind a socket, send and receive ipc over UDP", async function() { - server.enableLuaUdp = true; - server.luaUdpPort = udpServer.address().port; - await server._bindUdpSocket(); + it("should send packets to the Lua UDP port", async function() { + await server._startUdp(); server._state = "running"; + assert(server.hostUdpPort > 0, "hostUdpPort not set"); + + await server.sendUdp("hello"); + await server.sendUdp(Buffer.from("world")); + while (received.length < 2) { + await wait(1); + } + assert.equal(received[0][0].toString(), "hello"); + assert.equal(received[1][0].toString(), "world"); + assert.equal(received[0][1].port, server.hostUdpPort); + }); + it("should emit udp events for packets from the game", async function() { + await server._startUdp(); + let udpReceived = []; + let onUdp = data => udpReceived.push(data); + // Escaped characters in the channel name are unescaped + server.on("udp-udp?channel", onUdp); + let waiter = events.once(server, "udp-udp?channel"); + // Packets not from the Factorio server's port are ignored + let other = dgram.createSocket("udp4"); try { - assert(server.hostUdpPort > 0, "hostUdpPort not set"); - - await server.sendUdp("hello"); - await server.sendUdp(Buffer.from("world")); - while (received.length < 2) { - await wait(1); - } - assert.equal(received[0][0].toString(), "hello"); - assert.equal(received[1][0].toString(), "world"); - assert.equal(received[0][1].port, server.hostUdpPort); - - let ipcReceived = []; - let onIpc = content => ipcReceived.push(content); - server.on("ipc-udp_channel", onIpc); - let waiter = events.once(server, "ipc-udp_channel"); - // Packets not from the Factorio server's port are ignored - let other = dgram.createSocket("udp4"); await events.once(other.bind(0, "127.0.0.1"), "listening"); - other.send(Buffer.from('\f$ipc:udp_channel?j{"data":"bad"}'), server.hostUdpPort, "127.0.0.1"); - udpServer.send(Buffer.from('\f$ipc:udp_channel?j{"data":"spam"}'), server.hostUdpPort, "127.0.0.1"); + other.send(Buffer.from("udp\\x3fchannel?bad"), server.hostUdpPort, "127.0.0.1"); + sendFromGame("udp\\x3fchannel?spam?eggs"); await waiter; - server.off("ipc-udp_channel", onIpc); - other.close(); - assert.deepEqual(ipcReceived, [{ data: "spam" }]); } finally { - server._resetState(); - server.enableLuaUdp = false; + other.close(); + server.off("udp-udp?channel", onUdp); } - assert.equal(server._udpSocket, null); - assert.equal(server.hostUdpPort, undefined); + assert(Buffer.isBuffer(udpReceived[0]), "data is not a Buffer"); + assert.deepEqual(udpReceived.map(String), ["spam?eggs"]); }); - it("should warn on packets not in IPC format", async function() { - server.luaUdpPort = udpServer.address().port; - await server._bindUdpSocket(); - const logger = server._logger; - let warnings = []; - server._logger = { warn: msg => warnings.push(msg) }; + it("should warn on malformed packets and unhandled channels", async function() { + await server._startUdp(); + const { lines, restore } = captureLogger("warn"); try { - udpServer.send(Buffer.from("not ipc"), server.hostUdpPort, "127.0.0.1"); - while (warnings.length < 1) { + sendFromGame("no separator"); + sendFromGame("unhandled?data"); + while (lines.length < 2) { await wait(1); } - assert.equal(warnings[0], "Ignoring UDP packet not in IPC format: not ipc"); } finally { - server._logger = logger; - server._resetState(); + restore(); } + assert.deepEqual(lines, [ + 'Ignoring malformed UDP packet "no separator"', + "Warning: Unhandled udp-unhandled", + ]); }); - it("should emit error on malformed IPC packets", async function() { - server.luaUdpPort = udpServer.address().port; - await server._bindUdpSocket(); + it("should invoke handlers registered with handleUdp", async function() { + let handled = []; + server.handleUdp("handled", async data => { handled.push(data.toString()); }); + server.handleUdp("failing", async () => { throw new Error("boom"); }); + const { lines, restore } = captureLogger("error"); try { - let waiter = events.once(server, "error"); - udpServer.send(Buffer.from("\f$ipc:no_separator"), server.hostUdpPort, "127.0.0.1"); - let [err] = await waiter; - assert.equal(err.message, 'Malformed IPC line "\f$ipc:no_separator"'); + server.emit("udp-handled", Buffer.from("data")); + server.emit("udp-failing", Buffer.from("data")); + await new Promise(resolve => setImmediate(resolve)); } finally { - server._resetState(); + restore(); } + assert.deepEqual(handled, ["data"]); + assert.equal(lines.length, 1); + assert(lines[0].startsWith("Error handling udp event:\nError: boom"), lines[0]); }); it("should forward socket errors", async function() { - await server._bindUdpSocket(); - try { - let waiter = events.once(server, "error"); - server._udpSocket.emit("error", new Error("socket failed")); - let [err] = await waiter; - assert.equal(err.message, "socket failed"); - } finally { - server._resetState(); - } + await server._startUdp(); + let waiter = events.once(server, "error"); + server._udpSocket.emit("error", new Error("socket failed")); + let [err] = await waiter; + assert.equal(err.message, "socket failed"); }); it("should reject sendUdp when sending fails", async function() { - server.luaUdpPort = udpServer.address().port; - await server._bindUdpSocket(); + await server._startUdp(); server._state = "running"; - try { - // Larger than the maximum size of a UDP datagram - await assert.rejects(server.sendUdp(Buffer.alloc(70000)), { code: "EMSGSIZE" }); - } finally { - server._resetState(); - } + // Larger than the maximum size of a UDP datagram + await assert.rejects(server.sendUdp(Buffer.alloc(70000)), { code: "EMSGSIZE" }); }); - it("should close the socket if binding fails", async function() { + it("should throw if the socket is already open", async function() { + await server._startUdp(); + await assert.rejects(server._startUdp(), new Error("UDP socket is already open")); + }); + it("should log an error if opening the socket fails", async function() { const createSocket = dgram.createSocket; dgram.createSocket = (...args) => { const socket = createSocket(...args); socket.bind = () => process.nextTick(() => socket.emit("error", new Error("bind failed"))); return socket; }; + const { lines, restore } = captureLogger("error"); try { - await assert.rejects(server._bindUdpSocket(), new Error("bind failed")); + await server._startUdp(); } finally { dgram.createSocket = createSocket; + restore(); } assert.equal(server._udpSocket, null); + assert(lines[0].startsWith("Failed to open UDP socket:\nError: bind failed"), lines[0]); + }); + it("should close the socket when the server exits", async function() { + await server._startUdp(); + server._server = new events.EventEmitter(); + server._state = "stopping"; + server._watchExit(); + server._server.emit("exit", 0, null); + assert.equal(server._udpSocket, null); + assert.equal(server._state, "init"); + }); + it("should close the socket when the server is killed for hanging", async function() { + await server._startUdp(); + let killed = false; + server._server = { kill: () => { killed = true; } }; + const { restore } = captureLogger("error"); + try { + server.onStopTimeout(); + } finally { + restore(); + server._server = null; + } + assert(killed, "server was not killed"); + assert.equal(server._udpSocket, null); }); }); diff --git a/test/integration/instance.js b/test/integration/instance.js index 56d75e72..79278f80 100644 --- a/test/integration/instance.js +++ b/test/integration/instance.js @@ -158,24 +158,6 @@ describe("Clusterio Instance", function() { const userLeave = await getUser("JoiningPlayer"); assert(!userLeave.instances.has(instId), "Player is not shown as offline"); }); - it("should respond to a player join and leave event sent over UDP", async function() { - // The host port is passed to the module by updateInstanceData on startup - if (!hasLuaUdp() || !scriptCommandsEnabled) { - this.skip(); - } - await sendRcon(instId, - `/sc ${requireApi} api.send_udp("player_event",` + - "{ type='join', name='UdpPlayer' })" - ); - const userJoin = await getUser("UdpPlayer"); - assert(userJoin.instances.has(instId), "Player is not shown as online"); - await sendRcon(instId, - `/sc ${requireApi} api.send_udp("player_event",` + - "{ type='leave', name='UdpPlayer', reason='quit' })" - ); - const userLeave = await getUser("UdpPlayer"); - assert(!userLeave.instances.has(instId), "Player is not shown as offline"); - }); } it("should respond to a player ban and unban event", async function() { await execCtl(`${instSetConfig} factorio.sync_banlist bidirectional`); diff --git a/test/integration/server.js b/test/integration/server.js index e4e9ade2..8c3b8075 100644 --- a/test/integration/server.js +++ b/test/integration/server.js @@ -125,6 +125,8 @@ describe("Integration of host/src/server", function() { let mapPath = server.writePath("saves", "test.zip"); await assert.doesNotReject(fs.access(mapPath), "save is missing"); + // --enable-lua-udp was added in Factorio 2.0.59 + server.enableLuaUdp = lib.integerFullVersion(server.version) >= lib.integerFullVersion("2.0.59"); await server.start("test.zip"); }); }); @@ -159,6 +161,53 @@ describe("Integration of host/src/server", function() { }); }); + describe("Lua UDP", function() { + it("emits udp events for packets sent from the game", async function() { + slowTest(this); + if (!server.enableLuaUdp) { + this.skip(); + } + log("Lua UDP from game"); + + assert(server.hostUdpPort > 0, "host UDP socket is not open"); + let waiter = events.once(server, "udp-test_channel"); + await server.sendRcon(`/sc helpers.send_udp(${server.hostUdpPort}, "test_channel?hello", 0)`); + let [data] = await waiter; + assert.equal(data.toString(), "hello"); + }); + it("delivers packets sent with sendUdp to the game", async function() { + slowTest(this); + if (!server.enableLuaUdp) { + this.skip(); + } + log("Lua UDP to game"); + + await server.sendRcon( + "/sc script.on_event(defines.events.on_udp_packet_received, " + + "function(event) print('udp:' .. event.payload) end)" + ); + let pass = false; + function filter(output) { + if (output.message === "udp:hello") { + pass = true; + } + } + server.on("output", filter); + + await server.sendUdp("hello"); + for (let i = 0; i < 10; i++) { + await server.sendRcon("/sc helpers.recv_udp(0)"); + await lib.wait(10); + if (pass) { + break; + } + } + + server.off("output", filter); + assert(pass, "server did not output the packet payload"); + }); + }); + describe(".stop()", function() { it("stops the server", async function() { slowTest(this); @@ -179,8 +228,6 @@ describe("Integration of host/src/server", function() { slowTest(this); log(".startScenario()"); - // --enable-lua-udp was added in Factorio 2.0.59 - server.enableLuaUdp = lib.integerFullVersion(server.version) >= lib.integerFullVersion("2.0.59"); let pass = false; function filter(output) { if (output.message === "test_scenario init") { @@ -193,7 +240,6 @@ describe("Integration of host/src/server", function() { log(".stop()"); await server.stop(); - server.enableLuaUdp = false; server.off("output", filter); assert(pass, "server did not output line from test scenario");