@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.getMessage().toBuilder().clearMessageId().clearPublishTime();

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

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

      request.addMessages(message);
    }

    PublishResponse response = publisherStub().publish(request.build());
    return response.getMessageIdsCount();
  }
@Test
  public void publishOneMessage() throws IOException {
    initializeClient(TIMESTAMP_ATTRIBUTE, ID_ATTRIBUTE);
    String expectedTopic = TOPIC.getPath();
    PubsubMessage expectedPubsubMessage =
        PubsubMessage.newBuilder()
            .setData(ByteString.copyFrom(DATA.getBytes(StandardCharsets.UTF_8)))
            .putAllAttributes(ATTRIBUTES)
            .putAllAttributes(
                ImmutableMap.of(
                    TIMESTAMP_ATTRIBUTE, String.valueOf(MESSAGE_TIME_MS), ID_ATTRIBUTE, RECORD_ID))
            .build();
    final PublishRequest expectedRequest =
        PublishRequest.newBuilder()
            .setTopic(expectedTopic)
            .addAllMessages(ImmutableList.of(expectedPubsubMessage))
            .build();
    final PublishResponse response =
        PublishResponse.newBuilder().addAllMessageIds(ImmutableList.of(MESSAGE_ID)).build();

    final List<PublishRequest> requestsReceived = new ArrayList<>();
    PublisherImplBase publisherImplBase =
        new PublisherImplBase() {
          @Override
          public void publish(
              PublishRequest request, StreamObserver<PublishResponse> responseObserver) {
            requestsReceived.add(request);
            responseObserver.onNext(response);
            responseObserver.onCompleted();
          }
        };
    Server server =
        InProcessServerBuilder.forName(channelName).addService(publisherImplBase).build().start();
    try {
      OutgoingMessage actualMessage =
          OutgoingMessage.of(
              com.google.pubsub.v1.PubsubMessage.newBuilder()
                  .setData(ByteString.copyFromUtf8(DATA))
                  .putAllAttributes(ATTRIBUTES)
                  .build(),
              MESSAGE_TIME_MS,
              RECORD_ID,
              null);
      int n = client.publish(TOPIC, ImmutableList.of(actualMessage));
      assertEquals(1, n);
      assertEquals(expectedRequest, Iterables.getOnlyElement(requestsReceived));
    } finally {
      server.shutdownNow();
    }
  }