import Router from './router' ; import { IEndpoint , Msg } from './types' ; import { Buffer } from "buffer" ; import setAsap from 'setasap' ; type PendingMsg = [ IEndpoint , Buffer [ ] ] ; export default class Rep extends Router { private static bottom = Buffer . alloc ( 0 ) ; sendingReply : boolean ; ids : Buffer [ ] ; pending : PendingMsg [ ] ; constructor ( ) { super ( ) ; this . sendingReply = false ; this . ids = [ ] ; this . pending = [ ] ; } protected xsend ( msg ) { if ( ! this . sendingReply ) throw new Error ( "cannot send another reply" ) ; const withIds = [ ... this . ids , Rep . bottom , ... msg ] ; super . xsend ( withIds ) ; this . ids = [ ] ; const nextMsg = this . pending . shift ( ) ; if ( nextMsg ) setAsap ( ( ) => this . recvInternal ( nextMsg [ 0 ] , nextMsg [ 1 ] ) ) ; else this . sendingReply = false ; } private recvInternal ( endpoint , frames ) { while ( true ) { const frame = frames . shift ( ) ; if ( ! frame ) { this . ids = [ ] ; const nextMsg = this . pending . shift ( ) ; if ( nextMsg ) this . recvInternal ( nextMsg [ 0 ] , nextMsg [ 1 ] ) ; return ; } if ( frame . length === 0 ) { this . sendingReply = true ; this . emit ( 'message' , ... frames ) ; return ; } this . ids . push ( frame ) ; } } protected xxrecv ( endpoint , ... frames ) { if ( this . sendingReply ) this . pending . push ( [ endpoint , frames ] ) ; else this . recvInternal ( endpoint , frames ) ; } }