rename MessagesProcessor to MessageDispatcher (#1548)

* rename MessagesProcessor to MessageDispatcher

Fixes #1546

Author: pongad

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessagesProcessor.java renamed to google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java

@@ -51,8 +51,8 @@
  * Dispatches messages to a message receiver while handling the messages acking and lease
  * extensions.
  */
-class MessagesProcessor {
-  private static final Logger logger = LoggerFactory.getLogger(MessagesProcessor.class);
+class MessageDispatcher {
+  private static final Logger logger = LoggerFactory.getLogger(MessageDispatcher.class);
 
   private static final int INITIAL_ACK_DEADLINE_EXTENSION_SECONDS = 2;
   @VisibleForTesting static final Duration PENDING_ACKS_SEND_DELAY = Duration.millis(100);
@@ -63,7 +63,7 @@ class MessagesProcessor {
 
   private final Duration ackExpirationPadding;
   private final MessageReceiver receiver;
-  private final AcksProcessor acksProcessor;
+  private final AckProcessor ackProcessor;
 
   private final FlowController flowController;
   private final MessagesWaiter messagesWaiter;
@@ -199,14 +199,14 @@ public void onSuccess(AckReply reply) {
     }
   }
 
-  public interface AcksProcessor {
+  public interface AckProcessor {
     void sendAckOperations(
         List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions);
   }
 
-  MessagesProcessor(
+  MessageDispatcher(
       MessageReceiver receiver,
-      AcksProcessor acksProcessor,
+      AckProcessor ackProcessor,
       Duration ackExpirationPadding,
       Distribution ackLatencyDistribution,
       FlowController flowController,
@@ -215,7 +215,7 @@ void sendAckOperations(
     this.executor = executor;
     this.ackExpirationPadding = ackExpirationPadding;
     this.receiver = receiver;
-    this.acksProcessor = acksProcessor;
+    this.ackProcessor = ackProcessor;
     this.flowController = flowController;
     outstandingAckHandlers = new HashMap<>();
     pendingAcks = new HashSet<>();
@@ -466,6 +466,6 @@ private void processOutstandingAckOperations(
       }
     }
 
-    acksProcessor.sendAckOperations(acksToSend, modifyAckDeadlinesToSend);
+    ackProcessor.sendAckOperations(acksToSend, modifyAckDeadlinesToSend);
   }
 }

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java

@@ -22,8 +22,8 @@
 import com.google.api.stats.Distribution;
 import com.google.auth.Credentials;
 import com.google.cloud.Clock;
-import com.google.cloud.pubsub.spi.v1.MessagesProcessor.AcksProcessor;
-import com.google.cloud.pubsub.spi.v1.MessagesProcessor.PendingModifyAckDeadline;
+import com.google.cloud.pubsub.spi.v1.MessageDispatcher.AckProcessor;
+import com.google.cloud.pubsub.spi.v1.MessageDispatcher.PendingModifyAckDeadline;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.AbstractService;
 import com.google.common.util.concurrent.FutureCallback;
@@ -51,7 +51,7 @@
  * Implementation of {@link AbstractSubscriberConnection} based on Cloud Pub/Sub pull and
  * acknowledge operations.
  */
-final class PollingSubscriberConnection extends AbstractService implements AcksProcessor {
+final class PollingSubscriberConnection extends AbstractService implements AckProcessor {
   private static final int MAX_PER_REQUEST_CHANGES = 1000;
   private static final Duration DEFAULT_TIMEOUT = Duration.standardSeconds(10);
   private static final int DEFAULT_MAX_MESSAGES = 1000;
@@ -63,7 +63,7 @@ final class PollingSubscriberConnection extends AbstractService implements AcksP
   private final String subscription;
   private final ScheduledExecutorService executor;
   private final SubscriberFutureStub stub;
-  private final MessagesProcessor messagesProcessor;
+  private final MessageDispatcher messageDispatcher;
 
   public PollingSubscriberConnection(
       String subscription,
@@ -80,8 +80,8 @@ public PollingSubscriberConnection(
     stub =
         SubscriberGrpc.newFutureStub(channel)
             .withCallCredentials(MoreCallCredentials.from(credentials));
-    messagesProcessor =
-        new MessagesProcessor(
+    messageDispatcher =
+        new MessageDispatcher(
             receiver,
             this,
             ackExpirationPadding,
@@ -109,7 +109,7 @@ private void initialize() {
         new FutureCallback<Subscription>() {
           @Override
           public void onSuccess(Subscription result) {
-            messagesProcessor.setMessageDeadlineSeconds(result.getAckDeadlineSeconds());
+            messageDispatcher.setMessageDeadlineSeconds(result.getAckDeadlineSeconds());
             pullMessages(INITIAL_BACKOFF);
           }
 
@@ -122,7 +122,7 @@ public void onFailure(Throwable cause) {
 
   @Override
   protected void doStop() {
-    messagesProcessor.stop();
+    messageDispatcher.stop();
     notifyStopped();
   }
 
@@ -141,7 +141,7 @@ private void pullMessages(final Duration backoff) {
         new FutureCallback<PullResponse>() {
           @Override
           public void onSuccess(PullResponse pullResponse) {
-            messagesProcessor.processReceivedMessages(pullResponse.getReceivedMessagesList());
+            messageDispatcher.processReceivedMessages(pullResponse.getReceivedMessagesList());
             if (pullResponse.getReceivedMessagesCount() == 0) {
               // No messages in response, possibly caught up in backlog, we backoff to avoid
               // slamming the server.

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/StreamingSubscriberConnection.java

@@ -22,8 +22,8 @@
 import com.google.api.stats.Distribution;
 import com.google.auth.Credentials;
 import com.google.cloud.Clock;
-import com.google.cloud.pubsub.spi.v1.MessagesProcessor.AcksProcessor;
-import com.google.cloud.pubsub.spi.v1.MessagesProcessor.PendingModifyAckDeadline;
+import com.google.cloud.pubsub.spi.v1.MessageDispatcher.AckProcessor;
+import com.google.cloud.pubsub.spi.v1.MessageDispatcher.PendingModifyAckDeadline;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.AbstractService;
 import com.google.common.util.concurrent.FutureCallback;
@@ -49,7 +49,7 @@
 import org.slf4j.LoggerFactory;
 
 /** Implementation of {@link AbstractSubscriberConnection} based on Cloud Pub/Sub streaming pull. */
-final class StreamingSubscriberConnection extends AbstractService implements AcksProcessor {
+final class StreamingSubscriberConnection extends AbstractService implements AckProcessor {
   private static final Logger logger = LoggerFactory.getLogger(StreamingSubscriberConnection.class);
 
   private static final Duration INITIAL_CHANNEL_RECONNECT_BACKOFF = new Duration(100); // 100ms
@@ -62,7 +62,7 @@ final class StreamingSubscriberConnection extends AbstractService implements Ack
 
   private final String subscription;
   private final ScheduledExecutorService executor;
-  private final MessagesProcessor messagesProcessor;
+  private final MessageDispatcher messageDispatcher;
   private ClientCallStreamObserver<StreamingPullRequest> requestObserver;
 
   public StreamingSubscriberConnection(
@@ -80,16 +80,16 @@ public StreamingSubscriberConnection(
     this.executor = executor;
     this.credentials = credentials;
     this.channel = channel;
-    this.messagesProcessor =
-        new MessagesProcessor(
+    this.messageDispatcher =
+        new MessageDispatcher(
             receiver,
             this,
             ackExpirationPadding,
             ackLatencyDistribution,
             flowController,
             executor,
             clock);
-    messagesProcessor.setMessageDeadlineSeconds(streamAckDeadlineSeconds);
+    messageDispatcher.setMessageDeadlineSeconds(streamAckDeadlineSeconds);
   }
 
   @Override
@@ -101,7 +101,7 @@ protected void doStart() {
 
   @Override
   protected void doStop() {
-    messagesProcessor.stop();
+    messageDispatcher.stop();
     notifyStopped();
     requestObserver.onError(Status.CANCELLED.asException());
   }
@@ -123,7 +123,7 @@ public void beforeStart(ClientCallStreamObserver<StreamingPullRequest> requestOb
 
     @Override
     public void onNext(StreamingPullResponse response) {
-      messagesProcessor.processReceivedMessages(response.getReceivedMessagesList());
+      messageDispatcher.processReceivedMessages(response.getReceivedMessagesList());
       // Only if not shutdown we will request one more bundles of messages to be delivered.
       if (isAlive()) {
         requestObserver.request(1);
@@ -157,11 +157,11 @@ private void initialize() {
     logger.debug(
         "Initializing stream to subscription {} with deadline {}",
         subscription,
-        messagesProcessor.getMessageDeadlineSeconds());
+        messageDispatcher.getMessageDeadlineSeconds());
     requestObserver.onNext(
         StreamingPullRequest.newBuilder()
             .setSubscription(subscription)
-            .setStreamAckDeadlineSeconds(messagesProcessor.getMessageDeadlineSeconds())
+            .setStreamAckDeadlineSeconds(messageDispatcher.getMessageDeadlineSeconds())
             .build());
     requestObserver.request(1);
 
@@ -240,7 +240,7 @@ public void sendAckOperations(
   }
 
   public void updateStreamAckDeadline(int newAckDeadlineSeconds) {
-    messagesProcessor.setMessageDeadlineSeconds(newAckDeadlineSeconds);
+    messageDispatcher.setMessageDeadlineSeconds(newAckDeadlineSeconds);
     requestObserver.onNext(
         StreamingPullRequest.newBuilder()
             .setStreamAckDeadlineSeconds(newAckDeadlineSeconds)

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java

@@ -16,7 +16,7 @@
 
 package com.google.cloud.pubsub.spi.v1;
 
-import static com.google.cloud.pubsub.spi.v1.MessagesProcessor.PENDING_ACKS_SEND_DELAY;
+import static com.google.cloud.pubsub.spi.v1.MessageDispatcher.PENDING_ACKS_SEND_DELAY;
 import static org.junit.Assert.assertEquals;
 
 import com.google.cloud.pubsub.spi.v1.FakeSubscriberServiceImpl.ModifyAckDeadline;