use ExecutorProvider

Use ExecutorProvider instead of asking of an Executor directly.
This makes Publisher and Subscriber consistent with toolkit-generated
code.

This commit leaves Channels and Credentials alone for now,
since it is already getting big.

Author: pongad

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

@@ -19,6 +19,8 @@
 import com.google.api.gax.bundling.FlowController;
 import com.google.api.gax.core.RetrySettings;
 import com.google.api.gax.grpc.BundlingSettings;
+import com.google.api.gax.grpc.ExecutorProvider;
+import com.google.api.gax.grpc.InstantiatingExecutorProvider;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.common.base.Optional;
@@ -28,7 +30,6 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.pubsub.v1.PublishRequest;
 import com.google.pubsub.v1.PublishResponse;
 import com.google.pubsub.v1.PublisherGrpc;
@@ -43,10 +44,10 @@
 import io.grpc.netty.NegotiationType;
 import io.grpc.netty.NettyChannelBuilder;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadLocalRandom;
@@ -118,8 +119,6 @@ public static long getApiMaxBundleBytes() {
     return 10L * 1000L * 1000L; // 10 megabytes (https://en.wikipedia.org/wiki/Megabyte)
   }
 
-  private static final int DEFAULT_MIN_THREAD_POOL_SIZE = 5;
-
   private static final Logger logger = LoggerFactory.getLogger(Publisher.class);
 
   private final String topic;
@@ -143,6 +142,7 @@ public static long getApiMaxBundleBytes() {
 
   private final ScheduledExecutorService executor;
   private final AtomicBoolean shutdown;
+  private final List<AutoCloseable> closeables = new ArrayList<>();
   private final MessagesWaiter messagesWaiter;
   private ScheduledFuture<?> currentAlarmFuture;
 
@@ -160,15 +160,16 @@ private Publisher(Builder builder) throws IOException {
     messagesBundleLock = new ReentrantLock();
     activeAlarm = new AtomicBoolean(false);
     int numCores = Math.max(1, Runtime.getRuntime().availableProcessors());
-    executor =
-        builder.executor.isPresent()
-            ? builder.executor.get()
-            : Executors.newScheduledThreadPool(
-                numCores * DEFAULT_MIN_THREAD_POOL_SIZE,
-                new ThreadFactoryBuilder()
-                    .setDaemon(true)
-                    .setNameFormat("cloud-pubsub-publisher-thread-%d")
-                    .build());
+    executor = builder.executorProvider.getExecutor();
+    if (builder.executorProvider.shouldAutoClose()) {
+      closeables.add(
+          new AutoCloseable() {
+            @Override
+            public void close() throws IOException {
+              executor.shutdown();
+            }
+          });
+    }
     channels = new Channel[numCores];
     channelIndex = new AtomicRoundRobin(channels.length);
     for (int i = 0; i < numCores; i++) {
@@ -480,7 +481,7 @@ public PublisherStats getStats() {
    * should be invoked prior to deleting the {@link Publisher} object in order to ensure that no
    * pending messages are lost.
    */
-  public void shutdown() {
+  public void shutdown() throws Exception {
     if (shutdown.getAndSet(true)) {
       throw new IllegalStateException("Cannot shut down a publisher already shut-down.");
     }
@@ -489,6 +490,9 @@ public void shutdown() {
     }
     publishAllOutstanding();
     messagesWaiter.waitNoMessages();
+    for (AutoCloseable closeable : closeables) {
+      closeable.close();
+    }
   }
 
   private boolean hasBundlingBytes() {
@@ -550,6 +554,12 @@ public static final class Builder {
             .setMaxRpcTimeout(DEFAULT_RPC_TIMEOUT)
             .build();
 
+    private static final int THREADS_PER_CPU = 5;
+    static final ExecutorProvider DEFAULT_EXECUTOR_PROVIDER =
+        InstantiatingExecutorProvider.newBuilder()
+            .setExecutorThreadCount(THREADS_PER_CPU * Runtime.getRuntime().availableProcessors())
+            .build();
+
     String topic;
 
     // Bundling options
@@ -566,7 +576,7 @@ public static final class Builder {
     Optional<ManagedChannelBuilder<? extends ManagedChannelBuilder<?>>> channelBuilder =
         Optional.absent();
 
-    Optional<ScheduledExecutorService> executor = Optional.absent();
+    ExecutorProvider executorProvider = DEFAULT_EXECUTOR_PROVIDER;
 
     /** Constructs a new {@link Builder} using the given topic. */
     public static Builder newBuilder(TopicName topic) {
@@ -656,8 +666,8 @@ public Builder setRetrySettings(RetrySettings retrySettings) {
     }
 
     /** Gives the ability to set a custom executor to be used by the library. */
-    public Builder setExecutor(ScheduledExecutorService executor) {
-      this.executor = Optional.of(Preconditions.checkNotNull(executor));
+    public Builder setExecutorProvider(ExecutorProvider executorProvider) {
+      this.executorProvider = Preconditions.checkNotNull(executorProvider);
       return this;
     }
 

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

@@ -17,6 +17,8 @@
 package com.google.cloud.pubsub.spi.v1;
 
 import com.google.api.gax.bundling.FlowController;
+import com.google.api.gax.grpc.ExecutorProvider;
+import com.google.api.gax.grpc.InstantiatingExecutorProvider;
 import com.google.api.stats.Distribution;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
@@ -28,7 +30,6 @@
 import com.google.common.primitives.Ints;
 import com.google.common.util.concurrent.AbstractService;
 import com.google.common.util.concurrent.Service;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.pubsub.v1.SubscriptionName;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Status;
@@ -40,7 +41,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -131,6 +131,7 @@ public class Subscriber extends AbstractService {
   private final List<StreamingSubscriberConnection> streamingSubscriberConnections;
   private final List<PollingSubscriberConnection> pollingSubscriberConnections;
   private final Clock clock;
+  private final List<AutoCloseable> closeables = new ArrayList<>();
   private ScheduledFuture<?> ackDeadlineUpdater;
   private int streamAckDeadlineSeconds;
 
@@ -147,16 +148,16 @@ private Subscriber(Builder builder) throws IOException {
 
     flowController = new FlowController(builder.flowControlSettings, false);
 
-    numChannels = Math.max(1, Runtime.getRuntime().availableProcessors()) * CHANNELS_PER_CORE;
-    executor =
-        builder.executor.isPresent()
-            ? builder.executor.get()
-            : Executors.newScheduledThreadPool(
-                numChannels * THREADS_PER_CHANNEL,
-                new ThreadFactoryBuilder()
-                    .setDaemon(true)
-                    .setNameFormat("cloud-pubsub-subscriber-thread-%d")
-                    .build());
+    executor = builder.executorProvider.getExecutor();
+    if (builder.executorProvider.shouldAutoClose()) {
+      closeables.add(
+          new AutoCloseable() {
+            @Override
+            public void close() throws IOException {
+              executor.shutdown();
+            }
+          });
+    }
 
     channelBuilder =
         builder.channelBuilder.isPresent()
@@ -176,6 +177,7 @@ private Subscriber(Builder builder) throws IOException {
             : GoogleCredentials.getApplicationDefault()
                 .createScoped(SubscriberSettings.getDefaultServiceScopes());
 
+    numChannels = Math.max(1, Runtime.getRuntime().availableProcessors()) * CHANNELS_PER_CORE;
     streamingSubscriberConnections = new ArrayList<StreamingSubscriberConnection>(numChannels);
     pollingSubscriberConnections = new ArrayList<PollingSubscriberConnection>(numChannels);
   }
@@ -191,7 +193,14 @@ protected void doStart() {
   protected void doStop() {
     stopAllStreamingConnections();
     stopAllPollingConnections();
-    notifyStopped();
+    try {
+      for (AutoCloseable closeable : closeables) {
+        closeable.close();
+      }
+      notifyStopped();
+    } catch (Exception e) {
+      notifyFailed(e);
+    }
   }
 
   private void startStreamingConnections() {
@@ -379,6 +388,14 @@ public static final class Builder {
     private static final Duration MIN_ACK_EXPIRATION_PADDING = Duration.millis(100);
     private static final Duration DEFAULT_ACK_EXPIRATION_PADDING = Duration.millis(500);
 
+    static final ExecutorProvider DEFAULT_EXECUTOR_PROVIDER =
+        InstantiatingExecutorProvider.newBuilder()
+            .setExecutorThreadCount(
+                THREADS_PER_CHANNEL
+                    * CHANNELS_PER_CORE
+                    * Runtime.getRuntime().availableProcessors())
+            .build();
+
     String subscription;
     Optional<Credentials> credentials = Optional.absent();
     MessageReceiver receiver;
@@ -387,7 +404,7 @@ public static final class Builder {
 
     FlowController.Settings flowControlSettings = FlowController.Settings.DEFAULT;
 
-    Optional<ScheduledExecutorService> executor = Optional.absent();
+    ExecutorProvider executorProvider = DEFAULT_EXECUTOR_PROVIDER;
     Optional<ManagedChannelBuilder<? extends ManagedChannelBuilder<?>>> channelBuilder =
         Optional.absent();
     Optional<Clock> clock = Optional.absent();
@@ -459,8 +476,8 @@ public Builder setAckExpirationPadding(Duration ackExpirationPadding) {
     }
 
     /** Gives the ability to set a custom executor. */
-    public Builder setExecutor(ScheduledExecutorService executor) {
-      this.executor = Optional.of(executor);
+    public Builder setExecutorProvider(ExecutorProvider executorProvider) {
+      this.executorProvider = Preconditions.checkNotNull(executorProvider);
       return this;
     }