---

pongad · web-flow · commit 03870222381c · 2017-05-17T20:36:15.000+10:00
yaml --- r: 5697 b: refs/heads/master c: 5af4a46 h: refs/heads/master i: 5695: 67eb9ab
diff --git a/[refs] b/[refs]
@@ -1,5 +1,5 @@
 ---
-refs/heads/master: fdea40dce6514ac3dbb9687ed183afbefaf9d4d3
+refs/heads/master: 5af4a462bfc42a07e02966d51ddc55e22df94207
 refs/heads/travis: dae77e558b884bc1b165155482d76c8e40b0fca4
 refs/heads/gh-pages: f8ea70cdc599a5d39c2df480280877afb3bef9bd
 refs/tags/0.0.9: 22f1839238f66c39e67ed4dfdcd273b1ae2e8444
diff --git a/trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java b/trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java
@@ -99,11 +99,6 @@ public PollingSubscriberConnection(
   @Override
   protected void doStart() {
     logger.config("Starting subscriber.");
-    initialize();
-    notifyStarted();
-  }
-
-  private void initialize() {
     ListenableFuture<Subscription> subscriptionInfo =
         stub.withDeadlineAfter(DEFAULT_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS)
             .getSubscription(
@@ -116,6 +111,7 @@ private void initialize() {
           public void onSuccess(Subscription result) {
             messageDispatcher.setMessageDeadlineSeconds(result.getAckDeadlineSeconds());
             pullMessages(INITIAL_BACKOFF);
+            notifyStarted();
           }
 
           @Override
diff --git a/trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java b/trunk/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java
@@ -39,7 +39,6 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -244,10 +243,23 @@ public void close() {
       throw new IllegalStateException(e);
     }
 
-    // Streaming pull is not enabled on the service yet.
-    // startStreamingConnections();
-    startPollingConnections();
-    notifyStarted();
+    // When started, connections submit tasks to the executor.
+    // These tasks must finish before the connections can declare themselves running.
+    // If we have a single-thread executor and call startPollingConnections from the
+    // same executor, it will deadlock: the thread will be stuck waiting for connections
+    // to start but cannot start the connections.
+    // For this reason, we spawn a dedicated thread. Starting subscriber should be rare.
+    new Thread(new Runnable() {
+      @Override
+      public void run() {
+        try {
+          startPollingConnections();
+          notifyStarted();
+        } catch (Throwable t) {
+          notifyFailed(t);
+        }
+      }
+    }).start();
   }
 
   @Override
@@ -387,25 +399,11 @@ private void stopAllPollingConnections() {
 
   private void startConnections(
       List<? extends ApiService> connections, final ApiService.Listener connectionsListener) {
-    final CountDownLatch subscribersStarting = new CountDownLatch(numChannels);
-    for (final ApiService subscriber : connections) {
-      executor.submit(
-          new Runnable() {
-            @Override
-            public void run() {
-              subscriber.addListener(connectionsListener, executor);
-              try {
-                subscriber.startAsync().awaitRunning();
-              } finally {
-                subscribersStarting.countDown();
-              }
-            }
-          });
-    }
-    try {
-      subscribersStarting.await();
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
+    for (ApiService subscriber : connections) {
+      subscriber.addListener(connectionsListener, executor);
+      // Starting each connection submits a blocking task to the executor.
+      // We start connections one at a time to avoid swamping executor with blocking tasks.
+      subscriber.startAsync().awaitRunning();
     }
   }
 
@@ -415,26 +413,17 @@ private void stopConnections(List<? extends ApiService> connections) {
       liveConnections = new ArrayList<ApiService>(connections);
       connections.clear();
     }
-    final CountDownLatch connectionsStopping = new CountDownLatch(liveConnections.size());
-    for (final ApiService subscriberConnection : liveConnections) {
-      executor.submit(
-          new Runnable() {
-            @Override
-            public void run() {
-              try {
-                subscriberConnection.stopAsync().awaitTerminated();
-              } catch (IllegalStateException ignored) {
-                // It is expected for some connections to be already in state failed so stop will
-                // throw this expection.
-              }
-              connectionsStopping.countDown();
-            }
-          });
+    for (ApiService subscriber : liveConnections) {
+      subscriber.stopAsync();
     }
-    try {
-      connectionsStopping.await();
-    } catch (InterruptedException e) {
-      throw new IllegalStateException(e);
+    for (ApiService subscriber : liveConnections) {
+      try {
+        subscriber.awaitTerminated();
+      } catch (IllegalStateException e) {
+        // If the service fails, awaitTerminated will throw an exception.
+        // However, we could be stopping services because at least one
+        // has already failed, so we just ignore this exception.
+      }
     }
   }
 
