FirstElementCallable extends UnaryCallable<RequestT, ResponseT> { @Override public ApiFuture<ResponseT> futureCall(RequestT request, ApiCallContext context) { FirstElementResponseObserver<ResponseT> observer = new FirstElementResponseObserver<>(); streamingCallable.call(request, observer, context); return observer.getFuture(); } FirstElementCallable(ServerStreamingCallable<RequestT, ResponseT> streamingCallable); @Override ApiFuture<ResponseT> futureCall(RequestT request, ApiCallContext context);  }
@Test public void testHappyPath() throws InterruptedException, ExecutionException { ApiFuture<String> result = callable.futureCall("request"); MockServerStreamingCall<String, String> call = upstream.popLastCall(); Truth.assertThat(call.getController().isAutoFlowControlEnabled()).isFalse(); Truth.assertThat(call.getController().popLastPull()).isEqualTo(1); call.getController().getObserver().onResponse("response"); Truth.assertThat(result.get()).isEqualTo("response"); Truth.assertThat(call.getController().getObserver()).isNotNull(); }
@Test public void testEarlyTermination() throws Exception { ApiFuture<String> result = callable.futureCall("request"); MockServerStreamingCall<String, String> call = upstream.popLastCall(); Truth.assertThat(call.getController().isAutoFlowControlEnabled()).isFalse(); Truth.assertThat(call.getController().popLastPull()).isEqualTo(1); result.cancel(true); Truth.assertThat(call.getController().isCancelled()).isTrue(); call.getController() .getObserver() .onError(new RuntimeException("Some other upstream cancellation notice")); Throwable actualError = null; try { result.get(1, TimeUnit.SECONDS); } catch (Throwable e) { actualError = e; } Truth.assertThat(actualError).isInstanceOf(CancellationException.class); }
@Test public void testNoResults() throws Exception { ApiFuture<String> result = callable.futureCall("request"); MockServerStreamingCall<String, String> call = upstream.popLastCall(); Truth.assertThat(call.getController().isAutoFlowControlEnabled()).isFalse(); call.getController().getObserver().onComplete(); Truth.assertThat(result.get()).isNull(); }
@Test public void testErrorAfterResultIsIgnored() throws Exception { ApiFuture<String> result = callable.futureCall("request"); MockServerStreamingCall<String, String> call = upstream.popLastCall(); Truth.assertThat(call.getController().isAutoFlowControlEnabled()).isFalse(); call.getController().getObserver().onResponse("response"); call.getController() .getObserver() .onError(new RuntimeException("some error that will be ignored")); Truth.assertThat(result.get()).isEqualTo("response"); }