Use stream.finished for detecting end of streams

There are some corner cases with streams that cause them to not emit
finish close or end events which in turn means that code liseting for
these events directly may end up hanging forever.  Use stream.finished
to detect end of streams which will handle these cases correctly.

packages/create/create.js

@@ -15,6 +15,7 @@ const yargs = require("yargs");
 const { levels, logger, setLogLevel } = require("./logging");
 let dev = false;
 const scriptExt = process.platform === "win32" ? ".cmd" : "";
+const finished = util.promisify(stream.finished);
 
 
 const availablePlugins = [
@@ -480,7 +481,7 @@ async function downloadLinuxServer() {
 		const writeStream = fs.createWriteStream(tmpArchivePath);
 		res.pipe(writeStream);
 
-		await events.once(res, "end");
+		await finished(writeStream);
 
 		await fs.rename(tmpArchivePath, archivePath);
 		try {

packages/ctl/ctl.js

@@ -16,6 +16,8 @@ const setBlocking = require("set-blocking");
 const phin = require("phin");
 const os = require("os");
 const child_process = require("child_process");
+const stream = require("stream");
+const util = require("util");
 
 const libLink = require("@clusterio/lib/link");
 const libErrors = require("@clusterio/lib/errors");
@@ -30,6 +32,8 @@ const libHelpers = require("@clusterio/lib/helpers");
 const libFactorio = require("@clusterio/lib/factorio");
 const libFileOps = require("@clusterio/lib/file_ops");
 
+const finished = util.promisify(stream.finished);
+
 
 function print(...content) {
 	// eslint-disable-next-line no-console
@@ -640,12 +644,12 @@ instanceCommands.add(new libCommand.Command({
 			stream: true,
 		});
 
-		let stream;
+		let writeStream;
 		let tempFilename = args.save.replace(/(\.zip)?$/, ".tmp.zip");
 		while (true) {
 			try {
-				stream = fs.createWriteStream(tempFilename, { flags: "wx" });
-				await events.once(stream, "open");
+				writeStream = fs.createWriteStream(tempFilename, { flags: "wx" });
+				await events.once(writeStream, "open");
 				break;
 			} catch (err) {
 				if (err.code === "EEXIST") {
@@ -655,8 +659,8 @@ instanceCommands.add(new libCommand.Command({
 				}
 			}
 		}
-		response.pipe(stream);
-		await events.once(stream, "finish");
+		response.pipe(writeStream);
+		await finished(writeStream);
 
 		let filename = await libFileOps.findUnusedName(".", args.save, ".zip");
 		await fs.rename(tempFilename, filename);

packages/lib/build_mod.js

@@ -5,8 +5,12 @@ const fs = require("fs-extra");
 const JSZip = require("jszip");
 const klaw = require("klaw");
 const path = require("path");
+const stream = require("stream");
+const util = require("util");
 const yargs = require("yargs");
 
+const finished = util.promisify(stream.finished);
+
 
 async function buildMod(args, info) {
 	if (args.build) {
@@ -42,7 +46,7 @@ async function buildMod(args, info) {
 			let modPath = path.join(args.outputDir, `${modName}.zip`);
 			console.log(`Writing ${modPath}`);
 			let writeStream = zip.generateNodeStream().pipe(fs.createWriteStream(modPath));
-			await events.once(writeStream, "finish");
+			await finished(writeStream);
 
 		} else {
 			let modDir = path.join(args.outputDir, modName);

packages/lib/logging_utils.js

@@ -8,6 +8,8 @@
 const events = require("events");
 const fs = require("fs-extra");
 const path = require("path");
+const stream = require("stream");
+const util = require("util");
 const winston = require("winston");
 const Transport = require("winston-transport");
 const { LEVEL, MESSAGE } = require("triple-beam");
@@ -19,6 +21,8 @@ const libFileOps = require("./file_ops");
 const { levels, logger } = require("./logging");
 const libStream = require("./stream");
 
+const finished = util.promisify(stream.finished);
+
 
 /**
  * Format a parsed Factorio message with colors
@@ -435,7 +439,13 @@ async function queryLog(logDirectory, filter, index) {
 			}
 		});
 		fileStream.pipe(lineStream);
-		await events.once(lineStream, "close");
+		try {
+			await finished(lineStream);
+		} catch (err) {
+			if (log.length < filter.limit || err.code !== "ERR_STREAM_PREMATURE_CLOSE") {
+				throw err;
+			}
+		}
 
 		if (log.length >= filter.limit) {
 			break;
@@ -456,7 +466,7 @@ async function migrateLogs(log, directory, pattern) {
 	async function switchOutput(newOutputPath) {
 		if (output) {
 			output.end();
-			await events.once(output, "close");
+			await finished(output);
 		}
 
 		output = fs.createWriteStream(newOutputPath, { flags: "ax" });

packages/slave/slave.js

@@ -20,6 +20,7 @@ const events = require("events");
 const pidusage = require("pidusage");
 const setBlocking = require("set-blocking");
 const phin = require("phin");
+const stream = require("stream");
 const util = require("util");
 const version = require("./package").version;
 const winston = require("winston");
@@ -40,6 +41,8 @@ const { ConsoleTransport, levels, logger } = require("@clusterio/lib/logging");
 const libLoggingUtils = require("@clusterio/lib/logging_utils");
 const libHelpers = require("@clusterio/lib/helpers");
 
+const finished = util.promisify(stream.finished);
+
 
 const instanceRconCommandDuration = new libPrometheus.Histogram(
 	"clusterio_instance_rcon_command_duration_seconds",
@@ -1478,11 +1481,11 @@ class Slave extends libLink.Link {
 
 		let savesDir = path.join(instanceInfo.path, "saves");
 		let tempFilename = filename.replace(/(\.zip)?$/, ".tmp.zip");
-		let stream;
+		let writeStream;
 		while (true) {
 			try {
-				stream = fs.createWriteStream(path.join(savesDir, tempFilename), { flags: "wx" });
-				await events.once(stream, "open");
+				writeStream = fs.createWriteStream(path.join(savesDir, tempFilename), { flags: "wx" });
+				await events.once(writeStream, "open");
 				break;
 			} catch (err) {
 				if (err.code === "EEXIST") {
@@ -1492,8 +1495,8 @@ class Slave extends libLink.Link {
 				}
 			}
 		}
-		response.pipe(stream);
-		await events.once(stream, "finish");
+		response.pipe(writeStream);
+		await finished(writeStream);
 
 		filename = await libFileOps.findUnusedName(savesDir, filename, ".zip");
 		await fs.rename(path.join(savesDir, tempFilename), path.join(savesDir, filename));

test/lib/stream.js

@@ -3,16 +3,20 @@ const assert = require("assert").strict;
 const events = require("events");
 const fs = require("fs-extra");
 const path = require("path");
+const stream = require("stream");
+const util = require("util");
 
 const libStream = require("@clusterio/lib/stream");
 
+const finished = util.promisify(stream.finished);
+
 
 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;
+			let lineStream = new libStream.LineSplitter({ readableObjectMode: true });
+			lineStream.on("data", line => lines.push(line.toString("utf-8")));
+			return lineStream;
 		}
 
 		it("should split three lines", function() {
@@ -57,9 +61,9 @@ describe("lib/stream", function() {
 
 	describe("class ReverseLineSplitter", function() {
 		function createReverseSplitter(lines) {
-			let stream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
-			stream.on("data", line => lines.push(line.toString("utf-8")));
-			return stream;
+			let lineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
+			lineStream.on("data", line => lines.push(line.toString("utf-8")));
+			return lineStream;
 		}
 
 		it("should split three lines", function() {
@@ -107,16 +111,16 @@ describe("lib/stream", function() {
 			let content = "";
 			for (let i = 0; i < 10; i++) { content += String(i).repeat(10); }
 			await fs.outputFile(path.join("temp", "test", "reverse.txt"), content);
-			let stream = await libStream.createReverseReadStream(
+			let reverseStream = await libStream.createReverseReadStream(
 				path.join("temp", "test", "reverse.txt"),
 				{ encoding: "utf8", highWaterMark: 10 }
 			);
 			let index = 9;
-			stream.on("data", data => {
+			reverseStream.on("data", data => {
 				assert.equal(data, String(index).repeat(10));
 				index -= 1;
 			});
-			await events.once(stream, "end");
+			await finished(reverseStream);
 		});
 
 		it("should reverse the lines of a file", async function() {
@@ -125,7 +129,7 @@ describe("lib/stream", function() {
 				ws.write(Buffer.from(`${i}\n`));
 			}
 			ws.end();
-			await events.once(ws, "finish");
+			await finished(ws);
 
 			let lineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
 			let index = 99999;
@@ -139,7 +143,7 @@ describe("lib/stream", function() {
 				index -= 1;
 			});
 			rs.pipe(lineStream);
-			await events.once(lineStream, "end");
+			await finished(lineStream);
 
 			assert.equal(index, 0, "line check did not reach the end");
 		});