import config from '../../../../config/AgentConfig' ; import * as grpc from '@grpc/grpc-js' ; import { connectivityState } from '@grpc/grpc-js' ; import { createLogger } from '../../../../logging' ; import Client from './Client' ; import { TraceSegmentReportServiceClient } from '../../../../proto/language-agent/Tracing_grpc_pb' ; import AuthInterceptor from '../AuthInterceptor' ; import SegmentObjectAdapter from '../SegmentObjectAdapter' ; import { emitter } from '../../../../lib/EventEmitter' ; import Segment from '../../../../trace/context/Segment' ; const logger = createLogger ( __filename ) ; export default class TraceReportClient implements Client { private readonly reporterClient : TraceSegmentReportServiceClient ; private readonly buffer : Segment [ ] = [ ] ; private timeout ? : NodeJS . Timeout ; constructor ( ) { this . reporterClient = new TraceSegmentReportServiceClient ( config . collectorAddress , config . secure ? grpc . credentials . createSsl ( ) : grpc . credentials . createInsecure ( ) ) ; emitter . on ( 'segment-finished' , ( segment ) => { this . buffer . push ( segment ) ; this . timeout ?. ref ( ) ; } ) ; } get isConnected ( ) : boolean { return this . reporterClient ?. getChannel ( ) . getConnectivityState ( true ) === connectivityState . READY ; } start ( ) { const reportFunction = ( ) => { emitter . emit ( 'segments-sent' ) ; try { if ( this . buffer . length === 0 ) { return ; } const stream = this . reporterClient . collect ( AuthInterceptor ( ) , ( error , _ ) => { if ( error ) { logger . error ( 'Failed to report trace data' , error ) ; } } ) ; try { for ( const segment of this . buffer ) { if ( segment ) { if ( logger . _isDebugEnabled ) { logger . debug ( 'Sending segment ' , { segment } ) ; } stream . write ( new SegmentObjectAdapter ( segment ) ) ; } } } finally { this . buffer . length = 0 ; } stream . end ( ) ; } finally { this . timeout = setTimeout ( reportFunction , 1000 ) . unref ( ) ; } } ; this . timeout = setTimeout ( reportFunction , 1000 ) . unref ( ) ; } }