module AliMNS { export class NotifyRecv implements INotifyRecvBatch { public constructor ( mq ) { this . _mq = mq ; this . _emitter = new Events . EventEmitter ( ) ; if ( mq instanceof MQ ) { var account = mq . getAccount ( ) ; this . _ga = new GA ( account . getAccountId ( ) ) ; this . _ga . disableGA ( ! account . getGA ( ) ) ; } } public notifyRecv ( cb , waitSeconds ? , numOfMessages ? ) { this . _signalSTOP = false ; this . _timeoutCount = 0 ; this . notifyRecvInternal ( cb , waitSeconds , numOfMessages ) ; if ( this . _ga ) this . _ga . send ( "NotifyRecv.notifyRecv" , 0 , "" ) ; } public notifyStopP ( ) { if ( this . _signalSTOP ) return Promise . resolve ( this . _evStopped ) ; if ( this . _ga ) this . _ga . send ( "NotifyRecv.notifyStopP" , 0 , "" ) ; this . _signalSTOP = true ; return new Promise ( ( resolve ) => { this . _emitter . once ( this . _evStopped , ( ) => { resolve ( this . _evStopped ) ; } ) ; } ) ; } private notifyRecvInternal ( cb , waitSeconds , numOfMessages ? ) { if ( this . _signalSTOP ) { debug ( "notifyStopped" ) ; this . _emitter . emit ( this . _evStopped ) ; return ; } debug ( "notifyRecvInternal()" ) ; try { var mqBatch = this . _mq ; mqBatch . recvP ( waitSeconds , numOfMessages ) . done ( ( dataRecv ) => { try { debug ( dataRecv ) ; this . _timeoutCount = 0 ; if ( cb ( null , dataRecv ) ) { this . deleteP ( dataRecv ) . done ( null , ( ex ) => { console . log ( ex ) ; } ) ; } } catch ( ex ) { console . warn ( ex ) ; } this . notifyRecvInternal ( cb , waitSeconds , numOfMessages ) ; } , ( ex ) => { debug ( ex ) ; if ( ( ! ex . Error ) || ( ex . Error . Code !== "MessageNotExist" ) ) { cb ( ex , null ) ; } if ( ex ) { if ( ex . message === "timeout" ) { this . _timeoutCount ++ ; if ( this . _timeoutCount > this . _timeoutMax ) { cb ( new Error ( "NetworkBroken" ) , null ) ; } } else if ( ex . Error && ex . Error . Code === "MessageNotExist" ) { this . _timeoutCount = 0 ; } } process . nextTick ( ( ) => { this . notifyRecvInternal ( cb , waitSeconds , numOfMessages ) ; } ) ; } ) ; } catch ( ex ) { console . warn ( ex ) ; debug ( "Retry after 5 seconds" ) ; setTimeout ( ( ) => { this . notifyRecvInternal ( cb , waitSeconds , numOfMessages ) ; } , 5000 ) ; } } private deleteP ( dataRecv ) { if ( dataRecv ) { if ( dataRecv . Message ) { return this . _mq . deleteP ( dataRecv . Message . ReceiptHandle ) ; } else if ( dataRecv . Messages && dataRecv . Messages . Message ) { var rhs = [ ] ; for ( var i = 0 ; i < dataRecv . Messages . Message . length ; i ++ ) { rhs . push ( dataRecv . Messages . Message [ i ] . ReceiptHandle ) ; } var mqBatch = this . _mq ; return mqBatch . deleteP ( rhs ) ; } else { return Promise . resolve ( dataRecv ) ; } } else { return Promise . resolve ( dataRecv ) ; } } private _mq : IMQ ; private _signalSTOP = true ; private _evStopped = "AliMNS_MQ_NOTIFY_STOPPED" ; private _emitter : any ; private _timeoutCount = 0 ; private _timeoutMax = 128 ; private _ga : GA = null ; } }