PubsubGrpcClient extends PubsubClient { @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.message().toBuilder(); if (timestampAttribute != null) { message.putAttributes( timestampAttribute, String.valueOf(outgoingMessage.timestampMsSinceEpoch())); } if (idAttribute != null && !Strings.isNullOrEmpty(outgoingMessage.recordId())) { message.putAttributes(idAttribute, outgoingMessage.recordId()); } request.addMessages(message); } PublishResponse response = publisherStub().publish(request.build()); return response.getMessageIdsCount(); } @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 publishOneMessage() throws IOException { 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), 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, RECORD_ID); int n = client.publish(TOPIC, ImmutableList.of(actualMessage)); assertEquals(1, n); assertEquals(expectedRequest, Iterables.getOnlyElement(requestsReceived)); } finally { server.shutdownNow(); } }