class 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);

}