From 285d71e13d419c765267b16a40dc169590cc097e Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Sat, 11 Dec 2021 21:32:25 +0100 Subject: [PATCH] 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 | 104 +++++++++++++++++++++++++++++++++++++++++ test/lib/stream.js | 90 +++++++++++++++++++++++++++++++++++ 2 files changed, 194 insertions(+) diff --git a/packages/lib/stream.js b/packages/lib/stream.js index 535026a8..98787397 100644 --- a/packages/lib/stream.js +++ b/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, }; diff --git a/test/lib/stream.js b/test/lib/stream.js index 40d45fc3..721ed090 100644 --- a/test/lib/stream.js +++ b/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"); + }); + }); });