ceph/src/arrow/js/src/io/stream.ts

   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
   8 //
   9 //   http://www.apache.org/licenses/LICENSE-2.0
  10 //
  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
  16 // under the License.
  17
  18 import streamAdapters from './adapters';
  19 import { decodeUtf8 } from '../util/utf8';
  20 import { ITERATOR_DONE, Readable, Writable, AsyncQueue } from './interfaces';
  21 import { toUint8Array, joinUint8Arrays, ArrayBufferViewInput } from '../util/buffer';
  22
  23 import {
  24     isPromise, isFetchResponse,
  25     isIterable, isAsyncIterable,
  26     isReadableDOMStream, isReadableNodeStream
  27 } from '../util/compat';
  28
  29 /** @ignore */
  30 export type WritableSink<T> = Writable<T> | WritableStream<T> | NodeJS.WritableStream | null;
  31 /** @ignore */
  32 export type ReadableSource<T> = Readable<T> | PromiseLike<T> | AsyncIterable<T> | ReadableStream<T> | NodeJS.ReadableStream | null;
  33
  34 /** @ignore */
  35 export class AsyncByteQueue<T extends ArrayBufferViewInput = Uint8Array> extends AsyncQueue<Uint8Array, T> {
  36     public write(value: ArrayBufferViewInput | Uint8Array) {
  37         if ((value = toUint8Array(value)).byteLength > 0) {
  38             return super.write(value as T);
  39         }
  40     }
  41     public toString(sync: true): string;
  42     public toString(sync?: false): Promise<string>;
  43     public toString(sync = false) {
  44         return sync
  45             ? decodeUtf8(this.toUint8Array(true))
  46             : this.toUint8Array(false).then(decodeUtf8);
  47     }
  48     public toUint8Array(sync: true): Uint8Array;
  49     public toUint8Array(sync?: false): Promise<Uint8Array>;
  50     public toUint8Array(sync = false) {
  51         return sync ? joinUint8Arrays(this._values as any[])[0] : (async () => {
  52             const buffers = [];
  53             let byteLength = 0;
  54             for await (const chunk of this) {
  55                 buffers.push(chunk);
  56                 byteLength += chunk.byteLength;
  57             }
  58             return joinUint8Arrays(buffers, byteLength)[0];
  59         })();
  60     }
  61 }
  62
  63 /** @ignore */
  64 export class ByteStream implements IterableIterator<Uint8Array> {
  65     private source!: ByteStreamSource<Uint8Array>;
  66     constructor(source?: Iterable<ArrayBufferViewInput> | ArrayBufferViewInput) {
  67         if (source) {
  68             this.source = new ByteStreamSource(streamAdapters.fromIterable(source));
  69         }
  70     }
  71     [Symbol.iterator]() { return this; }
  72     public next(value?: any) { return this.source.next(value); }
  73     public throw(value?: any) { return this.source.throw(value); }
  74     public return(value?: any) { return this.source.return(value); }
  75     public peek(size?: number | null) { return this.source.peek(size); }
  76     public read(size?: number | null) { return this.source.read(size); }
  77 }
  78
  79 /** @ignore */
  80 export class AsyncByteStream implements Readable<Uint8Array>, AsyncIterableIterator<Uint8Array> {
  81     private source!: AsyncByteStreamSource<Uint8Array>;
  82     constructor(source?: PromiseLike<ArrayBufferViewInput> | Response | ReadableStream<ArrayBufferViewInput> | NodeJS.ReadableStream | AsyncIterable<ArrayBufferViewInput> | Iterable<ArrayBufferViewInput>) {
  83         if (source instanceof AsyncByteStream) {
  84             this.source = (source as AsyncByteStream).source;
  85         } else if (source instanceof AsyncByteQueue) {
  86             this.source = new AsyncByteStreamSource(streamAdapters.fromAsyncIterable(source));
  87         } else if (isReadableNodeStream(source)) {
  88             this.source = new AsyncByteStreamSource(streamAdapters.fromNodeStream(source));
  89         } else if (isReadableDOMStream<ArrayBufferViewInput>(source)) {
  90             this.source = new AsyncByteStreamSource(streamAdapters.fromDOMStream(source));
  91         } else if (isFetchResponse(source)) {
  92             this.source = new AsyncByteStreamSource(streamAdapters.fromDOMStream(source.body!));
  93         } else if (isIterable<ArrayBufferViewInput>(source)) {
  94             this.source = new AsyncByteStreamSource(streamAdapters.fromIterable(source));
  95         } else if (isPromise<ArrayBufferViewInput>(source)) {
  96             this.source = new AsyncByteStreamSource(streamAdapters.fromAsyncIterable(source));
  97         } else if (isAsyncIterable<ArrayBufferViewInput>(source)) {
  98             this.source = new AsyncByteStreamSource(streamAdapters.fromAsyncIterable(source));
  99         }
 100     }
 101     [Symbol.asyncIterator]() { return this; }
 102     public next(value?: any) { return this.source.next(value); }
 103     public throw(value?: any) { return this.source.throw(value); }
 104     public return(value?: any) { return this.source.return(value); }
 105     public get closed(): Promise<void> { return this.source.closed; }
 106     public cancel(reason?: any) { return this.source.cancel(reason); }
 107     public peek(size?: number | null) { return this.source.peek(size); }
 108     public read(size?: number | null) { return this.source.read(size); }
 109 }
 110
 111 /** @ignore */
 112 type ByteStreamSourceIterator<T> = Generator<T, null, { cmd: 'peek' | 'read'; size?: number | null }>;
 113 /** @ignore */
 114 type AsyncByteStreamSourceIterator<T> = AsyncGenerator<T, null, { cmd: 'peek' | 'read'; size?: number | null }>;
 115
 116 /** @ignore */
 117 class ByteStreamSource<T> {
 118     constructor(protected source: ByteStreamSourceIterator<T>) {}
 119     public cancel(reason?: any) { this.return(reason); }
 120     public peek(size?: number | null): T | null { return this.next(size, 'peek').value; }
 121     public read(size?: number | null): T | null { return this.next(size, 'read').value; }
 122     public next(size?: number | null, cmd: 'peek' | 'read' = 'read') { return this.source.next({ cmd, size }); }
 123     public throw(value?: any) { return Object.create((this.source.throw && this.source.throw(value)) || ITERATOR_DONE); }
 124     public return(value?: any) { return Object.create((this.source.return && this.source.return(value)) || ITERATOR_DONE); }
 125 }
 126
 127 /** @ignore */
 128 class AsyncByteStreamSource<T> implements Readable<T> {
 129
 130     private _closedPromise: Promise<void>;
 131     private _closedPromiseResolve?: (value?: any) => void;
 132     constructor (protected source: ByteStreamSourceIterator<T> | AsyncByteStreamSourceIterator<T>) {
 133         this._closedPromise = new Promise((r) => this._closedPromiseResolve = r);
 134     }
 135     public async cancel(reason?: any) { await this.return(reason); }
 136     public get closed(): Promise<void> { return this._closedPromise; }
 137     public async read(size?: number | null): Promise<T | null> { return (await this.next(size, 'read')).value; }
 138     public async peek(size?: number | null): Promise<T | null> { return (await this.next(size, 'peek')).value; }
 139     public async next(size?: number | null, cmd: 'peek' | 'read' = 'read') { return (await this.source.next({ cmd, size })); }
 140     public async throw(value?: any) {
 141         const result = (this.source.throw && await this.source.throw(value)) || ITERATOR_DONE;
 142         this._closedPromiseResolve && this._closedPromiseResolve();
 143         this._closedPromiseResolve = undefined;
 144         return Object.create(result);
 145     }
 146     public async return(value?: any) {
 147         const result = (this.source.return && await this.source.return(value)) || ITERATOR_DONE;
 148         this._closedPromiseResolve && this._closedPromiseResolve();
 149         this._closedPromiseResolve = undefined;
 150         return Object.create(result);
 151     }
 152 }