Improve the examples, some were missing

tkrotoff · greglo · commit 4218e2ed5e7d · 2016-12-15T12:12:36.000-08:00
See https://streams.spec.whatwg.org/#creating-examples
diff --git a/whatwg-streams/whatwg-streams-tests.ts b/whatwg-streams/whatwg-streams-tests.ts
@@ -1,20 +1,39 @@
-﻿function makeReadableWebSocketStream(url: string, protocols: string[]) {
-    const ws = new WebSocket(url, protocols);
-    ws.binaryType = "arraybuffer";
+﻿/// <reference types="node" />
 
-    return new ReadableStream({
-        start(controller) {
-            ws.onmessage = event => controller.enqueue(event.data);
-            ws.onclose = () => controller.close();
-            ws.onerror = () => controller.error(new Error("The WebSocket errored!"));
-        },
+// Examples taken from https://streams.spec.whatwg.org/#creating-examples
 
-        cancel() {
-            ws.close();
-        }
-    });
+// 8.1. A readable stream with an underlying push source (no backpressure support)
+
+{
+    function makeReadableWebSocketStream(url: string, protocols: string | string[]) {
+        const ws = new WebSocket(url, protocols);
+        ws.binaryType = "arraybuffer";
+
+        return new ReadableStream({
+            start(controller) {
+                ws.onmessage = event => controller.enqueue(event.data);
+                ws.onclose = () => controller.close();
+                ws.onerror = () => controller.error(new Error("The WebSocket errored!"));
+            },
+
+            cancel() {
+                ws.close();
+            }
+        });
+    }
+
+    const writableStream = new WritableStream();
+
+    const webSocketStream = makeReadableWebSocketStream("wss://example.com:443/", "protocol");
+
+    webSocketStream.pipeTo(writableStream)
+        .then(() => console.log("All data successfully written!"))
+        .catch(e => console.error("Something went wrong!", e));
 }
 
+
+// 8.2. A readable stream with an underlying push source and backpressure support
+
 function makeReadableBackpressureSocketStream(host: string, port: number) {
     const socket = createBackpressureSocket(host, port);
 
@@ -49,6 +68,9 @@ function makeReadableBackpressureSocketStream(host: string, port: number) {
     function createBackpressureSocket(host: string, port: number): any { };
 }
 
+
+// 8.3. A readable byte stream with an underlying push source (no backpressure support)
+
 const DEFAULT_CHUNK_SIZE = 65536;
 
 function makeUDPSocketStream(host: string, port: number) {
@@ -93,7 +115,100 @@ function makeUDPSocketStream(host: string, port: number) {
     function createUDPSocket(host: string, port: number): any { };
 }
 
-function makeWritableWebSocketStream(url: string, protocols: string[]) {
+
+// 8.4. A readable stream with an underlying pull source
+
+//const fs = require("pr/fs"); // https://github.com/jden/pr
+interface fs {
+    open(path: string | Buffer, flags: string | number): Promise<number>;
+    read(fd: number, buffer: Buffer, offset: number, length: number, position: number): Promise<number>;
+    write(fd: number, buffer: Buffer, offset: number, length: number): Promise<number>;
+    close(fd: number): Promise<void>;
+}
+let fs: fs;
+
+{
+    const CHUNK_SIZE = 1024;
+
+    function makeReadableFileStream(filename: string) {
+        let fd: number;
+        let position = 0;
+
+        return new ReadableStream({
+            start() {
+                return fs.open(filename, "r").then(result => {
+                    fd = result;
+                });
+            },
+
+            pull(controller) {
+                const buffer = new ArrayBuffer(CHUNK_SIZE);
+
+                return fs.read(fd, <any> buffer, 0, CHUNK_SIZE, position).then(bytesRead => {
+                    if (bytesRead === 0) {
+                        return fs.close(fd).then(() => controller.close());
+                    } else {
+                        position += bytesRead;
+                        controller.enqueue(new Uint8Array(buffer, 0, bytesRead));
+                    }
+                });
+            },
+
+            cancel() {
+                return fs.close(fd);
+            }
+        });
+    }
+}
+
+
+// 8.5. A readable byte stream with an underlying pull source
+
+{
+    //const fs = require("pr/fs"); // https://github.com/jden/pr
+    const DEFAULT_CHUNK_SIZE = 1024;
+
+    function makeReadableByteFileStream(filename: string) {
+        let fd: number;
+        let position = 0;
+
+        return new ReadableStream({
+            type: "bytes",
+
+            start() {
+                return fs.open(filename, "r").then(result => {
+                    fd = result;
+                });
+            },
+
+            pull(controller) {
+                // Even when the consumer is using the default reader, the auto-allocation
+                // feature allocates a buffer and passes it to us via byobRequest.
+                const v = controller.byobRequest.view;
+
+                return fs.read(fd, <any> v.buffer, v.byteOffset, v.byteLength, position).then(bytesRead => {
+                    if (bytesRead === 0) {
+                        return fs.close(fd).then(() => controller.close());
+                    } else {
+                        position += bytesRead;
+                        controller.byobRequest.respond(bytesRead);
+                    }
+                });
+            },
+
+            cancel() {
+                return fs.close(fd);
+            },
+
+            autoAllocateChunkSize: DEFAULT_CHUNK_SIZE
+        });
+    }
+}
+
+
+// 8.6. A writable stream with no backpressure or success signals
+
+function makeWritableWebSocketStream(url: string, protocols: string | string[]) {
     const ws = new WebSocket(url, protocols);
 
     return new WritableStream({
@@ -117,71 +232,123 @@ function makeWritableWebSocketStream(url: string, protocols: string[]) {
     });
 }
 
-function streamifyWebSocket(url: string, protocol: string) {
-    const ws = new WebSocket(url, protocol);
-    ws.binaryType = "arraybuffer";
+{
+    const readableStream = new ReadableStream();
+
+    const webSocketStream = makeWritableWebSocketStream("wss://example.com:443/", "protocol");
 
-    return {
-        readable: new ReadableStream(new WebSocketSource(ws)),
-        writable: new WritableStream(new WebSocketSink(ws))
-    };
+    readableStream.pipeTo(webSocketStream)
+        .then(() => console.log("All data successfully written!"))
+        .catch(e => console.error("Something went wrong!", e));
 }
 
-class WebSocketSource implements ReadableStreamSource {
-    private _ws: WebSocket
 
-    constructor(ws: WebSocket) {
-        this._ws = ws;
-    }
+// 8.7. A writable stream with backpressure and success signals
 
-    start(controller: ReadableStreamDefaultController) {
-        this._ws.onmessage = event => controller.enqueue(event.data);
-        this._ws.onclose = () => controller.close();
+{
+    //const fs = require("pr/fs"); // https://github.com/jden/pr
 
-        this._ws.addEventListener("error", () => {
-            controller.error(new Error("The WebSocket errored!"));
+    function makeWritableFileStream(filename: string) {
+        let fd: number;
+
+        return new WritableStream({
+            start() {
+                return fs.open(filename, "w").then(result => {
+                    fd = result;
+                });
+            },
+
+            write(chunk) {
+                return fs.write(fd, chunk, 0, chunk.length);
+            },
+
+            close() {
+                return fs.close(fd);
+            }
         });
     }
 
-    cancel() {
-        this._ws.close();
-    }
+    const fileStream = makeWritableFileStream("/example/path/on/fs.txt");
+    const writer = fileStream.getWriter();
+
+    writer.write("To stream, or not to stream\n");
+    writer.write("That is the question\n");
+
+    writer.close()
+        .then(() => console.log("chunks written and stream closed successfully!"))
+        .catch(e => console.error(e));
 }
 
-class WebSocketSink implements WritableStreamSink {
-    private _ws: WebSocket
 
-    constructor(ws: WebSocket) {
-        this._ws = ws;
-    }
+// 8.8. A { readable, writable } stream pair wrapping the same underlying resource
 
-    start(controller: WritableStreamDefaultController) {
-        this._ws.addEventListener("error", () => {
-            controller.error(new Error("The WebSocket errored!"));
-        });
+{
+    function streamifyWebSocket(url: string, protocol: string) {
+        const ws = new WebSocket(url, protocol);
+        ws.binaryType = "arraybuffer";
 
-        return new Promise<void>(resolve => this._ws.onopen = () => resolve());
+        return {
+            readable: new ReadableStream(new WebSocketSource(ws)),
+            writable: new WritableStream(new WebSocketSink(ws))
+        };
     }
 
-    write(chunk: any) {
-        this._ws.send(chunk);
-    }
+    class WebSocketSource implements ReadableStreamSource {
+        private _ws: WebSocket
+
+        constructor(ws: WebSocket) {
+            this._ws = ws;
+        }
 
-    close() {
-        return new Promise<void>((resolve, reject) => {
-            this._ws.onclose = () => resolve();
+        start(controller: ReadableStreamDefaultController) {
+            this._ws.onmessage = event => controller.enqueue(event.data);
+            this._ws.onclose = () => controller.close();
+
+            this._ws.addEventListener("error", () => {
+                controller.error(new Error("The WebSocket errored!"));
+            });
+        }
+
+        cancel() {
             this._ws.close();
-        });
+        }
     }
-}
 
-const streamyWS = streamifyWebSocket("wss://example.com:443/", "protocol");
-const writer = streamyWS.writable.getWriter();
-const reader = streamyWS.readable.getReader();
+    class WebSocketSink implements WritableStreamSink {
+        private _ws: WebSocket
+
+        constructor(ws: WebSocket) {
+            this._ws = ws;
+        }
+
+        start(controller: WritableStreamDefaultController) {
+            this._ws.addEventListener("error", () => {
+                controller.error(new Error("The WebSocket errored!"));
+            });
 
-writer.write("Hello");
-writer.write("web socket!");
+            return new Promise<void>(resolve => this._ws.onopen = () => resolve());
+        }
 
-reader.read().then(({ value, done }) => {
-    console.log("The web socket says: ", value);
-});
+        write(chunk: any) {
+            this._ws.send(chunk);
+        }
+
+        close() {
+            return new Promise<void>((resolve, reject) => {
+                this._ws.onclose = () => resolve();
+                this._ws.close();
+            });
+        }
+    }
+
+    const streamyWS = streamifyWebSocket("wss://example.com:443/", "protocol");
+    const writer = streamyWS.writable.getWriter();
+    const reader = streamyWS.readable.getReader();
+
+    writer.write("Hello");
+    writer.write("web socket!");
+
+    reader.read().then(({ value, done }) => {
+        console.log("The web socket says: ", value);
+    });
+}
