import { Msg , Req } from './nats' ; import { createInbox } from './util' ; import { NatsError } from './error' ; hidden export class MuxSubscriptions { baseInbox ! : string ; reqs : { [ key ] : Req ; } = { } ; length : number = 0 ; init ( ) { this . baseInbox = `${ createInbox ( ) }.` ; return this . baseInbox ; } add ( r ) { if ( ! isNaN ( r . received ) ) { r . received = 0 ; } this . length ++ ; this . reqs [ r . token ] = r ; } get ( token ) { if ( token in this . reqs ) { return this . reqs [ token ] ; } return null ; } all ( ) { let buf = [ ] ; for ( let token in this . reqs ) { let req = this . reqs [ token ] ; buf . push ( req ) ; } return buf ; } cancel ( r ) { if ( r && r . timeout ) { clearTimeout ( r . timeout ) ; delete r . timeout ; } if ( r . token in this . reqs ) { delete this . reqs [ r . token ] ; this . length -- ; } } close ( ) { let reqs = this . all ( ) ; for ( let i = 0 ; i < reqs . length ; i ++ ) { this . cancel ( reqs [ i ] ) ; } } getToken ( m ? ) { let s = '' ; if ( m ) { s = m . subject || '' ; } if ( s . indexOf ( this . baseInbox ) === 0 ) { return s . substring ( this . baseInbox . length ) ; } return null ; } dispatcher ( ) { let mux = this ; return ( error , m ) : void => { let token = mux . getToken ( m ) ; if ( token ) { let r = mux . get ( token ) ; if ( r ) { r . received ++ ; r . callback ( error , m ) ; if ( r . max && r . received >= r . max ) { mux . cancel ( r ) ; } } } } ; } ; }