PubsubGrpcClient extends PubsubClient { @Override public List<IncomingMessage> pull( long requestTimeMsSinceEpoch, SubscriptionPath subscription, int batchSize, boolean returnImmediately) throws IOException { PullRequest request = PullRequest.newBuilder() .setSubscription(subscription.getPath()) .setReturnImmediately(returnImmediately) .setMaxMessages(batchSize) .build(); PullResponse response = subscriberStub().pull(request); if (response.getReceivedMessagesCount() == 0) { return ImmutableList.of(); } List<IncomingMessage> incomingMessages = new ArrayList<>(response.getReceivedMessagesCount()); for (ReceivedMessage message : response.getReceivedMessagesList()) { PubsubMessage pubsubMessage = message.getMessage(); @Nullable Map<String, String> attributes = pubsubMessage.getAttributes(); String pubsubTimestampString = null; Timestamp timestampProto = pubsubMessage.getPublishTime(); if (timestampProto != null) { pubsubTimestampString = String.valueOf(timestampProto.getSeconds() + timestampProto.getNanos() / 1000L); } long timestampMsSinceEpoch = extractTimestamp(timestampAttribute, pubsubTimestampString, attributes); String ackId = message.getAckId(); checkState(!Strings.isNullOrEmpty(ackId)); @Nullable String recordId = null; if (idAttribute != null && attributes != null) { recordId = attributes.get(idAttribute); } if (Strings.isNullOrEmpty(recordId)) { recordId = pubsubMessage.getMessageId(); } incomingMessages.add( IncomingMessage.of( pubsubMessage, timestampMsSinceEpoch, requestTimeMsSinceEpoch, ackId, recordId)); } return incomingMessages; } @VisibleForTesting  PubsubGrpcClient(
      @Nullable String timestampAttribute,
      @Nullable String idAttribute,
      int timeoutSec,
      ManagedChannel publisherChannel,
      Credentials credentials); @Override void close(); @Override int publish(TopicPath topic, List<OutgoingMessage> outgoingMessages); @Override 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(); final PullRequest expectedRequest = PullRequest.newBuilder() .setSubscription(expectedSubscription) .setReturnImmediately(true) .setMaxMessages(10) .build(); Timestamp timestamp = Timestamp.newBuilder() .setSeconds(PUB_TIME / 1000) .setNanos((int) (PUB_TIME % 1000) * 1000) .build(); PubsubMessage expectedPubsubMessage = PubsubMessage.newBuilder() .setMessageId(MESSAGE_ID) .setData(ByteString.copyFrom(DATA.getBytes(StandardCharsets.UTF_8))) .setPublishTime(timestamp) .putAllAttributes(ATTRIBUTES) .putAllAttributes( ImmutableMap.of( TIMESTAMP_ATTRIBUTE, String.valueOf(MESSAGE_TIME), ID_ATTRIBUTE, RECORD_ID)) .build(); ReceivedMessage expectedReceivedMessage = ReceivedMessage.newBuilder().setMessage(expectedPubsubMessage).setAckId(ACK_ID).build(); final PullResponse response = PullResponse.newBuilder() .addAllReceivedMessages(ImmutableList.of(expectedReceivedMessage)) .build(); final List<PullRequest> requestsReceived = new ArrayList<>(); SubscriberImplBase subscriberImplBase = new SubscriberImplBase() { @Override public void pull(PullRequest request, StreamObserver<PullResponse> responseObserver) { requestsReceived.add(request); responseObserver.onNext(response); responseObserver.onCompleted(); } }; Server server = InProcessServerBuilder.forName(channelName).addService(subscriberImplBase).build().start(); try { 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(expectedRequest, Iterables.getOnlyElement(requestsReceived)); } finally { server.shutdownNow(); } }