Make Pub/Sub subscriber example simpler and more idiomatic (GoogleCloudPlatform#1534)

kamalaboulhosn · anguillanneuf · commit e3ca6d6aaabc · 2019-08-06T11:56:32.000-07:00
* Make Pub/Sub subscriber example more idiomatic by acking only once a message has been printed out.

* Add documentation and improve formatting

* Fix comment

* Simplify example to not use a BlockingQueue.
diff --git a/pubsub/cloud-client/src/main/java/com/example/pubsub/SubscriberExample.java b/pubsub/cloud-client/src/main/java/com/example/pubsub/SubscriberExample.java
@@ -24,21 +24,18 @@
 import com.google.cloud.pubsub.v1.Subscriber;
 import com.google.pubsub.v1.ProjectSubscriptionName;
 import com.google.pubsub.v1.PubsubMessage;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingDeque;
 
 public class SubscriberExample {
-
   // use the default project id
   private static final String PROJECT_ID = ServiceOptions.getDefaultProjectId();
 
-  private static final BlockingQueue<PubsubMessage> messages = new LinkedBlockingDeque<>();
-
   static class MessageReceiverExample implements MessageReceiver {
 
     @Override
     public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
-      messages.offer(message);
+      System.out.println(
+          "Message Id: " + message.getMessageId() + " Data: " + message.getData().toStringUtf8());
+      // Ack only after all work for the message is complete.
       consumer.ack();
     }
   }
@@ -55,16 +52,10 @@ public static void main(String... args) throws Exception {
       subscriber =
           Subscriber.newBuilder(subscriptionName, new MessageReceiverExample()).build();
       subscriber.startAsync().awaitRunning();
-      // Continue to listen to messages
-      while (true) {
-        PubsubMessage message = messages.take();
-        System.out.println("Message Id: " + message.getMessageId());
-        System.out.println("Data: " + message.getData().toStringUtf8());
-      }
-    } finally {
-      if (subscriber != null) {
-        subscriber.stopAsync();
-      }
+      // Allow the subscriber to run indefinitely unless an unrecoverable error occurs.
+      subscriber.awaitTerminated();
+    } catch (IllegalStateException e) {
+      System.out.println("Subscriber unexpectedly stopped: " + e);
     }
   }
 }
