import msgpack , { Codec } from 'msgpack-lite' ; import Buffered from './buffered' ; import { Metadata } from '../meta' ; import { EventEmitter } from 'events' ; const logger = require ( '../logger' ) ( 'client' ) ; class Response { private requestId : number ; private sent : boolean ; private encoder : NodeJS . WritableStream ; constructor ( encoder , requestId , private codec ) { this . encoder = encoder ; this . requestId = requestId ; } public send ( resp , isError ? ) { if ( this . sent ) { throw new Error ( `Response to id ${ this . requestId } already sent` ) ; } this . encoder . write ( msgpack . encode ( [ 1 , this . requestId , isError ? [ 0 , ( resp instanceof Error ) ? resp . message : resp ] : null , ! isError ? resp : null , ] , { codec : this . codec } ) ) ; this . sent = true ; } } interface ClientInfo { name : string ; version ? : object ; } export default class Client extends EventEmitter { private pending : Map < number , Function > = new Map ( ) ; private nextRequestId = 1 ; private encodeStream : any ; private decodeStream : any ; private reader : NodeJS . ReadableStream ; private writer : NodeJS . WritableStream ; private codec : Codec ; private clientInfo : ClientInfo ; constructor ( public readonly id ) { super ( ) ; const codec = this . setupCodec ( ) ; this . encodeStream = msgpack . createEncodeStream ( { codec } ) ; this . decodeStream = msgpack . createDecodeStream ( { codec } ) ; this . decodeStream . on ( 'data' , ( msg ) => { this . parseMessage ( msg ) ; } ) ; this . decodeStream . on ( 'end' , ( ) => { this . detach ( ) ; this . emit ( 'detach' ) ; } ) ; } public setClientInfo ( info ) { this . clientInfo = info ; } public get name ( ) : string { return this . clientInfo ? this . clientInfo . name : '' ; } private setupCodec ( ) { const codec = msgpack . createCodec ( ) ; Metadata . forEach ( ( { constructor } , id ) : void => { codec . addExtPacker ( id , constructor , ( obj ) => msgpack . encode ( obj . id ) ) ; codec . addExtUnpacker ( id , data => new constructor ( msgpack . decode ( data ) ) ) ; } ) ; this . codec = codec ; return this . codec ; } public attach ( writer , reader ) { this . encodeStream = this . encodeStream . pipe ( writer ) ; const buffered = new Buffered ( ) ; reader . pipe ( buffered ) . pipe ( this . decodeStream ) ; this . writer = writer ; this . reader = reader ; } public detach ( ) { this . encodeStream . unpipe ( this . writer ) ; this . reader . unpipe ( this . decodeStream ) ; } public request ( method , args ) { this . nextRequestId = this . nextRequestId + 1 ; this . encodeStream . write ( msgpack . encode ( [ 0 , this . nextRequestId , method , args ] , { codec : this . codec } ) ) ; return new Promise ( ( resolve , reject ) => { let resolved = false ; setTimeout ( ( ) => { if ( resolved ) return ; reject ( new Error ( `request "${ method }" timeout after 3000` ) ) ; } , 3000 ) ; this . pending . set ( this . nextRequestId , ( err , result ) => { resolved = true ; if ( err ) return reject ( err ) ; resolve ( result ) ; } ) ; } ) ; } public notify ( method , args ) { this . encodeStream . write ( msgpack . encode ( [ 2 , method , args ] , { codec : this . codec } ) ) ; } private parseMessage ( msg ) { const msgType = msg [ 0 ] ; if ( msgType === 0 ) { this . emit ( 'request' , msg [ 2 ] . toString ( ) , msg [ 3 ] , new Response ( this . encodeStream , msg [ 1 ] , this . codec ) ) ; } else if ( msgType === 1 ) { const id = msg [ 1 ] ; const handler = this . pending . get ( id ) ; this . pending . delete ( id ) ; handler ( msg [ 2 ] , msg [ 3 ] ) ; } else if ( msgType === 2 ) { this . emit ( 'notification' , msg [ 1 ] . toString ( ) , msg [ 2 ] || [ ] ) ; } else { this . encodeStream . write ( [ 1 , 0 , 'Invalid message type' , null ] ) ; } } }