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().isEmpty()) {
      return ImmutableList.of();
    }
    List<IncomingMessage> incomingMessages = new ArrayList<>(response.getReceivedMessages().size());
    for (ReceivedMessage message : response.getReceivedMessages()) {
      PubsubMessage pubsubMessage = message.getMessage();
      Map<String, String> attributes;
      if (pubsubMessage.getAttributes() != null) {
        attributes = pubsubMessage.getAttributes();
      } else {
        attributes = new HashMap<>();
      }

      // Payload.
      byte[] elementBytes = pubsubMessage.getData() == null ? null : pubsubMessage.decodeData();
      if (elementBytes == null) {
        elementBytes = new byte[0];
      }

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

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

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

      com.google.pubsub.v1.PubsubMessage.Builder protoMessage =
          com.google.pubsub.v1.PubsubMessage.newBuilder();
      protoMessage.setData(ByteString.copyFrom(elementBytes));
      protoMessage.putAllAttributes(attributes);
      protoMessage.setOrderingKey(
          (String) pubsubMessage.getUnknownKeys().getOrDefault("orderingKey", ""));
      incomingMessages.add(
          IncomingMessage.of(
              protoMessage.build(),
              timestampMsSinceEpoch,
              requestTimeMsSinceEpoch,
              ackId,
              recordId));
    }

    return incomingMessages;
  }

  @VisibleForTesting  PubsubJsonClient(
      @Nullable String timestampAttribute, @Nullable String idAttribute, Pubsub pubsub);

}