Skip to content

D
deno

张重言 / deno

通知 1
Star 0
Fork 0
D
deno

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

未验证 提交 8bcfc03d 编写于 作者: K Kitson Kelly 提交者： GitHub
浏览文件
操作

Rewrite streams (#4842)

上级 b270d6c8
展开全部 隐藏空白更改
内联 并排
Showing with 2008 addition and 4952 deletion
cli/js/globals.ts
浏览文件 @ 8bcfc03d
......@@ -22,7 +22,7 @@ import * as urlSearchParams from "./web/url_search_params.ts";
import * as workers from "./web/workers.ts";
import * as performanceUtil from "./web/performance.ts";
import * as request from "./web/request.ts";
import * as streams from "./web/streams/mod.ts";
import * as readableStream from "./web/streams/readable_stream.ts";
// These imports are not exposed and therefore are fine to just import the
// symbols required.
......@@ -223,7 +223,7 @@ export const windowOrWorkerGlobalScopeProperties = {
FormData: nonEnumerable(formData.FormDataImpl),
TextEncoder: nonEnumerable(textEncoding.TextEncoder),
TextDecoder: nonEnumerable(textEncoding.TextDecoder),
ReadableStream: nonEnumerable(streams.ReadableStream),
ReadableStream: nonEnumerable(readableStream.ReadableStreamImpl),
Request: nonEnumerable(request.Request),
Response: nonEnumerable(fetchTypes.Response),
performance: writable(new performanceUtil.Performance()),
......
cli/js/lib.deno.shared_globals.d.ts
浏览文件 @ 8bcfc03d
......@@ -245,6 +245,24 @@ interface ReadableStreamDefaultReader<R = any> {
releaseLock(): void;
}
interface ReadableStreamReader<R = any> {
cancel(): Promise<void>;
read(): Promise<ReadableStreamReadResult<R>>;
releaseLock(): void;
}
interface ReadableByteStreamControllerCallback {
(controller: ReadableByteStreamController): void | PromiseLike<void>;
}
interface UnderlyingByteSource {
autoAllocateChunkSize?: number;
cancel?: ReadableStreamErrorCallback;
pull?: ReadableByteStreamControllerCallback;
start?: ReadableByteStreamControllerCallback;
type: "bytes";
}
interface UnderlyingSource<R = any> {
cancel?: ReadableStreamErrorCallback;
pull?: ReadableStreamDefaultControllerCallback<R>;
......@@ -260,11 +278,35 @@ interface ReadableStreamDefaultControllerCallback<R> {
(controller: ReadableStreamDefaultController<R>): void | PromiseLike<void>;
}
interface ReadableStreamDefaultController<R> {
readonly desiredSize: number;
enqueue(chunk?: R): void;
interface ReadableStreamDefaultController<R = any> {
readonly desiredSize: number | null;
close(): void;
error(e?: any): void;
enqueue(chunk: R): void;
error(error?: any): void;
}
interface ReadableByteStreamController {
readonly byobRequest: undefined;
readonly desiredSize: number | null;
close(): void;
enqueue(chunk: ArrayBufferView): void;
error(error?: any): void;
}
interface PipeOptions {
preventAbort?: boolean;
preventCancel?: boolean;
preventClose?: boolean;
signal?: AbortSignal;
}
interface QueuingStrategySizeCallback<T = any> {
(chunk: T): number;
}
interface QueuingStrategy<T = any> {
highWaterMark?: number;
size?: QueuingStrategySizeCallback<T>;
}
/** This Streams API interface represents a readable stream of byte data. The
......@@ -273,16 +315,36 @@ interface ReadableStreamDefaultController<R> {
interface ReadableStream<R = any> {
readonly locked: boolean;
cancel(reason?: any): Promise<void>;
// TODO(ry) It doesn't seem like Chrome supports this.
getIterator(options?: { preventCancel?: boolean }): AsyncIterableIterator<R>;
// getReader(options: { mode: "byob" }): ReadableStreamBYOBReader;
getReader(): ReadableStreamDefaultReader<R>;
pipeThrough<T>(
{
writable,
readable,
}: {
writable: WritableStream<R>;
readable: ReadableStream<T>;
},
options?: PipeOptions
): ReadableStream<T>;
pipeTo(dest: WritableStream<R>, options?: PipeOptions): Promise<void>;
tee(): [ReadableStream<R>, ReadableStream<R>];
[Symbol.asyncIterator](options?: {
preventCancel?: boolean;
}): AsyncIterableIterator<R>;
}
declare const ReadableStream: {
declare var ReadableStream: {
prototype: ReadableStream;
// TODO(ry) This doesn't match lib.dom.d.ts
new <R = any>(src?: UnderlyingSource<R>): ReadableStream<R>;
new (
underlyingSource: UnderlyingByteSource,
strategy?: { highWaterMark?: number; size?: undefined }
): ReadableStream<Uint8Array>;
new <R = any>(
underlyingSource?: UnderlyingSource<R>,
strategy?: QueuingStrategy<R>
): ReadableStream<R>;
};
/** This Streams API interface provides a standard abstraction for writing streaming data to a destination, known as a sink. This object comes with built-in backpressure and queuing. */
......
cli/js/web/blob.ts
浏览文件 @ 8bcfc03d
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
import * as domTypes from "./dom_types.d.ts";
import { TextDecoder, TextEncoder } from "./text_encoding.ts";
import { build } from "../build.ts";
import { ReadableStream } from "./streams/mod.ts";
import { ReadableStreamImpl } from "./streams/readable_stream.ts";
export const bytesSymbol = Symbol("bytes");
......@@ -124,40 +123,36 @@ function processBlobParts(
return bytes;
}
function getStream(blobBytes: Uint8Array): domTypes.ReadableStream<Uint8Array> {
return new ReadableStream<Uint8Array>({
start: (
controller: domTypes.ReadableStreamDefaultController<Uint8Array>
): void => {
function getStream(blobBytes: Uint8Array): ReadableStream<ArrayBufferView> {
// TODO: Align to spec https://fetch.spec.whatwg.org/#concept-construct-readablestream
return new ReadableStreamImpl({
type: "bytes",
start: (controller: ReadableByteStreamController): void => {
controller.enqueue(blobBytes);
controller.close();
},
}) as domTypes.ReadableStream<Uint8Array>;
});
}
async function readBytes(
reader: domTypes.ReadableStreamReader<Uint8Array>
reader: ReadableStreamReader<ArrayBufferView>
): Promise<ArrayBuffer> {
const chunks: Uint8Array[] = [];
while (true) {
try {
const { done, value } = await reader.read();
if (!done && value instanceof Uint8Array) {
chunks.push(value);
} else if (done) {
const size = chunks.reduce((p, i) => p + i.byteLength, 0);
const bytes = new Uint8Array(size);
let offs = 0;
for (const chunk of chunks) {
bytes.set(chunk, offs);
offs += chunk.byteLength;
}
return Promise.resolve(bytes);
} else {
return Promise.reject(new TypeError());
const { done, value } = await reader.read();
if (!done && value instanceof Uint8Array) {
chunks.push(value);
} else if (done) {
const size = chunks.reduce((p, i) => p + i.byteLength, 0);
const bytes = new Uint8Array(size);
let offs = 0;
for (const chunk of chunks) {
bytes.set(chunk, offs);
offs += chunk.byteLength;
}
} catch (e) {
return Promise.reject(e);
return bytes;
} else {
throw new TypeError("Invalid reader result.");
}
}
}
......@@ -207,7 +202,7 @@ export class DenoBlob implements Blob {
});
}
stream(): domTypes.ReadableStream<Uint8Array> {
stream(): ReadableStream<ArrayBufferView> {
return getStream(this[bytesSymbol]);
}
......
cli/js/web/body.ts
浏览文件 @ 8bcfc03d
import * as blob from "./blob.ts";
import * as encoding from "./text_encoding.ts";
import * as domTypes from "./dom_types.d.ts";
import { ReadableStream } from "./streams/mod.ts";
import { ReadableStreamImpl } from "./streams/readable_stream.ts";
// only namespace imports work for now, plucking out what we need
const { TextEncoder, TextDecoder } = encoding;
const DenoBlob = blob.DenoBlob;
type ReadableStreamReader = domTypes.ReadableStreamReader;
interface ReadableStreamController {
enqueue(chunk: string | ArrayBuffer): void;
close(): void;
}
export type BodySource =
| Blob
| BufferSource
| FormData
| URLSearchParams
| domTypes.ReadableStream
| ReadableStream
| string;
function validateBodyType(owner: Body, bodySource: BodySource): boolean {
......@@ -39,7 +32,7 @@ function validateBodyType(owner: Body, bodySource: BodySource): boolean {
return true;
} else if (typeof bodySource === "string") {
return true;
} else if (bodySource instanceof ReadableStream) {
} else if (bodySource instanceof ReadableStreamImpl) {
return true;
} else if (bodySource instanceof FormData) {
return true;
......@@ -118,7 +111,7 @@ export const BodyUsedError =
"Failed to execute 'clone' on 'Body': body is already used";
export class Body implements domTypes.Body {
protected _stream: domTypes.ReadableStream<string | ArrayBuffer> | null;
protected _stream: ReadableStreamImpl<string | ArrayBuffer> | null;
constructor(protected _bodySource: BodySource, readonly contentType: string) {
validateBodyType(this, _bodySource);
......@@ -127,23 +120,23 @@ export class Body implements domTypes.Body {
this._stream = null;
}
get body(): domTypes.ReadableStream | null {
get body(): ReadableStream | null {
if (this._stream) {
return this._stream;
}
if (this._bodySource instanceof ReadableStream) {
if (this._bodySource instanceof ReadableStreamImpl) {
// @ts-ignore
this._stream = this._bodySource;
}
if (typeof this._bodySource === "string") {
const bodySource = this._bodySource;
this._stream = new ReadableStream({
start(controller: ReadableStreamController): void {
this._stream = new ReadableStreamImpl<string | ArrayBuffer>({
start(controller: ReadableStreamDefaultController): void {
controller.enqueue(bodySource);
controller.close();
},
}) as domTypes.ReadableStream<ArrayBuffer | string>;
});
}
return this._stream;
}
......@@ -320,7 +313,7 @@ export class Body implements domTypes.Body {
return Promise.resolve(
enc.encode(this._bodySource).buffer as ArrayBuffer
);
} else if (this._bodySource instanceof ReadableStream) {
} else if (this._bodySource instanceof ReadableStreamImpl) {
// @ts-ignore
return bufferFromStream(this._bodySource.getReader());
} else if (this._bodySource instanceof FormData) {
......
cli/js/web/dom_types.d.ts
浏览文件 @ 8bcfc03d
......@@ -17,14 +17,6 @@ and limitations under the License.
/* eslint-disable @typescript-eslint/no-explicit-any */
type BodyInit =
| Blob
| BufferSource
| FormData
| URLSearchParams
| ReadableStream
| string;
export type RequestInfo = Request | string;
export interface ProgressEventInit extends EventInit {
......@@ -261,82 +253,6 @@ export interface Body {
text(): Promise<string>;
}
export interface ReadableStreamReadDoneResult<T> {
done: true;
value?: T;
}
export interface ReadableStreamReadValueResult<T> {
done: false;
value: T;
}
export type ReadableStreamReadResult<T> =
| ReadableStreamReadValueResult<T>
| ReadableStreamReadDoneResult<T>;
export interface ReadableStreamDefaultReader<R = any> {
readonly closed: Promise<void>;
cancel(reason?: any): Promise<void>;
read(): Promise<ReadableStreamReadResult<R>>;
releaseLock(): void;
}
export interface PipeOptions {
preventAbort?: boolean;
preventCancel?: boolean;
preventClose?: boolean;
signal?: AbortSignal;
}
export interface UnderlyingSource<R = any> {
cancel?: ReadableStreamErrorCallback;
pull?: ReadableStreamDefaultControllerCallback<R>;
start?: ReadableStreamDefaultControllerCallback<R>;
type?: undefined;
}
export interface ReadableStreamErrorCallback {
(reason: any): void | PromiseLike<void>;
}
export interface ReadableStreamDefaultControllerCallback<R> {
(controller: ReadableStreamDefaultController<R>): void | PromiseLike<void>;
}
export interface ReadableStreamConstructor {
new <R = any>(source?: UnderlyingSource<R>): ReadableStream<R>;
}
export interface ReadableStream<R = any> {
readonly locked: boolean;
cancel(reason?: any): Promise<void>;
getReader(options: { mode: "byob" }): ReadableStreamBYOBReader;
getReader(): ReadableStreamDefaultReader<R>;
/* disabled for now
pipeThrough<T>(
{
writable,
readable
}: {
writable: WritableStream<R>;
readable: ReadableStream<T>;
},
options?: PipeOptions
): ReadableStream<T>;
pipeTo(dest: WritableStream<R>, options?: PipeOptions): Promise<void>;
*/
tee(): [ReadableStream<R>, ReadableStream<R>];
}
export interface ReadableStreamBYOBReader {
readonly closed: Promise<void>;
cancel(reason?: any): Promise<void>;
read<T extends ArrayBufferView>(
view: T
): Promise<ReadableStreamReadResult<T>>;
releaseLock(): void;
}
export interface WritableStream<W = any> {
readonly locked: boolean;
abort(reason?: any): Promise<void>;
......@@ -353,69 +269,6 @@ export interface WritableStreamDefaultWriter<W = any> {
write(chunk: W): Promise<void>;
}
export interface UnderlyingSource<R = any> {
cancel?: ReadableStreamErrorCallback;
pull?: ReadableStreamDefaultControllerCallback<R>;
start?: ReadableStreamDefaultControllerCallback<R>;
type?: undefined;
}
export interface UnderlyingByteSource {
autoAllocateChunkSize?: number;
cancel?: ReadableStreamErrorCallback;
pull?: ReadableByteStreamControllerCallback;
start?: ReadableByteStreamControllerCallback;
type: "bytes";
}
export interface ReadableStreamReader<R = any> {
cancel(reason: any): Promise<void>;
read(): Promise<ReadableStreamReadResult<R>>;
releaseLock(): void;
}
export interface ReadableStreamErrorCallback {
(reason: any): void | PromiseLike<void>;
}
export interface ReadableByteStreamControllerCallback {
(controller: ReadableByteStreamController): void | PromiseLike<void>;
}
export interface ReadableStreamDefaultControllerCallback<R> {
(controller: ReadableStreamDefaultController<R>): void | PromiseLike<void>;
}
export interface ReadableStreamDefaultController<R = any> {
readonly desiredSize: number | null;
close(): void;
enqueue(chunk: R): void;
error(error?: any): void;
}
export interface ReadableByteStreamController {
readonly byobRequest: ReadableStreamBYOBRequest | undefined;
readonly desiredSize: number | null;
close(): void;
enqueue(chunk: ArrayBufferView): void;
error(error?: any): void;
}
export interface ReadableStreamBYOBRequest {
readonly view: ArrayBufferView;
respond(bytesWritten: number): void;
respondWithNewView(view: ArrayBufferView): void;
}
export interface QueuingStrategy<T = any> {
highWaterMark?: number;
size?: QueuingStrategySizeCallback<T>;
}
export interface QueuingStrategySizeCallback<T = any> {
(chunk: T): number;
}
export interface RequestInit {
body?: BodyInit | null;
cache?: RequestCache;
......
cli/js/web/fetch.ts
浏览文件 @ 8bcfc03d
......@@ -28,14 +28,13 @@ function hasHeaderValueOf(s: string, value: string): boolean {
return new RegExp(`^${value}[\t\s]*;?`).test(s);
}
class Body
implements domTypes.Body, domTypes.ReadableStream<Uint8Array>, io.ReadCloser {
class Body implements domTypes.Body, ReadableStream<Uint8Array>, io.ReadCloser {
#bodyUsed = false;
#bodyPromise: Promise<ArrayBuffer> | null = null;
#data: ArrayBuffer | null = null;
#rid: number;
readonly locked: boolean = false; // TODO
readonly body: domTypes.ReadableStream<Uint8Array>;
readonly body: ReadableStream<Uint8Array>;
constructor(rid: number, readonly contentType: string) {
this.#rid = rid;
......@@ -234,15 +233,17 @@ class Body
return notImplemented();
}
getReader(options: { mode: "byob" }): domTypes.ReadableStreamBYOBReader;
getReader(): domTypes.ReadableStreamDefaultReader<Uint8Array>;
getReader():
| domTypes.ReadableStreamBYOBReader
| domTypes.ReadableStreamDefaultReader<Uint8Array> {
getIterator(_options?: {
preventCancel?: boolean;
}): AsyncIterableIterator<Uint8Array> {
return notImplemented();
}
tee(): [domTypes.ReadableStream, domTypes.ReadableStream] {
getReader(): ReadableStreamDefaultReader<Uint8Array> {
return notImplemented();
}
tee(): [ReadableStream, ReadableStream] {
return notImplemented();
}
......@@ -257,16 +258,16 @@ class Body
pipeThrough<T>(
_: {
writable: domTypes.WritableStream<Uint8Array>;
readable: domTypes.ReadableStream<T>;
readable: ReadableStream<T>;
},
_options?: domTypes.PipeOptions
): domTypes.ReadableStream<T> {
_options?: PipeOptions
): ReadableStream<T> {
return notImplemented();
}
pipeTo(
_dest: domTypes.WritableStream<Uint8Array>,
_options?: domTypes.PipeOptions
_options?: PipeOptions
): Promise<void> {
return notImplemented();
}
......
cli/js/web/request.ts
浏览文件 @ 8bcfc03d
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
import * as body from "./body.ts";
import * as domTypes from "./dom_types.d.ts";
import * as streams from "./streams/mod.ts";
const { ReadableStream } = streams;
import { ReadableStreamImpl } from "./streams/readable_stream.ts";
function byteUpperCase(s: string): string {
return String(s).replace(/[a-z]/g, function byteUpperCaseReplace(c): string {
......@@ -124,8 +122,8 @@ export class Request extends body.Body implements domTypes.Request {
let body2 = this._bodySource;
if (this._bodySource instanceof ReadableStream) {
const tees = (this._bodySource as domTypes.ReadableStream).tee();
if (this._bodySource instanceof ReadableStreamImpl) {
const tees = this._bodySource.tee();
this._stream = this._bodySource = tees[0];
body2 = tees[1];
}
......
cli/js/web/streams/internals.ts 0 → 100644
浏览文件 @ 8bcfc03d
此差异已折叠。
cli/js/web/streams/mod.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
export { SDReadableStream as ReadableStream } from "./readable-stream.ts";
/* TODO The following are currently unused so not exported for clarity.
export { WritableStream } from "./writable-stream.ts";
export { TransformStream } from "./transform-stream.ts";
export {
ByteLengthQueuingStrategy,
CountQueuingStrategy
} from "./strategies.ts";
*/
cli/js/web/streams/pipe-to.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// TODO reenable this code when we enable writableStreams and transport types
// // Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// // Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
// /**
// * streams/pipe-to - pipeTo algorithm implementation
// * Part of Stardazed
// * (c) 2018-Present by Arthur Langereis - @zenmumbler
// * https://github.com/stardazed/sd-streams
// */
// /* eslint-disable @typescript-eslint/no-explicit-any */
// // TODO reenable this lint here
// import * as rs from "./readable-internals.ts";
// import * as ws from "./writable-internals.ts";
// import * as shared from "./shared-internals.ts";
// import { ReadableStreamDefaultReader } from "./readable-stream-default-reader.ts";
// import { WritableStreamDefaultWriter } from "./writable-stream-default-writer.ts";
// import { PipeOptions } from "../dom_types.d.ts";
// import { Err } from "../errors.ts";
// // add a wrapper to handle falsy rejections
// interface ErrorWrapper {
// actualError: shared.ErrorResult;
// }
// export function pipeTo<ChunkType>(
// source: rs.SDReadableStream<ChunkType>,
// dest: ws.WritableStream<ChunkType>,
// options: PipeOptions
// ): Promise<void> {
// const preventClose = !!options.preventClose;
// const preventAbort = !!options.preventAbort;
// const preventCancel = !!options.preventCancel;
// const signal = options.signal;
// let shuttingDown = false;
// let latestWrite = Promise.resolve();
// const promise = shared.createControlledPromise<void>();
// // If IsReadableByteStreamController(this.[[readableStreamController]]) is true, let reader be either ! AcquireReadableStreamBYOBReader(this) or ! AcquireReadableStreamDefaultReader(this), at the user agent’s discretion.
// // Otherwise, let reader be ! AcquireReadableStreamDefaultReader(this).
// const reader = new ReadableStreamDefaultReader(source);
// const writer = new WritableStreamDefaultWriter(dest);
// let abortAlgorithm: () => any;
// if (signal !== undefined) {
// abortAlgorithm = (): void => {
// // TODO this should be a DOMException,
// // https://github.com/stardazed/sd-streams/blob/master/packages/streams/src/pipe-to.ts#L38
// const error = new errors.Aborted("Aborted");
// const actions: Array<() => Promise<void>> = [];
// if (preventAbort === false) {
// actions.push(() => {
// if (dest[shared.state_] === "writable") {
// return ws.writableStreamAbort(dest, error);
// }
// return Promise.resolve();
// });
// }
// if (preventCancel === false) {
// actions.push(() => {
// if (source[shared.state_] === "readable") {
// return rs.readableStreamCancel(source, error);
// }
// return Promise.resolve();
// });
// }
// shutDown(
// () => {
// return Promise.all(actions.map(a => a())).then(_ => undefined);
// },
// { actualError: error }
// );
// };
// if (signal.aborted === true) {
// abortAlgorithm();
// } else {
// signal.addEventListener("abort", abortAlgorithm);
// }
// }
// function onStreamErrored(
// stream: rs.SDReadableStream<ChunkType> | ws.WritableStream<ChunkType>,
// promise: Promise<void>,
// action: (error: shared.ErrorResult) => void
// ): void {
// if (stream[shared.state_] === "errored") {
// action(stream[shared.storedError_]);
// } else {
// promise.catch(action);
// }
// }
// function onStreamClosed(
// stream: rs.SDReadableStream<ChunkType> | ws.WritableStream<ChunkType>,
// promise: Promise<void>,
// action: () => void
// ): void {
// if (stream[shared.state_] === "closed") {
// action();
// } else {
// promise.then(action);
// }
// }
// onStreamErrored(source, reader[rs.closedPromise_].promise, error => {
// if (!preventAbort) {
// shutDown(() => ws.writableStreamAbort(dest, error), {
// actualError: error
// });
// } else {
// shutDown(undefined, { actualError: error });
// }
// });
// onStreamErrored(dest, writer[ws.closedPromise_].promise, error => {
// if (!preventCancel) {
// shutDown(() => rs.readableStreamCancel(source, error), {
// actualError: error
// });
// } else {
// shutDown(undefined, { actualError: error });
// }
// });
// onStreamClosed(source, reader[rs.closedPromise_].promise, () => {
// if (!preventClose) {
// shutDown(() =>
// ws.writableStreamDefaultWriterCloseWithErrorPropagation(writer)
// );
// } else {
// shutDown();
// }
// });
// if (
// ws.writableStreamCloseQueuedOrInFlight(dest) ||
// dest[shared.state_] === "closed"
// ) {
// // Assert: no chunks have been read or written.
// const destClosed = new TypeError();
// if (!preventCancel) {
// shutDown(() => rs.readableStreamCancel(source, destClosed), {
// actualError: destClosed
// });
// } else {
// shutDown(undefined, { actualError: destClosed });
// }
// }
// function awaitLatestWrite(): Promise<void> {
// const curLatestWrite = latestWrite;
// return latestWrite.then(() =>
// curLatestWrite === latestWrite ? undefined : awaitLatestWrite()
// );
// }
// function flushRemainder(): Promise<void> | undefined {
// if (
// dest[shared.state_] === "writable" &&
// !ws.writableStreamCloseQueuedOrInFlight(dest)
// ) {
// return awaitLatestWrite();
// } else {
// return undefined;
// }
// }
// function shutDown(action?: () => Promise<void>, error?: ErrorWrapper): void {
// if (shuttingDown) {
// return;
// }
// shuttingDown = true;
// if (action === undefined) {
// action = (): Promise<void> => Promise.resolve();
// }
// function finishShutDown(): void {
// action!().then(
// _ => finalize(error),
// newError => finalize({ actualError: newError })
// );
// }
// const flushWait = flushRemainder();
// if (flushWait) {
// flushWait.then(finishShutDown);
// } else {
// finishShutDown();
// }
// }
// function finalize(error?: ErrorWrapper): void {
// ws.writableStreamDefaultWriterRelease(writer);
// rs.readableStreamReaderGenericRelease(reader);
// if (signal && abortAlgorithm) {
// signal.removeEventListener("abort", abortAlgorithm);
// }
// if (error) {
// promise.reject(error.actualError);
// } else {
// promise.resolve(undefined);
// }
// }
// function next(): Promise<void> | undefined {
// if (shuttingDown) {
// return;
// }
// writer[ws.readyPromise_].promise.then(() => {
// rs.readableStreamDefaultReaderRead(reader).then(
// ({ value, done }) => {
// if (done) {
// return;
// }
// latestWrite = ws
// .writableStreamDefaultWriterWrite(writer, value!)
// .catch(() => {});
// next();
// },
// _error => {
// latestWrite = Promise.resolve();
// }
// );
// });
// }
// next();
// return promise.promise;
// }
cli/js/web/streams/queue-mixin.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
/* eslint-disable @typescript-eslint/no-explicit-any */
// TODO reenable this lint here
import { Queue, QueueImpl } from "./queue.ts";
import { isFiniteNonNegativeNumber } from "./shared-internals.ts";
export const queue_ = Symbol("queue_");
export const queueTotalSize_ = Symbol("queueTotalSize_");
export interface QueueElement<V> {
value: V;
size: number;
}
export interface QueueContainer<V> {
[queue_]: Queue<QueueElement<V>>;
[queueTotalSize_]: number;
}
export interface ByteQueueContainer {
[queue_]: Queue<{
buffer: ArrayBufferLike;
byteOffset: number;
byteLength: number;
}>;
[queueTotalSize_]: number;
}
export function dequeueValue<V>(container: QueueContainer<V>): V {
// Assert: container has[[queue]] and[[queueTotalSize]] internal slots.
// Assert: container.[[queue]] is not empty.
const pair = container[queue_].shift()!;
const newTotalSize = container[queueTotalSize_] - pair.size;
container[queueTotalSize_] = Math.max(0, newTotalSize); // < 0 can occur due to rounding errors.
return pair.value;
}
export function enqueueValueWithSize<V>(
container: QueueContainer<V>,
value: V,
size: number
): void {
// Assert: container has[[queue]] and[[queueTotalSize]] internal slots.
if (!isFiniteNonNegativeNumber(size)) {
throw new RangeError("Chunk size must be a non-negative, finite numbers");
}
container[queue_].push({ value, size });
container[queueTotalSize_] += size;
}
export function peekQueueValue<V>(container: QueueContainer<V>): V {
// Assert: container has[[queue]] and[[queueTotalSize]] internal slots.
// Assert: container.[[queue]] is not empty.
return container[queue_].front()!.value;
}
export function resetQueue<V>(
container: ByteQueueContainer | QueueContainer<V>
): void {
// Chrome (as of v67) has a steep performance cliff with large arrays
// and shift(), around about 50k elements. While this is an unusual case
// we use a simple wrapper around shift and push that is chunked to
// avoid this pitfall.
// @see: https://github.com/stardazed/sd-streams/issues/1
container[queue_] = new QueueImpl<any>();
// The code below can be used as a plain array implementation of the
// Queue interface.
// const q = [] as any;
// q.front = function() { return this[0]; };
// container[queue_] = q;
container[queueTotalSize_] = 0;
}
cli/js/web/streams/queue.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
const CHUNK_SIZE = 16384;
export interface Queue<T> {
push(t: T): void;
shift(): T | undefined;
front(): T | undefined;
readonly length: number;
}
export class QueueImpl<T> implements Queue<T> {
private readonly chunks_: T[][];
private readChunk_: T[];
private writeChunk_: T[];
private length_: number;
constructor() {
this.chunks_ = [[]];
this.readChunk_ = this.writeChunk_ = this.chunks_[0];
this.length_ = 0;
}
push(t: T): void {
this.writeChunk_.push(t);
this.length_ += 1;
if (this.writeChunk_.length === CHUNK_SIZE) {
this.writeChunk_ = [];
this.chunks_.push(this.writeChunk_);
}
}
front(): T | undefined {
if (this.length_ === 0) {
return undefined;
}
return this.readChunk_[0];
}
shift(): T | undefined {
if (this.length_ === 0) {
return undefined;
}
const t = this.readChunk_.shift();
this.length_ -= 1;
if (this.readChunk_.length === 0 && this.readChunk_ !== this.writeChunk_) {
this.chunks_.shift();
this.readChunk_ = this.chunks_[0];
}
return t;
}
get length(): number {
return this.length_;
}
}
cli/js/web/streams/readable-byte-stream-controller.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
/* eslint-disable @typescript-eslint/no-explicit-any */
// TODO reenable this lint here
import * as rs from "./readable-internals.ts";
import * as q from "./queue-mixin.ts";
import * as shared from "./shared-internals.ts";
import { ReadableStreamBYOBRequest } from "./readable-stream-byob-request.ts";
import { Queue } from "./queue.ts";
import { UnderlyingByteSource } from "../dom_types.d.ts";
export class ReadableByteStreamController
implements rs.SDReadableByteStreamController {
[rs.autoAllocateChunkSize_]: number | undefined;
[rs.byobRequest_]: rs.SDReadableStreamBYOBRequest | undefined;
[rs.cancelAlgorithm_]: rs.CancelAlgorithm;
[rs.closeRequested_]: boolean;
[rs.controlledReadableByteStream_]: rs.SDReadableStream<ArrayBufferView>;
[rs.pullAgain_]: boolean;
[rs.pullAlgorithm_]: rs.PullAlgorithm<ArrayBufferView>;
[rs.pulling_]: boolean;
[rs.pendingPullIntos_]: rs.PullIntoDescriptor[];
[rs.started_]: boolean;
[rs.strategyHWM_]: number;
[q.queue_]: Queue<{
buffer: ArrayBufferLike;
byteOffset: number;
byteLength: number;
}>;
[q.queueTotalSize_]: number;
constructor() {
throw new TypeError();
}
get byobRequest(): rs.SDReadableStreamBYOBRequest | undefined {
if (!rs.isReadableByteStreamController(this)) {
throw new TypeError();
}
if (
this[rs.byobRequest_] === undefined &&
this[rs.pendingPullIntos_].length > 0
) {
const firstDescriptor = this[rs.pendingPullIntos_][0];
const view = new Uint8Array(
firstDescriptor.buffer,
firstDescriptor.byteOffset + firstDescriptor.bytesFilled,
firstDescriptor.byteLength - firstDescriptor.bytesFilled
);
const byobRequest = Object.create(
ReadableStreamBYOBRequest.prototype
) as ReadableStreamBYOBRequest;
rs.setUpReadableStreamBYOBRequest(byobRequest, this, view);
this[rs.byobRequest_] = byobRequest;
}
return this[rs.byobRequest_];
}
get desiredSize(): number | null {
if (!rs.isReadableByteStreamController(this)) {
throw new TypeError();
}
return rs.readableByteStreamControllerGetDesiredSize(this);
}
close(): void {
if (!rs.isReadableByteStreamController(this)) {
throw new TypeError();
}
if (this[rs.closeRequested_]) {
throw new TypeError("Stream is already closing");
}
if (this[rs.controlledReadableByteStream_][shared.state_] !== "readable") {
throw new TypeError("Stream is closed or errored");
}
rs.readableByteStreamControllerClose(this);
}
enqueue(chunk: ArrayBufferView): void {
if (!rs.isReadableByteStreamController(this)) {
throw new TypeError();
}
if (this[rs.closeRequested_]) {
throw new TypeError("Stream is already closing");
}
if (this[rs.controlledReadableByteStream_][shared.state_] !== "readable") {
throw new TypeError("Stream is closed or errored");
}
if (!ArrayBuffer.isView(chunk)) {
throw new TypeError("chunk must be a valid ArrayBufferView");
}
// If ! IsDetachedBuffer(chunk.[[ViewedArrayBuffer]]) is true, throw a TypeError exception.
return rs.readableByteStreamControllerEnqueue(this, chunk);
}
error(error?: shared.ErrorResult): void {
if (!rs.isReadableByteStreamController(this)) {
throw new TypeError();
}
rs.readableByteStreamControllerError(this, error);
}
[rs.cancelSteps_](reason: shared.ErrorResult): Promise<void> {
if (this[rs.pendingPullIntos_].length > 0) {
const firstDescriptor = this[rs.pendingPullIntos_][0];
firstDescriptor.bytesFilled = 0;
}
q.resetQueue(this);
const result = this[rs.cancelAlgorithm_](reason);
rs.readableByteStreamControllerClearAlgorithms(this);
return result;
}
[rs.pullSteps_](
forAuthorCode: boolean
): Promise<IteratorResult<ArrayBufferView, any>> {
const stream = this[rs.controlledReadableByteStream_];
// Assert: ! ReadableStreamHasDefaultReader(stream) is true.
if (this[q.queueTotalSize_] > 0) {
// Assert: ! ReadableStreamGetNumReadRequests(stream) is 0.
const entry = this[q.queue_].shift()!;
this[q.queueTotalSize_] -= entry.byteLength;
rs.readableByteStreamControllerHandleQueueDrain(this);
const view = new Uint8Array(
entry.buffer,
entry.byteOffset,
entry.byteLength
);
return Promise.resolve(
rs.readableStreamCreateReadResult(view, false, forAuthorCode)
);
}
const autoAllocateChunkSize = this[rs.autoAllocateChunkSize_];
if (autoAllocateChunkSize !== undefined) {
let buffer: ArrayBuffer;
try {
buffer = new ArrayBuffer(autoAllocateChunkSize);
} catch (error) {
return Promise.reject(error);
}
const pullIntoDescriptor: rs.PullIntoDescriptor = {
buffer,
byteOffset: 0,
byteLength: autoAllocateChunkSize,
bytesFilled: 0,
elementSize: 1,
ctor: Uint8Array,
readerType: "default",
};
this[rs.pendingPullIntos_].push(pullIntoDescriptor);
}
const promise = rs.readableStreamAddReadRequest(stream, forAuthorCode);
rs.readableByteStreamControllerCallPullIfNeeded(this);
return promise;
}
}
export function setUpReadableByteStreamControllerFromUnderlyingSource(
stream: rs.SDReadableStream<ArrayBufferView>,
underlyingByteSource: UnderlyingByteSource,
highWaterMark: number
): void {
// Assert: underlyingByteSource is not undefined.
const controller = Object.create(
ReadableByteStreamController.prototype
) as ReadableByteStreamController;
const startAlgorithm = (): any => {
return shared.invokeOrNoop(underlyingByteSource, "start", [controller]);
};
const pullAlgorithm = shared.createAlgorithmFromUnderlyingMethod(
underlyingByteSource,
"pull",
[controller]
);
const cancelAlgorithm = shared.createAlgorithmFromUnderlyingMethod(
underlyingByteSource,
"cancel",
[]
);
let autoAllocateChunkSize = underlyingByteSource.autoAllocateChunkSize;
if (autoAllocateChunkSize !== undefined) {
autoAllocateChunkSize = Number(autoAllocateChunkSize);
if (
!shared.isInteger(autoAllocateChunkSize) ||
autoAllocateChunkSize <= 0
) {
throw new RangeError(
"autoAllocateChunkSize must be a positive, finite integer"
);
}
}
rs.setUpReadableByteStreamController(
stream,
controller,
startAlgorithm,
pullAlgorithm,
cancelAlgorithm,
highWaterMark,
autoAllocateChunkSize
);
}
cli/js/web/streams/readable-internals.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
此差异已折叠。
cli/js/web/streams/readable-stream-byob-reader.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
import * as rs from "./readable-internals.ts";
import * as shared from "./shared-internals.ts";
export class SDReadableStreamBYOBReader
implements rs.SDReadableStreamBYOBReader {
[rs.closedPromise_]: shared.ControlledPromise<void>;
[rs.ownerReadableStream_]: rs.SDReadableStream<ArrayBufferView> | undefined;
[rs.readIntoRequests_]: Array<
rs.ReadRequest<IteratorResult<ArrayBufferView>>
>;
constructor(stream: rs.SDReadableStream<ArrayBufferView>) {
if (!rs.isReadableStream(stream)) {
throw new TypeError();
}
if (
!rs.isReadableByteStreamController(stream[rs.readableStreamController_])
) {
throw new TypeError();
}
if (rs.isReadableStreamLocked(stream)) {
throw new TypeError("The stream is locked.");
}
rs.readableStreamReaderGenericInitialize(this, stream);
this[rs.readIntoRequests_] = [];
}
get closed(): Promise<void> {
if (!rs.isReadableStreamBYOBReader(this)) {
return Promise.reject(new TypeError());
}
return this[rs.closedPromise_].promise;
}
cancel(reason: shared.ErrorResult): Promise<void> {
if (!rs.isReadableStreamBYOBReader(this)) {
return Promise.reject(new TypeError());
}
const stream = this[rs.ownerReadableStream_];
if (stream === undefined) {
return Promise.reject(
new TypeError("Reader is not associated with a stream")
);
}
return rs.readableStreamCancel(stream, reason);
}
read(view: ArrayBufferView): Promise<IteratorResult<ArrayBufferView>> {
if (!rs.isReadableStreamBYOBReader(this)) {
return Promise.reject(new TypeError());
}
if (this[rs.ownerReadableStream_] === undefined) {
return Promise.reject(
new TypeError("Reader is not associated with a stream")
);
}
if (!ArrayBuffer.isView(view)) {
return Promise.reject(
new TypeError("view argument must be a valid ArrayBufferView")
);
}
// If ! IsDetachedBuffer(view.[[ViewedArrayBuffer]]) is true, return a promise rejected with a TypeError exception.
if (view.byteLength === 0) {
return Promise.reject(
new TypeError("supplied buffer view must be > 0 bytes")
);
}
return rs.readableStreamBYOBReaderRead(this, view, true);
}
releaseLock(): void {
if (!rs.isReadableStreamBYOBReader(this)) {
throw new TypeError();
}
if (this[rs.ownerReadableStream_] === undefined) {
throw new TypeError("Reader is not associated with a stream");
}
if (this[rs.readIntoRequests_].length > 0) {
throw new TypeError();
}
rs.readableStreamReaderGenericRelease(this);
}
}
cli/js/web/streams/readable-stream-byob-request.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
import * as rs from "./readable-internals.ts";
export class ReadableStreamBYOBRequest {
[rs.associatedReadableByteStreamController_]:
| rs.SDReadableByteStreamController
| undefined;
[rs.view_]: ArrayBufferView | undefined;
constructor() {
throw new TypeError();
}
get view(): ArrayBufferView {
if (!rs.isReadableStreamBYOBRequest(this)) {
throw new TypeError();
}
return this[rs.view_]!;
}
respond(bytesWritten: number): void {
if (!rs.isReadableStreamBYOBRequest(this)) {
throw new TypeError();
}
if (this[rs.associatedReadableByteStreamController_] === undefined) {
throw new TypeError();
}
// If! IsDetachedBuffer(this.[[view]].[[ViewedArrayBuffer]]) is true, throw a TypeError exception.
return rs.readableByteStreamControllerRespond(
this[rs.associatedReadableByteStreamController_]!,
bytesWritten
);
}
respondWithNewView(view: ArrayBufferView): void {
if (!rs.isReadableStreamBYOBRequest(this)) {
throw new TypeError();
}
if (this[rs.associatedReadableByteStreamController_] === undefined) {
throw new TypeError();
}
if (!ArrayBuffer.isView(view)) {
throw new TypeError("view parameter must be a TypedArray");
}
// If! IsDetachedBuffer(view.[[ViewedArrayBuffer]]) is true, throw a TypeError exception.
return rs.readableByteStreamControllerRespondWithNewView(
this[rs.associatedReadableByteStreamController_]!,
view
);
}
}
cli/js/web/streams/readable-stream-default-controller.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
/* eslint-disable @typescript-eslint/no-explicit-any */
// TODO reenable this lint here
import * as rs from "./readable-internals.ts";
import * as shared from "./shared-internals.ts";
import * as q from "./queue-mixin.ts";
import { Queue } from "./queue.ts";
import {
QueuingStrategySizeCallback,
UnderlyingSource,
} from "../dom_types.d.ts";
export class ReadableStreamDefaultController<OutputType>
implements rs.SDReadableStreamDefaultController<OutputType> {
[rs.cancelAlgorithm_]: rs.CancelAlgorithm;
[rs.closeRequested_]: boolean;
[rs.controlledReadableStream_]: rs.SDReadableStream<OutputType>;
[rs.pullAgain_]: boolean;
[rs.pullAlgorithm_]: rs.PullAlgorithm<OutputType>;
[rs.pulling_]: boolean;
[rs.strategyHWM_]: number;
[rs.strategySizeAlgorithm_]: QueuingStrategySizeCallback<OutputType>;
[rs.started_]: boolean;
[q.queue_]: Queue<q.QueueElement<OutputType>>;
[q.queueTotalSize_]: number;
constructor() {
throw new TypeError();
}
get desiredSize(): number | null {
return rs.readableStreamDefaultControllerGetDesiredSize(this);
}
close(): void {
if (!rs.isReadableStreamDefaultController(this)) {
throw new TypeError();
}
if (!rs.readableStreamDefaultControllerCanCloseOrEnqueue(this)) {
throw new TypeError(
"Cannot close, the stream is already closing or not readable"
);
}
rs.readableStreamDefaultControllerClose(this);
}
enqueue(chunk?: OutputType): void {
if (!rs.isReadableStreamDefaultController(this)) {
throw new TypeError();
}
if (!rs.readableStreamDefaultControllerCanCloseOrEnqueue(this)) {
throw new TypeError(
"Cannot enqueue, the stream is closing or not readable"
);
}
rs.readableStreamDefaultControllerEnqueue(this, chunk!);
}
error(e?: shared.ErrorResult): void {
if (!rs.isReadableStreamDefaultController(this)) {
throw new TypeError();
}
rs.readableStreamDefaultControllerError(this, e);
}
[rs.cancelSteps_](reason: shared.ErrorResult): Promise<void> {
q.resetQueue(this);
const result = this[rs.cancelAlgorithm_](reason);
rs.readableStreamDefaultControllerClearAlgorithms(this);
return result;
}
[rs.pullSteps_](
forAuthorCode: boolean
): Promise<IteratorResult<OutputType, any>> {
const stream = this[rs.controlledReadableStream_];
if (this[q.queue_].length > 0) {
const chunk = q.dequeueValue(this);
if (this[rs.closeRequested_] && this[q.queue_].length === 0) {
rs.readableStreamDefaultControllerClearAlgorithms(this);
rs.readableStreamClose(stream);
} else {
rs.readableStreamDefaultControllerCallPullIfNeeded(this);
}
return Promise.resolve(
rs.readableStreamCreateReadResult(chunk, false, forAuthorCode)
);
}
const pendingPromise = rs.readableStreamAddReadRequest(
stream,
forAuthorCode
);
rs.readableStreamDefaultControllerCallPullIfNeeded(this);
return pendingPromise;
}
}
export function setUpReadableStreamDefaultControllerFromUnderlyingSource<
OutputType
>(
stream: rs.SDReadableStream<OutputType>,
underlyingSource: UnderlyingSource<OutputType>,
highWaterMark: number,
sizeAlgorithm: QueuingStrategySizeCallback<OutputType>
): void {
// Assert: underlyingSource is not undefined.
const controller = Object.create(ReadableStreamDefaultController.prototype);
const startAlgorithm = (): any => {
return shared.invokeOrNoop(underlyingSource, "start", [controller]);
};
const pullAlgorithm = shared.createAlgorithmFromUnderlyingMethod(
underlyingSource,
"pull",
[controller]
);
const cancelAlgorithm = shared.createAlgorithmFromUnderlyingMethod(
underlyingSource,
"cancel",
[]
);
rs.setUpReadableStreamDefaultController(
stream,
controller,
startAlgorithm,
pullAlgorithm,
cancelAlgorithm,
highWaterMark,
sizeAlgorithm
);
}
cli/js/web/streams/readable-stream-default-reader.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
import * as rs from "./readable-internals.ts";
import * as shared from "./shared-internals.ts";
export class ReadableStreamDefaultReader<OutputType>
implements rs.SDReadableStreamReader<OutputType> {
[rs.closedPromise_]: shared.ControlledPromise<void>;
[rs.ownerReadableStream_]: rs.SDReadableStream<OutputType> | undefined;
[rs.readRequests_]: Array<rs.ReadRequest<IteratorResult<OutputType>>>;
constructor(stream: rs.SDReadableStream<OutputType>) {
if (!rs.isReadableStream(stream)) {
throw new TypeError();
}
if (rs.isReadableStreamLocked(stream)) {
throw new TypeError("The stream is locked.");
}
rs.readableStreamReaderGenericInitialize(this, stream);
this[rs.readRequests_] = [];
}
get closed(): Promise<void> {
if (!rs.isReadableStreamDefaultReader(this)) {
return Promise.reject(new TypeError());
}
return this[rs.closedPromise_].promise;
}
cancel(reason: shared.ErrorResult): Promise<void> {
if (!rs.isReadableStreamDefaultReader(this)) {
return Promise.reject(new TypeError());
}
const stream = this[rs.ownerReadableStream_];
if (stream === undefined) {
return Promise.reject(
new TypeError("Reader is not associated with a stream")
);
}
return rs.readableStreamCancel(stream, reason);
}
read(): Promise<IteratorResult<OutputType | undefined>> {
if (!rs.isReadableStreamDefaultReader(this)) {
return Promise.reject(new TypeError());
}
if (this[rs.ownerReadableStream_] === undefined) {
return Promise.reject(
new TypeError("Reader is not associated with a stream")
);
}
return rs.readableStreamDefaultReaderRead(this, true);
}
releaseLock(): void {
if (!rs.isReadableStreamDefaultReader(this)) {
throw new TypeError();
}
if (this[rs.ownerReadableStream_] === undefined) {
return;
}
if (this[rs.readRequests_].length !== 0) {
throw new TypeError("Cannot release a stream with pending read requests");
}
rs.readableStreamReaderGenericRelease(this);
}
}
cli/js/web/streams/readable-stream.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
/* eslint prefer-const: "off" */
// TODO remove this, surpressed because of
// 284:7 error 'branch1' is never reassigned. Use 'const' instead prefer-const
import * as rs from "./readable-internals.ts";
import * as shared from "./shared-internals.ts";
import {
QueuingStrategy,
QueuingStrategySizeCallback,
UnderlyingSource,
UnderlyingByteSource,
} from "../dom_types.d.ts";
import {
ReadableStreamDefaultController,
setUpReadableStreamDefaultControllerFromUnderlyingSource,
} from "./readable-stream-default-controller.ts";
import { ReadableStreamDefaultReader } from "./readable-stream-default-reader.ts";
import {
ReadableByteStreamController,
setUpReadableByteStreamControllerFromUnderlyingSource,
} from "./readable-byte-stream-controller.ts";
import { SDReadableStreamBYOBReader } from "./readable-stream-byob-reader.ts";
export class SDReadableStream<OutputType>
implements rs.SDReadableStream<OutputType> {
[shared.state_]: rs.ReadableStreamState;
[shared.storedError_]: shared.ErrorResult;
[rs.reader_]: rs.SDReadableStreamReader<OutputType> | undefined;
[rs.readableStreamController_]: rs.SDReadableStreamControllerBase<OutputType>;
constructor(
underlyingSource: UnderlyingByteSource,
strategy?: { highWaterMark?: number; size?: undefined }
);
constructor(
underlyingSource?: UnderlyingSource<OutputType>,
strategy?: QueuingStrategy<OutputType>
);
constructor(
underlyingSource: UnderlyingSource<OutputType> | UnderlyingByteSource = {},
strategy:
| QueuingStrategy<OutputType>
| { highWaterMark?: number; size?: undefined } = {}
) {
rs.initializeReadableStream(this);
const sizeFunc = strategy.size;
const stratHWM = strategy.highWaterMark;
const sourceType = underlyingSource.type;
if (sourceType === undefined) {
const sizeAlgorithm = shared.makeSizeAlgorithmFromSizeFunction(sizeFunc);
const highWaterMark = shared.validateAndNormalizeHighWaterMark(
stratHWM === undefined ? 1 : stratHWM
);
setUpReadableStreamDefaultControllerFromUnderlyingSource(
this,
underlyingSource as UnderlyingSource<OutputType>,
highWaterMark,
sizeAlgorithm
);
} else if (String(sourceType) === "bytes") {
if (sizeFunc !== undefined) {
throw new RangeError(
"bytes streams cannot have a strategy with a `size` field"
);
}
const highWaterMark = shared.validateAndNormalizeHighWaterMark(
stratHWM === undefined ? 0 : stratHWM
);
setUpReadableByteStreamControllerFromUnderlyingSource(
(this as unknown) as rs.SDReadableStream<ArrayBufferView>,
underlyingSource as UnderlyingByteSource,
highWaterMark
);
} else {
throw new RangeError(
"The underlying source's `type` field must be undefined or 'bytes'"
);
}
}
get locked(): boolean {
return rs.isReadableStreamLocked(this);
}
getReader(): rs.SDReadableStreamDefaultReader<OutputType>;
getReader(options: { mode?: "byob" }): rs.SDReadableStreamBYOBReader;
getReader(options?: {
mode?: "byob";
}):
| rs.SDReadableStreamDefaultReader<OutputType>
| rs.SDReadableStreamBYOBReader {
if (!rs.isReadableStream(this)) {
throw new TypeError();
}
if (options === undefined) {
options = {};
}
const { mode } = options;
if (mode === undefined) {
return new ReadableStreamDefaultReader(this);
} else if (String(mode) === "byob") {
return new SDReadableStreamBYOBReader(
(this as unknown) as rs.SDReadableStream<ArrayBufferView>
);
}
throw RangeError("mode option must be undefined or `byob`");
}
cancel(reason: shared.ErrorResult): Promise<void> {
if (!rs.isReadableStream(this)) {
return Promise.reject(new TypeError());
}
if (rs.isReadableStreamLocked(this)) {
return Promise.reject(new TypeError("Cannot cancel a locked stream"));
}
return rs.readableStreamCancel(this, reason);
}
tee(): [SDReadableStream<OutputType>, SDReadableStream<OutputType>] {
return readableStreamTee(this, false);
}
/* TODO reenable these methods when we bring in writableStreams and transport types
pipeThrough<ResultType>(
transform: rs.GenericTransformStream<OutputType, ResultType>,
options: PipeOptions = {}
): rs.SDReadableStream<ResultType> {
const { readable, writable } = transform;
if (!rs.isReadableStream(this)) {
throw new TypeError();
}
if (!ws.isWritableStream(writable)) {
throw new TypeError("writable must be a WritableStream");
}
if (!rs.isReadableStream(readable)) {
throw new TypeError("readable must be a ReadableStream");
}
if (options.signal !== undefined && !shared.isAbortSignal(options.signal)) {
throw new TypeError("options.signal must be an AbortSignal instance");
}
if (rs.isReadableStreamLocked(this)) {
throw new TypeError("Cannot pipeThrough on a locked stream");
}
if (ws.isWritableStreamLocked(writable)) {
throw new TypeError("Cannot pipeThrough to a locked stream");
}
const pipeResult = pipeTo(this, writable, options);
pipeResult.catch(() => {});
return readable;
}
pipeTo(
dest: ws.WritableStream<OutputType>,
options: PipeOptions = {}
): Promise<void> {
if (!rs.isReadableStream(this)) {
return Promise.reject(new TypeError());
}
if (!ws.isWritableStream(dest)) {
return Promise.reject(
new TypeError("destination must be a WritableStream")
);
}
if (options.signal !== undefined && !shared.isAbortSignal(options.signal)) {
return Promise.reject(
new TypeError("options.signal must be an AbortSignal instance")
);
}
if (rs.isReadableStreamLocked(this)) {
return Promise.reject(new TypeError("Cannot pipe from a locked stream"));
}
if (ws.isWritableStreamLocked(dest)) {
return Promise.reject(new TypeError("Cannot pipe to a locked stream"));
}
return pipeTo(this, dest, options);
}
*/
}
export function createReadableStream<OutputType>(
startAlgorithm: rs.StartAlgorithm,
pullAlgorithm: rs.PullAlgorithm<OutputType>,
cancelAlgorithm: rs.CancelAlgorithm,
highWaterMark?: number,
sizeAlgorithm?: QueuingStrategySizeCallback<OutputType>
): SDReadableStream<OutputType> {
if (highWaterMark === undefined) {
highWaterMark = 1;
}
if (sizeAlgorithm === undefined) {
sizeAlgorithm = (): number => 1;
}
// Assert: ! IsNonNegativeNumber(highWaterMark) is true.
const stream = Object.create(SDReadableStream.prototype) as SDReadableStream<
OutputType
>;
rs.initializeReadableStream(stream);
const controller = Object.create(
ReadableStreamDefaultController.prototype
) as ReadableStreamDefaultController<OutputType>;
rs.setUpReadableStreamDefaultController(
stream,
controller,
startAlgorithm,
pullAlgorithm,
cancelAlgorithm,
highWaterMark,
sizeAlgorithm
);
return stream;
}
export function createReadableByteStream<OutputType>(
startAlgorithm: rs.StartAlgorithm,
pullAlgorithm: rs.PullAlgorithm<OutputType>,
cancelAlgorithm: rs.CancelAlgorithm,
highWaterMark?: number,
autoAllocateChunkSize?: number
): SDReadableStream<OutputType> {
if (highWaterMark === undefined) {
highWaterMark = 0;
}
// Assert: ! IsNonNegativeNumber(highWaterMark) is true.
if (autoAllocateChunkSize !== undefined) {
if (
!shared.isInteger(autoAllocateChunkSize) ||
autoAllocateChunkSize <= 0
) {
throw new RangeError(
"autoAllocateChunkSize must be a positive, finite integer"
);
}
}
const stream = Object.create(SDReadableStream.prototype) as SDReadableStream<
OutputType
>;
rs.initializeReadableStream(stream);
const controller = Object.create(
ReadableByteStreamController.prototype
) as ReadableByteStreamController;
rs.setUpReadableByteStreamController(
(stream as unknown) as SDReadableStream<ArrayBufferView>,
controller,
startAlgorithm,
(pullAlgorithm as unknown) as rs.PullAlgorithm<ArrayBufferView>,
cancelAlgorithm,
highWaterMark,
autoAllocateChunkSize
);
return stream;
}
export function readableStreamTee<OutputType>(
stream: SDReadableStream<OutputType>,
cloneForBranch2: boolean
): [SDReadableStream<OutputType>, SDReadableStream<OutputType>] {
if (!rs.isReadableStream(stream)) {
throw new TypeError();
}
const reader = new ReadableStreamDefaultReader(stream);
let closedOrErrored = false;
let canceled1 = false;
let canceled2 = false;
let reason1: shared.ErrorResult;
let reason2: shared.ErrorResult;
let branch1: SDReadableStream<OutputType>;
let branch2: SDReadableStream<OutputType>;
let cancelResolve: (reason: shared.ErrorResult) => void;
const cancelPromise = new Promise<void>(
(resolve) => (cancelResolve = resolve)
);
const pullAlgorithm = (): Promise<void> => {
return rs
.readableStreamDefaultReaderRead(reader)
.then(({ value, done }) => {
if (done && !closedOrErrored) {
if (!canceled1) {
rs.readableStreamDefaultControllerClose(
branch1![
rs.readableStreamController_
] as ReadableStreamDefaultController<OutputType>
);
}
if (!canceled2) {
rs.readableStreamDefaultControllerClose(
branch2![
rs.readableStreamController_
] as ReadableStreamDefaultController<OutputType>
);
}
closedOrErrored = true;
}
if (closedOrErrored) {
return;
}
const value1 = value;
let value2 = value;
if (!canceled1) {
rs.readableStreamDefaultControllerEnqueue(
branch1![
rs.readableStreamController_
] as ReadableStreamDefaultController<OutputType>,
value1!
);
}
if (!canceled2) {
if (cloneForBranch2) {
value2 = shared.cloneValue(value2);
}
rs.readableStreamDefaultControllerEnqueue(
branch2![
rs.readableStreamController_
] as ReadableStreamDefaultController<OutputType>,
value2!
);
}
});
};
const cancel1Algorithm = (reason: shared.ErrorResult): Promise<void> => {
canceled1 = true;
reason1 = reason;
if (canceled2) {
const cancelResult = rs.readableStreamCancel(stream, [reason1, reason2]);
cancelResolve(cancelResult);
}
return cancelPromise;
};
const cancel2Algorithm = (reason: shared.ErrorResult): Promise<void> => {
canceled2 = true;
reason2 = reason;
if (canceled1) {
const cancelResult = rs.readableStreamCancel(stream, [reason1, reason2]);
cancelResolve(cancelResult);
}
return cancelPromise;
};
const startAlgorithm = (): undefined => undefined;
branch1 = createReadableStream(
startAlgorithm,
pullAlgorithm,
cancel1Algorithm
);
branch2 = createReadableStream(
startAlgorithm,
pullAlgorithm,
cancel2Algorithm
);
reader[rs.closedPromise_].promise.catch((error) => {
if (!closedOrErrored) {
rs.readableStreamDefaultControllerError(
branch1![
rs.readableStreamController_
] as ReadableStreamDefaultController<OutputType>,
error
);
rs.readableStreamDefaultControllerError(
branch2![
rs.readableStreamController_
] as ReadableStreamDefaultController<OutputType>,
error
);
closedOrErrored = true;
}
});
return [branch1, branch2];
}
cli/js/web/streams/readable_byte_stream_controller.ts 0 → 100644
浏览文件 @ 8bcfc03d
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
import {
BufferQueueItem,
CancelAlgorithm,
isDetachedBuffer,
isReadableByteStreamController,
PullAlgorithm,
resetQueue,
readableByteStreamControllerCallPullIfNeeded,
readableByteStreamControllerClearAlgorithms,
readableByteStreamControllerClose,
readableByteStreamControllerEnqueue,
readableByteStreamControllerError,
readableByteStreamControllerGetDesiredSize,
readableByteStreamControllerHandleQueueDrain,
readableStreamAddReadRequest,
readableStreamHasDefaultReader,
readableStreamGetNumReadRequests,
readableStreamCreateReadResult,
} from "./internals.ts";
import { ReadableStreamImpl } from "./readable_stream.ts";
import * as sym from "./symbols.ts";
import { assert } from "../../util.ts";
import { symbols } from "../../symbols.ts";
export class ReadableByteStreamControllerImpl
implements ReadableByteStreamController {
[sym.autoAllocateChunkSize]: number | undefined;
[sym.byobRequest]: undefined;
[sym.cancelAlgorithm]: CancelAlgorithm;
[sym.closeRequested]: boolean;
[sym.controlledReadableByteStream]: ReadableStreamImpl<Uint8Array>;
[sym.pullAgain]: boolean;
[sym.pullAlgorithm]: PullAlgorithm;
[sym.pulling]: boolean;
[sym.queue]: BufferQueueItem[];
[sym.queueTotalSize]: number;
[sym.started]: boolean;
[sym.strategyHWM]: number;
private constructor() {
throw new TypeError(
"ReadableByteStreamController's constructor cannot be called."
);
}
get byobRequest(): undefined {
return undefined;
}
get desiredSize(): number | null {
if (!isReadableByteStreamController(this)) {
throw new TypeError("Invalid ReadableByteStreamController.");
}
return readableByteStreamControllerGetDesiredSize(this);
}
close(): void {
if (!isReadableByteStreamController(this)) {
throw new TypeError("Invalid ReadableByteStreamController.");
}
if (this[sym.closeRequested]) {
throw new TypeError("Closed already requested.");
}
if (this[sym.controlledReadableByteStream][sym.state] !== "readable") {
throw new TypeError(
"ReadableByteStreamController's stream is not in a readable state."
);
}
readableByteStreamControllerClose(this);
}
enqueue(chunk: ArrayBufferView): void {
if (!isReadableByteStreamController(this)) {
throw new TypeError("Invalid ReadableByteStreamController.");
}
if (this[sym.closeRequested]) {
throw new TypeError("Closed already requested.");
}
if (this[sym.controlledReadableByteStream][sym.state] !== "readable") {
throw new TypeError(
"ReadableByteStreamController's stream is not in a readable state."
);
}
if (!ArrayBuffer.isView(chunk)) {
throw new TypeError(
"You can only enqueue array buffer views when using a ReadableByteStreamController"
);
}
if (isDetachedBuffer(chunk.buffer)) {
throw new TypeError("Cannot enqueue a view onto a detached ArrayBuffer");
}
readableByteStreamControllerEnqueue(this, chunk);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
error(error?: any): void {
if (!isReadableByteStreamController(this)) {
throw new TypeError("Invalid ReadableByteStreamController.");
}
readableByteStreamControllerError(this, error);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
[sym.cancelSteps](reason: any): PromiseLike<void> {
// 3.11.5.1.1 If this.[[pendingPullIntos]] is not empty,
resetQueue(this);
const result = this[sym.cancelAlgorithm](reason);
readableByteStreamControllerClearAlgorithms(this);
return result;
}
[sym.pullSteps](): Promise<ReadableStreamReadResult<Uint8Array>> {
const stream = this[sym.controlledReadableByteStream];
assert(readableStreamHasDefaultReader(stream));
if (this[sym.queueTotalSize] > 0) {
assert(readableStreamGetNumReadRequests(stream) === 0);
const entry = this[sym.queue].shift();
assert(entry);
this[sym.queueTotalSize] -= entry.size;
readableByteStreamControllerHandleQueueDrain(this);
const view = new Uint8Array(entry.value, entry.offset, entry.size);
return Promise.resolve(
readableStreamCreateReadResult(
view,
false,
stream[sym.reader]![sym.forAuthorCode]
)
);
}
// 3.11.5.2.5 If autoAllocateChunkSize is not undefined,
const promise = readableStreamAddReadRequest(stream);
readableByteStreamControllerCallPullIfNeeded(this);
return promise;
}
[symbols.customInspect](): string {
return `ReadableByteStreamController { byobRequest: ${String(
this.byobRequest
)}, desiredSize: ${String(this.desiredSize)} }`;
}
}
cli/js/web/streams/readable_stream.ts 0 → 100644
浏览文件 @ 8bcfc03d
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
import {
acquireReadableStreamDefaultReader,
initializeReadableStream,
isReadableStream,
isReadableStreamLocked,
isUnderlyingByteSource,
makeSizeAlgorithmFromSizeFunction,
readableStreamCancel,
ReadableStreamGenericReader,
readableStreamTee,
setUpReadableByteStreamControllerFromUnderlyingSource,
setUpReadableStreamDefaultControllerFromUnderlyingSource,
validateAndNormalizeHighWaterMark,
} from "./internals.ts";
import { ReadableByteStreamControllerImpl } from "./readable_byte_stream_controller.ts";
import { ReadableStreamAsyncIteratorPrototype } from "./readable_stream_async_iterator.ts";
import { ReadableStreamDefaultControllerImpl } from "./readable_stream_default_controller.ts";
import * as sym from "./symbols.ts";
import { symbols } from "../../symbols.ts";
import { notImplemented } from "../../util.ts";
// eslint-disable-next-line @typescript-eslint/no-explicit-any
export class ReadableStreamImpl<R = any> implements ReadableStream<R> {
[sym.disturbed]: boolean;
[sym.readableStreamController]:
| ReadableStreamDefaultControllerImpl<R>
| ReadableByteStreamControllerImpl;
[sym.reader]: ReadableStreamGenericReader<R> | undefined;
[sym.state]: "readable" | "closed" | "errored";
// eslint-disable-next-line @typescript-eslint/no-explicit-any
[sym.storedError]: any;
constructor(
underlyingSource: UnderlyingByteSource | UnderlyingSource<R> = {},
strategy:
| {
highWaterMark?: number;
size?: undefined;
}
| QueuingStrategy<R> = {}
) {
initializeReadableStream(this);
const { size } = strategy;
let { highWaterMark } = strategy;
const { type } = underlyingSource;
if (isUnderlyingByteSource(underlyingSource)) {
if (size !== undefined) {
throw new RangeError(
`When underlying source is "bytes", strategy.size must be undefined.`
);
}
highWaterMark = validateAndNormalizeHighWaterMark(highWaterMark ?? 0);
setUpReadableByteStreamControllerFromUnderlyingSource(
this,
underlyingSource,
highWaterMark
);
} else if (type === undefined) {
const sizeAlgorithm = makeSizeAlgorithmFromSizeFunction(size);
highWaterMark = validateAndNormalizeHighWaterMark(highWaterMark ?? 1);
setUpReadableStreamDefaultControllerFromUnderlyingSource(
this,
underlyingSource,
highWaterMark,
sizeAlgorithm
);
} else {
throw new RangeError(
`Valid values for underlyingSource are "bytes" or undefined. Received: "${type}".`
);
}
}
get locked(): boolean {
if (!isReadableStream(this)) {
throw new TypeError("Invalid ReadableStream.");
}
return isReadableStreamLocked(this);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
cancel(reason?: any): Promise<void> {
if (!isReadableStream(this)) {
return Promise.reject(new TypeError("Invalid ReadableStream."));
}
if (isReadableStreamLocked(this)) {
return Promise.reject(
new TypeError("Cannot cancel a locked ReadableStream.")
);
}
return readableStreamCancel(this, reason);
}
getIterator({
preventCancel,
}: { preventCancel?: boolean } = {}): AsyncIterableIterator<R> {
if (!isReadableStream(this)) {
throw new TypeError("Invalid ReadableStream.");
}
const reader = acquireReadableStreamDefaultReader(this);
const iterator = Object.create(ReadableStreamAsyncIteratorPrototype);
iterator[sym.asyncIteratorReader] = reader;
iterator[sym.preventCancel] = Boolean(preventCancel);
return iterator;
}
getReader({ mode }: { mode?: string } = {}): ReadableStreamDefaultReader<R> {
if (!isReadableStream(this)) {
throw new TypeError("Invalid ReadableStream.");
}
if (mode === undefined) {
return acquireReadableStreamDefaultReader(this, true);
}
mode = String(mode);
// 3.2.5.4.4 If mode is "byob", return ? AcquireReadableStreamBYOBReader(this, true).
throw new RangeError(`Unsupported mode "${mode}"`);
}
pipeThrough<T>(): // {
// writable,
// readable,
// }: {
// writable: WritableStream<R>;
// readable: ReadableStream<T>;
// },
// { preventClose, preventAbort, preventCancel, signal }: PipeOptions = {},
ReadableStream<T> {
return notImplemented();
// if (!isReadableStream(this)) {
// throw new TypeError("Invalid ReadableStream.");
// }
// if (!isWritableStream(writable)) {
// throw new TypeError("writable is not a valid WritableStream.");
// }
// if (!isReadableStream(readable)) {
// throw new TypeError("readable is not a valid ReadableStream.");
// }
// preventClose = Boolean(preventClose);
// preventAbort = Boolean(preventAbort);
// preventCancel = Boolean(preventCancel);
// if (signal && !(signal instanceof AbortSignalImpl)) {
// throw new TypeError("Invalid signal.");
// }
// if (isReadableStreamLocked(this)) {
// throw new TypeError("ReadableStream is locked.");
// }
// if (isWritableStreamLocked(writable)) {
// throw new TypeError("writable is locked.");
// }
// readableStreamPipeTo(
// this,
// writable,
// preventClose,
// preventAbort,
// preventCancel,
// signal,
// );
// return readable;
}
pipeTo(): // dest: WritableStream<R>,
// { preventClose, preventAbort, preventCancel, signal }: PipeOptions = {},
Promise<void> {
return notImplemented();
// if (!isReadableStream(this)) {
// return Promise.reject(new TypeError("Invalid ReadableStream."));
// }
// if (!isWritableStream(dest)) {
// return Promise.reject(
// new TypeError("dest is not a valid WritableStream."),
// );
// }
// preventClose = Boolean(preventClose);
// preventAbort = Boolean(preventAbort);
// preventCancel = Boolean(preventCancel);
// if (signal && !(signal instanceof AbortSignalImpl)) {
// return Promise.reject(new TypeError("Invalid signal."));
// }
// if (isReadableStreamLocked(this)) {
// return Promise.reject(new TypeError("ReadableStream is locked."));
// }
// if (isWritableStreamLocked(this)) {
// return Promise.reject(new TypeError("dest is locked."));
// }
// return readableStreamPipeTo(
// this,
// dest,
// preventClose,
// preventAbort,
// preventCancel,
// signal,
// );
}
tee(): [ReadableStreamImpl<R>, ReadableStreamImpl<R>] {
if (!isReadableStream(this)) {
throw new TypeError("Invalid ReadableStream.");
}
return readableStreamTee(this, false);
}
[symbols.customInspect](): string {
return `ReadableStream { locked: ${String(this.locked)} }`;
}
[Symbol.asyncIterator](
options: {
preventCancel?: boolean;
} = {}
): AsyncIterableIterator<R> {
return this.getIterator(options);
}
}
cli/js/web/streams/readable_stream_async_iterator.ts 0 → 100644
浏览文件 @ 8bcfc03d
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
import * as sym from "./symbols.ts";
import {
isReadableStreamAsyncIterator,
ReadableStreamAsyncIterator,
readableStreamCreateReadResult,
readableStreamReaderGenericCancel,
readableStreamReaderGenericRelease,
readableStreamDefaultReaderRead,
} from "./internals.ts";
import { assert } from "../../util.ts";
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const AsyncIteratorPrototype: AsyncIterableIterator<any> = Object.getPrototypeOf(
Object.getPrototypeOf(async function* () {}).prototype
);
export const ReadableStreamAsyncIteratorPrototype: ReadableStreamAsyncIterator = Object.setPrototypeOf(
{
next(
this: ReadableStreamAsyncIterator
// eslint-disable-next-line @typescript-eslint/no-explicit-any
): Promise<ReadableStreamReadResult<any>> {
if (!isReadableStreamAsyncIterator(this)) {
return Promise.reject(
new TypeError("invalid ReadableStreamAsyncIterator.")
);
}
const reader = this[sym.asyncIteratorReader];
if (!reader[sym.ownerReadableStream]) {
return Promise.reject(
new TypeError("reader owner ReadableStream is undefined.")
);
}
return readableStreamDefaultReaderRead(reader).then((result) => {
assert(typeof result === "object");
const { done } = result;
assert(typeof done === "boolean");
if (done) {
readableStreamReaderGenericRelease(reader);
}
const { value } = result;
return readableStreamCreateReadResult(value, done, true);
});
},
return(
this: ReadableStreamAsyncIterator,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
value?: any | PromiseLike<any>
// eslint-disable-next-line @typescript-eslint/no-explicit-any
): Promise<ReadableStreamReadResult<any>> {
if (!isReadableStreamAsyncIterator(this)) {
return Promise.reject(
new TypeError("invalid ReadableStreamAsyncIterator.")
);
}
const reader = this[sym.asyncIteratorReader];
if (!reader[sym.ownerReadableStream]) {
return Promise.reject(
new TypeError("reader owner ReadableStream is undefined.")
);
}
if (reader[sym.readRequests].length) {
return Promise.reject(
new TypeError("reader has outstanding read requests.")
);
}
if (!this[sym.preventCancel]) {
const result = readableStreamReaderGenericCancel(reader, value);
readableStreamReaderGenericRelease(reader);
return result.then(() =>
readableStreamCreateReadResult(value, true, true)
);
}
readableStreamReaderGenericRelease(reader);
return Promise.resolve(readableStreamCreateReadResult(value, true, true));
},
},
AsyncIteratorPrototype
);
cli/js/web/streams/readable_stream_default_controller.ts 0 → 100644
浏览文件 @ 8bcfc03d
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
import {
CancelAlgorithm,
dequeueValue,
isReadableStreamDefaultController,
Pair,
PullAlgorithm,
readableStreamAddReadRequest,
readableStreamClose,
readableStreamCreateReadResult,
readableStreamDefaultControllerCallPullIfNeeded,
readableStreamDefaultControllerCanCloseOrEnqueue,
readableStreamDefaultControllerClearAlgorithms,
readableStreamDefaultControllerClose,
readableStreamDefaultControllerEnqueue,
readableStreamDefaultControllerError,
readableStreamDefaultControllerGetDesiredSize,
resetQueue,
SizeAlgorithm,
} from "./internals.ts";
import { ReadableStreamImpl } from "./readable_stream.ts";
import * as sym from "./symbols.ts";
import { symbols } from "../../symbols.ts";
// eslint-disable-next-line @typescript-eslint/no-explicit-any
export class ReadableStreamDefaultControllerImpl<R = any>
implements ReadableStreamDefaultController<R> {
[sym.cancelAlgorithm]: CancelAlgorithm;
[sym.closeRequested]: boolean;
[sym.controlledReadableStream]: ReadableStreamImpl<R>;
[sym.pullAgain]: boolean;
[sym.pullAlgorithm]: PullAlgorithm;
[sym.pulling]: boolean;
[sym.queue]: Array<Pair<R>>;
[sym.queueTotalSize]: number;
[sym.started]: boolean;
[sym.strategyHWM]: number;
[sym.strategySizeAlgorithm]: SizeAlgorithm<R>;
private constructor() {
throw new TypeError(
"ReadableStreamDefaultController's constructor cannot be called."
);
}
get desiredSize(): number | null {
if (!isReadableStreamDefaultController(this)) {
throw new TypeError("Invalid ReadableStreamDefaultController.");
}
return readableStreamDefaultControllerGetDesiredSize(this);
}
close(): void {
if (!isReadableStreamDefaultController(this)) {
throw new TypeError("Invalid ReadableStreamDefaultController.");
}
if (!readableStreamDefaultControllerCanCloseOrEnqueue(this)) {
throw new TypeError(
"ReadableStreamDefaultController cannot close or enqueue."
);
}
readableStreamDefaultControllerClose(this);
}
enqueue(chunk: R): void {
if (!isReadableStreamDefaultController(this)) {
throw new TypeError("Invalid ReadableStreamDefaultController.");
}
if (!readableStreamDefaultControllerCanCloseOrEnqueue(this)) {
throw new TypeError("ReadableSteamController cannot enqueue.");
}
return readableStreamDefaultControllerEnqueue(this, chunk);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
error(error?: any): void {
if (!isReadableStreamDefaultController(this)) {
throw new TypeError("Invalid ReadableStreamDefaultController.");
}
readableStreamDefaultControllerError(this, error);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
[sym.cancelSteps](reason?: any): PromiseLike<void> {
resetQueue(this);
const result = this[sym.cancelAlgorithm](reason);
readableStreamDefaultControllerClearAlgorithms(this);
return result;
}
[sym.pullSteps](): Promise<ReadableStreamReadResult<R>> {
const stream = this[sym.controlledReadableStream];
if (this[sym.queue].length) {
const chunk = dequeueValue<R>(this);
if (this[sym.closeRequested] && this[sym.queue].length === 0) {
readableStreamDefaultControllerClearAlgorithms(this);
readableStreamClose(stream);
} else {
readableStreamDefaultControllerCallPullIfNeeded(this);
}
return Promise.resolve(
readableStreamCreateReadResult(
chunk,
false,
stream[sym.reader]![sym.forAuthorCode]
)
);
}
const pendingPromise = readableStreamAddReadRequest(stream);
readableStreamDefaultControllerCallPullIfNeeded(this);
return pendingPromise;
}
[symbols.customInspect](): string {
return `ReadableStreamDefaultController { desiredSize: ${String(
this.desiredSize
)} }`;
}
}
cli/js/web/streams/readable_stream_default_reader.ts 0 → 100644
浏览文件 @ 8bcfc03d
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
import {
Deferred,
isReadableStream,
isReadableStreamDefaultReader,
isReadableStreamLocked,
readableStreamDefaultReaderRead,
readableStreamReaderGenericCancel,
readableStreamReaderGenericInitialize,
readableStreamReaderGenericRelease,
} from "./internals.ts";
import { ReadableStreamImpl } from "./readable_stream.ts";
import * as sym from "./symbols.ts";
import { symbols } from "../../symbols.ts";
// eslint-disable-next-line @typescript-eslint/no-explicit-any
export class ReadableStreamDefaultReaderImpl<R = any>
implements ReadableStreamDefaultReader<R> {
[sym.closedPromise]: Deferred<void>;
[sym.forAuthorCode]: boolean;
[sym.ownerReadableStream]: ReadableStreamImpl<R>;
[sym.readRequests]: Array<Deferred<ReadableStreamReadResult<R>>>;
constructor(stream: ReadableStream<R>) {
if (!isReadableStream(stream)) {
throw new TypeError("stream is not a ReadableStream.");
}
if (isReadableStreamLocked(stream)) {
throw new TypeError("stream is locked.");
}
readableStreamReaderGenericInitialize(this, stream);
this[sym.readRequests] = [];
}
get closed(): Promise<void> {
if (!isReadableStreamDefaultReader(this)) {
return Promise.reject(
new TypeError("Invalid ReadableStreamDefaultReader.")
);
}
return (
this[sym.closedPromise].promise ??
Promise.reject(new TypeError("Invalid reader."))
);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
cancel(reason?: any): Promise<void> {
if (!isReadableStreamDefaultReader(this)) {
return Promise.reject(
new TypeError("Invalid ReadableStreamDefaultReader.")
);
}
if (!this[sym.ownerReadableStream]) {
return Promise.reject(new TypeError("Invalid reader."));
}
return readableStreamReaderGenericCancel(this, reason);
}
read(): Promise<ReadableStreamReadResult<R>> {
if (!isReadableStreamDefaultReader(this)) {
return Promise.reject(
new TypeError("Invalid ReadableStreamDefaultReader.")
);
}
if (!this[sym.ownerReadableStream]) {
return Promise.reject(new TypeError("Invalid reader."));
}
return readableStreamDefaultReaderRead(this);
}
releaseLock(): void {
if (!isReadableStreamDefaultReader(this)) {
throw new TypeError("Invalid ReadableStreamDefaultReader.");
}
if (this[sym.ownerReadableStream] === undefined) {
return;
}
if (this[sym.readRequests].length) {
throw new TypeError("Cannot release lock with pending read requests.");
}
readableStreamReaderGenericRelease(this);
}
[symbols.customInspect](): string {
return `ReadableStreamDefaultReader { closed: Promise }`;
}
}
cli/js/web/streams/shared-internals.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
此差异已折叠。
cli/js/web/streams/strategies.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
// Forked from https://github.com/stardazed/sd-streams/tree/8928cf04b035fd02fb1340b7eb541c76be37e546
// Copyright (c) 2018-Present by Arthur Langereis - @zenmumbler MIT
/* eslint-disable @typescript-eslint/no-explicit-any */
// TODO reenable this lint here
import { QueuingStrategy } from "../dom_types.d.ts";
export class ByteLengthQueuingStrategy
implements QueuingStrategy<ArrayBufferView> {
highWaterMark: number;
constructor(options: { highWaterMark: number }) {
this.highWaterMark = options.highWaterMark;
}
size(chunk: ArrayBufferView): number {
return chunk.byteLength;
}
}
export class CountQueuingStrategy implements QueuingStrategy<any> {
highWaterMark: number;
constructor(options: { highWaterMark: number }) {
this.highWaterMark = options.highWaterMark;
}
size(): number {
return 1;
}
}
cli/js/web/streams/symbols.ts 0 → 100644
浏览文件 @ 8bcfc03d
此差异已折叠。
cli/js/web/streams/transform-internals.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
此差异已折叠。
cli/js/web/streams/transform-stream.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
此差异已折叠。
cli/js/web/streams/writable-internals.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
此差异已折叠。
cli/js/web/streams/writable-stream.ts 已删除 100644 → 0
浏览文件 @ b270d6c8
此差异已折叠。
cli/js/web/util.ts
浏览文件 @ 8bcfc03d
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或