1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
import {
Deferred,
getDeferred,
initializeTransformStream,
invokeOrNoop,
isTransformStream,
makeSizeAlgorithmFromSizeFunction,
setFunctionName,
setUpTransformStreamDefaultControllerFromTransformer,
validateAndNormalizeHighWaterMark,
} from "./internals.ts";
import { ReadableStreamImpl } from "./readable_stream.ts";
import * as sym from "./symbols.ts";
import { TransformStreamDefaultControllerImpl } from "./transform_stream_default_controller.ts";
import { WritableStreamImpl } from "./writable_stream.ts";
import { customInspect, inspect } from "../console.ts";
// eslint-disable-next-line @typescript-eslint/no-explicit-any
export class TransformStreamImpl<I = any, O = any>
implements TransformStream<I, O> {
[sym.backpressure]?: boolean;
[sym.backpressureChangePromise]?: Deferred<void>;
[sym.readable]: ReadableStreamImpl<O>;
[sym.transformStreamController]: TransformStreamDefaultControllerImpl<I, O>;
[sym.writable]: WritableStreamImpl<I>;
constructor(
transformer: Transformer<I, O> = {},
writableStrategy: QueuingStrategy<I> = {},
readableStrategy: QueuingStrategy<O> = {}
) {
const writableSizeFunction = writableStrategy.size;
let writableHighWaterMark = writableStrategy.highWaterMark;
const readableSizeFunction = readableStrategy.size;
let readableHighWaterMark = readableStrategy.highWaterMark;
const writableType = transformer.writableType;
if (writableType !== undefined) {
throw new RangeError(
`Expected transformer writableType to be undefined, received "${String(
writableType
)}"`
);
}
const writableSizeAlgorithm = makeSizeAlgorithmFromSizeFunction(
writableSizeFunction
);
if (writableHighWaterMark === undefined) {
writableHighWaterMark = 1;
}
writableHighWaterMark = validateAndNormalizeHighWaterMark(
writableHighWaterMark
);
const readableType = transformer.readableType;
if (readableType !== undefined) {
throw new RangeError(
`Expected transformer readableType to be undefined, received "${String(
readableType
)}"`
);
}
const readableSizeAlgorithm = makeSizeAlgorithmFromSizeFunction(
readableSizeFunction
);
if (readableHighWaterMark === undefined) {
readableHighWaterMark = 1;
}
readableHighWaterMark = validateAndNormalizeHighWaterMark(
readableHighWaterMark
);
const startPromise = getDeferred<void>();
initializeTransformStream(
this,
startPromise.promise,
writableHighWaterMark,
writableSizeAlgorithm,
readableHighWaterMark,
readableSizeAlgorithm
);
// the brand check expects this, and the brand check occurs in the following
// but the property hasn't been defined.
Object.defineProperty(this, sym.transformStreamController, {
value: undefined,
writable: true,
configurable: true,
});
setUpTransformStreamDefaultControllerFromTransformer(this, transformer);
const startResult: void | PromiseLike<void> = invokeOrNoop(
transformer,
"start",
this[sym.transformStreamController]
);
startPromise.resolve(startResult);
}
get readable(): ReadableStream<O> {
if (!isTransformStream(this)) {
throw new TypeError("Invalid TransformStream.");
}
return this[sym.readable];
}
get writable(): WritableStream<I> {
if (!isTransformStream(this)) {
throw new TypeError("Invalid TransformStream.");
}
return this[sym.writable];
}
[customInspect](): string {
return `${this.constructor.name} {\n readable: ${inspect(
this.readable
)}\n writable: ${inspect(this.writable)}\n}`;
}
}
setFunctionName(TransformStreamImpl, "TransformStream");
|
