@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.getMessage().getData().toByteArray());
      pubsubMessage.setAttributes(getMessageAttributes(outgoingMessage));
      if (!outgoingMessage.getMessage().getOrderingKey().isEmpty()) {
        pubsubMessage.setOrderingKey(outgoingMessage.getMessage().getOrderingKey());
      }

      // N.B. publishTime and messageId are intentionally not set on the message that is published
      pubsubMessages.add(pubsubMessage);
    }
    PublishRequest request = new PublishRequest().setMessages(pubsubMessages);
    PublishResponse response =
        pubsub.projects().topics().publish(topic.getPath(), request).execute();
    return response.getMessageIds().size();
  }
@Test
  public void publishOneMessage() throws IOException {
    String expectedTopic = TOPIC.getPath();
    PubsubMessage expectedPubsubMessage =
        new PubsubMessage()
            .encodeData(DATA.getBytes(StandardCharsets.UTF_8))
            .setAttributes(
                ImmutableMap.<String, String>builder()
                    .put(TIMESTAMP_ATTRIBUTE, String.valueOf(MESSAGE_TIME))
                    .put(ID_ATTRIBUTE, RECORD_ID)
                    .put("k", "v")
                    .build())
            .set("orderingKey", ORDERING_KEY);
    PublishRequest expectedRequest =
        new PublishRequest().setMessages(ImmutableList.of(expectedPubsubMessage));
    PublishResponse expectedResponse =
        new PublishResponse().setMessageIds(ImmutableList.of(MESSAGE_ID));
    when((Object) mockPubsub.projects().topics().publish(expectedTopic, expectedRequest).execute())
        .thenReturn(expectedResponse);
    Map<String, String> attrs = new HashMap<>();
    attrs.put("k", "v");
    OutgoingMessage actualMessage =
        OutgoingMessage.of(
            com.google.pubsub.v1.PubsubMessage.newBuilder()
                .setData(ByteString.copyFromUtf8(DATA))
                .putAllAttributes(attrs)
                .setOrderingKey(ORDERING_KEY)
                .build(),
            MESSAGE_TIME,
            RECORD_ID,
            null);
    int n = client.publish(TOPIC, ImmutableList.of(actualMessage));
    assertEquals(1, n);
  }