import Deferred from '../Deferred' ; import LinkedQueue from '../LinkedQueue' ; import { PushAdapter , doneResult } from '../common' ; import fromDom from '../fromDom' ; import fromEmitter from '../fromEmitter' ; type Result < A > = IteratorResult < A > ; /**
 * Async iterable iterator with a non-optional [[return]] method.
 */ interface WrappedBalancer < A > extends AsyncIterableIterator < A > { return ( value ? ) : Promise < Result < A > > ; throw ? : undefined ; } interface Unpushed < A > { result : Result < A > ; defer : Deferred < Result < A > > ; } /**
 * Balances a push queue with a pull queue.
 */ export default class Channel < A > implements PushAdapter < A > { /** Pushed results waiting for pulls to resolve */ readonly pushBuffer : LinkedQueue < Unpushed < A > > ; /** Unresolved pulls waiting for results to be pushed */ readonly pullBuffer : LinkedQueue < Deferred < Result < A > > > ; /** Determines whether new values can be pushed or pulled */ private closed = false ; static fromDom = fromDom ( ( ) => new Channel ( ) ) ; static fromEmitter = fromEmitter ( ( ) => new Channel ( ) ) ; constructor ( pushLimit = 0 , pullLimit = 0 ) { this . pushBuffer = new LinkedQueue ( pushLimit ) ; this . pullBuffer = new LinkedQueue ( pullLimit ) ; } /**
     * Pull a promise of the next [[Result]].
     */ next ( ) { if ( this . closed ) { return Promise . resolve ( doneResult ) ; } if ( this . pushBuffer . length === 0 ) { const defer = new Deferred < Result < A > > ( ) ; this . pullBuffer . enqueue ( defer ) ; return defer . promise ; } const { result , defer } = this . pushBuffer . dequeue ( ) ; defer . resolve ( result ) ; if ( result . done ) { this . close ( ) ; } return defer . promise ; } param value param done throws push ( value , done = false ) { if ( this . closed ) { return Promise . resolve ( doneResult ) ; } const result = { value , done , } ; if ( this . pullBuffer . length > 0 ) { return this . pullBuffer . dequeue ( ) . resolve ( result ) ; } const defer = new Deferred < Result < A > > ( ) ; this . pushBuffer . enqueue ( { result , defer } ) ; return defer . promise ; } /**
     * Returns itself, since [[Balancer]] already implements the iterator protocol.
     */ [ Symbol . asyncIterator ] ( ) { return this ; } param value async return ( value ? ) { this . close ( ) ; return { done : true , value : value ! , } ; } close ( ) { if ( this . closed ) { return ; } this . closed = true ; this . pushBuffer . forEach ( ( { defer : { resolve } } ) => void resolve ( doneResult ) ) ; this . pushBuffer . clear ( ) ; this . pullBuffer . forEach ( ( { resolve } ) => void resolve ( doneResult ) ) ; this . pullBuffer . clear ( ) ; } param onReturn throws wrap ( onReturn ? ) { if ( this . closed ) { throw Error ( 'Balancer is closed' ) ; } return { [ Symbol . asyncIterator ] ( ) { return this ; } , next : ( ) => this . next ( ) , return : async ( value ? ) => { if ( onReturn ) { onReturn ( ) ; } return this . return ( value ) ; } , } ; } }