mirror of
https://github.com/gosticks/DefinitelyTyped.git
synced 2026-10-04 23:07:06 +00:00
Improve the examples, some were missing
See https://streams.spec.whatwg.org/#creating-examples
This commit is contained in:
Tanguy Krotoff
@@ -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();
|
||||
|
||||
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<void>(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<void>((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<void>(resolve => this._ws.onopen = () => resolve());
|
||||
}
|
||||
|
||||
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);
|
||||
});
|
||||
}
|
||||
|
||||
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);
|
||||
});
|
||||
Reference in New Issue
Block a user