Allow activities to be async

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 90
+              :fail-threshold 84
               :ns-exclude-regex [#"temporal.client.worker"]})

src/temporal/activity.clj

@@ -6,7 +6,8 @@
             [taoensso.nippy :as nippy]
             [promesa.core :as p]
             [temporal.internal.activity :as a]
-            [temporal.internal.utils :refer [->promise] :as u])
+            [temporal.internal.utils :as u]
+            [temporal.internal.promise])                    ;; needed for IPromise protocol extention
   (:import [io.temporal.workflow Workflow]
            [java.time Duration]))
 
@@ -30,8 +31,7 @@ the evaluation of the defactivity once the activity concludes.
    (let [act-name (a/get-annotation activity)
          stub (Workflow/newUntypedActivityStub (a/invoke-options-> options))]
      (log/trace "invoke:" activity "with" params options)
-     (-> (->promise
-          (.executeAsync stub act-name u/bytes-type (u/->objarray params)))
+     (-> (.executeAsync stub act-name u/bytes-type (u/->objarray params))
          (p/then nippy/thaw)))))
 
 (defmacro defactivity

src/temporal/internal/common.clj

@@ -1,6 +1,6 @@
 ;; Copyright © 2022 Manetu, Inc.  All rights reserved
 
-(ns temporal.internal.common
+(ns ^:no-doc temporal.internal.common
   (:require [temporal.internal.utils :as u])
   (:import [io.temporal.common RetryOptions RetryOptions$Builder]))
 

src/temporal/internal/promise.clj

@@ -0,0 +1,105 @@
+;; Copyright © 2022 Manetu, Inc.  All rights reserved
+
+(ns ^:no-doc temporal.internal.promise
+  (:require [promesa.protocols :as pt]
+            [temporal.internal.utils :refer [->Func] :as u])
+  (:import [clojure.lang IDeref IBlockingDeref]
+           [io.temporal.workflow Promise]
+           [java.util.concurrent CompletableFuture]))
+
+(deftype PromiseAdapter [^Promise p]
+  IDeref
+  (deref [_] (.get p))
+  IBlockingDeref
+  (deref [_ ms val] (.get p ms val)))
+
+(defmulti ->temporal type)
+
+(defmethod ->temporal Promise
+  [x] x)
+
+(defmethod ->temporal CompletableFuture
+  [^CompletableFuture x]
+  (reify Promise
+    (get [_] (.get x))
+    (isCompleted [_] (.isDone x))
+    (handle [_ f]
+      (try
+        (let [r (.get x)]
+          (.apply f r nil))
+        (catch Exception e
+          (.apply f nil e))))))
+
+(defmethod ->temporal :default
+  [x]
+  (reify Promise
+    (get [_] x)
+    (handle [_ f]
+      (.apply f x nil))
+    (isCompleted [_] true)))
+
+(def fw-identity (->Func identity))
+
+(extend-protocol pt/IPromise
+  PromiseAdapter
+  (-map
+    ([it f]
+     (.thenApply ^Promise (.p it) (->Func f)))
+
+    ([it f executor]))
+
+  (-bind
+    ([it f]
+     (.thenCompose ^Promise (.p it) (->Func f)))
+
+    ([it f executor]))
+
+  (-then
+    ([it f]
+     (pt/-promise (.thenCompose ^Promise (.p it) (->Func (comp ->temporal f)))))
+
+    ([it f executor]))
+
+  (-mapErr
+    ([it f]
+     (letfn [(handler [e]
+               (if (instance? Promise e)
+                 (f (.getCause ^Exception e))
+                 (f e)))]
+       (.exceptionally ^Promise (.p it) (->Func handler))))
+
+    ([it f executor]))
+
+  (-thenErr
+    ([it f]
+     (letfn [(handler [v e]
+               (if e
+                 (if (instance? Promise e)
+                   (pt/-promise (f (.getFailure e)))
+                   (pt/-promise (f e)))
+                 it))]
+       (as-> ^Promise (.p it) $$
+         (.handle $$ (->Func handler))
+         (.thenCompose $$ fw-identity))))
+
+    ([it f executor]))
+
+  (-handle
+    ([it f]
+     (as-> ^Promise (.p it) $$
+       (.handle $$ (->Func (comp ->temporal f)))
+       (.thenCompose $$ fw-identity)))
+
+    ([it f executor]))
+
+  (-finally
+    ([it f])
+
+    ([it f executor])))
+
+(extend-protocol pt/IPromiseFactory
+  Promise
+  (-promise [p] (->PromiseAdapter p))
+
+  PromiseAdapter
+  (-promise [p] p))

src/temporal/internal/utils.clj

@@ -3,11 +3,9 @@
 (ns ^:no-doc temporal.internal.utils
   (:require [clojure.string :as string]
             [taoensso.timbre :as log]
-            [taoensso.nippy :as nippy]
-            [promesa.core :as p])
+            [taoensso.nippy :as nippy])
   (:import [io.temporal.common.converter EncodedValues]
-           [io.temporal.workflow Promise
-            Functions$Func
+           [io.temporal.workflow Functions$Func
             Functions$Func1
             Functions$Func2
             Functions$Func3
@@ -95,18 +93,4 @@
       (f x1 x2 x3 x4 x5))
     Functions$Func6
     (apply [_ x1 x2 x3 x4 x5 x6]
-      (f x1 x2 x3 x4 x5 x6))))
-
-(defn promise-impl
-  [f]
-  (p/create
-   (fn [resolve reject]
-     (try
-       (let [^Promise p (f)]
-         (resolve (.get p)))
-       (catch Exception e
-         (reject e))))))
-
-(defmacro ->promise
-  [& body]
-  `(promise-impl (fn [] (do ~@body))))
+      (f x1 x2 x3 x4 x5 x6))))

src/temporal/promise.clj

@@ -0,0 +1,47 @@
+;; Copyright © 2022 Manetu, Inc.  All rights reserved
+
+(ns temporal.promise
+  "Methods for managing promises from pending activities from within workflows"
+  (:require [promesa.core :as p]
+            [temporal.internal.promise :as pt])
+  (:import [temporal.internal.promise PromiseAdapter]
+           [io.temporal.workflow Promise]))
+
+(defn- ->array
+  ^"[Lio.temporal.workflow.Promise;" [coll]
+  {:pre [(every? (partial instance? PromiseAdapter) coll)]}
+  (into-array Promise (map #(.p %) coll)))
+
+(defn all
+  "Returns Promise that becomes completed when all arguments are completed.
+
+*N.B. A single promise failure causes resulting promise to deliver the failure immediately.*
+
+Similar to [promesa/all](https://funcool.github.io/promesa/latest/promesa.core.html#var-all) but designed to work with
+promises returned from [[temporal.activity/invoke]] from within workflow context.
+
+```clojure
+(-> (all [(a/invoke activity-a ..) (a/invoke activity-b ..)])
+    (promesa.core/then (fn [[a-result b-result]] ...)))
+```
+"
+  [coll]
+  (-> (pt/->PromiseAdapter (Promise/allOf (->array coll)))
+      (p/then (fn [_]
+                (mapv deref coll)))))
+
+(defn race
+  "Returns Promise that becomes completed when any of the arguments are completed.
+
+*N.B. A single promise failure causes resulting promise to deliver the failure immediately.*
+
+Similar to [promesa/race](https://funcool.github.io/promesa/latest/promesa.core.html#var-race) but designed to work with
+promises returned from [[temporal.activity/invoke]] from within workflow context.
+
+```clojure
+(-> (race [(a/invoke activity-a ..) (a/invoke activity-b ..)])
+    (promesa.core/then (fn [fastest-result] ...)))
+```
+"
+  [coll]
+  (pt/->PromiseAdapter (Promise/anyOf (->array coll))))

test/temporal/test/concurrency.clj

@@ -0,0 +1,35 @@
+;; Copyright © 2022 Manetu, Inc.  All rights reserved
+
+(ns temporal.test.concurrency
+  (:require [clojure.test :refer :all]
+            [promesa.core :as p]
+            [taoensso.timbre :as log]
+            [temporal.client.core :as c]
+            [temporal.workflow :refer [defworkflow]]
+            [temporal.activity :refer [defactivity] :as a]
+            [temporal.promise :as pt]
+            [temporal.test.utils :as t]))
+
+(use-fixtures :once t/wrap-service)
+
+(defactivity echo-activity
+  [ctx args]
+  (log/info "activity:" args)
+  args)
+
+(defn invoke [x]
+  (a/invoke echo-activity x))
+
+(defworkflow concurrency-workflow
+  [ctx {:keys [args]}]
+  (log/info "workflow:" args)
+  @(-> (pt/all (map invoke (range 10)))
+       (p/then (fn [r]
+                 (log/info "r:" r)
+                 r))))
+
+(deftest the-test
+  (testing "Verifies that we can launch activities in parallel"
+    (let [workflow (t/create-workflow concurrency-workflow)]
+      (c/start workflow {})
+      (is (-> workflow c/get-result deref count (= 10))))))

test/temporal/test/race.clj

@@ -0,0 +1,38 @@
+;; Copyright © 2022 Manetu, Inc.  All rights reserved
+(ns temporal.test.race
+  (:require [clojure.test :refer :all]
+            [clojure.core.async :refer [go <!] :as async]
+            [promesa.core :as p]
+            [taoensso.timbre :as log]
+            [temporal.client.core :as c]
+            [temporal.workflow :refer [defworkflow]]
+            [temporal.activity :refer [defactivity] :as a]
+            [temporal.promise :as pt]
+            [temporal.test.utils :as t]))
+
+(use-fixtures :once t/wrap-service)
+
+(defactivity delayed-activity
+  [ctx {:keys [id delay] :as args}]
+  (log/info "activity:" args)
+  (go
+    (<! (async/timeout delay))
+    id))
+
+(defn invoke [x]
+  (a/invoke delayed-activity x))
+
+(defworkflow race-workflow
+  [ctx {:keys [args]}]
+  (log/info "workflow:" args)
+  ;; invoke activities with various synthetic delays.  The last entry, index 4, should be the fastest
+  @(-> (pt/race (map-indexed (fn [i x] (invoke {:id i :delay x})) [600 400 200 100 10]))
+       (p/then (fn [r]
+                 (log/info "r:" r)
+                 r))))
+
+(deftest the-test
+  (testing "Verifies that we can launch activities in parallel"
+    (let [workflow (t/create-workflow race-workflow)]
+      (c/start workflow {})
+      (is (-> workflow c/get-result deref (= 4))))))