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
|
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
import { DataType } from '../../type';
import { Vector } from '../../vector';
import { VectorType as V } from '../../interfaces';
import { Builder, BuilderOptions } from '../../builder/index';
/** @ignore */
export interface BuilderTransformOptions<T extends DataType = any, TNull = any> extends BuilderOptions<T, TNull> {
queueingStrategy?: 'bytes' | 'count';
dictionaryHashFunction?: (value: any) => string | number;
readableStrategy?: { highWaterMark?: number; size?: any; type?: 'bytes' };
writableStrategy?: { highWaterMark?: number; size?: any; type?: 'bytes' };
valueToChildTypeId?: (builder: Builder<T, TNull>, value: any, offset: number) => number;
}
/** @ignore */
export function builderThroughDOMStream<T extends DataType = any, TNull = any>(options: BuilderTransformOptions<T, TNull>) {
return new BuilderTransform(options);
}
/** @ignore */
export class BuilderTransform<T extends DataType = any, TNull = any> {
public readable: ReadableStream<V<T>>;
public writable: WritableStream<T['TValue'] | TNull>;
public _controller: ReadableStreamDefaultController<V<T>> | null;
private _numChunks = 0;
private _finished = false;
private _bufferedSize = 0;
private _builder: Builder<T, TNull>;
private _getSize: (builder: Builder<T, TNull>) => number;
constructor(options: BuilderTransformOptions<T, TNull>) {
// Access properties by string indexers to defeat closure compiler
const {
['readableStrategy']: readableStrategy,
['writableStrategy']: writableStrategy,
['queueingStrategy']: queueingStrategy = 'count',
...builderOptions
} = options;
this._controller = null;
this._builder = Builder.new<T, TNull>(builderOptions);
this._getSize = queueingStrategy !== 'bytes' ? chunkLength : chunkByteLength;
const { ['highWaterMark']: readableHighWaterMark = queueingStrategy === 'bytes' ? 2 ** 14 : 1000 } = { ...readableStrategy };
const { ['highWaterMark']: writableHighWaterMark = queueingStrategy === 'bytes' ? 2 ** 14 : 1000 } = { ...writableStrategy };
this['readable'] = new ReadableStream<V<T>>({
['cancel']: () => { this._builder.clear(); },
['pull']: (c) => { this._maybeFlush(this._builder, this._controller = c); },
['start']: (c) => { this._maybeFlush(this._builder, this._controller = c); },
}, {
'highWaterMark': readableHighWaterMark,
'size': queueingStrategy !== 'bytes' ? chunkLength : chunkByteLength,
});
this['writable'] = new WritableStream({
['abort']: () => { this._builder.clear(); },
['write']: () => { this._maybeFlush(this._builder, this._controller); },
['close']: () => { this._maybeFlush(this._builder.finish(), this._controller); },
}, {
'highWaterMark': writableHighWaterMark,
'size': (value: T['TValue'] | TNull) => this._writeValueAndReturnChunkSize(value),
});
}
private _writeValueAndReturnChunkSize(value: T['TValue'] | TNull) {
const bufferedSize = this._bufferedSize;
this._bufferedSize = this._getSize(this._builder.append(value));
return this._bufferedSize - bufferedSize;
}
private _maybeFlush(builder: Builder<T, TNull>, controller: ReadableStreamDefaultController<V<T>> | null) {
if (controller === null) { return; }
if (this._bufferedSize >= controller.desiredSize!) {
++this._numChunks && this._enqueue(controller, builder.toVector());
}
if (builder.finished) {
if (builder.length > 0 || this._numChunks === 0) {
++this._numChunks && this._enqueue(controller, builder.toVector());
}
if (!this._finished && (this._finished = true)) {
this._enqueue(controller, null);
}
}
}
private _enqueue(controller: ReadableStreamDefaultController<V<T>>, chunk: V<T> | null) {
this._bufferedSize = 0;
this._controller = null;
chunk === null ? controller.close() : controller.enqueue(chunk);
}
}
/** @ignore */ const chunkLength = <T extends DataType = any>(chunk: Vector<T> | Builder<T>) => chunk.length;
/** @ignore */ const chunkByteLength = <T extends DataType = any>(chunk: Vector<T> | Builder<T>) => chunk.byteLength;
|
