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

@@ -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,
 };

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,
+};

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);
 	});
 }
 

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) {

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"]);
+		});
+	});
+});