Change AckReplyConsumer to expose individual methods for replying (#1899)

* Change AckReplyConsumer to expose individual methods for replying to a
message outcome as opposed to use an enum.

Author: davidtorres

google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/PubSubExample.java

@@ -502,7 +502,7 @@ public void run(Tuple<SubscriptionName, Long> params) throws Exception {
                         @Override
                         public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
                             messageCount.incrementAndGet();
-                            consumer.accept(AckReply.ACK);
+                            consumer.ack();
                         }
                     };
             SubscriptionName subscriptionName = params.x();

…s/CreateSubscriptionAndPullMessages.java …reateSubscriptionAndConsumeMessages.javagoogle-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndPullMessages.java renamed to google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndConsumeMessages.java google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndPullMessages.java renamed to google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndConsumeMessages.java

@@ -16,7 +16,6 @@
 
 package com.google.cloud.examples.pubsub.snippets;
 
-import com.google.cloud.pubsub.spi.v1.AckReply;
 import com.google.cloud.pubsub.spi.v1.AckReplyConsumer;
 import com.google.cloud.pubsub.spi.v1.MessageReceiver;
 import com.google.cloud.pubsub.spi.v1.Subscriber;
@@ -31,7 +30,7 @@
  * A snippet for Google Cloud Pub/Sub showing how to create a Pub/Sub pull subscription and
  * asynchronously pull messages from it.
  */
-public class CreateSubscriptionAndPullMessages {
+public class CreateSubscriptionAndConsumeMessages {
 
   public static void main(String... args) throws Exception {
     // [START async_pull_subscription]
@@ -47,7 +46,7 @@ public static void main(String... args) throws Exception {
           @Override
           public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
             System.out.println("got message: " + message.getData().toStringUtf8());
-            consumer.accept(AckReply.ACK);
+            consumer.ack();
           }
         };
     Subscriber subscriber = null;

google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/MessageReceiverSnippets.java

@@ -22,7 +22,6 @@
 
 package com.google.cloud.examples.pubsub.snippets;
 
-import com.google.cloud.pubsub.spi.v1.AckReply;
 import com.google.cloud.pubsub.spi.v1.AckReplyConsumer;
 import com.google.cloud.pubsub.spi.v1.MessageReceiver;
 import com.google.pubsub.v1.PubsubMessage;
@@ -51,9 +50,9 @@ public MessageReceiver messageReceiver() {
     MessageReceiver receiver = new MessageReceiver() {
       public void receiveMessage(final PubsubMessage message, final AckReplyConsumer consumer) {
         if (blockingQueue.offer(message)) {
-          consumer.accept(AckReply.ACK);
+          consumer.ack();
         } else {
-          consumer.accept(AckReply.NACK);
+          consumer.nack();
         }
       }
     };

google-cloud-pubsub/README.md

@@ -158,7 +158,6 @@ With Pub/Sub you can pull messages from a subscription. Add the following import
 file:
 
 ```java
-import com.google.cloud.pubsub.spi.v1.AckReply;
 import com.google.cloud.pubsub.spi.v1.AckReplyConsumer;
 import com.google.cloud.pubsub.spi.v1.MessageReceiver;
 import com.google.cloud.pubsub.spi.v1.Subscriber;
@@ -175,7 +174,7 @@ MessageReceiver receiver =
       @Override
       public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
         System.out.println("got message: " + message.getData().toStringUtf8());
-        consumer.accept(AckReply.ACK, null);
+        consumer.ack();
       }
     };
 Subscriber subscriber = null;
@@ -204,7 +203,7 @@ try {
 In
 [CreateTopicAndPublishMessages.java](../google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateTopicAndPublishMessages.java)
 and
-[CreateSubscriptionAndPullMessages.java](../google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndPullMessages.java)
+[CreateSubscriptionAndConsumeMessages.java](../google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndConsumeMessages.java)
 we put together all the code shown above into two programs. The programs assume that you are
 running on Compute Engine, App Engine Flexible or from your own desktop.
 

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

@@ -18,10 +18,17 @@
 
 /**
  * Accepts a reply, sending it to the service.
- *
- * <p>Both the interface and its method is named after the Java 8's {@code Consumer} interface
- * to make migration to Java 8 and adopting its patterns easier.
  */
 public interface AckReplyConsumer {
-  void accept(AckReply ackReply);
+  /**
+   * Acknowledges that the message has been successfully processed. The service will not send the
+   * message again.
+   */
+  void ack();
+  
+  /**
+   * Signals that the message has not been successfully processed. The service should resend the
+   * message.
+   */
+  void nack();
 }

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

@@ -154,6 +154,12 @@ public String toString() {
     }
   }
 
+  /** Internal representation of a reply to a Pubsub message, to be sent back to the service. */
+  public enum AckReply {
+    ACK,
+    NACK
+  }
+
   /**
    * Handles callbacks for acking/nacking messages from the {@link
    * com.google.cloud.pubsub.spi.v1.MessageReceiver}.
@@ -295,8 +301,13 @@ public void processReceivedMessages(List<com.google.pubsub.v1.ReceivedMessage> r
       final AckReplyConsumer consumer =
           new AckReplyConsumer() {
             @Override
-            public void accept(AckReply reply) {
-              response.set(reply);
+            public void ack() {
+              response.set(AckReply.ACK);
+            }
+
+            @Override
+            public void nack() {
+              response.set(AckReply.NACK);
             }
           };
       Futures.addCallback(response, ackHandler);

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

@@ -22,7 +22,8 @@
 public interface MessageReceiver {
   /**
    * Called when a message is received by the subscriber. The implementation must arrange for {@link
-   * AckReplyConsumer#accept} to be called after processing the {@code message}.
+   * AckReplyConsumer#ack()} or {@link
+   * AckReplyConsumer#nack()} to be called after processing the {@code message}.
    *
    * <p>This {@code MessageReceiver} passes all messages to a {@code BlockingQueue}.
    * This method can be called concurrently from multiple threads,
@@ -34,9 +35,9 @@ public interface MessageReceiver {
    * MessageReceiver receiver = new MessageReceiver() {
    *   public void receiveMessage(final PubsubMessage message, final AckReplyConsumer consumer) {
    *     if (blockingQueue.offer(message)) {
-   *       consumer.accept(AckReply.ACK, null);
+   *       consumer.ack();
    *     } else {
-   *       consumer.accept(AckReply.NACK, null);
+   *       consumer.nack();
    *     }
    *   }
    * };

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

@@ -55,9 +55,9 @@
  *
  * <p>A {@link Subscriber} allows you to provide an implementation of a {@link MessageReceiver
  * receiver} to which messages are going to be delivered as soon as they are received by the
- * subscriber. The delivered messages then can be {@link AckReply#ACK acked} or {@link AckReply#NACK
- * nacked} at will as they get processed by the receiver. Nacking a messages implies a later
- * redelivery of such message.
+ * subscriber. The delivered messages then can be {@link AckReplyConsumer#ack() acked} or {@link
+ * AckReplyConsumer#nack() nacked} at will as they get processed by the receiver. Nacking a messages
+ * implies a later redelivery of such message.
  *
  * <p>The subscriber handles the ack management, by automatically extending the ack deadline while
  * the message is being processed, to then issue the ack or nack of such message when the processing

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

@@ -21,7 +21,6 @@
 
 import com.google.api.gax.core.SettableApiFuture;
 import com.google.cloud.ServiceOptions;
-import com.google.cloud.pubsub.spi.v1.AckReply;
 import com.google.cloud.pubsub.spi.v1.AckReplyConsumer;
 import com.google.cloud.pubsub.spi.v1.MessageReceiver;
 import com.google.cloud.pubsub.spi.v1.Publisher;
@@ -116,9 +115,9 @@ public void testPublishSubscribe() throws Exception {
                   public void receiveMessage(
                       final PubsubMessage message, final AckReplyConsumer consumer) {
                     if (received.set(message)) {
-                      consumer.accept(AckReply.ACK);
+                      consumer.ack();
                     } else {
-                      consumer.accept(AckReply.NACK);
+                      consumer.nack();
                     }
                   }
                 })