TracedServerStreamingCallable extends ServerStreamingCallable<RequestT, ResponseT> { @Override public void call( RequestT request, ResponseObserver<ResponseT> responseObserver, ApiCallContext context) { ApiTracer tracer = tracerFactory.newTracer(context.getTracer(), spanName, OperationType.ServerStreaming); TracedResponseObserver<ResponseT> tracedObserver = new TracedResponseObserver<>(tracer, responseObserver); context = context.withTracer(tracer); try { innerCallable.call(request, tracedObserver, context); } catch (RuntimeException e) { tracedObserver.onError(e); throw e; } } TracedServerStreamingCallable(
      @Nonnull ServerStreamingCallable<RequestT, ResponseT> innerCallable,
      @Nonnull ApiTracerFactory tracerFactory,
      @Nonnull SpanName spanName); @Override void call(
      RequestT request, ResponseObserver<ResponseT> responseObserver, ApiCallContext context);  }
@Test public void testTracerCreated() { tracedCallable.call("test", responseObserver, callContext); verify(tracerFactory, times(1)) .newTracer(parentTracer, SPAN_NAME, OperationType.ServerStreaming); }
@Test public void testResponseNotify() { tracedCallable.call("test", responseObserver, callContext); MockServerStreamingCall<String, String> innerCall = innerCallable.popLastCall(); innerCall.getController().getObserver().onResponse("response1"); innerCall.getController().getObserver().onResponse("response2"); assertThat(responseObserver.popNextResponse()).isEqualTo("response1"); assertThat(responseObserver.popNextResponse()).isEqualTo("response2"); verify(tracer, times(2)).responseReceived(); }
@Test public void testOperationCancelled() { tracedCallable.call("test", responseObserver, callContext); responseObserver.getController().cancel(); MockServerStreamingCall<String, String> innerCall = innerCallable.popLastCall(); innerCall .getController() .getObserver() .onError( new CancelledException("fake cancel", null, new FakeStatusCode(Code.CANCELLED), false)); assertThat(responseObserver.isDone()).isTrue(); verify(tracer, times(1)).operationCancelled(); }
@Test public void testOperationFinish() { tracedCallable.call("test", responseObserver, callContext); MockServerStreamingCall<String, String> innerCall = innerCallable.popLastCall(); innerCall.getController().getObserver().onComplete(); assertThat(responseObserver.isDone()).isTrue(); verify(tracer, times(1)).operationSucceeded(); }
@Test public void testOperationFail() { RuntimeException expectedError = new RuntimeException("expected error"); tracedCallable.call("test", responseObserver, callContext); MockServerStreamingCall<String, String> innerCall = innerCallable.popLastCall(); innerCall.getController().getObserver().onError(expectedError); assertThat(responseObserver.getFinalError()).isEqualTo(expectedError); verify(tracer, times(1)).operationFailed(expectedError); }
@Test @SuppressWarnings("unchecked") public void testSyncError() { RuntimeException expectedError = new RuntimeException("expected error"); ServerStreamingCallable<String, String> innerCallable = mock(ServerStreamingCallable.class); doThrow(expectedError) .when(innerCallable) .call( eq("test"), (ResponseObserver<String>) any(ResponseObserver.class), any(ApiCallContext.class)); tracedCallable = new TracedServerStreamingCallable<>(innerCallable, tracerFactory, SPAN_NAME); try { tracedCallable.call("test", responseObserver, callContext); } catch (RuntimeException e) { } verify(tracer, times(1)).operationFailed(expectedError); }