Add timeout option to signal receive interface

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

Author: ghaskins

src/temporal/signals.clj

@@ -32,12 +32,19 @@
     (rx state signal-name)))
 
 (defn <!
-  "Light-weight/parking receive of a single message"
+  "Light-weight/parking receive of a single message with an optional timeout"
   ([state] (<! state ::default))
-  ([state signal-name]
-   (log/trace "waiting on:" signal-name)
-   (w/await #(not (is-empty? state signal-name)))
-   (rx state signal-name)))
+  ([state signal-name] (<! state signal-name nil))
+  ([state signal-name timeout]
+   (log/trace "waiting on:" signal-name "with timeout" timeout)
+   (let [pred #(not (is-empty? state signal-name))]
+     (if (some? timeout)
+       (do
+         (when (w/await timeout pred)
+           (rx state signal-name)))
+       (do
+         (w/await pred)
+         (rx state signal-name))))))
 
 (defn >!
   "Sends `payload` to `workflow-id` via signal `signal-name`."

src/temporal/workflow.clj

@@ -6,15 +6,16 @@
             [temporal.internal.utils :as u]
             [temporal.internal.workflow :as w])
   (:import [io.temporal.workflow Workflow]
-           [java.util.function Supplier]))
+           [java.util.function Supplier]
+           [java.time Duration]))
 
 (defn get-info
   "Return info about the current workflow"
   []
   (w/get-info))
 
 (defn- ->supplier
-  [f]
+  ^Supplier [f]
   (reify Supplier
     (get [_]
       (f))))
@@ -23,7 +24,7 @@
   "Efficiently parks the workflow until 'pred' evaluates to true.  Re-evaluates on each state transition"
   ([pred]
    (Workflow/await (->supplier pred)))
-  ([duration pred]
+  ([^Duration duration pred]
    (Workflow/await duration (->supplier pred))))
 
 (defmacro defworkflow

test/temporal/test/signal_timeout.clj

@@ -0,0 +1,34 @@
+;; Copyright © 2022 Manetu, Inc.  All rights reserved
+
+(ns temporal.test.signal-timeout
+  (:require [clojure.test :refer :all]
+            [taoensso.timbre :as log]
+            [temporal.client.core :refer [>!] :as c]
+            [temporal.signals :refer [<!]]
+            [temporal.workflow :refer [defworkflow]]
+            [temporal.test.utils :as t])
+  (:import [java.time Duration]))
+
+(use-fixtures :once t/wrap-service)
+
+(def signal-name ::signal)
+
+(defworkflow timeout-workflow
+  [ctx {:keys [signals] :as args}]
+  (log/info "timeout-workflow:" args)
+  (or (<! signals signal-name (Duration/ofSeconds 1))
+      :timed-out))
+
+(defn create []
+  (let [wf (c/create-workflow (t/get-client) timeout-workflow {:task-queue t/task-queue})]
+    (c/start wf nil)
+    wf))
+
+(deftest the-test
+  (testing "Verifies that signals may timeout properly"
+    (let [wf (create)]
+      (is (= @(c/get-result wf) :timed-out))))
+  (testing "Verifies that signals are received properly even when a timeout is requested"
+    (let [wf (create)]
+      (>! wf ::signal "Hi, Bob")
+      (is (= @(c/get-result wf) "Hi, Bob")))))