diff --git a/whatwg-streams/whatwg-streams-tests.ts b/whatwg-streams/whatwg-streams-tests.ts index ce5eb2b256..b1d48f51bb 100644 --- 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"; +/// - 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; + read(fd: number, buffer: Buffer, offset: number, length: number, position: number): Promise; + write(fd: number, buffer: Buffer, offset: number, length: number): Promise; + close(fd: number): Promise; +} +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, 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, 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(); - return { - readable: new ReadableStream(new WebSocketSource(ws)), - writable: new WritableStream(new WebSocketSink(ws)) - }; + const webSocketStream = makeWritableWebSocketStream("wss://example.com:443/", "protocol"); + + 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 + +{ + function streamifyWebSocket(url: string, protocol: string) { + const ws = new WebSocket(url, protocol); + ws.binaryType = "arraybuffer"; + + return { + readable: new ReadableStream(new WebSocketSource(ws)), + writable: new WritableStream(new WebSocketSink(ws)) + }; } - start(controller: WritableStreamDefaultController) { - this._ws.addEventListener("error", () => { - controller.error(new Error("The WebSocket errored!")); - }); + class WebSocketSource implements ReadableStreamSource { + private _ws: WebSocket - return new Promise(resolve => this._ws.onopen = () => resolve()); - } + constructor(ws: WebSocket) { + this._ws = ws; + } - write(chunk: any) { - this._ws.send(chunk); - } + start(controller: ReadableStreamDefaultController) { + this._ws.onmessage = event => controller.enqueue(event.data); + this._ws.onclose = () => controller.close(); - close() { - return new Promise((resolve, reject) => { - this._ws.onclose = () => resolve(); + this._ws.addEventListener("error", () => { + controller.error(new Error("The WebSocket errored!")); + }); + } + + cancel() { this._ws.close(); - }); + } } + + 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!")); + }); + + return new Promise(resolve => this._ws.onopen = () => resolve()); + } + + write(chunk: any) { + this._ws.send(chunk); + } + + close() { + return new Promise((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); + }); } - -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); -}); \ No newline at end of file