pubsub: stop Subscriber asynchronously (#3144)

It is hard to do this completely async, but we can use a background
thread so it doesn't block user code.

Fixes #3065.
Fixes #3134.

Author: pongad

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

@@ -296,7 +296,7 @@ protected void doStart() {
               @Override
               public void run() {
                 try {
-                    startStreamingConnections();
+                  startStreamingConnections();
                   notifyStarted();
                 } catch (Throwable t) {
                   notifyFailed(t);
@@ -308,16 +308,23 @@ public void run() {
 
   @Override
   protected void doStop() {
-    // stop connection is no-op if connections haven't been started.
-    stopAllStreamingConnections();
-    try {
-      for (AutoCloseable closeable : closeables) {
-        closeable.close();
-      }
-      notifyStopped();
-    } catch (Exception e) {
-      notifyFailed(e);
-    }
+    new Thread(
+            new Runnable() {
+              @Override
+              public void run() {
+                try {
+                  // stop connection is no-op if connections haven't been started.
+                  stopAllStreamingConnections();
+                  for (AutoCloseable closeable : closeables) {
+                    closeable.close();
+                  }
+                  notifyStopped();
+                } catch (Exception e) {
+                  notifyFailed(e);
+                }
+              }
+            })
+        .start();
   }
 
   private void startStreamingConnections() throws IOException {