TraceSegmentServiceClient implements BootService, IConsumer<TraceSegment>, TracingContextListener, GRPCChannelListener { @Override public void consume(List<TraceSegment> data) { if (CONNECTED.equals(status)) { final GRPCStreamServiceStatus status = new GRPCStreamServiceStatus(false); StreamObserver<SegmentObject> upstreamSegmentStreamObserver = serviceStub.withDeadlineAfter( Config.Collector.GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS ).collect(new StreamObserver<Commands>() { @Override public void onNext(Commands commands) { ServiceManager.INSTANCE.findService(CommandService.class) .receiveCommand(commands); } @Override public void onError( Throwable throwable) { status.finished(); if (LOGGER.isErrorEnable()) { LOGGER.error( throwable, "Send UpstreamSegment to collector fail with a grpc internal exception." ); } ServiceManager.INSTANCE .findService(GRPCChannelManager.class) .reportError(throwable); } @Override public void onCompleted() { status.finished(); } }); try { for (TraceSegment segment : data) { SegmentObject upstreamSegment = segment.transform(); upstreamSegmentStreamObserver.onNext(upstreamSegment); } } catch (Throwable t) { LOGGER.error(t, "Transform and send UpstreamSegment to collector fail."); } upstreamSegmentStreamObserver.onCompleted(); status.wait4Finish(); segmentUplinkedCounter += data.size(); } else { segmentAbandonedCounter += data.size(); } printUplinkStatus(); }  @Override void prepare(); @Override void boot(); @Override void onComplete(); @Override void shutdown(); @Override void init(); @Override void consume(List<TraceSegment> data); @Override void onError(List<TraceSegment> data, Throwable t); @Override void onExit(); @Override void afterFinished(TraceSegment traceSegment); @Override void statusChanged(GRPCChannelStatus status);  }
@Test public void testSendTraceSegmentWithoutException() throws InvalidProtocolBufferException { grpcServerRule.getServiceRegistry().addService(serviceImplBase); AbstractSpan firstEntrySpan = ContextManager.createEntrySpan("/testFirstEntry", null); firstEntrySpan.setComponent(ComponentsDefine.TOMCAT); Tags.HTTP.METHOD.set(firstEntrySpan, "GET"); Tags.URL.set(firstEntrySpan, "127.0.0.1:8080"); SpanLayer.asHttp(firstEntrySpan); ContextManager.stopSpan(); serviceClient.consume(storage.getTraceSegments()); assertThat(upstreamSegments.size(), is(1)); SegmentObject traceSegmentObject = upstreamSegments.get(0); assertThat(traceSegmentObject.getSpans(0).getRefsCount(), is(0)); assertThat(traceSegmentObject.getSpansCount(), is(1)); SpanObject spanObject = traceSegmentObject.getSpans(0); assertThat(spanObject.getSpanType(), is(SpanType.Entry)); assertThat(spanObject.getSpanId(), is(0)); assertThat(spanObject.getParentSpanId(), is(-1)); }
@Test public void testSendTraceSegmentWithException() throws InvalidProtocolBufferException { grpcServerRule.getServiceRegistry().addService(serviceImplBase); AbstractSpan firstEntrySpan = ContextManager.createEntrySpan("/testFirstEntry", null); firstEntrySpan.setComponent(ComponentsDefine.TOMCAT); Tags.HTTP.METHOD.set(firstEntrySpan, "GET"); Tags.URL.set(firstEntrySpan, "127.0.0.1:8080"); SpanLayer.asHttp(firstEntrySpan); ContextManager.stopSpan(); grpcServerRule.getServer().shutdownNow(); serviceClient.consume(storage.getTraceSegments()); assertThat(upstreamSegments.size(), is(0)); boolean reconnect = Whitebox.getInternalState( ServiceManager.INSTANCE.findService(GRPCChannelManager.class), "reconnect"); assertThat(reconnect, is(true)); }