ServerStreamingAttemptCallable implements Callable<Void> { public void start() { Preconditions.checkState(!isStarted, "Already started"); outerObserver.onStart( new StreamController() { @Override public void disableAutoInboundFlowControl() { Preconditions.checkState( !isStarted, "Can't disable auto flow control once the stream is started"); autoFlowControl = false; } @Override public void request(int count) { onRequest(count); } @Override public void cancel() { onCancel(); } }); if (autoFlowControl) { synchronized (lock) { pendingRequests = Integer.MAX_VALUE; } } isStarted = true; Duration totalTimeout = outerRetryingFuture.getAttemptSettings().getGlobalSettings().getTotalTimeout(); if (totalTimeout != null && context != null && context.getTimeout() == null) { context = context.withTimeout(totalTimeout); } call(); } ServerStreamingAttemptCallable(
      ServerStreamingCallable<RequestT, ResponseT> innerCallable,
      StreamResumptionStrategy<RequestT, ResponseT> resumptionStrategy,
      RequestT initialRequest,
      ApiCallContext context,
      ResponseObserver<ResponseT> outerObserver); void start(); @Override Void call();  }
@Test public void testUserProvidedContextTimeout() { Mockito.doReturn(NoopApiTracer.getInstance()).when(mockedCallContext).getTracer(); Mockito.doReturn(Duration.ofHours(5)).when(mockedCallContext).getTimeout(); Mockito.doReturn(Duration.ofHours(5)).when(mockedCallContext).getStreamWaitTimeout(); ServerStreamingAttemptCallable<String, String> callable = createCallable(mockedCallContext); callable.start(); Mockito.verify(mockedCallContext, Mockito.times(1)).getTimeout(); Mockito.verify(mockedCallContext, Mockito.never()).withTimeout(totalTimeout); Mockito.verify(mockedCallContext, Mockito.times(1)).getStreamWaitTimeout(); Mockito.verify(mockedCallContext, Mockito.never()) .withStreamWaitTimeout(Mockito.any(Duration.class)); Truth.assertThat(observer.controller).isNotNull(); MockServerStreamingCall<String, String> call = innerCallable.popLastCall(); Truth.assertThat(call.getController().isAutoFlowControlEnabled()).isTrue(); Truth.assertThat(call.getRequest()).isEqualTo("request"); call.getController().getObserver().onResponse("response1"); call.getController().getObserver().onResponse("response2"); call.getController().getObserver().onComplete(); Truth.assertThat(observer.responses).containsExactly("response1", "response2").inOrder(); fakeRetryingFuture.assertSuccess(); }
@Test public void testNoUserProvidedContextTimeout() { Mockito.doReturn(NoopApiTracer.getInstance()).when(mockedCallContext).getTracer(); Mockito.doReturn(null).when(mockedCallContext).getTimeout(); Mockito.doReturn(null).when(mockedCallContext).getStreamWaitTimeout(); Mockito.doReturn(mockedCallContext).when(mockedCallContext).withTimeout(totalTimeout); Mockito.doReturn(mockedCallContext) .when(mockedCallContext) .withStreamWaitTimeout(Mockito.any(Duration.class)); ServerStreamingAttemptCallable<String, String> callable = createCallable(mockedCallContext); callable.start(); Mockito.verify(mockedCallContext, Mockito.times(1)).getTimeout(); Mockito.verify(mockedCallContext, Mockito.times(1)).withTimeout(totalTimeout); Mockito.verify(mockedCallContext, Mockito.times(1)).getStreamWaitTimeout(); Mockito.verify(mockedCallContext, Mockito.times(1)) .withStreamWaitTimeout(Mockito.any(Duration.class)); Truth.assertThat(observer.controller).isNotNull(); MockServerStreamingCall<String, String> call = innerCallable.popLastCall(); Truth.assertThat(call.getController().isAutoFlowControlEnabled()).isTrue(); Truth.assertThat(call.getRequest()).isEqualTo("request"); call.getController().getObserver().onResponse("response1"); call.getController().getObserver().onResponse("response2"); call.getController().getObserver().onComplete(); Truth.assertThat(observer.responses).containsExactly("response1", "response2").inOrder(); fakeRetryingFuture.assertSuccess(); }
@Test public void testNoErrorsAutoFlow() { ServerStreamingAttemptCallable<String, String> callable = createCallable(); callable.start(); Truth.assertThat(observer.controller).isNotNull(); MockServerStreamingCall<String, String> call = innerCallable.popLastCall(); Truth.assertThat(call.getController().isAutoFlowControlEnabled()).isTrue(); Truth.assertThat(call.getRequest()).isEqualTo("request"); call.getController().getObserver().onResponse("response1"); call.getController().getObserver().onResponse("response2"); call.getController().getObserver().onComplete(); Truth.assertThat(observer.responses).containsExactly("response1", "response2").inOrder(); fakeRetryingFuture.assertSuccess(); }
@Test public void testNoErrorsManualFlow() { observer = new AccumulatingObserver(false); ServerStreamingAttemptCallable<String, String> callable = createCallable(); callable.start(); Truth.assertThat(observer.controller).isNotNull(); MockServerStreamingCall<String, String> call = innerCallable.popLastCall(); Truth.assertThat(call.getController().isAutoFlowControlEnabled()).isFalse(); Truth.assertThat(call.getRequest()).isEqualTo("request"); observer.controller.request(1); Truth.assertThat(call.getController().popLastPull()).isEqualTo(1); call.getController().getObserver().onResponse("response1"); observer.controller.request(1); Truth.assertThat(call.getController().popLastPull()).isEqualTo(1); call.getController().getObserver().onResponse("response2"); call.getController().getObserver().onComplete(); Truth.assertThat(observer.responses).containsExactly("response1", "response2").inOrder(); fakeRetryingFuture.assertSuccess(); }
@Test public void testCancel() { observer = new AccumulatingObserver(false); ServerStreamingAttemptCallable<String, String> callable = createCallable(); callable.start(); observer.controller.request(1); Truth.assertThat(observer.controller).isNotNull(); MockServerStreamingCall<String, String> call = innerCallable.popLastCall(); Truth.assertThat(call).isNotNull(); Truth.assertThat(call.getController().isAutoFlowControlEnabled()).isFalse(); observer.controller.cancel(); Truth.assertThat(call.getController().isCancelled()).isTrue(); RuntimeException innerException = new RuntimeException("Some internal representation of cancel"); call.getController().getObserver().onError(innerException); Throwable outerError = null; try { fakeRetryingFuture.getAttemptResult().get(1, TimeUnit.SECONDS); } catch (ExecutionException e) { outerError = e.getCause(); } catch (Throwable e) { outerError = e; } Truth.assertThat(outerError).isInstanceOf(ServerStreamingAttemptException.class); Truth.assertThat(outerError.getCause()).isInstanceOf(CancellationException.class); boolean includesMeInStackTrace = false; for (StackTraceElement e : outerError.getCause().getStackTrace()) { if (ServerStreamingAttemptCallableTest.class.getName().equals(e.getClassName())) { includesMeInStackTrace = true; break; } } Truth.assertWithMessage("Cancel caller included in stack trace") .that(includesMeInStackTrace) .isTrue(); }