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.message().getData().toByteArray()); pubsubMessage.setAttributes(getMessageAttributes(outgoingMessage)); if (!outgoingMessage.message().getOrderingKey().isEmpty()) { pubsubMessage.setOrderingKey(outgoingMessage.message().getOrderingKey()); } 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 timestampAttribute, @Nullable String idAttribute, Pubsub pubsub); @Override void close(); @Override int publish(TopicPath topic, List<OutgoingMessage> outgoingMessages); @Override @SuppressWarnings("ProtoFieldNullComparison") List<IncomingMessage> pull(
      long requestTimeMsSinceEpoch,
      SubscriptionPath subscription,
      int batchSize,
      boolean returnImmediately); @Override void acknowledge(SubscriptionPath subscription, List<String> ackIds); @Override void modifyAckDeadline(
      SubscriptionPath subscription, List<String> ackIds, int deadlineSeconds); @Override void createTopic(TopicPath topic); @Override void deleteTopic(TopicPath topic); @Override List<TopicPath> listTopics(ProjectPath project); @Override void createSubscription(
      TopicPath topic, SubscriptionPath subscription, int ackDeadlineSeconds); @Override void deleteSubscription(SubscriptionPath subscription); @Override List<SubscriptionPath> listSubscriptions(ProjectPath project, TopicPath topic); @Override int ackDeadlineSeconds(SubscriptionPath subscription); @Override boolean isEOF(); static final PubsubClientFactory FACTORY; }
@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); int n = client.publish(TOPIC, ImmutableList.of(actualMessage)); assertEquals(1, n); }
@Test public void publishOneMessageWithOnlyTimestampAndIdAttributes() 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) .build()); 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); OutgoingMessage actualMessage = OutgoingMessage.of( com.google.pubsub.v1.PubsubMessage.newBuilder() .setData(ByteString.copyFromUtf8(DATA)) .build(), MESSAGE_TIME, RECORD_ID); int n = client.publish(TOPIC, ImmutableList.of(actualMessage)); assertEquals(1, n); }
@Test public void publishOneMessageWithNoTimestampOrIdAttribute() throws IOException { client = new PubsubJsonClient(null, null, mockPubsub); String expectedTopic = TOPIC.getPath(); PubsubMessage expectedPubsubMessage = new PubsubMessage() .encodeData(DATA.getBytes(StandardCharsets.UTF_8)) .setAttributes(ImmutableMap.<String, String>builder().put("k", "v").build()); 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) .build(), MESSAGE_TIME, RECORD_ID); int n = client.publish(TOPIC, ImmutableList.of(actualMessage)); assertEquals(1, n); }