add Publisher.Settings

pongad · pongad · commit 2beb34efab4e · 2017-01-09T16:04:04.000+11:00
This class is used to configure Publisher.
User-visible options are made public.
Options that should be automatically populated,
like user credentials,
are left package-private for now.
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PubSub.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PubSub.java
@@ -221,7 +221,7 @@ public static ListOption pageToken(String pageToken) {
    */
   Future<AsyncPage<Topic>> listTopicsAsync(ListOption... options);
 
-  Publisher getPublisher(TopicInfo topic) throws IOException;
+  Publisher newPublisher(String topic, Publisher.Settings settings) throws PubSubException;
 
   /**
    * Creates a new subscription.
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PubSubImpl.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PubSubImpl.java
@@ -286,13 +286,13 @@ public Future<AsyncPage<Topic>> listTopicsAsync(ListOption... options) {
   }
 
   @Override
-  public Publisher getPublisher(TopicInfo topic) throws IOException {
-    // TODO(pongad): Provide a way to pass in the rest of the options.
-    String topicName =
-        PublisherClient.formatTopicName(getOptions().getProjectId(), topic.getName());
-    return Publisher.Builder.newBuilder(topicName)
-        .setCredentials(getOptions().getCredentials())
-        .build();
+  public Publisher newPublisher(String topic, Publisher.Settings settings) throws PubSubException {
+    String topicName = PublisherClient.formatTopicName(getOptions().getProjectId(), topic);
+    try {
+      return new PublisherImpl(topicName, settings);
+    } catch (IOException e) {
+      throw new PubSubException(e, false);
+    }
   }
 
   @Override
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/Publisher.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/Publisher.java
@@ -20,12 +20,12 @@
 import com.google.api.gax.grpc.BundlingSettings;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auto.value.AutoValue;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.pubsub.v1.PubsubMessage;
 import io.grpc.ManagedChannelBuilder;
-import java.io.IOException;
 import java.util.concurrent.ScheduledExecutorService;
 import org.joda.time.Duration;
 
@@ -79,28 +79,6 @@
  * </pre>
  */
 public interface Publisher {
-  String PUBSUB_API_ADDRESS = "pubsub.googleapis.com";
-  String PUBSUB_API_SCOPE = "https://www.googleapis.com/auth/pubsub";
-
-  // API limits.
-  int MAX_BUNDLE_MESSAGES = 1000;
-  int MAX_BUNDLE_BYTES = 10 * 1000 * 1000; // 10 megabytes (https://en.wikipedia.org/wiki/Megabyte)
-
-  // Meaningful defaults.
-  long DEFAULT_MAX_BUNDLE_MESSAGES = 100L;
-  long DEFAULT_MAX_BUNDLE_BYTES = 1000L; // 1 kB
-  Duration DEFAULT_MAX_BUNDLE_DURATION = new Duration(1); // 1ms
-  Duration DEFAULT_REQUEST_TIMEOUT = new Duration(10 * 1000); // 10 seconds
-  Duration MIN_SEND_BUNDLE_DURATION = new Duration(10 * 1000); // 10 seconds
-  Duration MIN_REQUEST_TIMEOUT = new Duration(10); // 10 milliseconds
-
-  BundlingSettings DEFAULT_BUNDLING_SETTINGS =
-      BundlingSettings.newBuilder()
-          .setDelayThreshold(DEFAULT_MAX_BUNDLE_DURATION)
-          .setRequestByteThreshold(DEFAULT_MAX_BUNDLE_BYTES)
-          .setElementCountThreshold(DEFAULT_MAX_BUNDLE_MESSAGES)
-          .build();
-
   /** Topic to which the publisher publishes to. */
   String getTopic();
 
@@ -161,130 +139,115 @@ public interface Publisher {
    */
   void shutdown();
 
-  /** A builder of {@link Publisher}s. */
-  final class Builder {
-    String topic;
+  @AutoValue
+  public abstract class Settings {
+    static final String PUBSUB_API_ADDRESS = "pubsub.googleapis.com";
+    static final String PUBSUB_API_SCOPE = "https://www.googleapis.com/auth/pubsub";
 
-    // Bundling options
-    BundlingSettings bundlingSettings = DEFAULT_BUNDLING_SETTINGS;
+    // API limits.
+    static final int MAX_BUNDLE_MESSAGES = 1000;
+    static final int MAX_BUNDLE_BYTES =
+        10 * 1000 * 1000; // 10 megabytes (https://en.wikipedia.org/wiki/Megabyte)
 
-    // Client-side flow control options
-    FlowController.Settings flowControlSettings = FlowController.Settings.DEFAULT;
-    boolean failOnFlowControlLimits = false;
+    // Meaningful defaults.
+    static final long DEFAULT_MAX_BUNDLE_MESSAGES = 100L;
+    static final long DEFAULT_MAX_BUNDLE_BYTES = 1000L; // 1 kB
+    static final Duration DEFAULT_MAX_BUNDLE_DURATION = new Duration(1); // 1ms
+    static final Duration DEFAULT_REQUEST_TIMEOUT = new Duration(10 * 1000); // 10 seconds
+    static final Duration MIN_SEND_BUNDLE_DURATION = new Duration(10 * 1000); // 10 seconds
+    static final Duration MIN_REQUEST_TIMEOUT = new Duration(10); // 10 milliseconds
 
-    // Send bundle deadline
-    Duration sendBundleDeadline = MIN_SEND_BUNDLE_DURATION;
+    static final BundlingSettings DEFAULT_BUNDLING_SETTINGS =
+        BundlingSettings.newBuilder()
+            .setDelayThreshold(DEFAULT_MAX_BUNDLE_DURATION)
+            .setRequestByteThreshold(DEFAULT_MAX_BUNDLE_BYTES)
+            .setElementCountThreshold(DEFAULT_MAX_BUNDLE_MESSAGES)
+            .build();
 
-    // RPC options
-    Duration requestTimeout = DEFAULT_REQUEST_TIMEOUT;
+    public static Settings DEFAULT = newBuilder().build();
 
-    // Channels and credentials
-    Optional<Credentials> userCredentials = Optional.absent();
-    Optional<ManagedChannelBuilder<? extends ManagedChannelBuilder<?>>> channelBuilder =
-        Optional.absent();
+    public abstract BundlingSettings getBundlingSettings();
 
-    Optional<ScheduledExecutorService> executor = Optional.absent();
+    public abstract FlowController.Settings getFlowControlSettings();
 
-    /** Constructs a new {@link Builder} using the given topic. */
-    public static Builder newBuilder(String topic) {
-      return new Builder(topic);
-    }
+    public abstract boolean getFailOnFlowControlLimits();
 
-    Builder(String topic) {
-      this.topic = Preconditions.checkNotNull(topic);
-    }
+    abstract Duration getSendBundleDeadline();
 
-    /**
-     * Credentials to authenticate with.
-     *
-     * <p>Must be properly scoped for accessing Cloud Pub/Sub APIs.
-     */
-    public Builder setCredentials(Credentials userCredentials) {
-      this.userCredentials = Optional.of(Preconditions.checkNotNull(userCredentials));
-      return this;
-    }
+    abstract Duration getRequestTimeout();
 
-    /**
-     * ManagedChannelBuilder to use to create Channels.
-     *
-     * <p>Must point at Cloud Pub/Sub endpoint.
-     */
-    public Builder setChannelBuilder(
-        ManagedChannelBuilder<? extends ManagedChannelBuilder<?>> channelBuilder) {
-      this.channelBuilder =
-          Optional.<ManagedChannelBuilder<? extends ManagedChannelBuilder<?>>>of(
-              Preconditions.checkNotNull(channelBuilder));
-      return this;
-    }
+    abstract Optional<Credentials> getUserCredentials();
 
-    // Bundling options
-    public Builder setBundlingSettings(BundlingSettings bundlingSettings) {
-      Preconditions.checkNotNull(bundlingSettings);
-      Preconditions.checkNotNull(bundlingSettings.getElementCountThreshold());
-      Preconditions.checkArgument(bundlingSettings.getElementCountThreshold() > 0);
-      Preconditions.checkNotNull(bundlingSettings.getRequestByteThreshold());
-      Preconditions.checkArgument(bundlingSettings.getRequestByteThreshold() > 0);
-      Preconditions.checkNotNull(bundlingSettings.getDelayThreshold());
-      Preconditions.checkArgument(bundlingSettings.getDelayThreshold().getMillis() > 0);
-
-      Preconditions.checkArgument(
-          bundlingSettings.getElementCountLimit() == null,
-          "elementCountLimit option not honored by current implementation");
-      Preconditions.checkArgument(
-          bundlingSettings.getRequestByteLimit() == null,
-          "requestByteLimit option not honored by current implementation");
-      Preconditions.checkArgument(
-          bundlingSettings.getBlockingCallCountThreshold() == null,
-          "blockingCallCountThreshold option not honored by current implementation");
-
-      this.bundlingSettings = bundlingSettings;
-      return this;
-    }
+    abstract Optional<ManagedChannelBuilder<? extends ManagedChannelBuilder<?>>>
+        getChannelBuilder();
 
-    // Flow control options
+    abstract Optional<ScheduledExecutorService> getExecutor();
 
-    /** Sets the flow control settings. */
-    public Builder setFlowControlSettings(FlowController.Settings flowControlSettings) {
-      this.flowControlSettings = Preconditions.checkNotNull(flowControlSettings);
-      return this;
+    public static Builder newBuilder() {
+      return new AutoValue_Publisher_Settings.Builder()
+          .setFlowControlSettings(FlowController.Settings.DEFAULT)
+          .setFailOnFlowControlLimits(false)
+          .setSendBundleDeadline(MIN_SEND_BUNDLE_DURATION)
+          .setRequestTimeout(DEFAULT_REQUEST_TIMEOUT)
+          .setUserCredentials(Optional.<Credentials>absent())
+          .setChannelBuilder(
+              Optional.<ManagedChannelBuilder<? extends ManagedChannelBuilder<?>>>absent())
+          .setExecutor(Optional.<ScheduledExecutorService>absent())
+          .setBundlingSettings(DEFAULT_BUNDLING_SETTINGS);
     }
 
-    /**
-     * Whether to fail publish when reaching any of the flow control limits, with either a {@link
-     * RequestByteMaxOutstandingReachedException} or {@link
-     * ElementCountMaxOutstandingReachedException} as appropriate.
-     *
-     * <p>If set to false, then publish operations will block the current thread until the
-     * outstanding requests go under the limits.
-     */
-    public Builder setFailOnFlowControlLimits(boolean fail) {
-      failOnFlowControlLimits = fail;
-      return this;
-    }
+    @AutoValue.Builder
+    abstract static class Builder {
+      public abstract Builder setBundlingSettings(BundlingSettings value);
 
-    /** Maximum time to attempt sending (and retrying) a bundle of messages before giving up. */
-    public Builder setSendBundleDeadline(Duration deadline) {
-      Preconditions.checkArgument(deadline.compareTo(MIN_SEND_BUNDLE_DURATION) >= 0);
-      sendBundleDeadline = deadline;
-      return this;
-    }
+      public abstract Builder setFlowControlSettings(FlowController.Settings value);
 
-    // Runtime options
-    /** Time to wait for a publish call to return from the server. */
-    public Builder setRequestTimeout(Duration timeout) {
-      Preconditions.checkArgument(timeout.compareTo(MIN_REQUEST_TIMEOUT) >= 0);
-      requestTimeout = timeout;
-      return this;
-    }
+      public abstract Builder setFailOnFlowControlLimits(boolean value);
 
-    /** 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));
-      return this;
-    }
+      abstract Builder setSendBundleDeadline(Duration value);
+
+      abstract Builder setRequestTimeout(Duration value);
+
+      abstract Builder setUserCredentials(Optional<Credentials> value);
+
+      Builder setUserCredentials(Credentials value) {
+        return setUserCredentials(Optional.of(value));
+      }
+
+      abstract Builder setChannelBuilder(
+          Optional<ManagedChannelBuilder<? extends ManagedChannelBuilder<?>>> value);
+
+      Builder setChannelBuilder(ManagedChannelBuilder<? extends ManagedChannelBuilder<?>> value) {
+        return setChannelBuilder(
+            Optional.<ManagedChannelBuilder<? extends ManagedChannelBuilder<?>>>of(value));
+      }
+
+      abstract Builder setExecutor(Optional<ScheduledExecutorService> value);
+
+      Builder setExecutor(ScheduledExecutorService value) {
+        return setExecutor(Optional.of(value));
+      }
+
+      abstract Settings autoBuild();
+
+      public Settings build() {
+        Settings settings = autoBuild();
+        Preconditions.checkArgument(
+            settings.getBundlingSettings().getElementCountLimit() == null,
+            "elementCountLimit option not honored by current implementation");
+        Preconditions.checkArgument(
+            settings.getBundlingSettings().getRequestByteLimit() == null,
+            "requestByteLimit option not honored by current implementation");
+        Preconditions.checkArgument(
+            settings.getBundlingSettings().getBlockingCallCountThreshold() == null,
+            "blockingCallCountThreshold option not honored by current implementation");
 
-    public Publisher build() throws IOException {
-      return new PublisherImpl(this);
+        Preconditions.checkArgument(
+            settings.getRequestTimeout().compareTo(MIN_REQUEST_TIMEOUT) >= 0);
+        Preconditions.checkArgument(
+            settings.getSendBundleDeadline().compareTo(MIN_SEND_BUNDLE_DURATION) >= 0);
+        return settings;
+      }
     }
   }
 }
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PublisherImpl.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/PublisherImpl.java
@@ -90,29 +90,28 @@ final class PublisherImpl implements Publisher {
   private final Duration sendBundleDeadline;
   private ScheduledFuture<?> currentAlarmFuture;
 
-  PublisherImpl(Builder builder) throws IOException {
-    topic = builder.topic;
-
-    maxBundleMessages = builder.bundlingSettings.getElementCountThreshold();
-    maxBundleBytes = builder.bundlingSettings.getRequestByteThreshold();
-    maxBundleDuration = builder.bundlingSettings.getDelayThreshold();
+  PublisherImpl(String topic, Settings settings) throws IOException {
+    this.topic = topic;
+    maxBundleMessages = settings.getBundlingSettings().getElementCountThreshold();
+    maxBundleBytes = settings.getBundlingSettings().getRequestByteThreshold();
+    maxBundleDuration = settings.getBundlingSettings().getDelayThreshold();
     hasBundlingBytes = maxBundleBytes > 0;
 
-    flowControlSettings = builder.flowControlSettings;
-    failOnFlowControlLimits = builder.failOnFlowControlLimits;
+    flowControlSettings = settings.getFlowControlSettings();
+    failOnFlowControlLimits = settings.getFailOnFlowControlLimits();
     this.flowController = new FlowController(flowControlSettings, failOnFlowControlLimits);
 
-    sendBundleDeadline = builder.sendBundleDeadline;
+    sendBundleDeadline = settings.getSendBundleDeadline();
 
-    requestTimeout = builder.requestTimeout;
+    requestTimeout = settings.getRequestTimeout();
 
     messagesBundle = new LinkedList<>();
     messagesBundleLock = new ReentrantLock();
     activeAlarm = new AtomicBoolean(false);
     int numCores = Math.max(1, Runtime.getRuntime().availableProcessors());
     executor =
-        builder.executor.isPresent()
-            ? builder.executor.get()
+        settings.getExecutor().isPresent()
+            ? settings.getExecutor().get()
             : Executors.newScheduledThreadPool(
                 numCores * DEFAULT_MIN_THREAD_POOL_SIZE,
                 new ThreadFactoryBuilder()
@@ -123,20 +122,20 @@ final class PublisherImpl implements Publisher {
     channelIndex = new AtomicLong(0);
     for (int i = 0; i < numCores; i++) {
       channels[i] =
-          builder.channelBuilder.isPresent()
-              ? builder.channelBuilder.get().build()
-              : NettyChannelBuilder.forAddress(PUBSUB_API_ADDRESS, 443)
+          settings.getChannelBuilder().isPresent()
+              ? settings.getChannelBuilder().get().build()
+              : NettyChannelBuilder.forAddress(Publisher.Settings.PUBSUB_API_ADDRESS, 443)
                   .negotiationType(NegotiationType.TLS)
                   .sslContext(GrpcSslContexts.forClient().ciphers(null).build())
                   .executor(executor)
                   .build();
     }
     credentials =
         MoreCallCredentials.from(
-            builder.userCredentials.isPresent()
-                ? builder.userCredentials.get()
+            settings.getUserCredentials().isPresent()
+                ? settings.getUserCredentials().get()
                 : GoogleCredentials.getApplicationDefault()
-                    .createScoped(Collections.singletonList(PUBSUB_API_SCOPE)));
+                    .createScoped(Collections.singletonList(Publisher.Settings.PUBSUB_API_SCOPE)));
     shutdown = new AtomicBoolean(false);
     messagesWaiter = new MessagesWaiter();
   }
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/PublisherImplTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/PublisherImplTest.java
