class PubsubGrpcClient extends PubsubClient {

  @Override
  public int publish(TopicPath topic, List<OutgoingMessage> outgoingMessages) throws IOException {
    PublishRequest.Builder request = PublishRequest.newBuilder().setTopic(topic.getPath());
    for (OutgoingMessage outgoingMessage : outgoingMessages) {
      PubsubMessage.Builder message = outgoingMessage.message().toBuilder();

      if (timestampAttribute != null) {
        message.putAttributes(
            timestampAttribute, String.valueOf(outgoingMessage.timestampMsSinceEpoch()));
      }

      if (idAttribute != null && !Strings.isNullOrEmpty(outgoingMessage.recordId())) {
        message.putAttributes(idAttribute, outgoingMessage.recordId());
      }

      request.addMessages(message);
    }

    PublishResponse response = publisherStub().publish(request.build());
    return response.getMessageIdsCount();
  }

  @VisibleForTesting  PubsubGrpcClient(
      @Nullable String timestampAttribute,
      @Nullable String idAttribute,
      int timeoutSec,
      ManagedChannel publisherChannel,
      Credentials credentials);

}