import { IGBCallInfo , IGBCreateCallResult , IGBCallEvent , IGBCallEnded , IGBClientMessage , } from '../proto' ; import { Subject } from 'rxjs/Subject' ; export interface ICallHandle { write ? ( msg ) : void ; on ? ( eventId , callback ) : void ; off ? ( eventId ) : void ; end ? ( ) : void ; terminate ? ( ) : void ; } export class Call implements ICallHandle { public disposed : Subject < Call > = new Subject < Call > ( ) ; private eventHandlers : { [ id ] : ( ( arg : any ) => void ) [ ] ; } = { } ; private endEmitted : boolean = false ; private responseBuilder : any ; private requestBuilder : any ; constructor ( public clientId , public clientServiceId , private info , private callMeta , private callback , private send ) { this . requestBuilder = callMeta . resolvedRequestType . build ( ) ; this . responseBuilder = callMeta . resolvedResponseType . build ( ) ; } public on ( eventId , callback ) { let handlers = this . eventHandlers [ eventId ] ; if ( ! handlers ) { handlers = [ ] ; this . eventHandlers [ eventId ] = handlers ; } handlers . push ( callback ) ; } public off ( eventId ) { delete this . eventHandlers [ eventId ] ; } public handleCreateResponse ( msg ) { if ( msg . result === 0 ) { return ; } if ( msg . error_details && msg . error_details . length ) { this . terminateWithError ( msg . error_details ) ; } else { this . terminateWithError ( 'Error ' + msg . result ) ; } } public handleEnded ( msg ) { this . dispose ( ) ; } public handleEvent ( msg ) { let data = null ; if ( msg . json_data && msg . json_data . length ) { data = JSON . parse ( msg . json_data ) ; } else if ( msg . bin_data && ( msg . bin_data . limit || msg . bin_data . length ) ) { data = this . decodeResponseData ( msg . bin_data ) ; } this . emit ( msg . event , data ) ; if ( this . callback ) { if ( msg . event === 'error' ) { this . terminateWithError ( data ) ; } else if ( msg . event === 'data' ) { this . terminateWithData ( data ) ; } } } public end ( ) { this . send ( { call_send : { call_id : this . clientId , service_id : this . clientServiceId , is_end : true , } , } ) ; } public terminate ( ) { this . send ( { call_end : { call_id : this . clientId , service_id : this . clientServiceId , } , } ) ; this . dispose ( ) ; } public write ( msg ) { if ( ! this . callMeta . requestStream ) { throw new Error ( 'Cannot write to a non-streaming request.' ) ; } if ( typeof msg !== 'object' ) { throw new Error ( 'Can only write objects to streaming requests.' ) ; } this . send ( { call_send : { call_id : this . clientId , service_id : this . clientServiceId , bin_data : this . requestBuilder . encode ( msg ) , } , } ) ; } public dispose ( ) { if ( ! this . endEmitted ) { this . emit ( 'end' , null ) ; } this . disposed . next ( this ) ; } private decodeResponseData ( data ) { return this . responseBuilder . decode ( data ) ; } private terminateWithError ( error ) { if ( this . callback ) { this . callback ( error , null ) ; } else { this . emit ( 'error' , error ) ; } this . dispose ( ) ; } private terminateWithData ( data ) { this . callback ( null , data ) ; this . dispose ( ) ; } private emit ( eventId , arg ) { if ( eventId === 'end' ) { this . endEmitted = true ; } let handlers = this . eventHandlers [ eventId ] ; if ( ! handlers ) { return ; } for ( let handler of handlers ) { handler ( arg ) ; } } }