Test for multiple disconnections

Author: Eryk Solecki

packages/host/src/lib/serviceDiscovery/backupingStream.ts

@@ -1,69 +1,73 @@
-import { createReadStream, Stats, fstat, fstatSync, Dir, WriteStream, createWriteStream, openSync, watch } from "fs";
+import { FileHandle, open } from "fs/promises";
 import { Duplex, DuplexOptions } from "stream";
 
-type BackupingStreamOptions = Pick<DuplexOptions, "encoding">
+type BackupingStreamOptions = Pick<DuplexOptions, "encoding" | "highWaterMark">
 
 class BackupingStream extends Duplex {
-    readonly backupDir: Dir;
-    private writeStream: WriteStream;
-    private bytesRead;
-    private size: number;
-    private fd: number;
-    // private readStream: ReadStream;
+    private writeHandle: FileHandle;
+    private readHandle: FileHandle;
+    bytesWritten: number;
+    bytesRead: number;
+    readonly backupFile: string;
 
-    constructor(backupDir: Dir, opts?: BackupingStreamOptions) {
-        super({ ...opts, highWaterMark: 0 });
-        const backupFilePath = `${backupDir.path}/tmp1`;
-
-        this.backupDir = backupDir;
-        this.fd = openSync(backupFilePath, "w+");
+    private constructor(backupFile: string, writeHandle: FileHandle,
+        readHandle: FileHandle, opts?: BackupingStreamOptions) {
+        super({ ...opts });
+        this.backupFile = backupFile;
+        this.writeHandle = writeHandle;
+        this.bytesWritten = 0;
         this.bytesRead = 0;
-        this.size = fstatSync(this.fd).size;
-        this.writeStream = createWriteStream("", { fd: this.fd, encoding: opts?.encoding, autoClose: false });
-        this.attachBackupReadableEvent(this.fd, backupFilePath);
+        this.readHandle = readHandle;
+    }
 
-        this.on("backupReadable", async () => {
-            this.pushFromBackup();
-        });
+    static async create(backupDir: string, opts?: BackupingStreamOptions) {
+        const backupFile = `${backupDir}/tmp1`;
+        const backupWrite = await open(backupFile, "w");
+        const backupRead = await open(backupFile, "r");
+
+        return new BackupingStream(backupFile, backupWrite, backupRead, opts);
     }
-    _write(chunk: any, encoding: BufferEncoding, callback: (error?: Error | null | undefined) => void) {
-        if (this.canSkipBackup()) {
+
+    async _write(chunk: any, encoding: BufferEncoding,
+        callback: (error?: Error | null | undefined) => void): Promise<void> {
+        if (this.bytesInBackup() <= 0 && this.readableLength < this.readableHighWaterMark) {
             this.push(chunk, encoding);
             callback();
             return;
         }
-        this.createBackup(chunk, callback);
-    }
-    _read(_size: number) { }
-
-    canSkipBackup() { return this.readableFlowing === true && this.backupIsEmpty(); }
-    bytesInBackup() { return this.writeStream.bytesWritten - this.bytesRead; }
-    backupIsEmpty() { return this.bytesInBackup() <= 0; }
+        try {
+            const { bytesWritten } = await this.writeBuckup(chunk, encoding);
 
-    createBackup(chunk: any, callback: (error?: Error | null | undefined) => void) {
-        this.writeStream.write(chunk, callback);
+            this.bytesWritten += bytesWritten;
+            callback();
+        } catch (err: any) {
+            callback(err);
+        }
     }
+    async _read(size: number): Promise<void> {
+        if (this.bytesInBackup() <= 0) return;
 
-    private attachBackupReadableEvent(fd: number, backupFilePath: string) {
-        watch(backupFilePath, { persistent: false }, (event) => {
-            if (event !== "change") return;
-            fstat(fd, (err, stats: Stats) => {
-                if (this.size === stats.size) return;
-                this.size = stats.size;
-                this.emit("backupReadable");
-            });
+        const { bytesRead, buffer } = await this.readHandle.read({
+            buffer: Buffer.alloc(size),
+            position: this.bytesRead,
+            length: size,
         });
+
+        this.bytesRead += bytesRead;
+        this.push(buffer.subarray(0, bytesRead));
     }
 
-    private pushFromBackup(size?: number) {
-        const readStream = createReadStream("", { fd: this.fd, autoClose: false, start: this.bytesRead })
-            .on("readable", () => {
-                let chunk;
+    bytesInBackup() { return this.bytesWritten - this.bytesRead; }
+
+    writeBuckup(chunk: any, encoding: BufferEncoding) {
+        if (typeof chunk === "string") {
+            return this.writeHandle.write(chunk, this.bytesWritten, encoding);
+        }
+        return this.writeHandle.write(chunk, 0, undefined, this.bytesWritten);
+    }
 
-                while ((chunk = readStream.read(size)) !== null) {
-                    if (!this.push(chunk)) break;
-                }
-            });
+    async close() {
+        return Promise.all([this.writeHandle.close(), this.readHandle.close()]);
     }
 }
 

packages/host/test/serviceDiscovery/backupingStream.spec.ts

@@ -0,0 +1,183 @@
+import { createReadStream } from "fs";
+import BackupingStream from "../../src/lib/serviceDiscovery/backupingStream";
+import { FileHandle, mkdir, open } from "fs/promises";
+import { resolve } from "path";
+import { PassThrough } from "stream";
+import { createInterface } from "readline";
+import { EOL } from "os";
+
+const testText = `Lorem ipsum dolor sit amet, consectetur adipiscing elit. Duis eget nisl viverra, efficitur eros quis, sodales eros. Suspendisse feugiat ac ipsum non aliquam. Quisque dapibus nisi libero. Fusce euismod lacus vitae eros suscipit semper in non dui. Sed quis porttitor elit. Nulla a sapien mi. Suspendisse tempus vitae lectus a lobortis. Phasellus et gravida purus, et porta mauris. Integer urna nulla, semper nec leo id, ultrices bibendum odio. Orci varius natoque penatibus et magnis dis parturient montes, nascetur ridiculus mus. Aenean eu iaculis dui. Ut ac dolor aliquet, placerat sapien vitae, hendrerit purus.
+
+Proin porta cursus erat, ut dignissim felis. Suspendisse potenti. Duis et aliquet odio. Aliquam congue blandit dolor non dapibus. Vivamus eros dolor, vulputate vitae arcu vel, pretium porttitor lorem. Vivamus eget quam quis arcu lacinia malesuada id a sem. Maecenas in ligula orci. Sed dapibus, urna vel malesuada dictum, lectus ex convallis ante, eget tempor lorem ante ac lectus. Pellentesque habitant morbi tristique senectus et netus et malesuada fames ac turpis egestas. Suspendisse enim ante, laoreet at urna vel, malesuada vestibulum nulla. Morbi orci metus, imperdiet in mattis a, tincidunt vitae tortor. Vivamus at consectetur mauris.
+
+In eu nulla luctus, lacinia nulla sit amet, condimentum leo. Vivamus lectus ex, venenatis in nulla eu, congue rutrum sapien. Cras suscipit sed elit non blandit. Nullam volutpat facilisis elit at laoreet. Aliquam mattis quis sem eget rutrum. Duis sapien dui, tincidunt nec ipsum id, rutrum pulvinar purus. Maecenas ut euismod purus, vel porta lacus. Vestibulum sit amet nulla sit amet diam rhoncus pharetra nec at purus. Donec mattis diam sit amet nisi tincidunt accumsan. Donec a nunc sed tortor dictum finibus. In ultricies lectus non urna convallis mollis. Pellentesque aliquam metus ante, laoreet porttitor mauris sodales et. Orci varius natoque penatibus et magnis dis parturient montes, nascetur ridiculus mus. Nullam quis laoreet nisl. Praesent pulvinar, orci sit amet finibus semper, nibh risus laoreet augue, at blandit tellus nulla ut odio.
+
+Vestibulum bibendum est a massa porta, vitae tempus enim malesuada. Pellentesque fringilla leo non vehicula pretium. Ut maximus ipsum at diam egestas, sit amet volutpat erat iaculis. Fusce dictum ipsum quam, sit amet hendrerit elit gravida a. Nam efficitur leo in mollis eleifend. Cras quis lorem pretium, condimentum libero sed, lacinia neque. Vestibulum tincidunt quam ut congue consectetur. Nullam congue varius bibendum. Maecenas consequat arcu quis magna rhoncus posuere. Cras vitae lectus ut lectus pharetra pulvinar varius vel justo.
+
+Ut congue purus ac lorem scelerisque, id congue eros venenatis. Sed rutrum vitae odio vel malesuada. Aliquam vel nibh at ipsum sodales ultricies. Quisque facilisis neque sem, id accumsan leo gravida vel. Vestibulum eros libero, tempus ac arcu vel, tincidunt pulvinar magna. Aliquam dictum ornare magna sed iaculis. Integer vulputate metus at iaculis eleifend. Aenean vestibulum ut sapien id malesuada. Sed eu faucibus lectus, vitae gravida metus. Mauris et iaculis felis, vitae egestas metus. Integer semper, erat in gravida vehicula, justo neque suscipit est, porta dapibus est magna ac metus.`;
+
+let backupDir: string;
+let testFilePath: string;
+
+beforeAll(async () => {
+    backupDir = resolve(process.cwd(), "./dist/test/");
+    await mkdir(backupDir, { recursive: true });
+
+    testFilePath = resolve(process.cwd(), "./test/serviceDiscovery/testFile.txt");
+});
+
+test("Basic read/write", async () => {
+    const backupingStream = await BackupingStream.create(backupDir, { encoding: "ascii" });
+
+    const readingFinished = new Promise(res => { backupingStream.on("readable", () => { res(backupingStream.read()); }); });
+
+    backupingStream.write(testText);
+    const result = await readingFinished;
+
+    expect(result).toBe(testText);
+    await backupingStream.close();
+});
+
+test("Simple piped read/write", async () => {
+    const provider = new PassThrough({ encoding: "ascii" });
+    const consumer = new PassThrough({ encoding: "ascii" });
+    const backupingStream = await BackupingStream.create(backupDir, { encoding: "ascii" });
+
+    provider.pipe(backupingStream).pipe(consumer);
+
+    const readPromise = new Promise(res => consumer.on("readable", () => {
+        res(consumer.read());
+    }));
+
+    provider.write(testText);
+    const consumerRead = await readPromise;
+
+    expect(consumerRead).toBe(testText);
+    await backupingStream.close();
+});
+
+test("Write to backup", async () => {
+    const backupingStream = await BackupingStream.create(backupDir, { highWaterMark: 0 });
+
+    const source = createReadStream(testFilePath);
+    const readLine = createInterface({ input: source });
+
+    let lastLineWritten = new Promise<void>(res => { res(); });
+
+    const inputFinished = new Promise<void>((res) => {
+        readLine
+            .on("close", res)
+            .on("line", (input: string) => {
+                lastLineWritten = new Promise(wrtieRes => {
+                    backupingStream.write(input);
+                    backupingStream.write(EOL, undefined, () => wrtieRes());
+                });
+            });
+    });
+
+    await inputFinished;
+    await lastLineWritten;
+
+    const [inputFile, backupFile] = await Promise.all([open(testFilePath), open(backupingStream.backupFile)]);
+    const [inputBuff, outputBuff] = await Promise.all([inputFile.readFile(), backupFile.readFile()]);
+
+    const equals = inputBuff.equals(outputBuff);
+
+    expect(equals).toBeTruthy();
+    await Promise.all([backupingStream.close(), inputFile.close(), backupFile.close()]);
+});
+
+const waitForFileToReachSize = async (file: FileHandle, size: number) => {
+    let timeoutOccured = false;
+    const timeout = setTimeout(() => { timeoutOccured = true; }, 200);
+
+    while (!timeoutOccured) {
+        const stats = await file.stat();
+
+        if (stats.size === size) break;
+    }
+    clearTimeout(timeout);
+};
+
+test("Pipe to consumer from backup", async () => {
+    const backupingStream = await BackupingStream.create(backupDir);
+
+    const source = createReadStream(testFilePath);
+    const readLine = createInterface({ input: source });
+
+    let lastLineWritten = new Promise<void>(res => { res(); });
+
+    const inputFinished = new Promise<void>((res) => {
+        readLine
+            .on("close", res)
+            .on("line", (input: string) => {
+                lastLineWritten = new Promise(wrtieRes => {
+                    backupingStream.write(input);
+                    backupingStream.write(EOL, undefined, () => wrtieRes());
+                });
+            });
+    });
+
+    await inputFinished;
+    await lastLineWritten;
+
+    const outputFilePath = resolve(backupDir, "./testBackupingStream_output.txt");
+    const outputFile = await open(outputFilePath, "w+");
+    const consumer = outputFile.createWriteStream();
+
+    backupingStream.pipe(consumer);
+
+    const [inputFile, resultFile] = await Promise.all([open(testFilePath), open(outputFilePath)]);
+
+    const inputFileStat = await inputFile.stat();
+
+    await waitForFileToReachSize(outputFile, inputFileStat.size);
+
+    const [inputBuff, resultBuff] = await Promise.all([inputFile.readFile(), resultFile.readFile()]);
+
+    const equals = inputBuff.equals(resultBuff);
+
+    expect(equals).toBeTruthy();
+
+    await Promise.all([backupingStream.close(), inputFile.close(), outputFile.close(), resultFile.close()]);
+});
+
+test("Multiple disconnections of consumer", async () => {
+    const outputPath = resolve(backupDir, "./testBackupingStream_long_output.txt");
+    const [source, output] = await Promise.all([open(testFilePath, "r"), open(outputPath, "w+")]);
+    const sourceSize = (await source.stat()).size;
+
+    const provider = source.createReadStream({ highWaterMark: 50 });
+    const backupingStream = await BackupingStream.create(backupDir, { highWaterMark: 100 });
+    const consumer = output.createWriteStream();
+
+    const providerEnd = new Promise(res => { provider.on("end", res); });
+
+    let readLen = 0;
+    let piped = true;
+    let switchLenght = sourceSize / 5;
+
+    provider.on("data", (chunk) => {
+        readLen += chunk.length;
+
+        if (readLen > switchLenght) {
+            // eslint-disable-next-line no-unused-expressions
+            piped ? backupingStream.unpipe() : backupingStream.pipe(consumer);
+            piped = !piped;
+            switchLenght += sourceSize / 5;
+        }
+    });
+
+    provider.pipe(backupingStream).pipe(consumer);
+
+    await providerEnd;
+    await waitForFileToReachSize(output, sourceSize);
+
+    const [sourceBuff, outputBuff] = await Promise.all([source.readFile(), output.readFile()]);
+
+    const equals = sourceBuff.equals(outputBuff);
+
+    expect(equals).toBeTruthy();
+
+    await Promise.all([backupingStream.close(), source.close(), output.close()]);
+});