Changing the AckReplyConsumer interface (#1758)

davidtorres · pongad · commit a2ca9924be8f · 2017-03-21T11:11:21.000+11:00
Changing the AckReplyConsumer interface to comply to just the Java 8 Consumer interface.
It really is not useful to be able to set an exception as ack reply, since the result
is the same as nack, if we ever require another result then we can just
add one more value to the AckReply enum.

Also adding a fail-safe catch so if the receiver ever throws an
exception we will interpret that as a nack and keep going.

Example fixes to comply with changes to the AckReplyConsumer interface.
diff --git a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndPullMessages.java b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndPullMessages.java
@@ -46,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, null);
+            consumer.accept(AckReply.ACK);
           }
         };
     Subscriber subscriber = null;
diff --git a/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/MessageReceiverSnippets.java b/google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/MessageReceiverSnippets.java
@@ -51,9 +51,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, null);
+          consumer.accept(AckReply.ACK);
         } else {
-          consumer.accept(AckReply.NACK, null);
+          consumer.accept(AckReply.NACK);
         }
       }
     };
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/AckReplyConsumer.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/AckReplyConsumer.java
@@ -19,9 +19,9 @@
 /**
  * Accepts a reply, sending it to the service.
  *
- * <p>Both the interface and its method is named after the Java 8's {@code BiConsumer} interface
+ * <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, Throwable t);
+  void accept(AckReply ackReply);
 }
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java
@@ -285,20 +285,20 @@ public void processReceivedMessages(List<com.google.pubsub.v1.ReceivedMessage> r
       final AckReplyConsumer consumer =
           new AckReplyConsumer() {
             @Override
-            public void accept(AckReply reply, Throwable t) {
-              if (reply != null) {
-                response.set(reply);
-              } else {
-                response.setException(t);
-              }
+            public void accept(AckReply reply) {
+              response.set(reply);
             }
           };
       Futures.addCallback(response, ackHandler);
       executor.submit(
           new Runnable() {
             @Override
             public void run() {
-              receiver.receiveMessage(message, consumer);
+              try {
+                receiver.receiveMessage(message, consumer);
+              } catch (Exception e) {
+                response.setException(e);
+              }
             }
           });
     }
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/FakeSubscriberServiceImpl.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/FakeSubscriberServiceImpl.java
@@ -49,7 +49,8 @@
 class FakeSubscriberServiceImpl extends SubscriberImplBase {
   private final AtomicBoolean subscriptionInitialized = new AtomicBoolean(false);
   private String subscription = "";
-  private final AtomicInteger messageAckDeadline = new AtomicInteger(Subscriber.MIN_ACK_DEADLINE_SECONDS);
+  private final AtomicInteger messageAckDeadline =
+      new AtomicInteger(Subscriber.MIN_ACK_DEADLINE_SECONDS);
   private final List<Stream> openedStreams = new ArrayList<>();
   private final List<Stream> closedStreams = new ArrayList<>();
   private final List<String> acks = new ArrayList<>();
@@ -235,7 +236,9 @@ public void getSubscription(
 
   @Override
   public void pull(PullRequest request, StreamObserver<PullResponse> responseObserver) {
-    receivedPullRequest.add(request);
+    synchronized (receivedPullRequest) {
+      receivedPullRequest.add(request);
+    }
     try {
       responseObserver.onNext(pullResponses.take());
       responseObserver.onCompleted();
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java
@@ -84,14 +84,14 @@ static class TestReceiver implements MessageReceiver {
         new LinkedBlockingQueue<>();
     private AckReply ackReply = AckReply.ACK;
     private Optional<CountDownLatch> messageCountLatch = Optional.absent();
-    private Optional<Throwable> error = Optional.absent();
+    private Optional<RuntimeException> error = Optional.absent();
     private boolean explicitAckReplies;
 
     void setReply(AckReply ackReply) {
       this.ackReply = ackReply;
     }
 
-    void setErrorReply(Throwable error) {
+    void setErrorReply(RuntimeException error) {
       this.error = Optional.of(error);
     }
 
@@ -111,18 +111,20 @@ void waitForExpectedMessages() throws InterruptedException {
 
     @Override
     public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
-      if (explicitAckReplies) {
-        try {
-          outstandingMessageReplies.put(consumer);
-        } catch (InterruptedException e) {
-          throw new IllegalStateException(e);
+      try {
+        if (explicitAckReplies) {
+          try {
+            outstandingMessageReplies.put(consumer);
+          } catch (InterruptedException e) {
+            throw new IllegalStateException(e);
+          }
+        } else {
+          replyTo(consumer);
+        }
+      } finally {
+        if (messageCountLatch.isPresent()) {
+          messageCountLatch.get().countDown();
         }
-      } else {
-        replyTo(consumer);
-      }
-
-      if (messageCountLatch.isPresent()) {
-        messageCountLatch.get().countDown();
       }
     }
 
@@ -145,9 +147,9 @@ public void replyAllOutstandingMessage() {
 
     private void replyTo(AckReplyConsumer reply) {
       if (error.isPresent()) {
-        reply.accept(null, error.get());
+        throw error.get();
       } else {
-        reply.accept(ackReply, null);
+        reply.accept(ackReply);
       }
     }
   }
@@ -207,7 +209,7 @@ public void testNackSingleMessage() throws Exception {
 
   @Test
   public void testReceiverError_NacksMessage() throws Exception {
-    testReceiver.setErrorReply(new Exception("Can't process message"));
+    testReceiver.setErrorReply(new RuntimeException("Can't process message"));
 
     Subscriber subscriber = startSubscriber(getTestSubscriberBuilder(testReceiver));
 

Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ public static void main(String... args) throws Exception {`
`46`	`46`	`@Override`
`47`	`47`	`public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {`
`48`	`48`	`System.out.println("got message: " + message.getData().toStringUtf8());`
`49`		`- consumer.accept(AckReply.ACK, null);`
	`49`	`+ consumer.accept(AckReply.ACK);`
`50`	`50`	`}`
`51`	`51`	`};`
`52`	`52`	`Subscriber subscriber = null;`
Original file line number	Diff line number	Diff line change
`@@ -51,9 +51,9 @@ public MessageReceiver messageReceiver() {`
`51`	`51`	`MessageReceiver receiver = new MessageReceiver() {`
`52`	`52`	`public void receiveMessage(final PubsubMessage message, final AckReplyConsumer consumer) {`
`53`	`53`	`if (blockingQueue.offer(message)) {`
`54`		`- consumer.accept(AckReply.ACK, null);`
	`54`	`+ consumer.accept(AckReply.ACK);`
`55`	`55`	`} else {`
`56`		`- consumer.accept(AckReply.NACK, null);`
	`56`	`+ consumer.accept(AckReply.NACK);`
`57`	`57`	`}`
`58`	`58`	`}`
`59`	`59`	`};`
Original file line number	Diff line number	Diff line change
`@@ -19,9 +19,9 @@`
`19`	`19`	`/**`
`20`	`20`	`* Accepts a reply, sending it to the service.`
`21`	`21`	`*`
`22`		`- * <p>Both the interface and its method is named after the Java 8's {@code BiConsumer} interface`
	`22`	`+ * <p>Both the interface and its method is named after the Java 8's {@code Consumer} interface`
`23`	`23`	`* to make migration to Java 8 and adopting its patterns easier.`
`24`	`24`	`*/`
`25`	`25`	`public interface AckReplyConsumer {`
`26`		`- void accept(AckReply ackReply, Throwable t);`
	`26`	`+ void accept(AckReply ackReply);`
`27`	`27`	`}`