Rename ExecutorProvider to ExecutorFactory, refactor shutodown and serialization

Author: mziccard

gcloud-java-core/src/main/java/com/google/cloud/GrpcServiceOptions.java

@@ -16,12 +16,17 @@
 
 package com.google.cloud;
 
+import static com.google.common.base.MoreObjects.firstNonNull;
+
 import com.google.cloud.spi.ServiceRpcFactory;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 
 import io.grpc.internal.SharedResourceHolder;
 import io.grpc.internal.SharedResourceHolder.Resource;
 
+import java.io.IOException;
+import java.io.ObjectInputStream;
 import java.util.Objects;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -40,10 +45,13 @@ public abstract class GrpcServiceOptions<ServiceT extends Service<OptionsT>, Ser
     extends ServiceOptions<ServiceT, ServiceRpcT, OptionsT> {
 
   private static final long serialVersionUID = 6415982522610509549L;
+  private final String executorFactoryClassName;
   private final int initialTimeout;
   private final double timeoutMultiplier;
   private final int maxTimeout;
 
+  private transient ExecutorFactory executorFactory;
+
   /**
    * Shared thread pool executor.
    */
@@ -65,45 +73,41 @@ public void close(ScheduledExecutorService instance) {
       };
 
   /**
-   * An interface that provides access to a scheduled executor service.
+   * An interface for {@link ScheduledExecutorService} factories. Implementations of this interface
+   * can be used to provide an user-defined scheduled executor to execute requests. Any
+   * implementation of this interface must override the {@code get()} method to return the desired
+   * executor. The {@code release(executor)} method should be overriden to free resources used by
+   * the executor (if needed) according to application's logic.
+   *
+   * <p>Implementation must provide a public no-arg constructor. Loading of a factory implementation
+   * is done via {@link java.util.ServiceLoader}.
    */
-  public interface ExecutorProvider {
+  public interface ExecutorFactory {
 
     /**
-     * Returns the scheduled executor service.
+     * Gets a scheduled executor service instance.
      */
     ScheduledExecutorService get();
 
     /**
-     * Shuts down the scheduled executor service if it is no longer used.
+     * Releases resources used by the executor and possibly shuts it down.
      */
-    void shutdown();
+    void release(ScheduledExecutorService executor);
   }
 
-  /**
-   * An interface that provides access to a scheduled executor service.
-   */
-  private static class DefaultExecutorProvider implements ExecutorProvider {
-
-    private ScheduledExecutorService service;
-    private boolean shutdown = false;
+  @VisibleForTesting
+  static class DefaultExecutorFactory implements ExecutorFactory {
 
-    private DefaultExecutorProvider() {}
+    private static final DefaultExecutorFactory INSTANCE = new DefaultExecutorFactory();
 
     @Override
-    public synchronized ScheduledExecutorService get() {
-      if (service == null && !shutdown) {
-        service = SharedResourceHolder.get(EXECUTOR);
-      }
-      return service;
+    public ScheduledExecutorService get() {
+      return SharedResourceHolder.get(EXECUTOR);
     }
 
     @Override
-    public synchronized void shutdown() {
-      if (service != null && !shutdown) {
-        shutdown = true;
-        service = SharedResourceHolder.release(EXECUTOR, service);
-      }
+    public synchronized void release(ScheduledExecutorService executor) {
+      SharedResourceHolder.release(EXECUTOR, executor);
     }
   }
 
@@ -120,6 +124,7 @@ protected abstract static class Builder<ServiceT extends Service<OptionsT>, Serv
       B extends Builder<ServiceT, ServiceRpcT, OptionsT, B>>
       extends ServiceOptions.Builder<ServiceT, ServiceRpcT, OptionsT, B> {
 
+    private ExecutorFactory executorFactory;
     private int initialTimeout = 20_000;
     private double timeoutMultiplier = 1.5;
     private int maxTimeout = 100_000;
@@ -128,6 +133,7 @@ protected Builder() {}
 
     protected Builder(GrpcServiceOptions<ServiceT, ServiceRpcT, OptionsT> options) {
       super(options);
+      executorFactory = options.executorFactory;
       initialTimeout = options.initialTimeout;
       timeoutMultiplier = options.timeoutMultiplier;
       maxTimeout = options.maxTimeout;
@@ -136,6 +142,17 @@ protected Builder(GrpcServiceOptions<ServiceT, ServiceRpcT, OptionsT> options) {
     @Override
     protected abstract GrpcServiceOptions<ServiceT, ServiceRpcT, OptionsT> build();
 
+    /**
+     * Sets the scheduled executor factory. This method can be used to provide an user-defined
+     * scheduled executor to execute requests.
+     *
+     * @return the builder
+     */
+    public B executorFactory(ExecutorFactory executorFactory) {
+      this.executorFactory = executorFactory;
+      return self();
+    }
+
     /**
      * Sets the timeout for the initial RPC, in milliseconds. Subsequent calls will use this value
      * adjusted according to {@link #timeoutMultiplier(double)}. Default value is 20000.
@@ -180,6 +197,9 @@ protected GrpcServiceOptions(
       Class<? extends ServiceRpcFactory<ServiceRpcT, OptionsT>> rpcFactoryClass, Builder<ServiceT,
       ServiceRpcT, OptionsT, ?> builder) {
     super(serviceFactoryClass, rpcFactoryClass, builder);
+    executorFactory = firstNonNull(builder.executorFactory,
+        getFromServiceLoader(ExecutorFactory.class, DefaultExecutorFactory.INSTANCE));
+    executorFactoryClassName = executorFactory.getClass().getName();
     initialTimeout = builder.initialTimeout;
     timeoutMultiplier = builder.timeoutMultiplier;
     maxTimeout = builder.maxTimeout <= initialTimeout ? initialTimeout : builder.maxTimeout;
@@ -188,8 +208,8 @@ protected GrpcServiceOptions(
   /**
    * Returns a scheduled executor service provider.
    */
-  protected ExecutorProvider executorProvider() {
-    return new DefaultExecutorProvider();
+  protected ExecutorFactory executorFactory() {
+    return executorFactory;
   }
 
   /**
@@ -217,13 +237,20 @@ public int maxTimeout() {
 
   @Override
   protected int baseHashCode() {
-    return Objects.hash(super.baseHashCode(), initialTimeout, timeoutMultiplier, maxTimeout);
+    return Objects.hash(super.baseHashCode(), executorFactoryClassName, initialTimeout,
+        timeoutMultiplier, maxTimeout);
   }
 
   protected boolean baseEquals(GrpcServiceOptions<?, ?, ?> other) {
     return super.baseEquals(other)
+        && Objects.equals(executorFactoryClassName, other.executorFactoryClassName)
         && Objects.equals(initialTimeout, other.initialTimeout)
         && Objects.equals(timeoutMultiplier, other.timeoutMultiplier)
         && Objects.equals(maxTimeout, other.maxTimeout);
   }
+
+  private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException {
+    input.defaultReadObject();
+    executorFactory = newInstance(executorFactoryClassName);
+  }
 }

gcloud-java-core/src/test/java/com/google/cloud/GrpcServiceOptionsTest.java

@@ -18,21 +18,30 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import com.google.cloud.GrpcServiceOptions.DefaultExecutorFactory;
+import com.google.cloud.GrpcServiceOptions.ExecutorFactory;
 import com.google.cloud.spi.ServiceRpcFactory;
 
+import org.easymock.EasyMock;
 import org.junit.Test;
 
 import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
 
 public class GrpcServiceOptionsTest {
 
+  private static final ExecutorFactory MOCK_EXECUTOR_FACTORY =
+      EasyMock.createMock(ExecutorFactory.class);
   private static final TestGrpcServiceOptions OPTIONS = TestGrpcServiceOptions.builder()
       .projectId("project-id")
       .initialTimeout(1234)
       .timeoutMultiplier(1.6)
       .maxTimeout(5678)
+      .executorFactory(MOCK_EXECUTOR_FACTORY)
       .build();
   private static final TestGrpcServiceOptions DEFAULT_OPTIONS =
       TestGrpcServiceOptions.builder().projectId("project-id").build();
@@ -138,9 +147,11 @@ public void testBuilder() {
     assertEquals(1234, OPTIONS.initialTimeout());
     assertEquals(1.6, OPTIONS.timeoutMultiplier(), 0.0);
     assertEquals(5678, OPTIONS.maxTimeout());
+    assertSame(MOCK_EXECUTOR_FACTORY, OPTIONS.executorFactory());
     assertEquals(20000, DEFAULT_OPTIONS.initialTimeout());
     assertEquals(1.5, DEFAULT_OPTIONS.timeoutMultiplier(), 0.0);
     assertEquals(100000, DEFAULT_OPTIONS.maxTimeout());
+    assertTrue(DEFAULT_OPTIONS.executorFactory() instanceof DefaultExecutorFactory);
   }
 
   @Test
@@ -182,11 +193,26 @@ public void testBuilderInvalidMaxTimeout() {
   public void testBaseEquals() {
     assertEquals(OPTIONS, OPTIONS_COPY);
     assertNotEquals(DEFAULT_OPTIONS, OPTIONS);
+    TestGrpcServiceOptions options = OPTIONS.toBuilder()
+        .executorFactory(new DefaultExecutorFactory())
+        .build();
+    assertNotEquals(OPTIONS, options);
   }
 
   @Test
   public void testBaseHashCode() {
     assertEquals(OPTIONS.hashCode(), OPTIONS_COPY.hashCode());
     assertNotEquals(DEFAULT_OPTIONS.hashCode(), OPTIONS.hashCode());
+    TestGrpcServiceOptions options = OPTIONS.toBuilder()
+        .executorFactory(new DefaultExecutorFactory())
+        .build();
+    assertNotEquals(OPTIONS.hashCode(), options.hashCode());
+  }
+
+  @Test
+  public void testDefaultExecutorFactory() {
+    ExecutorFactory executorFactory = new DefaultExecutorFactory();
+    ScheduledExecutorService executorService = executorFactory.get();
+    assertSame(executorService, executorFactory.get());
   }
 }

gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/spi/DefaultPubSubRpc.java

@@ -21,7 +21,7 @@
 import com.google.api.gax.grpc.ApiException;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.AuthCredentials;
-import com.google.cloud.GrpcServiceOptions.ExecutorProvider;
+import com.google.cloud.GrpcServiceOptions.ExecutorFactory;
 import com.google.cloud.RetryParams;
 import com.google.cloud.pubsub.PubSubException;
 import com.google.cloud.pubsub.PubSubOptions;
@@ -64,12 +64,14 @@
 import java.io.IOException;
 import java.util.Set;
 import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
 
 public class DefaultPubSubRpc implements PubSubRpc {
 
   private final PublisherApi publisherApi;
   private final SubscriberApi subscriberApi;
-  private final ExecutorProvider executorProvider;
+  private final ScheduledExecutorService executor;
+  private final ExecutorFactory executorFactory;
 
   private static final class InternalPubSubOptions extends PubSubOptions {
 
@@ -80,18 +82,19 @@ private InternalPubSubOptions(PubSubOptions options) {
     }
 
     @Override
-    protected ExecutorProvider executorProvider() {
-      return super.executorProvider();
+    protected ExecutorFactory executorFactory() {
+      return super.executorFactory();
     }
   }
 
   public DefaultPubSubRpc(PubSubOptions options) throws IOException {
-    executorProvider = new InternalPubSubOptions(options).executorProvider();
+    executorFactory = new InternalPubSubOptions(options).executorFactory();
+    executor = executorFactory.get();
     try {
       PublisherSettings.Builder pubBuilder =
-          PublisherSettings.defaultBuilder().provideExecutorWith(executorProvider.get(), false);
+          PublisherSettings.defaultBuilder().provideExecutorWith(executor, false);
       SubscriberSettings.Builder subBuilder =
-          SubscriberSettings.defaultBuilder().provideExecutorWith(executorProvider.get(), false);
+          SubscriberSettings.defaultBuilder().provideExecutorWith(executor, false);
       // todo(mziccard): PublisherSettings should support null/absent credentials for testing
       if (options.host().contains("localhost")
           || options.authCredentials().equals(AuthCredentials.noAuth())) {
@@ -233,6 +236,6 @@ public Future<Empty> modify(ModifyPushConfigRequest request) {
   public void close() throws Exception {
     subscriberApi.close();
     publisherApi.close();
-    executorProvider.shutdown();
+    executorFactory.release(executor);
   }
 }