class PubsubJsonClient extends PubsubClient {

  @Override
  public int publish(TopicPath topic, List<OutgoingMessage> outgoingMessages) throws IOException {
    List<PubsubMessage> pubsubMessages = new ArrayList<>(outgoingMessages.size());
    for (OutgoingMessage outgoingMessage : outgoingMessages) {
      PubsubMessage pubsubMessage =
          new PubsubMessage().encodeData(outgoingMessage.message().getData().toByteArray());
      pubsubMessage.setAttributes(getMessageAttributes(outgoingMessage));
      if (!outgoingMessage.message().getOrderingKey().isEmpty()) {
        pubsubMessage.setOrderingKey(outgoingMessage.message().getOrderingKey());
      }
      pubsubMessages.add(pubsubMessage);
    }
    PublishRequest request = new PublishRequest().setMessages(pubsubMessages);
    PublishResponse response =
        pubsub.projects().topics().publish(topic.getPath(), request).execute();
    return response.getMessageIds().size();
  }

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

}