@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);
}