Generalized concept of AppendableBuffer

Author: calebsander

compiled/download.js

@@ -40,7 +40,7 @@ export declare type ValueCallback<E> = (err: Error | null, value: E | null) => v
 export declare type TypeAndValueCallback<E> = (err: Error | null, type: Type<E> | null, value: E | null) => void;
 /**
  * Writes the contents of `type.toBuffer()` ([[Type.toBuffer]])
- * followed by a null byte to a writable stream.
+ * to a writable stream and then closes the stream.
  * Calls `callback` when done.
  *
  * Example:
@@ -65,7 +65,7 @@ export declare type TypeAndValueCallback<E> = (err: Error | null, type: Type<E>
 export declare function writeType({type, outStream}: WriteParams<any>, callback?: ErrCallback): Writable;
 /**
  * Writes the contents of `type.valueBuffer(value)` ([[Type.valueBuffer]])
- * followed by a null byte to a writable stream.
+ * to a writable stream and then closes the stream.
  * Calls `callback` when done.
  *
  * Example:
@@ -91,7 +91,7 @@ export declare function writeValue<E>({type, value, outStream}: WriteTypeValuePa
 /**
  * Writes the contents of `type.toBuffer()` ([[Type.toBuffer]]),
  * followed by the contents of `type.valueBuffer(value)` ([[Type.valueBuffer]]),
- * and then a null byte to a writable stream.
+ * to a writable stream and then closes the stream.
  * Calls `callback` when done.
  *
  * Example:

compiled/upload-download.js

@@ -6,18 +6,16 @@ const http = require("http");
 const stream_1 = require("stream");
 const zlib = require("zlib");
 const accepts = require("accepts");
+const appendable_stream_1 = require("./lib/appendable-stream");
 const assert_1 = require("./lib/assert");
-const buffer_stream_1 = require("./lib/buffer-stream");
-const growable_buffer_1 = require("./lib/growable-buffer");
 const r = require("./read");
 const abstract_1 = require("./types/abstract");
 function toArrayBuffer(buffer) {
     return buffer.buffer.slice(buffer.byteOffset, buffer.byteOffset + buffer.byteLength);
 }
-const WRITABLE_STREAMS = [stream_1.Writable, stream_1.Duplex, http.OutgoingMessage];
 /**
  * Writes the contents of `type.toBuffer()` ([[Type.toBuffer]])
- * followed by a null byte to a writable stream.
+ * to a writable stream and then closes the stream.
  * Calls `callback` when done.
  *
  * Example:
@@ -41,30 +39,25 @@ const WRITABLE_STREAMS = [stream_1.Writable, stream_1.Duplex, http.OutgoingMessa
  */
 function writeType({ type, outStream }, callback) {
     assert_1.default.instanceOf(type, abstract_1.default);
-    assert_1.default.instanceOf(outStream, WRITABLE_STREAMS);
     if (callback === undefined)
         callback = () => { };
     assert_1.default.instanceOf(callback, Function);
-    let typeBuffer;
+    const typeStream = new appendable_stream_1.default(outStream);
+    outStream.on('error', callback);
     try {
-        typeBuffer = type.toBuffer();
+        type.addToBuffer(typeStream);
+        outStream.on('finish', () => callback(null));
     }
     catch (err) {
         callback(err);
-        return outStream;
     }
-    const typeStream = new buffer_stream_1.default(typeBuffer);
-    return typeStream.pipe(outStream)
-        .on('error', function (err) {
-        this.end();
-        callback(err);
-    })
-        .on('finish', () => callback(null));
+    typeStream.end();
+    return outStream;
 }
 exports.writeType = writeType;
 /**
  * Writes the contents of `type.valueBuffer(value)` ([[Type.valueBuffer]])
- * followed by a null byte to a writable stream.
+ * to a writable stream and then closes the stream.
  * Calls `callback` when done.
  *
  * Example:
@@ -88,30 +81,26 @@ exports.writeType = writeType;
  */
 function writeValue({ type, value, outStream }, callback) {
     assert_1.default.instanceOf(type, abstract_1.default);
-    assert_1.default.instanceOf(outStream, WRITABLE_STREAMS);
     if (callback === undefined)
         callback = () => { };
     assert_1.default.instanceOf(callback, Function);
-    const valueBuffer = new growable_buffer_1.default;
+    const valueStream = new appendable_stream_1.default(outStream);
+    outStream.on('error', callback);
     try {
-        type.writeValue(valueBuffer, value);
+        type.writeValue(valueStream, value);
+        outStream.on('finish', () => callback(null));
     }
     catch (err) {
         callback(err);
-        return outStream;
     }
-    return new buffer_stream_1.default(valueBuffer).pipe(outStream)
-        .on('error', function (err) {
-        this.end();
-        callback(err);
-    })
-        .on('finish', () => callback(null));
+    valueStream.end();
+    return outStream;
 }
 exports.writeValue = writeValue;
 /**
  * Writes the contents of `type.toBuffer()` ([[Type.toBuffer]]),
  * followed by the contents of `type.valueBuffer(value)` ([[Type.valueBuffer]]),
- * and then a null byte to a writable stream.
+ * to a writable stream and then closes the stream.
  * Calls `callback` when done.
  *
  * Example:
@@ -135,26 +124,20 @@ exports.writeValue = writeValue;
  */
 function writeTypeAndValue({ type, value, outStream }, callback) {
     assert_1.default.instanceOf(type, abstract_1.default);
-    assert_1.default.instanceOf(outStream, WRITABLE_STREAMS);
     if (callback === undefined)
         callback = () => { };
     assert_1.default.instanceOf(callback, Function);
-    let typeBuffer;
+    const typeValueStream = new appendable_stream_1.default(outStream);
+    outStream.on('error', callback);
     try {
-        typeBuffer = type.toBuffer();
+        type.addToBuffer(typeValueStream);
+        type.writeValue(typeValueStream, value);
+        outStream.on('finish', () => callback(null));
     }
     catch (err) {
         callback(err);
-        return outStream;
     }
-    const typeStream = new buffer_stream_1.default(typeBuffer);
-    typeStream.pipe(outStream, { end: false })
-        .on('error', function () {
-        this.end();
-    });
-    typeStream.on('bs-written', () => {
-        writeValue({ type, value, outStream }, callback);
-    });
+    typeValueStream.end();
     return outStream;
 }
 exports.writeTypeAndValue = writeTypeAndValue;

compiled/upload.js

@@ -0,0 +1,38 @@
+/// <reference types="node" />
+import { Writable } from 'stream';
+import AppendableBuffer from './appendable';
+/**
+ * A wrapper around a writable stream
+ * to implement [[AppendableBuffer]].
+ * The stream must be explicitly closed
+ * by calling [[AppendableStream.end]]
+ * after all bytes have been written.
+ */
+export default class AppendableStream implements AppendableBuffer {
+    private readonly outStream;
+    private writtenBytes;
+    /**
+     * @param outStream The underlying writable stream
+     */
+    constructor(outStream: Writable);
+    /**
+     * Appends a byte to the end
+     * of the written data
+     * @param value The unsigned byte value to add
+     */
+    add(value: number): this;
+    /**
+     * Appends a contiguous set of bytes
+     * to the end of the written data
+     * @param buffer The bytes to add
+     */
+    addAll(buffer: ArrayBuffer): this;
+    /**
+     * Closes the underlying stream
+     */
+    end(): void;
+    /**
+     * The number of bytes that have been written
+     */
+    readonly length: number;
+}

dist/io.d.ts

@@ -0,0 +1,59 @@
+"use strict";
+Object.defineProperty(exports, "__esModule", { value: true });
+const http_1 = require("http");
+const stream_1 = require("stream");
+const assert_1 = require("./assert");
+const WRITABLE_STREAMS = [stream_1.Writable, stream_1.Duplex, http_1.OutgoingMessage];
+/**
+ * A wrapper around a writable stream
+ * to implement [[AppendableBuffer]].
+ * The stream must be explicitly closed
+ * by calling [[AppendableStream.end]]
+ * after all bytes have been written.
+ */
+class AppendableStream {
+    /**
+     * @param outStream The underlying writable stream
+     */
+    constructor(outStream) {
+        assert_1.default.instanceOf(outStream, WRITABLE_STREAMS);
+        this.outStream = outStream;
+        this.writtenBytes = 0;
+    }
+    /**
+     * Appends a byte to the end
+     * of the written data
+     * @param value The unsigned byte value to add
+     */
+    add(value) {
+        assert_1.default.integer(value);
+        assert_1.default.between(0, value, 0x100, 'Not a byte: ' + String(value));
+        this.outStream.write(Buffer.from([value]));
+        this.writtenBytes++;
+        return this;
+    }
+    /**
+     * Appends a contiguous set of bytes
+     * to the end of the written data
+     * @param buffer The bytes to add
+     */
+    addAll(buffer) {
+        assert_1.default.instanceOf(buffer, ArrayBuffer);
+        this.outStream.write(Buffer.from(buffer));
+        this.writtenBytes += buffer.byteLength;
+        return this;
+    }
+    /**
+     * Closes the underlying stream
+     */
+    end() {
+        this.outStream.end();
+    }
+    /**
+     * The number of bytes that have been written
+     */
+    get length() {
+        return this.writtenBytes;
+    }
+}
+exports.default = AppendableStream;

dist/io.js

@@ -0,0 +1,26 @@
+/**
+ * A "writable" interface, sufficient
+ * to be able to write type and value bytes.
+ * Implemented by [[GrowableBuffer]], as well as
+ * [[AppendableStream]] (a wrapper around a writable stream).
+ */
+export default interface AppendableBuffer {
+    /**
+     * The number of bytes that have been written
+     */
+    readonly length: number;
+    /**
+     * Adds a byte after the end
+     * of the written data
+     * @param value The unsigned byte value to add
+     */
+    add(value: number): this;
+    /**
+     * Adds a contiguous set of bytes
+     * after the end of the written data
+     * @param buffer The bytes to add.
+     * The byte at position `i` in `buffer` will be written to
+     * position `this.length + i`.
+     */
+    addAll(buffer: ArrayBuffer): this;
+}

dist/lib/appendable-stream.d.ts

@@ -0,0 +1,2 @@
+"use strict";
+Object.defineProperty(exports, "__esModule", { value: true });

dist/lib/appendable-stream.js

@@ -1,5 +1,6 @@
 /**
- * Efficiently computes `Math.floor(n / 8)`
+ * Efficiently computes `Math.floor(n / 8)`,
+ * for positive `n`
  * @param n The number in question
  */
 export declare function dividedByEight(n: number): number;
@@ -8,3 +9,8 @@ export declare function dividedByEight(n: number): number;
  * @param n The number in question
  */
 export declare function modEight(n: number): number;
+/**
+ * Efficiently computes `n * 8`
+ * @param n The number in question
+ */
+export declare function timesEight(n: number): number;