/**
 * Promise-based object stream with seamless support for back-pressure and error
 * handling, written in Typescript.
 *
 * Copyright (C) 2015 Martin Poelstra
 * License: MIT
 */ import BaseError from "./BaseError" ; import { filter , map , Transform } from "./Transform" ; import { defer , Deferred , noop , swallowErrors , track , TrackedPromise , } from "./util" ; /**
 * Required methods for both readable and writable parts of a stream.
 */ export interface Common < T > { return result ( ) : Promise < void > ; param reason abort ( reason ? ) : void ; return aborted ( ) : Promise < never > ; return aborted ( aborter ) : void ; } /**
 * Required methods for the readable part of a stream.
 */ export interface Readable < T > extends Common < T > { param reader param ender param aborter return forEach ( reader , ender ? , aborter ? ) : Promise < void > ; } /**
 * Required methods for the writable part of a stream.
 */ export interface Writable < T > extends Common < T > { param value return write ( value ) : Promise < void > ; param error param result return end ( error ? , result ? ) : Promise < void > ; } export interface CommonStream < T > { return isEnding ( ) : boolean ; return isEnded ( ) : boolean ; return isEndingOrEnded ( ) : boolean ; return hasReader ( ) : boolean ; } /**
 * Readable part of a generic Stream, which contains handy helpers such as
 * .map() in addition to the basic requirements of a Readable interface.
 */ export interface ReadableStream < T > extends Readable < T > , CommonStream < T > { param mapper param ender param aborter return map < R > ( mapper , ender ? , aborter ? ) : ReadableStream < R > ; param filterer param ender param aborter return filter ( filterer , ender ? , aborter ? ) : ReadableStream < T > ; param reducer param initial return reduce ( reducer , initial ? ) : Promise < T > ; param reducer param initial return reduce < R > ( reducer , initial ) : Promise < R > ; param reducer param initial return reduce < R > ( reducer , initial ) : Promise < R > ; return toArray ( ) : Promise < T [ ] > ; param writable return pipe < R extends Writable < T > > ( writable ) : R ; param transformer return transform < R > ( transformer ) : ReadableStream < R > ; } /**
 * Writable part of a generic Stream, which contains handy helpers such as
 * .mappedBy() in addition to the basic requirements of a Writable interface.
 */ export interface WritableStream < T > extends Writable < T > , CommonStream < T > { param writer param ender param aborter return writeEach ( writer , ender ? , aborter ? ) : Promise < void > ; mappedBy < X > ( mapper ) : WritableStream < X > ; filterBy ( filterer ) : WritableStream < T > ; } /**
 * Used when writing to an already-ended stream.
 */ export class WriteAfterEndError extends BaseError { constructor ( ) { super ( "WriteAfterEndError" , "stream already ended" ) ; } } /**
 * Used when read callback(s) have already been attached.
 */ export class AlreadyHaveReaderError extends BaseError { constructor ( ) { super ( "AlreadyHaveReaderError" , "stream already has a reader" ) ; } } /**
 * Special internal 'error' value to indicate normal stream end.
 */ const EOF = new Error ( "eof" ) ; /**
 * Special end-of-stream value, optionally signalling an error.
 */ class Eof { public error ? : Error ; public result ? : PromiseLike < void > ; param error param result constructor ( error ? , result ? ) { this . error = error ; this . result = result ; } } /**
 * Written value-promise and a function to resolve the corresponding `write()`
 * call's return promise.
 */ interface WriteItem < T > { /**
     * Resolver `write()`'s returned promise
     */ resolveWrite : ( done ? : void | PromiseLike < void > ) => void ; /**
     * Promise for value passed to `write()`.
     * Either a special Eof value, or a value of type `T`.
     */ value : Eof | TrackedPromise < T > ; } /**
 * Object stream with seamless support for backpressure, ending and error
 * handling.
 */ export class Stream < T > implements ReadableStream < T > , WritableStream < T > { /**
     * Writers waiting for `_reader` to retrieve and process their value.
     */ private _writers : WriteItem < T > [ ] = [ ] ; /**
     * Read handler that is called for every written value, as set by
     * `forEach()`.
     */ private _reader ? : ( value : T ) => void | PromiseLike < void > ; /**
     * End handler that is called when the stream is ended, as set by
     * `forEach()`. Note that `forEach()` installs a default handler if the user
     * did not supply one.
     * Set to 'undefined' when it has been called.
     */ private _ender ? : ( error ? : Error ) => void | PromiseLike < void > ; /**
     * Abort handler that is called when the stream is aborted, as set by
     * `forEach()` (can be undefined).
     * Set to 'undefined' when it has been called.
     */ private _aborter ? : ( error : Error ) => void ; /**
     * When a written value is being processed by the `_reader`, this property
     * is set to a promise that resolves when the reader's returned PromiseLike is
     * resolved (or rejected).
     */ private _readBusy ? : TrackedPromise < void > ; /**
     * Set to an instance of an Eof object, containing optional error and final
     * result of this stream. Set when `end()` is called.
     */ private _ending ? : Eof ; /**
     * Set to an instance of an Eof object, containing optional error and final
     * result of this stream. Set when `_ender` is being called but not finished
     * yet, unset when `_ended` is set.
     */ private _endPending ? : Eof ; /**
     * Set to the error passed to `end()` (or the special value `eof`) when the
     * stream has ended, and the operation was confirmed by the `_ender`.
     */ private _ended ? : Error ; /**
     * Set to a rejected promise when the stream is explicitly `abort()`'ed.
     */ private _abortPromise ? : Promise < void > ; /**
     * Error given in abort() method
     */ private _abortReason ? : Error ; /**
     * Resolved to a rejection when `abort()` is called.
     */ private _abortDeferred : Deferred < never > = defer < never > ( ) ; /**
     * Resolved to the result of calling `_ender`, then the `result` property of
     * the end-of-stream value.
     */ private _resultDeferred : Deferred < void > = defer ( ) ; param value return public write ( value ) { if ( value === undefined ) { return Promise . reject ( new TypeError ( "cannot write void value, use end() to end the stream" ) ) ; } const writeDone = defer ( ) ; this . _writers . push ( { resolveWrite : writeDone . resolve , value : track < T > ( Promise . resolve ( value ) ) , } ) ; this . _pump ( ) ; return writeDone . promise ; } param error param result return public end ( error ? , endedResult ? ) { if ( ! ( error === undefined || error === null || error instanceof Error ) ) { return Promise . reject ( new TypeError ( "invalid argument to end(): must be undefined, null or Error object" ) ) ; } const eof = new Eof ( error , endedResult ) ; if ( ! this . _ending && ! this . _ended ) { this . _ending = eof ; } const writeDone = defer ( ) ; const item = { resolveWrite : writeDone . resolve , value : eof , } ; this . _writers . push ( item ) ; this . _pump ( ) ; return writeDone . promise ; } param reader param ender param aborter return public forEach ( reader , ender ? , aborter ? ) { if ( this . hasReader ( ) ) { return Promise . reject ( new AlreadyHaveReaderError ( ) ) ; } if ( ! ender ) { ender = noop ; } this . _reader = reader ; this . _ender = ender ; this . _aborter = aborter ; this . _pump ( ) ; return this . result ( ) ; } param reason public abort ( reason ? ) { if ( this . _abortPromise ) { return ; } if ( ! reason ) { reason = new Error ( "aborted" ) ; } this . _abortDeferred . reject ( reason ) ; this . _abortPromise = this . _abortDeferred . promise ; this . _abortReason = reason ; this . _pump ( ) ; } return public aborted ( ) { return this . _abortDeferred . promise ; } return public result ( ) { return this . _resultDeferred . promise ; } return public isEnding ( ) { return ! ! this . _ending ; } return public isEnded ( ) { return ! ! this . _ended ; } return public isEndingOrEnded ( ) { return this . isEnding ( ) || this . isEnded ( ) ; } return public hasReader ( ) { return ! ! this . _reader ; } param mapper param ender param aborter return public map < R > ( mapper , ender ? , aborter ? ) { const output = new Stream < R > ( ) ; map ( this , output , mapper , ender , aborter ) ; return output ; } param filterer param ender param aborter return public filter ( filterer , ender ? , aborter ? ) { const output = new Stream < T > ( ) ; filter ( this , output , filterer , ender , aborter ) ; return output ; } param reducer param initial return public reduce ( reducer , initial ? ) ; param reducer param initial return public reduce < R > ( reducer , initial ) ; param reducer param initial return public reduce < R > ( reducer , initial ) { let haveAccumulator = arguments . length === 2 ; let accumulator = initial ; let index = 0 ; return this . forEach ( ( value ) : void | PromiseLike < void > => { if ( ! haveAccumulator ) { accumulator = value ; haveAccumulator = true ; index ++ ; } else { return Promise . resolve ( reducer ( accumulator , value , index ++ , this ) ) . then ( ( newAccumulator ) => ( accumulator = newAccumulator ) ) ; } } ) . then ( ( ) => { if ( ! haveAccumulator ) { return Promise . reject < R > ( new TypeError ( "cannot reduce() empty stream without initial value" ) ) ; } return accumulator ; } ) ; } return public toArray ( ) { const result = [ ] ; return this . forEach ( ( value ) => { result . push ( value ) ; } ) . then ( ( ) => result ) ; } param writable return public pipe < R extends Writable < T > > ( writable ) { writable . aborted ( ) . catch ( ( err ) => this . abort ( err ) ) ; this . aborted ( ) . catch ( ( err ) => writable . abort ( err ) ) ; this . forEach ( ( value ) => writable . write ( value ) , ( error ? ) => writable . end ( error , this . result ( ) ) ) ; return writable ; } param transformer return public transform < R > ( transformer ) { const output = new Stream < R > ( ) ; transformer ( this , output ) ; return output ; } param writer param ender param aborter return public writeEach ( writer , ender ? , aborter ? ) { /**
         * Call aborter (if any) and convert any thrown error into
         * unhandled rejection. Unset aborter to prevent calling it
         * again later.
         */ const callAborter = ( abortReason ) => { if ( ! aborter ) { return ; } try { const callback = aborter ; aborter = undefined ; callback ( abortReason ) ; } catch ( aborterError ) { Promise . reject ( aborterError ) ; } } ; const worker = async ( ) => { try { while ( ! this . _abortPromise ) { const value = await writer ( ) ; if ( value === undefined ) { break ; } else { await this . write ( value as T ) ; } } } catch ( writeError ) { this . abort ( writeError ) ; } finally { let endError = this . _abortReason ; if ( this . _abortReason ) { callAborter ( this . _abortReason ) ; } if ( ender ) { try { await ( this . _abortReason ? ender ( this . _abortReason ) : ender ( ) ) ; } catch ( error ) { endError = error ; } } await this . end ( endError ) ; } } ; Promise . resolve ( ) . then ( worker ) . catch ( ( error ) => this . abort ( error ) ) ; this . aborted ( ) . catch ( callAborter ) ; return this . result ( ) ; } public mappedBy < X > ( mapper ) { const input = new Stream < X > ( ) ; map ( input , this , mapper ) ; return input ; } public filterBy ( filterer ) { const input = new Stream < T > ( ) ; filter ( input , this , filterer ) ; return input ; } see result see writeEach param data return public static from < T > ( data ) ; see result see writeEach param data return public static from < T > ( data ) ; see result see writeEach param data return public static from < T > ( data ) ; see result see writeEach param data return public static from < T > ( data ) ; see result see writeEach param data return public static from < T > ( data ) { const stream = new Stream < T > ( ) ; let i = 0 ; if ( Array . isArray ( data ) ) { stream . writeEach ( ( ) => data [ i ++ ] ) ; } else { Promise . resolve < T [ ] | PromiseLike < T > [ ] > ( data ) . then ( ( resolvedArray ) => { stream . writeEach ( ( ) => resolvedArray [ i ++ ] ) ; } ) ; } return stream ; } /**
     * Bound callback to be passed as handlers to Promise.then()
     */ private _pumper = ( ) => this . _pump ( ) ; /**
     * Pump written values to `_reader` and `_ender`, resolve results of
     * `write()` and `end()`.
     */ private _pump ( ) { if ( this . _abortPromise && this . _aborter ) { swallowErrors ( this . _abortPromise . catch ( this . _aborter ) ) ; this . _aborter = undefined ; } if ( this . _readBusy ) { if ( this . _readBusy . isPending ) { return ; } this . _writers . shift ( ) ! . resolveWrite ( this . _readBusy . promise ) ; if ( this . _endPending ) { const result = this . _endPending . result ; this . _ended = this . _endPending . error || EOF ; this . _ending = undefined ; this . _endPending = undefined ; this . _aborter = undefined ; let p ; if ( result ) { p = this . _readBusy . promise . then ( undefined , noop ) . then ( ( ) => result ) ; } else { p = this . _readBusy . promise . then ( ( ) => { if ( this . _ended !== EOF ) { throw this . _ended ; } } ) ; } this . _resultDeferred . resolve ( p ) ; } this . _readBusy = undefined ; } if ( this . _ended ) { while ( this . _writers . length > 0 ) { const writer = this . _writers . shift ( ) ! ; writer . resolveWrite ( Promise . reject ( new WriteAfterEndError ( ) ) ) ; } return ; } if ( this . _abortPromise ) { while ( this . _writers . length > 0 ) { const writer = this . _writers [ 0 ] ; if ( writer . value instanceof Eof ) { break ; } swallowErrors ( writer . value . promise ) ; writer . resolveWrite ( this . _abortPromise ) ; this . _writers . shift ( ) ; } } if ( this . _writers . length === 0 || ! this . _reader ) { return ; } const writer = this . _writers [ 0 ] ; if ( ! ( writer . value instanceof Eof ) && writer . value . isPending ) { writer . value . promise . then ( this . _pumper , this . _pumper ) ; return ; } if ( ! ( writer . value instanceof Eof ) && writer . value . isRejected ) { writer . resolveWrite ( writer . value . promise as PromiseLike < any > ) ; this . _writers . shift ( ) ; Promise . resolve ( ) . then ( this . _pumper ) ; return ; } if ( writer . value instanceof Eof ) { const eof = writer . value ; if ( this . _ended || this . _endPending ) { throw new Error ( "assertion failed" ) ; } this . _endPending = eof ; const ender = this . _ender ! ; this . _ender = undefined ; const enderArg = this . _abortPromise ? this . _abortReason : eof . error ; this . _readBusy = track ( Promise . resolve ( eof ) . then ( ( eofValue ) => ender ( enderArg ) ) ) ; } else { this . _readBusy = track ( writer . value . promise . then ( this . _reader ) ) ; } this . _readBusy . promise . then ( this . _pumper , this . _pumper ) ; } } export default Stream ;