Implement local activities (#273)

This PR adds local activities support in Java client.

Author: meiliang86

build.gradle

@@ -189,6 +189,7 @@ task registerDomain(type:JavaExec) {
 
 test {
     dependsOn 'registerDomain'
+    dependsOn 'checkLicenseMain'
     testLogging {
         events "passed", "skipped", "failed"
         exceptionFormat "full"

src/main/java/com/uber/cadence/activity/ActivityOptions.java

@@ -22,6 +22,7 @@
 import com.uber.cadence.common.MethodRetry;
 import com.uber.cadence.common.RetryOptions;
 import java.time.Duration;
+import java.util.Objects;
 
 /** Options used to configure how an activity is invoked. */
 public final class ActivityOptions {
@@ -43,11 +44,12 @@ public static ActivityOptions merge(ActivityMethod a, MethodRetry r, ActivityOpt
     }
     return new ActivityOptions.Builder()
         .setScheduleToCloseTimeout(
-            merge(a.scheduleToCloseTimeoutSeconds(), o.getScheduleToCloseTimeout()))
+            mergeDuration(a.scheduleToCloseTimeoutSeconds(), o.getScheduleToCloseTimeout()))
         .setScheduleToStartTimeout(
-            merge(a.scheduleToStartTimeoutSeconds(), o.getScheduleToStartTimeout()))
-        .setStartToCloseTimeout(merge(a.startToCloseTimeoutSeconds(), o.getStartToCloseTimeout()))
-        .setHeartbeatTimeout(merge(a.heartbeatTimeoutSeconds(), o.getHeartbeatTimeout()))
+            mergeDuration(a.scheduleToStartTimeoutSeconds(), o.getScheduleToStartTimeout()))
+        .setStartToCloseTimeout(
+            mergeDuration(a.startToCloseTimeoutSeconds(), o.getStartToCloseTimeout()))
+        .setHeartbeatTimeout(mergeDuration(a.heartbeatTimeoutSeconds(), o.getHeartbeatTimeout()))
         .setTaskList(
             o.getTaskList() != null
                 ? o.getTaskList()
@@ -278,37 +280,26 @@ public boolean equals(Object o) {
     if (o == null || getClass() != o.getClass()) return false;
 
     ActivityOptions that = (ActivityOptions) o;
-
-    if (heartbeatTimeout != null
-        ? !heartbeatTimeout.equals(that.heartbeatTimeout)
-        : that.heartbeatTimeout != null) return false;
-    if (scheduleToCloseTimeout != null
-        ? !scheduleToCloseTimeout.equals(that.scheduleToCloseTimeout)
-        : that.scheduleToCloseTimeout != null) return false;
-    if (scheduleToStartTimeout != null
-        ? !scheduleToStartTimeout.equals(that.scheduleToStartTimeout)
-        : that.scheduleToStartTimeout != null) return false;
-    if (startToCloseTimeout != null
-        ? !startToCloseTimeout.equals(that.startToCloseTimeout)
-        : that.startToCloseTimeout != null) return false;
-    if (taskList != null ? !taskList.equals(that.taskList) : that.taskList != null) return false;
-    return retryOptions != null
-        ? retryOptions.equals(that.retryOptions)
-        : that.retryOptions == null;
+    return Objects.equals(heartbeatTimeout, that.heartbeatTimeout)
+        && Objects.equals(scheduleToCloseTimeout, that.scheduleToCloseTimeout)
+        && Objects.equals(scheduleToStartTimeout, that.scheduleToStartTimeout)
+        && Objects.equals(startToCloseTimeout, that.startToCloseTimeout)
+        && Objects.equals(taskList, that.taskList)
+        && Objects.equals(retryOptions, that.retryOptions);
   }
 
   @Override
   public int hashCode() {
-    int result = heartbeatTimeout != null ? heartbeatTimeout.hashCode() : 0;
-    result = 31 * result + (scheduleToCloseTimeout != null ? scheduleToCloseTimeout.hashCode() : 0);
-    result = 31 * result + (scheduleToStartTimeout != null ? scheduleToStartTimeout.hashCode() : 0);
-    result = 31 * result + (startToCloseTimeout != null ? startToCloseTimeout.hashCode() : 0);
-    result = 31 * result + (taskList != null ? taskList.hashCode() : 0);
-    result = 31 * result + (retryOptions != null ? retryOptions.hashCode() : 0);
-    return result;
+    return Objects.hash(
+        heartbeatTimeout,
+        scheduleToCloseTimeout,
+        scheduleToStartTimeout,
+        startToCloseTimeout,
+        taskList,
+        retryOptions);
   }
 
-  private static Duration merge(int annotationSeconds, Duration options) {
+  static Duration mergeDuration(int annotationSeconds, Duration options) {
     if (options == null) {
       if (annotationSeconds == 0) {
         return null;

src/main/java/com/uber/cadence/activity/LocalActivityOptions.java

@@ -0,0 +1,139 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.activity;
+
+import static com.uber.cadence.internal.common.OptionsUtils.roundUpToSeconds;
+
+import com.uber.cadence.common.MethodRetry;
+import com.uber.cadence.common.RetryOptions;
+import java.time.Duration;
+import java.util.Objects;
+
+/** Options used to configure how an local activity is invoked. */
+public final class LocalActivityOptions {
+
+  /**
+   * Used to merge annotation and options. Options takes precedence. Returns options with all
+   * defaults filled in.
+   */
+  public static LocalActivityOptions merge(
+      ActivityMethod a, MethodRetry r, LocalActivityOptions o) {
+    if (a == null) {
+      if (r == null) {
+        return new LocalActivityOptions.Builder(o).validateAndBuildWithDefaults();
+      }
+      RetryOptions mergedR = RetryOptions.merge(r, o.getRetryOptions());
+      return new LocalActivityOptions.Builder()
+          .setRetryOptions(mergedR)
+          .validateAndBuildWithDefaults();
+    }
+    if (o == null) {
+      o = new LocalActivityOptions.Builder().build();
+    }
+    return new LocalActivityOptions.Builder()
+        .setScheduleToCloseTimeout(
+            ActivityOptions.mergeDuration(
+                a.scheduleToCloseTimeoutSeconds(), o.getScheduleToCloseTimeout()))
+        .setRetryOptions(RetryOptions.merge(r, o.getRetryOptions()))
+        .validateAndBuildWithDefaults();
+  }
+
+  public static final class Builder {
+    private Duration scheduleToCloseTimeout;
+    private RetryOptions retryOptions;
+
+    public Builder() {}
+
+    /** Copy Builder fields from the options. */
+    public Builder(LocalActivityOptions options) {
+      if (options == null) {
+        return;
+      }
+      this.scheduleToCloseTimeout = options.getScheduleToCloseTimeout();
+      this.retryOptions = options.retryOptions;
+    }
+
+    /** Overall timeout workflow is willing to wait for activity to complete. */
+    public Builder setScheduleToCloseTimeout(Duration scheduleToCloseTimeout) {
+      this.scheduleToCloseTimeout = scheduleToCloseTimeout;
+      return this;
+    }
+
+    /**
+     * RetryOptions that define how activity is retried in case of failure. Default is null which is
+     * no reties.
+     */
+    public Builder setRetryOptions(RetryOptions retryOptions) {
+      this.retryOptions = retryOptions;
+      return this;
+    }
+
+    public LocalActivityOptions build() {
+      return new LocalActivityOptions(scheduleToCloseTimeout, retryOptions);
+    }
+
+    public LocalActivityOptions validateAndBuildWithDefaults() {
+      RetryOptions ro = null;
+      if (retryOptions != null) {
+        ro = new RetryOptions.Builder(retryOptions).validateBuildWithDefaults();
+      }
+      return new LocalActivityOptions(roundUpToSeconds(scheduleToCloseTimeout), ro);
+    }
+  }
+
+  private final Duration scheduleToCloseTimeout;
+  private final RetryOptions retryOptions;
+
+  private LocalActivityOptions(Duration scheduleToCloseTimeout, RetryOptions retryOptions) {
+    this.scheduleToCloseTimeout = scheduleToCloseTimeout;
+    this.retryOptions = retryOptions;
+  }
+
+  public Duration getScheduleToCloseTimeout() {
+    return scheduleToCloseTimeout;
+  }
+
+  public RetryOptions getRetryOptions() {
+    return retryOptions;
+  }
+
+  @Override
+  public String toString() {
+    return "LocalActivityOptions{"
+        + "scheduleToCloseTimeout="
+        + scheduleToCloseTimeout
+        + ", retryOptions="
+        + retryOptions
+        + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+
+    LocalActivityOptions that = (LocalActivityOptions) o;
+    return Objects.equals(scheduleToCloseTimeout, that.scheduleToCloseTimeout)
+        && Objects.equals(retryOptions, that.retryOptions);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(scheduleToCloseTimeout, retryOptions);
+  }
+}

src/main/java/com/uber/cadence/common/RetryOptions.java

@@ -18,6 +18,8 @@
 package com.uber.cadence.common;
 
 import com.google.common.base.Defaults;
+import com.uber.cadence.workflow.ActivityFailureException;
+import com.uber.cadence.workflow.ChildWorkflowFailureException;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collections;
@@ -27,6 +29,7 @@
 public final class RetryOptions {
 
   private static final double DEFAULT_BACKOFF_COEFFICIENT = 2.0;
+  private static final int DEFAULT_MAXIMUM_MULTIPLIER = 100;
 
   /**
    * Merges annotation with explicitly provided RetryOptions. If there is conflict RetryOptions
@@ -403,4 +406,32 @@ private Class<? extends Throwable>[] merge(
     }
     return null;
   }
+
+  public long calculateSleepTime(long attempt) {
+    double coefficient =
+        backoffCoefficient == 0d ? DEFAULT_BACKOFF_COEFFICIENT : backoffCoefficient;
+    double sleepMillis = Math.pow(coefficient, attempt - 1) * initialInterval.toMillis();
+    if (maximumInterval == null) {
+      return (long) Math.min(sleepMillis, initialInterval.toMillis() * DEFAULT_MAXIMUM_MULTIPLIER);
+    }
+    return Math.min((long) sleepMillis, maximumInterval.toMillis());
+  }
+
+  public boolean shouldRethrow(Throwable e, long attempt, long elapsed, long sleepTime) {
+    if (e instanceof ActivityFailureException || e instanceof ChildWorkflowFailureException) {
+      e = e.getCause();
+    }
+    if (doNotRetry != null) {
+      for (Class<? extends Throwable> doNotRetry : doNotRetry) {
+        if (doNotRetry.equals(e.getClass())) {
+          return true;
+        }
+      }
+    }
+    // Attempt that failed.
+    if (maximumAttempts != 0 && attempt >= maximumAttempts) {
+      return true;
+    }
+    return expiration != null && elapsed + sleepTime >= expiration.toMillis();
+  }
 }