Add support for LocalActivity

Signed-off-by: Greg Haskins <[email protected]>

Author: ghaskins

project.clj

@@ -30,5 +30,5 @@
   :cloverage {:runner :eftest
               :runner-opts {:multithread? false
                             :fail-fast? true}
-              :fail-threshold 87
+              :fail-threshold 86
               :ns-exclude-regex [#"temporal.client.worker"]})

src/temporal/activity.clj

@@ -13,6 +13,13 @@
 
 (def ^:no-doc default-invoke-options {:start-to-close-timeout (Duration/ofSeconds 3)})
 
+(defn- complete-invoke
+  [activity result]
+  (log/trace activity "completed with" (count result) "bytes")
+  (let [r (nippy/thaw result)]
+    (log/trace activity "results:" r)
+    r))
+
 (defn invoke
   "
 Invokes 'activity' with 'params' from within a workflow context.  Returns a promise that when derefed will resolve to
@@ -32,11 +39,31 @@ the evaluation of the defactivity once the activity concludes.
          stub (Workflow/newUntypedActivityStub (a/invoke-options-> options))]
      (log/trace "invoke:" activity "with" params options)
      (-> (.executeAsync stub act-name u/bytes-type (u/->objarray params))
-         (p/then (fn [r]
-                   (log/trace activity "completed with" (count r) "bytes")
-                   (let [r (nippy/thaw r)]
-                     (log/trace activity "results:" r)
-                     r)))
+         (p/then (partial complete-invoke activity))
+         (p/catch (fn [e]
+                    (log/error e)
+                    (throw e)))))))
+
+(defn local-invoke
+  "
+Invokes 'activity' with 'params' from within a workflow context as a [Local Activity](https://docs.temporal.io/concepts/what-is-a-local-activity/).  Returns a promise that when
+derefed will resolve to the evaluation of the defactivity once the activity concludes.
+
+```clojure
+(defactivity my-activity
+   [ctx {:keys [foo] :as args}]
+   ...)
+
+(local-invoke my-activity {:foo \"bar\"} {:start-to-close-timeout (Duration/ofSeconds 3))
+```
+"
+  ([activity params] (invoke activity params default-invoke-options))
+  ([activity params options]
+   (let [act-name (a/get-annotation activity)
+         stub (Workflow/newUntypedLocalActivityStub (a/local-invoke-options-> options))]
+     (log/trace "local-invoke:" activity "with" params options)
+     (-> (.executeAsync stub act-name u/bytes-type (u/->objarray params))
+         (p/then (partial complete-invoke activity))
          (p/catch (fn [e]
                     (log/error e)
                     (throw e)))))))

src/temporal/internal/activity.clj

@@ -9,17 +9,28 @@
             [temporal.internal.utils :as u]
             [temporal.internal.common :as common])
   (:import [io.temporal.activity Activity ActivityInfo DynamicActivity]
-           [io.temporal.activity ActivityOptions ActivityOptions$Builder]
+           [io.temporal.activity ActivityOptions ActivityOptions$Builder LocalActivityOptions LocalActivityOptions$Builder]
            [clojure.core.async.impl.channels ManyToManyChannel]))
 
 (def invoke-option-spec
   {:start-to-close-timeout #(.setStartToCloseTimeout ^ActivityOptions$Builder %1  %2)
-   :retry-options          #(.setRetryOptions %1 (common/retry-options-> %2))})
+   :retry-options          #(.setRetryOptions ^ActivityOptions$Builder %1 (common/retry-options-> %2))})
 
 (defn invoke-options->
   ^ActivityOptions [params]
   (u/build (ActivityOptions/newBuilder) invoke-option-spec params))
 
+(def local-invoke-option-spec
+  {:start-to-close-timeout    #(.setStartToCloseTimeout ^LocalActivityOptions$Builder %1  %2)
+   :schedule-to-close-timeout #(.setScheduleToCloseTimeout ^LocalActivityOptions$Builder %1  %2)
+   :retry-options             #(.setRetryOptions ^LocalActivityOptions$Builder %1 (common/retry-options-> %2))
+   :do-not-include-args       #(.setDoNotIncludeArgumentsIntoMarker ^LocalActivityOptions$Builder %1 %2)
+   :local-retry-threshold     #(.setLocalRetryThreshold ^LocalActivityOptions$Builder %1 %2)})
+
+(defn local-invoke-options->
+  ^LocalActivityOptions [params]
+  (u/build (LocalActivityOptions/newBuilder) local-invoke-option-spec params))
+
 (extend-protocol p/Datafiable
   ActivityInfo
   (datafy [d]

test/temporal/test/local_activity.clj

@@ -0,0 +1,27 @@
+;; Copyright © 2022 Manetu, Inc.  All rights reserved
+
+(ns temporal.test.local-activity
+  (:require [clojure.test :refer :all]
+            [taoensso.timbre :as log]
+            [temporal.client.core :as c]
+            [temporal.workflow :refer [defworkflow]]
+            [temporal.activity :refer [defactivity] :as a]
+            [temporal.test.utils :as t]))
+
+(use-fixtures :once t/wrap-service)
+
+(defactivity greet-activity
+  [ctx {:keys [name] :as args}]
+  (log/info "greet-activity:" args)
+  (str "Hi, " name))
+
+(defworkflow greeter-workflow
+  [ctx {:keys [args]}]
+  (log/info "greeter-workflow:" args)
+  @(a/local-invoke greet-activity args (assoc a/default-invoke-options :do-not-include-args true)))
+
+(deftest the-test
+  (testing "Verifies that we can round-trip through start"
+    (let [workflow (t/create-workflow greeter-workflow)]
+      (c/start workflow {:name "Bob"})
+      (is (= @(c/get-result workflow) "Hi, Bob")))))