Add ReverseLineSplitter

Add splitter based on LineSplitter which splits lines in reverse from a
reverse stream created from the added createReverseReadStream function.
Enables line processing of files starting from the last line and going
towards the first.

packages/lib/stream.js

@@ -59,6 +59,110 @@ class LineSplitter extends stream.Transform {
 	}
 }
 
+/**
+ * Transform splitting a reverse stream into lines
+ *
+ * Splits up a reverse stream of chunks created by {@link
+ * module:lib/stream.createReverseReadStream} by newlines and passes on the
+ * lines one by one in reverse order removing the line feed and optional
+ * carriage return from the lines.
+ *
+ * @example
+ * let revLineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
+ * revLineStream.on("data", line => { console.log(line.toString("utf8")); });
+ * let revFileStream = await libStream.createReverseReadStream(path);
+ * revFileStream.pipe(revLineStream);
+ *
+ * @memberof module:lib/stream
+ */
+class ReverseLineSplitter extends stream.Transform {
+	constructor(options) {
+		super(options);
+		this._partial = null;
+	}
+
+	_transform(chunk, encoding, callback) {
+		if (this._partial) {
+			chunk = Buffer.concat([chunk, this._partial]);
+			this._partial = null;
+		}
+
+		while (chunk.length) {
+			let next = chunk.lastIndexOf("\n", -2);
+			if (next === -1) {
+				this._partial = chunk;
+				break;
+			}
+
+			let end = chunk.length - 1;
+			if (chunk[end] === "\n".charCodeAt(0)) {
+				end -= 1;
+				// Eat carriage return as well if present
+				if (end - next > 0 && chunk[end] === "\r".charCodeAt(0)) {
+					end -= 1;
+				}
+			}
+
+			let line = chunk.slice(next + 1, end + 1);
+			chunk = chunk.slice(0, next + 1);
+			this.push(line);
+		}
+		callback();
+	}
+
+	_flush(callback) {
+		if (this._partial) {
+			let end = this._partial.length - 1;
+			if (this._partial[end] === "\n".charCodeAt(0)) {
+				end -= 1;
+				// Eat carriage return as well if present
+				if (end > 0 && this._partial[end] === "\r".charCodeAt(0)) {
+					end -= 1;
+				}
+			}
+			this.push(this._partial.slice(0, end + 1));
+			this._partial = null;
+		}
+		callback();
+	}
+}
+
+
+/**
+ * Create a file read stream running in reverse
+ *
+ * Like fs.createReadStream but provides data chunks starting from the end
+ * of the file and progressing towards the start of the file.  The chunks
+ * are not reversed, thus the file content can be recreated by concatenating
+ * the chunks together in the reverse order they are read.
+ *
+ * @param {string} path -
+ *     Path to file to open. May be anything accepted by fs.open.
+ * @param {object} options -
+ *     Options to pass to fs.createReadStream.  Values passed for `fd` and
+ *     `fs` will be ignored.
+ * @memberof module:lib/stream
+ */
+async function createReverseReadStream(path, options) {
+	const fileFd = await fs.open(path, "r");
+	const fileSize = (await fs.fstat(fileFd)).size;
+	let filePosition = fileSize;
+	const reverseFs = {
+		read(fd, buffer, offset, length, position, callback) {
+			assert(fd === fileFd);
+			assert(position === undefined);
+			length = Math.min(length, filePosition);
+			filePosition -= length;
+			return fs.read(fd, buffer, offset, length, filePosition, callback);
+		},
+		open() { assert(false); },
+		close(fd, callback) { return fs.close(fd, callback); },
+	};
+	return fs.createReadStream("", { ...options, fd: fileFd, fs: reverseFs });
+}
+
 module.exports = {
 	LineSplitter,
+	ReverseLineSplitter,
+	createReverseReadStream,
 };

test/lib/stream.js

@@ -54,4 +54,94 @@ describe("lib/stream", function() {
 			assert.deepEqual(lines, ["part 1 part 2 part 3"]);
 		});
 	});
+
+	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;
+		}
+
+		it("should split three lines", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("line 1\nline 2\nline 3\n"));
+			ls.end();
+			assert.deepEqual(lines, ["line 3", "line 2", "line 1"]);
+		});
+		it("should split three Windows line endings lines", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("line 1\r\nline 2\r\nline 3\r\n"));
+			ls.end();
+			assert.deepEqual(lines, ["line 3", "line 2", "line 1"]);
+		});
+		it("should handle mixed line endings", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("1\n\n2\r\n\r\n3\n4"));
+			ls.end();
+			assert.deepEqual(lines, ["4", "3", "", "2", "", "1"]);
+		});
+		it("should give the first line on .end()", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("line a\nline b"));
+			assert.deepEqual(lines, ["line b"]);
+			ls.end();
+			assert.deepEqual(lines, ["line b", "line a"]);
+		});
+		it("should handled partial lines", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("part 3\n"));
+			ls.write(Buffer.from(" part 2 "));
+			ls.write(Buffer.from("part 1"));
+			ls.end();
+			assert.deepEqual(lines, ["part 1 part 2 part 3"]);
+		});
+	});
+
+	describe("createReverseReadStream()", function() {
+		it("should read chunks in a file in reverse", async 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(
+				path.join("temp", "test", "reverse.txt"),
+				{ encoding: "utf8", highWaterMark: 10 }
+			);
+			let index = 9;
+			stream.on("data", data => {
+				assert.equal(data, String(index).repeat(10));
+				index -= 1;
+			});
+			await events.once(stream, "end");
+		});
+
+		it("should reverse the lines of a file", async function() {
+			let ws = fs.createWriteStream(path.join("temp", "test", "reverse.txt"));
+			for (let i = 1; i < 100000; i++) {
+				ws.write(Buffer.from(`${i}\n`));
+			}
+			ws.end();
+			await events.once(ws, "finish");
+
+			let lineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
+			let index = 99999;
+			let rs = await libStream.createReverseReadStream(path.join("temp", "test", "reverse.txt"));
+			lineStream.on("data", line => {
+				let n = Number(line.toString());
+				if (index !== n) {
+					rs.destroy();
+					assert.equal(n, index);
+				}
+				index -= 1;
+			});
+			rs.pipe(lineStream);
+			await events.once(lineStream, "end");
+
+			assert.equal(index, 0, "line check did not reach the end");
+		});
+	});
 });