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.elementBytes);

      Map<String, String> attributes = outgoingMessage.attributes;
      if ((timestampLabel != null || idLabel != null) && attributes == null) {
        attributes = new TreeMap<>();
      }
      if (attributes != null) {
        pubsubMessage.setAttributes(attributes);
      }

      if (timestampLabel != null) {
        attributes.put(timestampLabel, String.valueOf(outgoingMessage.timestampMsSinceEpoch));
      }

      if (idLabel != null && !Strings.isNullOrEmpty(outgoingMessage.recordId)) {
        attributes.put(idLabel, outgoingMessage.recordId);
      }

      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 timestampLabel,
      @Nullable String idLabel,
      Pubsub pubsub);

}