From 01128979db9b9afaff0dda3137c24104f6d38f05 Mon Sep 17 00:00:00 2001
From: Tau <tau@nowhere>
Date: Wed, 11 Mar 2020 17:23:44 -0400
Subject: [PATCH] util: Add promise-based stream wrapper

You'd think something like this would exist on npm somewhere.
---
 src/util/stream/index.ts  |  4 ++
 src/util/stream/reader.ts | 78 +++++++++++++++++++++++++++++++++++++++
 src/util/stream/stream.ts | 41 ++++++++++++++++++++
 src/util/stream/types.ts  | 13 +++++++
 src/util/stream/writer.ts | 25 +++++++++++++
 5 files changed, 161 insertions(+)
 create mode 100644 src/util/stream/index.ts
 create mode 100644 src/util/stream/reader.ts
 create mode 100644 src/util/stream/stream.ts
 create mode 100644 src/util/stream/types.ts
 create mode 100644 src/util/stream/writer.ts

diff --git a/src/util/stream/index.ts b/src/util/stream/index.ts
new file mode 100644
index 0000000..57b712c
--- /dev/null
+++ b/src/util/stream/index.ts
@@ -0,0 +1,4 @@
+import IoByteStream from "./stream";
+
+export * from "./types";
+export default IoByteStream;
diff --git a/src/util/stream/reader.ts b/src/util/stream/reader.ts
new file mode 100644
index 0000000..3ae187d
--- /dev/null
+++ b/src/util/stream/reader.ts
@@ -0,0 +1,78 @@
+import { Readable } from "stream";
+
+export default function makeReader(stm: Readable) {
+  let remain = Buffer.alloc(0);
+  let busy = false;
+
+  return function read(nbytes: number): Promise<Buffer> {
+    return new Promise((resolve, reject) => {
+      if (busy) {
+        return reject(new Error("Already reading"));
+      }
+
+      //
+      // Synchronous completion case
+      //
+
+      const result = Buffer.alloc(nbytes);
+      const step = Math.min(remain.length, nbytes);
+
+      remain.copy(result, 0, 0, step);
+      remain = remain.slice(step);
+
+      if (step === nbytes) {
+        return resolve(result);
+      }
+
+      //
+      // Asynchronous completion case
+      //
+
+      let pos = step;
+      let finish: () => void;
+
+      const onReadable = () => {
+        while (true) {
+          const chunk = stm.read() as (Buffer | null);
+
+          if (chunk === null) {
+            break;
+          }
+
+          const step = Math.min(result.length - pos, chunk.length);
+
+          chunk.copy(result, pos, 0, step);
+          remain = chunk.slice(step);
+          pos += step;
+
+          if (pos === result.length) {
+            finish();
+            resolve(result);
+          }
+        }
+      };
+
+      const onEnd = () => {
+        finish();
+        resolve(result.slice(0, pos));
+      };
+
+      const onError = error => {
+        finish();
+        reject(error);
+      };
+
+      finish = () => {
+        stm.off("end", onEnd);
+        stm.off("error", onError);
+        stm.off("readable", onReadable);
+        busy = false;
+      };
+
+      busy = true;
+      stm.on("end", onEnd);
+      stm.on("error", onError);
+      stm.on("readable", onReadable);
+    });
+  };
+}
diff --git a/src/util/stream/stream.ts b/src/util/stream/stream.ts
new file mode 100644
index 0000000..2696f65
--- /dev/null
+++ b/src/util/stream/stream.ts
@@ -0,0 +1,41 @@
+import { Duplex } from "stream";
+
+import makeReader from "./reader";
+import { ByteStream } from "./types";
+import makeWriter from "./writer";
+
+export default class IoByteStream implements ByteStream {
+  private _stm?: Duplex;
+  private readonly _read: (nbytes: number) => Promise<Buffer>;
+  private readonly _write: (buf: Buffer) => Promise<void>;
+
+  constructor(stm: Duplex) {
+    this._stm = stm;
+    this._read = makeReader(stm);
+    this._write = makeWriter(stm);
+  }
+
+  close() {
+    if (this._stm !== undefined) {
+      this._stm.destroy();
+    }
+
+    this._stm = undefined;
+  }
+
+  read(nbytes: number): Promise<Buffer> {
+    if (this._stm === undefined) {
+      throw new Error("Stream is closed");
+    }
+
+    return this._read(nbytes);
+  }
+
+  write(buf: Buffer): Promise<void> {
+    if (this._stm === undefined) {
+      throw new Error("Stream is closed");
+    }
+
+    return this._write(buf);
+  }
+}
diff --git a/src/util/stream/types.ts b/src/util/stream/types.ts
new file mode 100644
index 0000000..b414e1e
--- /dev/null
+++ b/src/util/stream/types.ts
@@ -0,0 +1,13 @@
+export interface ReadableByteStream {
+  close(): void;
+
+  read(nbytes: number): Promise<Buffer>;
+}
+
+export interface WritableByteStream {
+  close(): void;
+
+  write(buf: Buffer): Promise<void>;
+}
+
+export type ByteStream = ReadableByteStream & WritableByteStream;
diff --git a/src/util/stream/writer.ts b/src/util/stream/writer.ts
new file mode 100644
index 0000000..92096c8
--- /dev/null
+++ b/src/util/stream/writer.ts
@@ -0,0 +1,25 @@
+import { Writable } from "stream";
+
+export default function makeWriter(stm: Writable) {
+  let busy = false;
+
+  return function write(buf: Buffer): Promise<void> {
+    return new Promise((resolve, reject) => {
+      if (busy) {
+        return reject(new Error("Already writing"));
+      }
+
+      busy = true;
+
+      stm.write(buf, error => {
+        busy = false;
+
+        if (error !== undefined) {
+          reject(error);
+        } else {
+          resolve();
+        }
+      });
+    });
+  };
+}