GrpcDirectServerStreamingCallable extends ServerStreamingCallable<RequestT, ResponseT> { @Override public void call( RequestT request, ResponseObserver<ResponseT> responseObserver, ApiCallContext context) { Preconditions.checkNotNull(request); Preconditions.checkNotNull(responseObserver); ClientCall<RequestT, ResponseT> call = GrpcClientCalls.newCall(descriptor, context); GrpcDirectStreamController<RequestT, ResponseT> controller = new GrpcDirectStreamController<>(call, responseObserver); controller.start(request); } GrpcDirectServerStreamingCallable(MethodDescriptor<RequestT, ResponseT> descriptor); @Override void call(
      RequestT request, ResponseObserver<ResponseT> responseObserver, ApiCallContext context);  }
@Test public void testBadContext() { streamingCallable = GrpcCallableFactory.createServerStreamingCallable( GrpcCallSettings.create(METHOD_SERVER_STREAMING_RECOGNIZE), streamingCallSettings, clientContext .toBuilder() .setDefaultCallContext(FakeCallContext.createDefault()) .build()); CountDownLatch latch = new CountDownLatch(1); MoneyObserver observer = new MoneyObserver(true, latch); try { streamingCallable.call(DEFAULT_REQUEST, observer); Assert.fail("Callable should have thrown an exception"); } catch (IllegalArgumentException expected) { Truth.assertThat(expected) .hasMessageThat() .contains("context must be an instance of GrpcCallContext"); } }
@Test public void testServerStreamingStart() throws Exception { CountDownLatch latch = new CountDownLatch(1); MoneyObserver moneyObserver = new MoneyObserver(true, latch); streamingCallable.call(DEFAULT_REQUEST, moneyObserver); Truth.assertThat(moneyObserver.controller).isNotNull(); }
@Test public void testServerStreaming() throws Exception { CountDownLatch latch = new CountDownLatch(2); MoneyObserver moneyObserver = new MoneyObserver(true, latch); streamingCallable.call(DEFAULT_REQUEST, moneyObserver); latch.await(20, TimeUnit.SECONDS); Truth.assertThat(moneyObserver.error).isNull(); Truth.assertThat(moneyObserver.response).isEqualTo(DEFAULT_RESPONSE); }
@Test public void testManualFlowControl() throws Exception { CountDownLatch latch = new CountDownLatch(2); MoneyObserver moneyObserver = new MoneyObserver(false, latch); streamingCallable.call(DEFAULT_REQUEST, moneyObserver); latch.await(500, TimeUnit.MILLISECONDS); Truth.assertWithMessage("Received response before requesting it") .that(moneyObserver.response) .isNull(); moneyObserver.controller.request(1); latch.await(500, TimeUnit.MILLISECONDS); Truth.assertThat(moneyObserver.response).isEqualTo(DEFAULT_RESPONSE); Truth.assertThat(moneyObserver.completed).isTrue(); }
@Test public void testCancelClientCall() throws Exception { CountDownLatch latch = new CountDownLatch(1); MoneyObserver moneyObserver = new MoneyObserver(false, latch); streamingCallable.call(ASYNC_REQUEST, moneyObserver); moneyObserver.controller.cancel(); moneyObserver.controller.request(1); latch.await(500, TimeUnit.MILLISECONDS); Truth.assertThat(moneyObserver.error).isInstanceOf(CancellationException.class); Truth.assertThat(moneyObserver.error).hasMessageThat().isEqualTo("User cancelled stream"); }
@Test public void testOnResponseError() throws Throwable { CountDownLatch latch = new CountDownLatch(1); MoneyObserver moneyObserver = new MoneyObserver(true, latch); streamingCallable.call(ERROR_REQUEST, moneyObserver); latch.await(500, TimeUnit.MILLISECONDS); Truth.assertThat(moneyObserver.error).isInstanceOf(ApiException.class); Truth.assertThat(((ApiException) moneyObserver.error).getStatusCode().getCode()) .isEqualTo(StatusCode.Code.INVALID_ARGUMENT); Truth.assertThat(moneyObserver.error) .hasMessageThat() .isEqualTo("io.grpc.StatusRuntimeException: INVALID_ARGUMENT: red must be positive"); }
@Test public void testObserverErrorCancelsCall() throws Throwable { final RuntimeException expectedCause = new RuntimeException("some error"); final SettableApiFuture<Throwable> actualErrorF = SettableApiFuture.create(); ResponseObserver<Money> moneyObserver = new StateCheckingResponseObserver<Money>() { @Override protected void onStartImpl(StreamController controller) {} @Override protected void onResponseImpl(Money response) { throw expectedCause; } @Override protected void onErrorImpl(Throwable t) { actualErrorF.set(t); } @Override protected void onCompleteImpl() { actualErrorF.set(null); } }; streamingCallable.call(DEFAULT_REQUEST, moneyObserver); Throwable actualError = actualErrorF.get(500, TimeUnit.MILLISECONDS); Truth.assertThat(actualError).isInstanceOf(ApiException.class); Truth.assertThat(((ApiException) actualError).getStatusCode().getCode()) .isEqualTo(StatusCode.Code.CANCELLED); Truth.assertThat(actualError.getCause()).isInstanceOf(StatusRuntimeException.class); Truth.assertThat(actualError.getCause().getCause()).isSameInstanceAs(expectedCause); }
@Test public void testBlockingServerStreaming() throws Exception { Color request = Color.newBuilder().setRed(0.5f).build(); ServerStream<Money> response = streamingCallable.call(request); List<Money> responseData = Lists.newArrayList(response); Money expected = Money.newBuilder().setCurrencyCode("USD").setUnits(127).build(); Truth.assertThat(responseData).containsExactly(expected); }