TracedClientStreamingCallable extends ClientStreamingCallable<RequestT, ResponseT> { @Override public ApiStreamObserver<RequestT> clientStreamingCall( ApiStreamObserver<ResponseT> responseObserver, ApiCallContext context) { ApiTracer tracer = tracerFactory.newTracer(context.getTracer(), spanName, OperationType.ClientStreaming); context = context.withTracer(tracer); AtomicReference<Throwable> cancellationCauseHolder = new AtomicReference<>(null); try { ApiStreamObserver<ResponseT> innerResponseObserver = new TracedResponseObserver<>(tracer, responseObserver, cancellationCauseHolder); ApiStreamObserver<RequestT> innerRequestObserver = innerCallable.clientStreamingCall(innerResponseObserver, context); return new TracedRequestObserver<>(tracer, innerRequestObserver, cancellationCauseHolder); } catch (RuntimeException e) { tracer.operationFailed(e); throw e; } } TracedClientStreamingCallable(
      @Nonnull ClientStreamingCallable<RequestT, ResponseT> innerCallable,
      @Nonnull ApiTracerFactory tracerFactory,
      @Nonnull SpanName spanName); @Override ApiStreamObserver<RequestT> clientStreamingCall(
      ApiStreamObserver<ResponseT> responseObserver, ApiCallContext context);  }
@Test public void testTracerCreated() { tracedCallable.clientStreamingCall(outerResponseObsever, callContext); verify(tracerFactory, times(1)) .newTracer(parentTracer, SPAN_NAME, OperationType.ClientStreaming); }
@Test public void testCallContextPropagated() { ImmutableMap<String, List<String>> extraHeaders = ImmutableMap.<String, List<String>>of("header1", ImmutableList.of("value1")); ApiCallContext newCallContext = callContext.withExtraHeaders(extraHeaders); tracedCallable.clientStreamingCall(outerResponseObsever, newCallContext); assertThat(innerCallable.callContext.getExtraHeaders()).isEqualTo(extraHeaders); }
@Test public void testOperationCancelled() { ApiStreamObserver<String> clientStream = tracedCallable.clientStreamingCall(outerResponseObsever, callContext); clientStream.onError(new CancellationException("explicitly cancelled")); innerCallable.responseObserver.onError( new CancelledException( "fake exception that would be generated by a client cancelling the rpc", null, FakeStatusCode.of(Code.CANCELLED), false)); verify(tracer, times(1)).operationCancelled(); }
@Test public void testOperationFinished() { tracedCallable.clientStreamingCall(outerResponseObsever, callContext); innerCallable.responseObserver.onNext("ignored"); innerCallable.responseObserver.onCompleted(); verify(tracer, times(1)).operationSucceeded(); }
@Test public void testOperationFailed() { RuntimeException expectedError = new RuntimeException("fake error"); tracedCallable.clientStreamingCall(outerResponseObsever, callContext); innerCallable.responseObserver.onError(expectedError); verify(tracer, times(1)).operationFailed(expectedError); }
@Test public void testSyncError() { RuntimeException expectedError = new RuntimeException("fake error"); innerCallable.syncError = expectedError; try { tracedCallable.clientStreamingCall(outerResponseObsever, callContext); } catch (RuntimeException e) { } verify(tracer, times(1)).operationFailed(expectedError); }
@Test public void testRequestNotify() { ApiStreamObserver<String> requestStream = tracedCallable.clientStreamingCall(outerResponseObsever, callContext); requestStream.onNext("request1"); requestStream.onNext("request2"); innerCallable.responseObserver.onNext("response"); innerCallable.responseObserver.onCompleted(); verify(tracer, times(2)).requestSent(); assertThat(outerResponseObsever.completed).isTrue(); assertThat(innerCallable.requestObserver.messages).containsExactly("request1", "request2"); }