Refactor API call and connection settings for gRPC services

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

@@ -135,30 +135,30 @@ public final void copyTo(byte[] target) {
   /**
    * Creates a {@code ByteArray} object given an array of bytes. The bytes are copied.
    */
-  public final static ByteArray copyFrom(byte[] bytes) {
+  public static final ByteArray copyFrom(byte[] bytes) {
     return new ByteArray(ByteString.copyFrom(bytes));
   }
 
   /**
    * Creates a {@code ByteArray} object given a string. The string is encoded in {@code UTF-8}. The
    * bytes are copied.
    */
-  public final static ByteArray copyFrom(String string) {
+  public static final ByteArray copyFrom(String string) {
     return new ByteArray(ByteString.copyFrom(string, StandardCharsets.UTF_8));
   }
 
   /**
    * Creates a {@code ByteArray} object given a {@link ByteBuffer}. The bytes are copied.
    */
-  public final static ByteArray copyFrom(ByteBuffer bytes) {
+  public static final ByteArray copyFrom(ByteBuffer bytes) {
     return new ByteArray(ByteString.copyFrom(bytes));
   }
 
   /**
    * Creates a {@code ByteArray} object given an {@link InputStream}. The stream is read into the
    * created object.
    */
-  public final static ByteArray copyFrom(InputStream input) throws IOException {
+  public static final ByteArray copyFrom(InputStream input) throws IOException {
     return new ByteArray(ByteString.readFrom(input));
   }
 }

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

@@ -18,13 +18,20 @@
 
 import static com.google.common.base.MoreObjects.firstNonNull;
 
+import com.google.api.gax.core.ConnectionSettings;
+import com.google.api.gax.core.RetrySettings;
+import com.google.api.gax.grpc.ApiCallSettings;
+import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.spi.ServiceRpcFactory;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.google.common.net.HostAndPort;
 
 import io.grpc.internal.SharedResourceHolder;
 import io.grpc.internal.SharedResourceHolder.Resource;
 
+import org.joda.time.Duration;
+
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.util.Objects;
@@ -160,8 +167,8 @@ public B executorFactory(ExecutorFactory<ScheduledExecutorService> executorFacto
      * 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.
      *
-     * @throws IllegalArgumentException if the provided timeout is &lt; 0
      * @return the builder
+     * @throws IllegalArgumentException if the provided timeout is &lt; 0
      */
     public B initialTimeout(int initialTimeout) {
       Preconditions.checkArgument(initialTimeout > 0, "Initial timeout must be > 0");
@@ -173,8 +180,8 @@ public B initialTimeout(int initialTimeout) {
      * Sets the timeout multiplier. This value is used to compute the timeout for a retried RPC.
      * Timeout is computed as {@code timeoutMultiplier * previousTimeout}. Default value is 1.5.
      *
-     * @throws IllegalArgumentException if the provided timeout multiplier is &lt; 0
      * @return the builder
+     * @throws IllegalArgumentException if the provided timeout multiplier is &lt; 0
      */
     public B timeoutMultiplier(double timeoutMultiplier) {
       Preconditions.checkArgument(timeoutMultiplier >= 1.0, "Timeout multiplier must be >= 1");
@@ -216,6 +223,38 @@ protected ExecutorFactory<ScheduledExecutorService> executorFactory() {
     return executorFactory;
   }
 
+  /**
+   * Returns a builder for API call settings.
+   */
+  protected ApiCallSettings.Builder apiCallSettings() {
+    // todo(mziccard): specify timeout these settings:
+    // retryParams().retryMaxAttempts(), retryParams().retryMinAttempts()
+    final RetrySettings.Builder builder = RetrySettings.newBuilder()
+        .setTotalTimeout(Duration.millis(retryParams().totalRetryPeriodMillis()))
+        .setInitialRpcTimeout(Duration.millis(initialTimeout()))
+        .setRpcTimeoutMultiplier(timeoutMultiplier())
+        .setMaxRpcTimeout(Duration.millis(maxTimeout()))
+        .setInitialRetryDelay(Duration.millis(retryParams().initialRetryDelayMillis()))
+        .setRetryDelayMultiplier(retryParams().retryDelayBackoffFactor())
+        .setMaxRetryDelay(Duration.millis(retryParams().maxRetryDelayMillis()));
+    return ApiCallSettings.newBuilder().setRetrySettingsBuilder(builder);
+  }
+
+  /**
+   * Returns a builder for connection-related settings.
+   */
+  protected ConnectionSettings.Builder connectionSettings() {
+    HostAndPort hostAndPort = HostAndPort.fromString(host());
+    ConnectionSettings.Builder builder = ConnectionSettings.newBuilder()
+        .setServiceAddress(hostAndPort.getHostText())
+        .setPort(hostAndPort.getPort());
+    GoogleCredentials credentials = authCredentials().credentials();
+    if (credentials != null) {
+      builder.provideCredentialsWith(credentials.createScoped(scopes()));
+    }
+    return builder;
+  }
+
   /**
    * Returns the timeout for the initial RPC, in milliseconds. Subsequent calls will use this value
    * adjusted according to {@link #timeoutMultiplier()}. Default value is 20000.

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

@@ -39,13 +39,13 @@ public class MonitoredResourceDescriptor implements Serializable {
   private static final long serialVersionUID = -3702077512777687441L;
   public static final Function<com.google.api.MonitoredResourceDescriptor,
       MonitoredResourceDescriptor> FROM_PB_FUNCTION =
-          new Function<com.google.api.MonitoredResourceDescriptor, MonitoredResourceDescriptor>() {
-            @Override
-            public MonitoredResourceDescriptor apply(
-                com.google.api.MonitoredResourceDescriptor pb) {
-              return fromPb(pb);
-            }
-          };
+      new Function<com.google.api.MonitoredResourceDescriptor, MonitoredResourceDescriptor>() {
+        @Override
+        public MonitoredResourceDescriptor apply(
+            com.google.api.MonitoredResourceDescriptor pb) {
+          return fromPb(pb);
+        }
+      };
 
   private final String type;
   private final String name;

gcloud-java-logging/src/main/java/com/google/cloud/logging/LoggingOptions.java

@@ -20,6 +20,7 @@
 import com.google.cloud.logging.spi.DefaultLoggingRpc;
 import com.google.cloud.logging.spi.LoggingRpc;
 import com.google.cloud.logging.spi.LoggingRpcFactory;
+import com.google.cloud.logging.spi.v2.LoggingServiceV2Settings;
 import com.google.common.collect.ImmutableSet;
 
 import java.io.IOException;
@@ -31,7 +32,8 @@ public class LoggingOptions extends GrpcServiceOptions<Logging, LoggingRpc, Logg
   private static final long serialVersionUID = -2996451684945061075L;
   private static final String LOGGING_SCOPE = "https://www.googleapis.com/auth/logging.admin";
   private static final Set<String> SCOPES = ImmutableSet.of(LOGGING_SCOPE);
-  private static final String DEFAULT_HOST = "https://logging.googleapis.com";
+  private static final String DEFAULT_HOST = LoggingServiceV2Settings.getDefaultServiceAddress()
+      + ':' + LoggingServiceV2Settings.getDefaultServicePort();
 
   public static class DefaultLoggingFactory implements LoggingFactory {
     private static final LoggingFactory INSTANCE = new DefaultLoggingFactory();

gcloud-java-logging/src/main/java/com/google/cloud/logging/spi/DefaultLoggingRpc.java

@@ -18,13 +18,11 @@
 
 import static com.google.common.base.MoreObjects.firstNonNull;
 
-import com.google.api.gax.core.RetrySettings;
+import com.google.api.gax.core.ConnectionSettings;
 import com.google.api.gax.grpc.ApiCallSettings;
 import com.google.api.gax.grpc.ApiException;
-import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.AuthCredentials;
 import com.google.cloud.GrpcServiceOptions.ExecutorFactory;
-import com.google.cloud.RetryParams;
 import com.google.cloud.logging.LoggingException;
 import com.google.cloud.logging.LoggingOptions;
 import com.google.cloud.logging.spi.v2.ConfigServiceV2Api;
@@ -65,8 +63,6 @@
 import io.grpc.netty.NegotiationType;
 import io.grpc.netty.NettyChannelBuilder;
 
-import org.joda.time.Duration;
-
 import java.io.IOException;
 import java.util.Set;
 import java.util.concurrent.Future;
@@ -94,10 +90,21 @@ private InternalLoggingOptions(LoggingOptions options) {
     protected ExecutorFactory<ScheduledExecutorService> executorFactory() {
       return super.executorFactory();
     }
+
+    @Override
+    protected ApiCallSettings.Builder apiCallSettings() {
+      return super.apiCallSettings();
+    }
+
+    @Override
+    protected ConnectionSettings.Builder connectionSettings() {
+      return super.connectionSettings();
+    }
   }
 
   public DefaultLoggingRpc(LoggingOptions options) throws IOException {
-    executorFactory = new InternalLoggingOptions(options).executorFactory();
+    InternalLoggingOptions internalOptions = new InternalLoggingOptions(options);
+    executorFactory = internalOptions.executorFactory();
     executor = executorFactory.get();
     String libraryName = options.libraryName();
     String libraryVersion = firstNonNull(options.libraryVersion(), "");
@@ -121,18 +128,15 @@ public DefaultLoggingRpc(LoggingOptions options) throws IOException {
         logBuilder.provideChannelWith(channel, true);
         metricsBuilder.provideChannelWith(channel, true);
       } else {
-        GoogleCredentials credentials = options.authCredentials().credentials();
-        confBuilder.provideChannelWith(
-            credentials.createScoped(ConfigServiceV2Settings.getDefaultServiceScopes()));
-        logBuilder.provideChannelWith(
-            credentials.createScoped(LoggingServiceV2Settings.getDefaultServiceScopes()));
-        metricsBuilder.provideChannelWith(
-            credentials.createScoped(MetricsServiceV2Settings.getDefaultServiceScopes()));
+        ConnectionSettings connectionSettings = internalOptions.connectionSettings().build();
+        confBuilder.provideChannelWith(connectionSettings);
+        logBuilder.provideChannelWith(connectionSettings);
+        metricsBuilder.provideChannelWith(connectionSettings);
       }
-      ApiCallSettings.Builder callBuilder = apiCallSettings(options);
-      confBuilder.applyToAllApiMethods(callBuilder);
-      logBuilder.applyToAllApiMethods(callBuilder);
-      metricsBuilder.applyToAllApiMethods(callBuilder);
+      ApiCallSettings.Builder callSettingsBuilder = internalOptions.apiCallSettings();
+      confBuilder.applyToAllApiMethods(callSettingsBuilder);
+      logBuilder.applyToAllApiMethods(callSettingsBuilder);
+      metricsBuilder.applyToAllApiMethods(callSettingsBuilder);
       configApi = ConfigServiceV2Api.create(confBuilder.build());
       loggingApi = LoggingServiceV2Api.create(logBuilder.build());
       metricsApi = MetricsServiceV2Api.create(metricsBuilder.build());
@@ -141,21 +145,6 @@ public DefaultLoggingRpc(LoggingOptions options) throws IOException {
     }
   }
 
-  private static ApiCallSettings.Builder apiCallSettings(LoggingOptions options) {
-    // todo(mziccard): specify timeout these settings:
-    // retryParams.retryMaxAttempts(), retryParams.retryMinAttempts()
-    RetryParams retryParams = options.retryParams();
-    final RetrySettings.Builder builder = RetrySettings.newBuilder()
-        .setTotalTimeout(Duration.millis(retryParams.totalRetryPeriodMillis()))
-        .setInitialRpcTimeout(Duration.millis(options.initialTimeout()))
-        .setRpcTimeoutMultiplier(options.timeoutMultiplier())
-        .setMaxRpcTimeout(Duration.millis(options.maxTimeout()))
-        .setInitialRetryDelay(Duration.millis(retryParams.initialRetryDelayMillis()))
-        .setRetryDelayMultiplier(retryParams.retryDelayBackoffFactor())
-        .setMaxRetryDelay(Duration.millis(retryParams.maxRetryDelayMillis()));
-    return ApiCallSettings.newBuilder().setRetrySettingsBuilder(builder);
-  }
-
   private static <V> Future<V> translate(ListenableFuture<V> from, final boolean idempotent,
       int... returnNullOn) {
     final Set<Integer> returnNullOnSet = Sets.newHashSetWithExpectedSize(returnNullOn.length);

gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSubOptions.java

@@ -32,7 +32,8 @@ public class PubSubOptions extends GrpcServiceOptions<PubSub, PubSubRpc, PubSubO
   private static final long serialVersionUID = 5640180400046623305L;
   private static final String PUBSUB_SCOPE = "https://www.googleapis.com/auth/pubsub";
   private static final Set<String> SCOPES = ImmutableSet.of(PUBSUB_SCOPE);
-  private static final String DEFAULT_HOST = PublisherSettings.getDefaultServiceAddress();
+  private static final String DEFAULT_HOST = PublisherSettings.getDefaultServiceAddress()
+      + ':' + PublisherSettings.getDefaultServicePort();
 
   public static class DefaultPubSubFactory implements PubSubFactory {
     private static final PubSubFactory INSTANCE = new DefaultPubSubFactory();

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

@@ -19,13 +19,10 @@
 import static com.google.common.base.MoreObjects.firstNonNull;
 
 import com.google.api.gax.core.ConnectionSettings;
-import com.google.api.gax.core.RetrySettings;
 import com.google.api.gax.grpc.ApiCallSettings;
 import com.google.api.gax.grpc.ApiException;
-import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.AuthCredentials;
 import com.google.cloud.GrpcServiceOptions.ExecutorFactory;
-import com.google.cloud.RetryParams;
 import com.google.cloud.pubsub.PubSubException;
 import com.google.cloud.pubsub.PubSubOptions;
 import com.google.cloud.pubsub.spi.v1.PublisherApi;
@@ -64,8 +61,6 @@
 import io.grpc.netty.NegotiationType;
 import io.grpc.netty.NettyChannelBuilder;
 
-import org.joda.time.Duration;
-
 import java.io.IOException;
 import java.util.Set;
 import java.util.concurrent.Future;
@@ -92,6 +87,16 @@ private InternalPubSubOptions(PubSubOptions options) {
     protected ExecutorFactory<ScheduledExecutorService> executorFactory() {
       return super.executorFactory();
     }
+
+    @Override
+    protected ApiCallSettings.Builder apiCallSettings() {
+      return super.apiCallSettings();
+    }
+
+    @Override
+    protected ConnectionSettings.Builder connectionSettings() {
+      return super.connectionSettings();
+    }
   }
 
   private static final class PullFutureImpl
@@ -119,7 +124,8 @@ public void onFailure(Throwable error) {
   }
 
   public DefaultPubSubRpc(PubSubOptions options) throws IOException {
-    executorFactory = new InternalPubSubOptions(options).executorFactory();
+    InternalPubSubOptions internalOptions = new InternalPubSubOptions(options);
+    executorFactory = internalOptions.executorFactory();
     executor = executorFactory.get();
     String libraryName = options.libraryName();
     String libraryVersion = firstNonNull(options.libraryVersion(), "");
@@ -139,46 +145,20 @@ public DefaultPubSubRpc(PubSubOptions options) throws IOException {
         pubBuilder.provideChannelWith(channel, true);
         subBuilder.provideChannelWith(channel, true);
       } else {
-        GoogleCredentials credentials = options.authCredentials().credentials();
-        ConnectionSettings pubConnectionSettings = ConnectionSettings.newBuilder()
-            .setServiceAddress(options.host())
-            .setPort(PublisherSettings.getDefaultServicePort())
-            .provideCredentialsWith(
-                credentials.createScoped(PublisherSettings.getDefaultServiceScopes()))
-            .build();
-        ConnectionSettings subConnectionSettings = ConnectionSettings.newBuilder()
-            .setServiceAddress(options.host())
-            .setPort(SubscriberSettings.getDefaultServicePort())
-            .provideCredentialsWith(
-                credentials.createScoped(SubscriberSettings.getDefaultServiceScopes()))
-            .build();
-        pubBuilder.provideChannelWith(pubConnectionSettings);
-        subBuilder.provideChannelWith(subConnectionSettings);
+        ConnectionSettings connectionSettings = internalOptions.connectionSettings().build();
+        pubBuilder.provideChannelWith(connectionSettings);
+        subBuilder.provideChannelWith(connectionSettings);
       }
-      pubBuilder.applyToAllApiMethods(apiCallSettings(options));
-      subBuilder.applyToAllApiMethods(apiCallSettings(options));
+      ApiCallSettings.Builder callSettingsBuilder = internalOptions.apiCallSettings();
+      pubBuilder.applyToAllApiMethods(callSettingsBuilder);
+      subBuilder.applyToAllApiMethods(callSettingsBuilder);
       publisherApi = PublisherApi.create(pubBuilder.build());
       subscriberApi = SubscriberApi.create(subBuilder.build());
     } catch (Exception ex) {
       throw new IOException(ex);
     }
   }
 
-  private static ApiCallSettings.Builder apiCallSettings(PubSubOptions options) {
-    // TODO: specify timeout these settings:
-    // retryParams.retryMaxAttempts(), retryParams.retryMinAttempts()
-    RetryParams retryParams = options.retryParams();
-    final RetrySettings.Builder builder = RetrySettings.newBuilder()
-        .setTotalTimeout(Duration.millis(retryParams.totalRetryPeriodMillis()))
-        .setInitialRpcTimeout(Duration.millis(options.initialTimeout()))
-        .setRpcTimeoutMultiplier(options.timeoutMultiplier())
-        .setMaxRpcTimeout(Duration.millis(options.maxTimeout()))
-        .setInitialRetryDelay(Duration.millis(retryParams.initialRetryDelayMillis()))
-        .setRetryDelayMultiplier(retryParams.retryDelayBackoffFactor())
-        .setMaxRetryDelay(Duration.millis(retryParams.maxRetryDelayMillis()));
-    return ApiCallSettings.newBuilder().setRetrySettingsBuilder(builder);
-  }
-
   private static <V> ListenableFuture<V> translate(ListenableFuture<V> from,
       final boolean idempotent, int... returnNullOn) {
     final Set<Integer> returnNullOnSet = Sets.newHashSetWithExpectedSize(returnNullOn.length);