1 // Licensed to the Apache Software Foundation (ASF) under one
2 // or more contributor license agreements. See the NOTICE file
3 // distributed with this work for additional information
4 // regarding copyright ownership. The ASF licenses this file
5 // to you under the Apache License, Version 2.0 (the
6 // "License"); you may not use this file except in compliance
7 // with the License. You may obtain a copy of the License at
9 // http://www.apache.org/licenses/LICENSE-2.0
11 // Unless required by applicable law or agreed to in writing,
12 // software distributed under the License is distributed on an
13 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 // KIND, either express or implied. See the License for the
15 // specific language governing permissions and limitations
18 import { DataType } from '../../type';
19 import { Vector } from '../../vector';
20 import { VectorType as V } from '../../interfaces';
21 import { Builder, BuilderOptions } from '../../builder/index';
24 export interface BuilderTransformOptions<T extends DataType = any, TNull = any> extends BuilderOptions<T, TNull> {
25 queueingStrategy?: 'bytes' | 'count';
26 dictionaryHashFunction?: (value: any) => string | number;
27 readableStrategy?: { highWaterMark?: number; size?: any; type?: 'bytes' };
28 writableStrategy?: { highWaterMark?: number; size?: any; type?: 'bytes' };
29 valueToChildTypeId?: (builder: Builder<T, TNull>, value: any, offset: number) => number;
33 export function builderThroughDOMStream<T extends DataType = any, TNull = any>(options: BuilderTransformOptions<T, TNull>) {
34 return new BuilderTransform(options);
38 export class BuilderTransform<T extends DataType = any, TNull = any> {
40 public readable: ReadableStream<V<T>>;
41 public writable: WritableStream<T['TValue'] | TNull>;
42 public _controller: ReadableStreamDefaultController<V<T>> | null;
44 private _numChunks = 0;
45 private _finished = false;
46 private _bufferedSize = 0;
47 private _builder: Builder<T, TNull>;
48 private _getSize: (builder: Builder<T, TNull>) => number;
50 constructor(options: BuilderTransformOptions<T, TNull>) {
52 // Access properties by string indexers to defeat closure compiler
55 ['readableStrategy']: readableStrategy,
56 ['writableStrategy']: writableStrategy,
57 ['queueingStrategy']: queueingStrategy = 'count',
61 this._controller = null;
62 this._builder = Builder.new<T, TNull>(builderOptions);
63 this._getSize = queueingStrategy !== 'bytes' ? chunkLength : chunkByteLength;
65 const { ['highWaterMark']: readableHighWaterMark = queueingStrategy === 'bytes' ? 2 ** 14 : 1000 } = { ...readableStrategy };
66 const { ['highWaterMark']: writableHighWaterMark = queueingStrategy === 'bytes' ? 2 ** 14 : 1000 } = { ...writableStrategy };
68 this['readable'] = new ReadableStream<V<T>>({
69 ['cancel']: () => { this._builder.clear(); },
70 ['pull']: (c) => { this._maybeFlush(this._builder, this._controller = c); },
71 ['start']: (c) => { this._maybeFlush(this._builder, this._controller = c); },
73 'highWaterMark': readableHighWaterMark,
74 'size': queueingStrategy !== 'bytes' ? chunkLength : chunkByteLength,
77 this['writable'] = new WritableStream({
78 ['abort']: () => { this._builder.clear(); },
79 ['write']: () => { this._maybeFlush(this._builder, this._controller); },
80 ['close']: () => { this._maybeFlush(this._builder.finish(), this._controller); },
82 'highWaterMark': writableHighWaterMark,
83 'size': (value: T['TValue'] | TNull) => this._writeValueAndReturnChunkSize(value),
87 private _writeValueAndReturnChunkSize(value: T['TValue'] | TNull) {
88 const bufferedSize = this._bufferedSize;
89 this._bufferedSize = this._getSize(this._builder.append(value));
90 return this._bufferedSize - bufferedSize;
93 private _maybeFlush(builder: Builder<T, TNull>, controller: ReadableStreamDefaultController<V<T>> | null) {
94 if (controller === null) { return; }
95 if (this._bufferedSize >= controller.desiredSize!) {
96 ++this._numChunks && this._enqueue(controller, builder.toVector());
98 if (builder.finished) {
99 if (builder.length > 0 || this._numChunks === 0) {
100 ++this._numChunks && this._enqueue(controller, builder.toVector());
102 if (!this._finished && (this._finished = true)) {
103 this._enqueue(controller, null);
108 private _enqueue(controller: ReadableStreamDefaultController<V<T>>, chunk: V<T> | null) {
109 this._bufferedSize = 0;
110 this._controller = null;
111 chunk === null ? controller.close() : controller.enqueue(chunk);
115 /** @ignore */ const chunkLength = <T extends DataType = any>(chunk: Vector<T> | Builder<T>) => chunk.length;
116 /** @ignore */ const chunkByteLength = <T extends DataType = any>(chunk: Vector<T> | Builder<T>) => chunk.byteLength;