class PubsubJsonClient extends PubsubClient {

  @Override
  public List<IncomingMessage> pull(
      long requestTimeMsSinceEpoch,
      SubscriptionPath subscription,
      int batchSize,
      boolean returnImmediately) throws IOException {
    PullRequest request = new PullRequest()
        .setReturnImmediately(returnImmediately)
        .setMaxMessages(batchSize);
    PullResponse response = pubsub.projects()
                                  .subscriptions()
                                  .pull(subscription.getPath(), request)
                                  .execute();
    if (response.getReceivedMessages() == null || response.getReceivedMessages().size() == 0) {
      return ImmutableList.of();
    }
    List<IncomingMessage> incomingMessages = new ArrayList<>(response.getReceivedMessages().size());
    for (ReceivedMessage message : response.getReceivedMessages()) {
      PubsubMessage pubsubMessage = message.getMessage();
      @Nullable Map<String, String> attributes = pubsubMessage.getAttributes();

      // Payload.
      byte[] elementBytes = pubsubMessage.decodeData();

      // Timestamp.
      long timestampMsSinceEpoch =
          extractTimestamp(timestampLabel, message.getMessage().getPublishTime(), attributes);

      // Ack id.
      String ackId = message.getAckId();
      checkState(!Strings.isNullOrEmpty(ackId));

      // Record id, if any.
      @Nullable String recordId = null;
      if (idLabel != null && attributes != null) {
        recordId = attributes.get(idLabel);
      }
      if (Strings.isNullOrEmpty(recordId)) {
        // Fall back to the Pubsub provided message id.
        recordId = pubsubMessage.getMessageId();
      }

      incomingMessages.add(new IncomingMessage(elementBytes, attributes, timestampMsSinceEpoch,
                                               requestTimeMsSinceEpoch, ackId, recordId));
    }

    return incomingMessages;
  }

  @VisibleForTesting  PubsubJsonClient(
      @Nullable String timestampLabel,
      @Nullable String idLabel,
      Pubsub pubsub);

}