From 8de76b5cfe36d5193fe6fe95ed64ff96c7e9737f Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Sat, 11 Dec 2021 21:26:22 +0100 Subject: [PATCH] Refactor LineSplitter Pull the LineSplitter class out of lib/factorio to its own stream module and refactor it to be based on stream.Transform. This allows composing with other streams by piping it around. --- packages/lib/factorio/server.js | 91 +++------------------------------ packages/lib/stream.js | 64 +++++++++++++++++++++++ test/integration/index.js | 14 ++--- test/lib/factorio/server.js | 37 -------------- test/lib/stream.js | 57 +++++++++++++++++++++ 5 files changed, 135 insertions(+), 128 deletions(-) create mode 100644 packages/lib/stream.js create mode 100644 test/lib/stream.js diff --git a/packages/lib/factorio/server.js b/packages/lib/factorio/server.js index e1766281..ba7dd424 100644 --- a/packages/lib/factorio/server.js +++ b/packages/lib/factorio/server.js @@ -13,6 +13,7 @@ const libFileOps = require("../file_ops"); const libIni = require("../ini"); const { logger } = require("../logging"); const { escapeRegExp } = require("../helpers"); +const { LineSplitter } = require("../stream"); /** @@ -194,82 +195,6 @@ async function generatePassword(length) { } } -/** - * Line splitter for chunked data - * - * Splits a stream of bytes into lines terminated by either line feed or - * carriage return followed by linefeed. Each line sepparated is passed - * to the callback one at a time without the line terminator included. - * - * Data is streamed via the data method, and once the stream ends the - * end method should be called. This ensures an unterminated line at - * the end of the stream is passed to the callback. - * - * @memberof module:lib/factorio - * @private - * @inner - */ -class LineSplitter { - - /** - * Create a line splitter - * - * @param {function(Buffer)} callback - Function called for every line. - */ - constructor(callback) { - this._callback = callback; - this._partial = null; - } - - /** - * Stream in bytes - * - * Input bytes to split lines over. Will invoke the callback for every - * line found in buf and keep any potential partial line at the end of - * buf for the next invocation. - * - * @param {Buffer} buf - Data to stream in. - */ - data(buf) { - if (this._partial) { - buf = Buffer.concat([this._partial, buf]); - this._partial = null; - } - - while (buf.length) { - let end = buf.indexOf("\n"); - if (end === -1) { - this._partial = buf; - break; - } - - let next = end + 1; - // Eat carriage return as well if present - if (end >= 1 && buf[end-1] === "\r".charCodeAt(0)) { - end -= 1; - } - - let line = buf.slice(0, end); - buf = buf.slice(next); - this._callback(line); - } - } - - /** - * Mark end of stream - * - * Signal the end of the bytestream. If the previous invocation of the - * data method ended with a partial line this will invoke the callback - * with this line. - */ - end() { - if (this._partial) { - this._callback(this._partial); - this._partial = null; - } - } -} - /** * Interpret lines of output from Factorio * @@ -563,9 +488,6 @@ class FactorioServer extends events.EventEmitter { // Array of possible causes for an unexpected shutdown this._unexpected = []; - this._stdout = new LineSplitter((line) => { this._handleOutput(line, "stdout"); }); - this._stderr = new LineSplitter((line) => { this._handleOutput(line, "stderr"); }); - // Track autosaving this._runningAutosave = null; this.on("_autosave", name => { @@ -761,10 +683,12 @@ class FactorioServer extends events.EventEmitter { } _attachStdio() { - this._server.stdout.on("data", chunk => { this._stdout.data(chunk); }); - this._server.stdout.on("close", () => { this._stdout.end(); }); - this._server.stderr.on("data", chunk => { this._stderr.data(chunk); }); - this._server.stderr.on("close", () => { this._stderr.end(); }); + let stdout = new LineSplitter({ readableObjectMode: true }); + stdout.on("data", line => { this._handleOutput(line, "stdout"); }); + this._server.stdout.pipe(stdout); + let stderr = new LineSplitter({ readableObjectMode: true }); + stderr.on("data", line => { this._handleOutput(line, "stderr"); }); + this._server.stderr.pipe(stderr); } _watchExit() { @@ -1201,6 +1125,5 @@ module.exports = { _findVersion: findVersion, _randomDynamicPort: randomDynamicPort, _generatePassword: generatePassword, - _LineSplitter: LineSplitter, _parseOutput: parseOutput, }; diff --git a/packages/lib/stream.js b/packages/lib/stream.js new file mode 100644 index 00000000..535026a8 --- /dev/null +++ b/packages/lib/stream.js @@ -0,0 +1,64 @@ +"use strict"; +const assert = require("assert"); +const fs = require("fs-extra"); +const stream = require("stream"); + + +/** + * Transform splitting a stream into lines + * + * Splits up a stream of chunks by newlines and passes on the lines one by + * one removing the line feed and optional carriage return from the lines. + * + * @example + * let lineStream = new libStream.LineSplitter({ readableObjectMode: true }); + * lineStream.on("data", line => { console.log(line.toString("utf8")); }); + * let fileStream = fs.createReadStream(path); + * fileStream.pipe(lineStream); + * + * @memberof module:lib/stream + */ +class LineSplitter extends stream.Transform { + constructor(options) { + super(options); + this._partial = null; + } + + _transform(chunk, encoding, callback) { + if (this._partial) { + chunk = Buffer.concat([this._partial, chunk]); + this._partial = null; + } + + while (chunk.length) { + let end = chunk.indexOf("\n"); + if (end === -1) { + this._partial = chunk; + break; + } + + let next = end + 1; + // Eat carriage return as well if present + if (end >= 1 && chunk[end-1] === "\r".charCodeAt(0)) { + end -= 1; + } + + let line = chunk.slice(0, end); + chunk = chunk.slice(next); + this.push(line); + } + callback(); + } + + _flush(callback) { + if (this._partial) { + this.push(this._partial); + this._partial = null; + } + callback(); + } +} + +module.exports = { + LineSplitter, +}; diff --git a/test/integration/index.js b/test/integration/index.js index 0bc08be8..b254e05f 100644 --- a/test/integration/index.js +++ b/test/integration/index.js @@ -9,7 +9,7 @@ const util = require("util"); const events = require("events"); const libLink = require("@clusterio/lib/link"); -const server = require("@clusterio/lib/factorio/server"); +const { LineSplitter } = require("@clusterio/lib/stream"); const { ConsoleTransport, logger } = require("@clusterio/lib/logging"); const libLoggingUtils = require("@clusterio/lib/logging_utils"); @@ -133,18 +133,18 @@ function spawn(name, cmd, waitFor) { console.log(cmd); let parts = cmd.split(" "); let process = child_process.spawn(parts[0], parts.slice(1), { cwd: path.join("temp", "test") }); - let stdout = new server._LineSplitter((line) => { + let stdout = new LineSplitter({ readableObjectMode: true }); + stdout.on("data", line => { line = line.toString("utf8"); if (waitFor.test(line)) { resolve(process); } console.log(name, line); }); - let stderr = new server._LineSplitter((line) => { console.log(name, line.toString("utf8")); }); - process.stdout.on("data", chunk => { stdout.data(chunk); }); - process.stdout.on("close", () => { stdout.end(); }); - process.stderr.on("data", chunk => { stderr.data(chunk); }); - process.stderr.on("close", () => { stderr.end(); }); + let stderr = new LineSplitter({ readableObjectMode: true }); + stderr.on("data", line => { console.log(name, line.toString("utf8")); }); + process.stdout.pipe(stdout); + process.stderr.pipe(stderr); }); } diff --git a/test/lib/factorio/server.js b/test/lib/factorio/server.js index c62186a3..c6d05bab 100644 --- a/test/lib/factorio/server.js +++ b/test/lib/factorio/server.js @@ -119,43 +119,6 @@ describe("lib/factorio/server", function() { }); }); - describe("class LineSplitter", function() { - function createSplitter(lines) { - return new libFactorio._LineSplitter(line => lines.push(line.toString("utf-8"))); - } - - it("should split three lines", function() { - let lines = []; - let ls = createSplitter(lines); - ls.data(Buffer.from("line 1\nline 2\nline 3\n")); - ls.end(); - assert.deepEqual(lines, ["line 1", "line 2", "line 3"]); - }); - it("should split three Windows line endings lines", function() { - let lines = []; - let ls = createSplitter(lines); - ls.data(Buffer.from("line 1\r\nline 2\r\nline 3\r\n")); - assert.deepEqual(lines, ["line 1", "line 2", "line 3"]); - }); - it("should give the last non-terminated line on .end()", function() { - let lines = []; - let ls = createSplitter(lines); - ls.data(Buffer.from("line a\nline b")); - assert.deepEqual(lines, ["line a"]); - ls.end(); - assert.deepEqual(lines, ["line a", "line b"]); - }); - it("should handled partial lines", function() { - let lines = []; - let ls = createSplitter(lines); - ls.data(Buffer.from("part 1")); - ls.data(Buffer.from(" part 2 ")); - ls.data(Buffer.from("part 3\n")); - ls.end(); - assert.deepEqual(lines, ["part 1 part 2 part 3"]); - }); - }); - describe("parseOutput()", function() { it("should parse the test lines", function() { for (let [line, reference] of testLines) { diff --git a/test/lib/stream.js b/test/lib/stream.js new file mode 100644 index 00000000..40d45fc3 --- /dev/null +++ b/test/lib/stream.js @@ -0,0 +1,57 @@ +"use strict"; +const assert = require("assert").strict; +const events = require("events"); +const fs = require("fs-extra"); +const path = require("path"); + +const libStream = require("@clusterio/lib/stream"); + + +describe("lib/stream", function() { + describe("class LineSplitter", function() { + function createSplitter(lines) { + let stream = new libStream.LineSplitter({ readableObjectMode: true }); + stream.on("data", line => lines.push(line.toString("utf-8"))); + return stream; + } + + it("should split three lines", function() { + let lines = []; + let ls = createSplitter(lines); + ls.write(Buffer.from("line 1\nline 2\nline 3\n")); + ls.end(); + assert.deepEqual(lines, ["line 1", "line 2", "line 3"]); + }); + it("should split three Windows line endings lines", function() { + let lines = []; + let ls = createSplitter(lines); + ls.write(Buffer.from("line 1\r\nline 2\r\nline 3\r\n")); + ls.end(); + assert.deepEqual(lines, ["line 1", "line 2", "line 3"]); + }); + it("should handle mixed line endings", function() { + let lines = []; + let ls = createSplitter(lines); + ls.write(Buffer.from("1\n\n2\r\n\r\n3\n4")); + ls.end(); + assert.deepEqual(lines, ["1", "", "2", "", "3", "4"]); + }); + it("should give the last non-terminated line on .end()", function() { + let lines = []; + let ls = createSplitter(lines); + ls.write(Buffer.from("line a\nline b")); + assert.deepEqual(lines, ["line a"]); + ls.end(); + assert.deepEqual(lines, ["line a", "line b"]); + }); + it("should handled partial lines", function() { + let lines = []; + let ls = createSplitter(lines); + ls.write(Buffer.from("part 1")); + ls.write(Buffer.from(" part 2 ")); + ls.write(Buffer.from("part 3\n")); + ls.end(); + assert.deepEqual(lines, ["part 1 part 2 part 3"]); + }); + }); +});