PubsubJsonClient extends PubsubClient { @Override @SuppressWarnings("ProtoFieldNullComparison") public List<IncomingMessage> pull( long requestTimeMsSinceEpoch, SubscriptionPath subscription, int batchSize, boolean returnImmediately) throws IOException { PullRequest request = new PullRequest().setReturnImmediately(returnImmediately).setMaxMessages(batchSize); PullResponse response = pubsub.projects().subscriptions().pull(subscription.getPath(), request).execute(); if (response.getReceivedMessages() == null || response.getReceivedMessages().isEmpty()) { return ImmutableList.of(); } List<IncomingMessage> incomingMessages = new ArrayList<>(response.getReceivedMessages().size()); for (ReceivedMessage message : response.getReceivedMessages()) { PubsubMessage pubsubMessage = message.getMessage(); Map<String, String> attributes; if (pubsubMessage.getAttributes() != null) { attributes = pubsubMessage.getAttributes(); } else { attributes = new HashMap<>(); } byte[] elementBytes = pubsubMessage.getData() == null ? null : pubsubMessage.decodeData(); if (elementBytes == null) { elementBytes = new byte[0]; } long timestampMsSinceEpoch = extractTimestamp(timestampAttribute, message.getMessage().getPublishTime(), attributes); String ackId = message.getAckId(); checkState(!Strings.isNullOrEmpty(ackId)); @Nullable String recordId = null; if (idAttribute != null) { recordId = attributes.get(idAttribute); } if (Strings.isNullOrEmpty(recordId)) { recordId = pubsubMessage.getMessageId(); } com.google.pubsub.v1.PubsubMessage.Builder protoMessage = com.google.pubsub.v1.PubsubMessage.newBuilder(); protoMessage.setData(ByteString.copyFrom(elementBytes)); protoMessage.putAllAttributes(attributes); if (pubsubMessage.getOrderingKey() != null) { protoMessage.setOrderingKey(pubsubMessage.getOrderingKey()); } else { protoMessage.setOrderingKey(""); } incomingMessages.add( IncomingMessage.of( protoMessage.build(), timestampMsSinceEpoch, requestTimeMsSinceEpoch, ackId, recordId)); } return incomingMessages; } @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 pullOneMessage() throws IOException { String expectedSubscription = SUBSCRIPTION.getPath(); PullRequest expectedRequest = new PullRequest().setReturnImmediately(true).setMaxMessages(10); PubsubMessage expectedPubsubMessage = new PubsubMessage() .setMessageId(MESSAGE_ID) .encodeData(DATA.getBytes(StandardCharsets.UTF_8)) .setPublishTime(String.valueOf(PUB_TIME)) .setAttributes( ImmutableMap.of( TIMESTAMP_ATTRIBUTE, String.valueOf(MESSAGE_TIME), ID_ATTRIBUTE, RECORD_ID)) .setOrderingKey(ORDERING_KEY); ReceivedMessage expectedReceivedMessage = new ReceivedMessage().setMessage(expectedPubsubMessage).setAckId(ACK_ID); PullResponse expectedResponse = new PullResponse().setReceivedMessages(ImmutableList.of(expectedReceivedMessage)); when((Object) (mockPubsub .projects() .subscriptions() .pull(expectedSubscription, expectedRequest) .execute())) .thenReturn(expectedResponse); List<IncomingMessage> acutalMessages = client.pull(REQ_TIME, SUBSCRIPTION, 10, true); assertEquals(1, acutalMessages.size()); IncomingMessage actualMessage = acutalMessages.get(0); assertEquals(ACK_ID, actualMessage.ackId()); assertEquals(DATA, actualMessage.message().getData().toStringUtf8()); assertEquals(RECORD_ID, actualMessage.recordId()); assertEquals(REQ_TIME, actualMessage.requestTimeMsSinceEpoch()); assertEquals(MESSAGE_TIME, actualMessage.timestampMsSinceEpoch()); assertEquals(ORDERING_KEY, actualMessage.message().getOrderingKey()); }
@Test public void pullOneMessageEmptyAttributes() throws IOException { client = new PubsubJsonClient(null, null, mockPubsub); String expectedSubscription = SUBSCRIPTION.getPath(); PullRequest expectedRequest = new PullRequest().setReturnImmediately(true).setMaxMessages(10); PubsubMessage expectedPubsubMessage = new PubsubMessage() .setMessageId(MESSAGE_ID) .encodeData(DATA.getBytes(StandardCharsets.UTF_8)) .setPublishTime(String.valueOf(PUB_TIME)); ReceivedMessage expectedReceivedMessage = new ReceivedMessage().setMessage(expectedPubsubMessage).setAckId(ACK_ID); PullResponse expectedResponse = new PullResponse().setReceivedMessages(ImmutableList.of(expectedReceivedMessage)); when((Object) (mockPubsub .projects() .subscriptions() .pull(expectedSubscription, expectedRequest) .execute())) .thenReturn(expectedResponse); List<IncomingMessage> acutalMessages = client.pull(REQ_TIME, SUBSCRIPTION, 10, true); assertEquals(1, acutalMessages.size()); IncomingMessage actualMessage = acutalMessages.get(0); assertEquals(ACK_ID, actualMessage.ackId()); assertEquals(DATA, actualMessage.message().getData().toStringUtf8()); assertEquals(REQ_TIME, actualMessage.requestTimeMsSinceEpoch()); assertEquals(PUB_TIME, actualMessage.timestampMsSinceEpoch()); }
@Test public void pullOneMessageWithNoData() throws IOException { String expectedSubscription = SUBSCRIPTION.getPath(); PullRequest expectedRequest = new PullRequest().setReturnImmediately(true).setMaxMessages(10); PubsubMessage expectedPubsubMessage = new PubsubMessage() .setMessageId(MESSAGE_ID) .setPublishTime(String.valueOf(PUB_TIME)) .setAttributes( ImmutableMap.of( TIMESTAMP_ATTRIBUTE, String.valueOf(MESSAGE_TIME), ID_ATTRIBUTE, RECORD_ID)); ReceivedMessage expectedReceivedMessage = new ReceivedMessage().setMessage(expectedPubsubMessage).setAckId(ACK_ID); PullResponse expectedResponse = new PullResponse().setReceivedMessages(ImmutableList.of(expectedReceivedMessage)); Mockito.when( (Object) (mockPubsub .projects() .subscriptions() .pull(expectedSubscription, expectedRequest) .execute())) .thenReturn(expectedResponse); List<IncomingMessage> acutalMessages = client.pull(REQ_TIME, SUBSCRIPTION, 10, true); assertEquals(1, acutalMessages.size()); IncomingMessage actualMessage = acutalMessages.get(0); assertArrayEquals(new byte[0], actualMessage.message().getData().toByteArray()); }