BufferingPullSubscriber implements PullSubscriber { @Override public List<SequencedMessage> pull() throws StatusException { @Nullable StatusException maybeError = error.get(); if (maybeError != null) { throw maybeError; } ArrayList<SequencedMessage> collection = new ArrayList<>(); messages.drainTo(collection); long bytes = collection.stream().mapToLong(SequencedMessage::getSizeBytes).sum(); underlying.allowFlow( FlowControlRequest.newBuilder() .setAllowedBytes(bytes) .setAllowedMessages(collection.size()) .build()); return ImmutableList.copyOf(collection); } BufferingPullSubscriber(SubscriberFactory factory, FlowControlSettings settings);   BufferingPullSubscriber(
      SubscriberFactory factory, FlowControlSettings settings, Offset initialLocation); @Override List<SequencedMessage> pull(); @Override void close();  }
@Test public void emptyPull() throws StatusException { assertThat(subscriber.pull(), empty()); }
@Test public void pullEmptiesForNext() throws StatusException { SequencedMessage message1 = newMessage(Timestamps.EPOCH, Offset.of(10), 10); SequencedMessage message2 = newMessage(Timestamps.EPOCH, Offset.of(11), 10); messageConsumer.accept(ImmutableList.of(toWrapper(message1), toWrapper(message2))); assertThat(subscriber.pull(), containsInAnyOrder(message1, message2)); assertThat(subscriber.pull(), empty()); }
@Test public void multipleBatchesAggregatedReturnsTokens() throws StatusException { SequencedMessage message1 = newMessage(Timestamps.EPOCH, Offset.of(10), 10); SequencedMessage message2 = newMessage(Timestamps.EPOCH, Offset.of(11), 20); SequencedMessage message3 = newMessage(Timestamps.EPOCH, Offset.of(12), 30); messageConsumer.accept(ImmutableList.of(toWrapper(message1), toWrapper(message2))); messageConsumer.accept(ImmutableList.of(toWrapper(message3))); assertThat(subscriber.pull(), containsInAnyOrder(message1, message2, message3)); assertThat(subscriber.pull(), empty()); FlowControlRequest flowControlRequest = FlowControlRequest.newBuilder().setAllowedMessages(3).setAllowedBytes(60).build(); verify(underlying).allowFlow(flowControlRequest); }