ServerStream implements Iterable<V> { @InternalApi("For use by ServerStreamingCallable only.") ResponseObserver<V> observer() { return observer; } @InternalApi("For use by ServerStreamingCallable only.")  ServerStream(); @Override @Nonnull Iterator<V> iterator(); boolean isReceiveReady(); void cancel();  }
@Test public void testEmptyStream() { stream.observer().onComplete(); Truth.assertThat(Lists.newArrayList(stream)).isEmpty(); }
@Test public void testMultipleItemStream() throws Exception { Future<Void> producerFuture = executor.submit( new Callable<Void>() { @Override public Void call() { for (int i = 0; i < 5; i++) { int requestCount = controller.popLastPull(); Truth.assertWithMessage("ServerStream should request one item at a time") .that(requestCount) .isEqualTo(1); stream.observer().onResponse(i); } stream.observer().onComplete(); return null; } }); Future<List<Integer>> consumerFuture = executor.submit( new Callable<List<Integer>>() { @Override public List<Integer> call() { return Lists.newArrayList(stream); } }); producerFuture.get(60, TimeUnit.SECONDS); List<Integer> results = consumerFuture.get(); Truth.assertThat(results).containsExactly(0, 1, 2, 3, 4); }
@Test public void testErrorPropagation() { ClassCastException e = new ClassCastException("fake error"); stream.observer().onError(e); Throwable actualError = null; try { Truth.assertThat(Lists.newArrayList(stream)).isNotNull(); } catch (Throwable t) { actualError = t; } Truth.assertThat(actualError).hasMessageThat().contains(e.getMessage()); Truth.assertThat(actualError).isEqualTo(e); }