add new allSettled handler

Signed-off-by: Bailey Kocin <bkocin@crossbeam.com>

Author: Bailey Kocin

src/temporal/client/schedule.clj

@@ -50,7 +50,7 @@
    ```"
   [^ScheduleClient client schedule-id options]
   (let [schedule (s/schedule-> options)
-        schedule-options (s/schedule-options-> options)]
+        schedule-options (s/schedule-options-> (:schedule options))]
     (log/tracef "create schedule:" schedule-id)
     (.createSchedule client schedule-id schedule schedule-options)))
 

src/temporal/internal/promise.clj

@@ -1,8 +1,7 @@
 ;; Copyright © Manetu, Inc.  All rights reserved
 
 (ns ^:no-doc temporal.internal.promise
-  (:require [taoensso.timbre :as log]
-            [promesa.protocols :as pt]
+  (:require [promesa.protocols :as pt]
             [temporal.internal.utils :refer [->Func] :as u])
   (:import [clojure.lang IDeref IBlockingDeref]
            [io.temporal.workflow Promise]

src/temporal/internal/schedule.clj

@@ -1,6 +1,5 @@
 (ns ^:no-doc temporal.internal.schedule
-  (:require [clojure.walk :refer [stringify-keys]]
-            [temporal.internal.utils :as u]
+  (:require [temporal.internal.utils :as u]
             [temporal.internal.workflow :as w])
   (:import [io.temporal.api.enums.v1 ScheduleOverlapPolicy]
            [io.temporal.client.schedules

src/temporal/promise.clj

@@ -30,6 +30,31 @@ promises returned from [[temporal.activity/invoke]] from within workflow context
       (p/then (fn [_]
                 (mapv deref coll)))))
 
+(defn allSettled
+  "Returns Promise that becomes completed when all arguments are completed, even in the face of errors.
+
+*N.B. You must handle the exceptions in the returned promises when done*
+
+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.
+
+For more Java SDK samples example look here:
+   https://github.com/temporalio/samples-java/tree/main/core/src/main/java/io/temporal/samples/batch
+
+```clojure
+(-> (allSettled [(a/invoke activity-a ..) (a/invoke activity-b ..)])
+    (promesa.core/then (fn [[a-result b-result]] ...)))
+```
+"
+  [coll]
+  (letfn [(wait! [^Promise p] (try (.get p) (catch Exception _)))]
+    (-> (into-array Promise (mapv wait! (->array coll)))
+        (Promise/allOf)
+        (pt/->PromiseAdapter)
+        ;; The promises are all completed at this point,
+        ;; this is just to use the promesa library
+        (p/then (fn [_] (mapv deref coll))))))
+
 (defn race
   "Returns Promise that becomes completed when any of the arguments are completed.
 
@@ -54,4 +79,4 @@ promises returned from [[temporal.activity/invoke]] from within workflow context
 (defn rejected
   "Returns a new, rejected promise"
   [^Exception e]
-  (Workflow/newFailedPromise e))
+  (Workflow/newFailedPromise e))

test/temporal/test/concurrency.clj

@@ -32,4 +32,37 @@
   (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))))))
+      (is (-> workflow c/get-result deref count (= 10))))))
+
+(defactivity all-settled-activity
+  [ctx args] args)
+
+(defworkflow all-settled-workflow
+  [args]
+  @(-> (pt/all (map #(a/invoke all-settled-activity %) (range 10)))
+       (p/then (fn [r] r))
+       (p/catch (fn [e] (:args (ex-data e))))))
+
+(defactivity error-prone-activity
+  [ctx args]
+  (when (= args 5)
+    (throw (ex-info "error on 5" {:args args})))
+  args)
+
+(defworkflow error-prone-workflow
+  [args]
+  @(-> (pt/all (map #(a/invoke error-prone-activity %) (range 10)))
+       (p/then (fn [r] r))
+       (p/catch (fn [e] (:args (ex-data e))))))
+
+(deftest test-all-settled
+  (testing "Testing that allSettled waits for all the activities to complete
+            just like `p/all` does in spite of errors"
+    (let [workflow (t/create-workflow all-settled-workflow)]
+      (c/start workflow {})
+      (is (-> workflow c/get-result deref count (= 10)))))
+  (testing "Testing that allSettled waits for all the activities to complete
+            despite error and can return the errors"
+    (let [workflow (t/create-workflow error-prone-workflow)]
+      (c/start workflow {})
+      (is (-> workflow c/get-result deref (= 5))))))