@reactivex/ix-esnext-esm/asynciterable/asynciterablex.js

import { AsyncSink } from './../asyncsink';
import { bindCallback } from '../internal/bindcallback';
import { identityAsync } from '../internal/identity';
import { toLength } from '../internal/tolength';
import { isArrayLike, isIterable, isAsyncIterable, isReadableNodeStream, isWritableNodeStream } from '../internal/isiterable';
/**
 * This class serves as the base for all operations which support [Symbol.asyncIterator].
 */
export class AsyncIterableX {
    async forEach(projection, thisArg) {
        const fn = bindCallback(projection, thisArg, 2);
        let i = 0;
        for await (let item of this) {
            await fn(item, i++);
        }
    }
    pipe(...args) {
        let i = -1;
        let n = args.length;
        let acc = this;
        let as = AsyncIterableX.as;
        while (++i < n) {
            acc = as(args[i](acc));
        }
        return acc;
    }
    tee() {
        return this._getDOMStream().tee();
    }
    pipeTo(writable, options) {
        return this._getDOMStream().pipeTo(writable, options);
    }
    pipeThrough(duplex, options) {
        return this._getDOMStream().pipeThrough(duplex, options);
    }
    _getDOMStream() {
        return this._DOMStream || (this._DOMStream = this.publish().toDOMStream());
    }
    /** @nocollapse */
    static as(source) {
        /* tslint:disable */
        if (source instanceof AsyncIterableX) {
            return source;
        }
        if (typeof source === 'string') {
            return new OfAsyncIterable([source]);
        }
        if (isIterable(source) || isAsyncIterable(source)) {
            return new FromAsyncIterable(source, identityAsync);
        }
        if (isPromise(source)) {
            return new FromPromiseIterable(source, identityAsync);
        }
        if (isObservable(source)) {
            return new FromObservableAsyncIterable(source, identityAsync);
        }
        if (isArrayLike(source)) {
            return new FromArrayIterable(source, identityAsync);
        }
        return new OfAsyncIterable([source]);
        /* tslint:enable */
    }
    /** @nocollapse */
    static from(source, selector = identityAsync, thisArg) {
        const fn = bindCallback(selector, thisArg, 2);
        /* tslint:disable */
        if (isIterable(source) || isAsyncIterable(source)) {
            return new FromAsyncIterable(source, fn);
        }
        if (isPromise(source)) {
            return new FromPromiseIterable(source, fn);
        }
        if (isObservable(source)) {
            return new FromObservableAsyncIterable(source, fn);
        }
        if (isArrayLike(source)) {
            return new FromArrayIterable(source, fn);
        }
        throw new TypeError('Input type not supported');
        /* tslint:enable */
    }
    /** @nocollapse */
    static of(...args) {
        //tslint:disable-next-line
        return new OfAsyncIterable(args);
    }
}
class FromArrayIterable extends AsyncIterableX {
    constructor(source, selector) {
        super();
        this._source = source;
        this._selector = selector;
    }
    async *[Symbol.asyncIterator]() {
        let i = 0;
        const length = toLength(this._source.length);
        while (i < length) {
            yield await this._selector(this._source[i], i++);
        }
    }
}
class FromAsyncIterable extends AsyncIterableX {
    constructor(source, selector) {
        super();
        this._source = source;
        this._selector = selector;
    }
    async *[Symbol.asyncIterator]() {
        let i = 0;
        for await (let item of this._source) {
            yield await this._selector(item, i++);
        }
    }
}
class FromPromiseIterable extends AsyncIterableX {
    constructor(source, selector) {
        super();
        this._source = source;
        this._selector = selector;
    }
    async *[Symbol.asyncIterator]() {
        const item = await this._source;
        yield await this._selector(item, 0);
    }
}
class FromObservableAsyncIterable extends AsyncIterableX {
    constructor(observable, selector) {
        super();
        this._observable = observable;
        this._selector = selector;
    }
    async *[Symbol.asyncIterator]() {
        const sink = new AsyncSink();
        const subscription = this._observable.subscribe({
            next(value) {
                sink.write(value);
            },
            error(err) {
                sink.error(err);
            },
            complete() {
                sink.end();
            }
        });
        let i = 0;
        try {
            for (let next; !(next = await sink.next()).done;) {
                yield await this._selector(next.value, i++);
            }
        }
        finally {
            subscription.unsubscribe();
        }
    }
}
function isPromise(x) {
    return x != null && Object(x) === x && typeof x['then'] === 'function';
}
function isObservable(x) {
    return x != null && Object(x) === x && typeof x['subscribe'] === 'function';
}
class OfAsyncIterable extends AsyncIterableX {
    constructor(args) {
        super();
        this._args = args;
    }
    async *[Symbol.asyncIterator]() {
        for (let item of this._args) {
            yield item;
        }
    }
}
try {
    (isBrowser => {
        if (isBrowser) {
            return;
        }
        const as = AsyncIterableX.as;
        AsyncIterableX.prototype.pipe = nodePipe;
        const readableOpts = (x, opts = x._writableState || { objectMode: true }) => opts;
        function nodePipe(...args) {
            let i = -1;
            let end;
            let n = args.length;
            let prev = this;
            let next;
            while (++i < n) {
                next = args[i];
                if (typeof next === 'function') {
                    prev = as(next(prev));
                }
                else if (isWritableNodeStream(next)) {
                    ({ end = true } = args[i + 1] || {});
                    // prettier-ignore
                    return isReadableNodeStream(prev) ? prev.pipe(next, { end }) :
                        prev.toNodeStream(readableOpts(next)).pipe(next, { end });
                }
            }
            return prev;
        }
    })(typeof window === 'object' && typeof document === 'object' && document.nodeType === 9);
}
catch (e) {
    /* */
}

//# sourceMappingURL=asynciterablex.mjs.map