import { Observable , Operation , NextLink , FetchResult } from 'apollo-link' ; export type BatchHandler = ( operations : Operation [ ] , forward ? : ( NextLink | undefined ) [ ] ) => Observable < FetchResult [ ] > | null ; export interface BatchableRequest { operation : Operation ; forward ? : NextLink ; observable ? : Observable < FetchResult > ; next ? : Array < ( result : FetchResult ) => void > ; error ? : Array < ( error : Error ) => void > ; complete ? : Array < ( ) => void > ; } export class OperationBatcher { public queuedRequests : Map < string , BatchableRequest [ ] > ; private batchInterval : number ; private batchMax : number ; private batchHandler : BatchHandler ; private batchKey : ( operation : Operation ) => string ; constructor ( { batchInterval , batchMax , batchHandler , batchKey , } ) { this . queuedRequests = new Map ( ) ; this . batchInterval = batchInterval ; this . batchMax = batchMax || 0 ; this . batchHandler = batchHandler ; this . batchKey = batchKey || ( ( ) => '' ) ; } public enqueueRequest ( request ) { const requestCopy = { ... request , } ; let queued = false ; const key = this . batchKey ( request . operation ) ; if ( ! requestCopy . observable ) { requestCopy . observable = new Observable < FetchResult > ( observer => { if ( ! this . queuedRequests . has ( key ) ) { this . queuedRequests . set ( key , [ ] ) ; } if ( ! queued ) { this . queuedRequests . get ( key ) . push ( requestCopy ) ; queued = true ; } requestCopy . next = requestCopy . next || [ ] ; if ( observer . next ) requestCopy . next . push ( observer . next . bind ( observer ) ) ; requestCopy . error = requestCopy . error || [ ] ; if ( observer . error ) requestCopy . error . push ( observer . error . bind ( observer ) ) ; requestCopy . complete = requestCopy . complete || [ ] ; if ( observer . complete ) requestCopy . complete . push ( observer . complete . bind ( observer ) ) ; if ( this . queuedRequests . get ( key ) . length === 1 ) { this . scheduleQueueConsumption ( key ) ; } if ( this . queuedRequests . get ( key ) . length === this . batchMax ) { this . consumeQueue ( key ) ; } } ) ; } return requestCopy . observable ; } public consumeQueue ( key ? ) { const requestKey = key || '' ; const queuedRequests = this . queuedRequests . get ( requestKey ) ; if ( ! queuedRequests ) { return ; } this . queuedRequests . delete ( requestKey ) ; const requests = queuedRequests . map ( queuedRequest => queuedRequest . operation ) ; const forwards = queuedRequests . map ( queuedRequest => queuedRequest . forward ) ; const observables = [ ] ; const nexts = [ ] ; const errors = [ ] ; const completes = [ ] ; queuedRequests . forEach ( ( batchableRequest , index ) => { observables . push ( batchableRequest . observable ) ; nexts . push ( batchableRequest . next ) ; errors . push ( batchableRequest . error ) ; completes . push ( batchableRequest . complete ) ; } ) ; const batchedObservable = this . batchHandler ( requests , forwards ) || Observable . of ( ) ; const onError = error => { errors . forEach ( rejecters => { if ( rejecters ) { rejecters . forEach ( e => e ( error ) ) ; } } ) ; } ; batchedObservable . subscribe ( { next : results => { if ( ! Array . isArray ( results ) ) { results = [ results ] ; } if ( nexts . length !== results . length ) { const error = new Error ( `server returned results with length ${ results . length }, expected length of ${ nexts . length }` ) ; ( error as any ) . result = results ; return onError ( error ) ; } results . forEach ( ( result , index ) => { if ( nexts [ index ] ) { nexts [ index ] . forEach ( next => next ( result ) ) ; } } ) ; } , error : onError , complete : ( ) => { completes . forEach ( complete => { if ( complete ) { complete . forEach ( c => c ( ) ) ; } } ) ; } , } ) ; return observables ; } private scheduleQueueConsumption ( key ? ) { const requestKey = key || '' ; setTimeout ( ( ) => { if ( this . queuedRequests . get ( requestKey ) && this . queuedRequests . get ( requestKey ) . length ) { this . consumeQueue ( requestKey ) ; } } , this . batchInterval ) ; } }