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 =
          PubsubMessage.newBuilder()
                       .setData(ByteString.copyFrom(outgoingMessage.elementBytes));

      if (outgoingMessage.attributes != null) {
        message.putAllAttributes(outgoingMessage.attributes);
      }

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

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

      request.addMessages(message);
    }

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

  @VisibleForTesting  PubsubGrpcClient(
      @Nullable String timestampLabel,
      @Nullable String idLabel,
      int timeoutSec,
      ManagedChannel publisherChannel,
      Credentials credentials);

}