Add register-signal-handler! and signal-channel abstraction

This patch breaks out the signal handling such that the core.async abstraction
surrounding signals is now optional.  There is a new function:

           temporal.signals/register-signal-handler!

that offers low-level access to the signal callback.  The design was inspired
by the work of Thomas Moerman (https://github.com/tmoerman) in the Query support

(See 4f052ae)

This work prompted another cleanup that I had been meaning to do: the removal of the 'ctx'
from the defworkflow.  The original design had both defworkflow and defactivity receiving
a user supplied 'ctx', but this never made sense for workflows since the context invites
non-determinism.

So, the new signature of defworkflow is simply a single-arity function that receives the
arguments directly, rather than within a {:keys [signals args]} map.  We still support
2-arity clients (for now) with a backwards compatibility check within the macro.  This
may be removed in a future release, so we print a WARN to signal the developer of the
deprecation.

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

Author: ghaskins

doc/clients.md

@@ -33,7 +33,7 @@ As a simple example, for the following Workflow implementation:
   (str "Hi, " name))
 
 (defworkflow greeter-workflow
-  [ctx {:keys [args]}]
+  [args]
   (log/info "greeter-workflow:" args)
   @(a/invoke greet-activity args))
 ```

doc/testing.md

@@ -25,7 +25,7 @@ You can use the provided environment with a Clojure unit testing framework of yo
   (str "Hi, " name))
 
 (defworkflow greeter-workflow
-  [ctx {:keys [args]}]
+  [args]
   (log/info "greeter-workflow:" args)
   @(a/invoke greet-activity args))
 

doc/workers.md

@@ -25,7 +25,7 @@ As a simple example, let's say we want our Worker to be able to execute the foll
   (str "Hi, " name))
 
 (defworkflow greeter-workflow
-  [ctx {:keys [args]}]
+  [args]
   (log/info "greeter-workflow:" args)
   @(a/invoke greet-activity args))
 ```

doc/workflows.md

@@ -10,7 +10,7 @@ In this Clojure SDK programming model, a Temporal Workflow is a function declare
 
 ```clojure
 (defworkflow my-workflow
-   [ctx params]
+   [params]
    ...)
 ```
 
@@ -24,7 +24,7 @@ A Workflow implementation consists of defining a (defworkflow) function.  The pl
 (require '[temporal.workflow :refer [defworkflow]])
 
 (defworkflow my-workflow
-    [ctx {{:keys [foo]} :args}]
+    [{:keys [foo]}]
     ...)
 ```
 
@@ -65,7 +65,7 @@ In this Clojure SDK, developers manage Workflows with the following flow:
 
 ```clojure
 (defworkflow my-workflow
-  [ctx {{:keys [foo]} :args}] 
+  [{:keys [foo]}] 
   ...)
 
 (let [w (create-workflow client my-workflow {:task-queue "MyTaskQueue"})]
@@ -155,24 +155,45 @@ Your Workflow may send or receive [signals](https://cljdoc.org/d/io.github.manet
 
 #### Receiving Signals
 
-Your Workflow may either block waiting with signals with [temporal.signals/<!](https://cljdoc.org/d/io.github.manetu/temporal-sdk/CURRENT/api/temporal.signals#%3C!) or use the non-blocking [temporal.signals/poll](https://cljdoc.org/d/io.github.manetu/temporal-sdk/CURRENT/api/temporal.signals#poll).  Either way, your Workflow needs to obtain the `signals` context provided in the Workflow request map.
+##### Channel Abstraction using 'signal-chan'
 
-##### Example
+This SDK provides a [core.async](https://github.com/clojure/core.async) inspired abstraction on [Temporal Signals](https://docs.temporal.io/workflows#signal) called signal-channels. To use signal channels, your Workflow may either block waiting with signals with [temporal.signals/<!](https://cljdoc.org/d/io.github.manetu/temporal-sdk/CURRENT/api/temporal.signals#%3C!) or use the non-blocking [temporal.signals/poll](https://cljdoc.org/d/io.github.manetu/temporal-sdk/CURRENT/api/temporal.signals#poll).  Either way, your Workflow needs to first obtain the `signal-chan` context obtained by  [temporal.signals/create-signal-chan](https://cljdoc.org/d/io.github.manetu/temporal-sdk/CURRENT/api/temporal.signals#create-signal-chan).
+
+###### Example
 
 ```clojure
+(require `[temporal.signals :as s])
+
 (defworkflow my-workflow
-  [ctx {:keys [signals]}]
-  (let [message (<! signals "MySignal")]
+  [args]
+  (let [signals (s/create-signal-chan)
+        message (<! signals "MySignal")]
     ...))
 ```
 
+##### Raw Signal Handling
+
+Alternatively, you may opt to handle signals directly with [temporal.signals/register-signal-handler!](https://cljdoc.org/d/io.github.manetu/temporal-sdk/CURRENT/api/temporal.signals#register-signal-handler!)
+
+###### Example
+
+```clojure
+(defworkflow my-workflow
+  [args]
+  (let [state (atom 0)]
+    (s/register-signal-handler! (fn [signal-name args]
+                                  (swap! state inc)))
+    (w/await (fn [] (> @state 1)))
+    @state))
+```
+
 ### Temporal Queries
 
 Your Workflow may respond to [queries](https://cljdoc.org/d/io.github.manetu/temporal-sdk/CURRENT/api/temporal.client.core#query).
 
 A temporal query is similar to a temporal signal, both are messages sent to a running Workflow. 
 The difference is that a signal intends to change the behaviour of the Workflow, whereas a query intends to inspect the current state of the Workflow.
-Querying the state of a Workflow implies that the Workflow must maintain state while running, typically in a clojure [atom](https://clojuredocs.org/clojure.core/atom).
+Querying the state of a Workflow implies that the Workflow must maintain state while running, typically in a clojure [atom](https://clojuredocs.org/clojure.core/atom) or [ref](https://clojure.org/reference/refs).
 
 #### Registering a Query handler
 
@@ -181,7 +202,7 @@ The query handler is a function that has a reference to the Workflow state, usua
 
 ```clojure
 (defworkflow stateful-workflow
-  [ctx {:keys [signals] {:keys [init] :as args} :args :as params}]
+  [{:keys [init] :as args}]
   (let [state (atom init)]
     (register-query-handler! (fn [query-type args]
                                (when (= query-type :my-query)

src/temporal/internal/signals.clj

@@ -21,13 +21,16 @@
              (.add ch payload)
              (assoc s signal-name ch)))))
 
-(defn create
-  "Registers the calling workflow to receive signals and returns a context variable to be passed back later"
+(defn register-signal-handler!
+  [f]
+  (Workflow/registerListener
+   (reify
+     DynamicSignalHandler
+     (handle [_ signal-name args]
+       (f signal-name (u/->args args))))))
+
+(defn create-signal-chan
   []
   (let [state (atom {})]
-    (Workflow/registerListener
-     (reify
-       DynamicSignalHandler
-       (handle [_ signal-name args]
-         (-handle state signal-name (u/->args args)))))
+    (register-signal-handler! (partial -handle state))
     state))

src/temporal/internal/utils.clj

@@ -75,11 +75,16 @@
     (verify-registered-fns data)
     (m/index-by :name data)))
 
+(defn find-dispatch
+  "Finds any dispatch descriptor named 't' that carry metadata 'marker'"
+  [dispatch-table t]
+  (or (get dispatch-table t)
+      (throw (ex-info "workflow/activity not found" {:function t}))))
+
 (defn find-dispatch-fn
   "Finds any functions named 't' that carry metadata 'marker'"
   [dispatch-table t]
-  (:fn (or (get dispatch-table t)
-           (throw (ex-info "workflow/activity not found" {:function t})))))
+  (:fn (find-dispatch dispatch-table t)))
 
 (defn import-dispatch
   [marker coll]

src/temporal/internal/workflow.clj

@@ -53,10 +53,13 @@
   [ctx dispatch args]
   (try
     (let [{:keys [workflow-type workflow-id]} (get-info)
-          f (u/find-dispatch-fn dispatch workflow-type)
+          d (u/find-dispatch dispatch workflow-type)
+          f (:fn d)
           a (u/->args args)
           _ (log/trace workflow-id "calling" f "with args:" a)
-          r (f ctx {:args a :signals (s/create)})]
+          r (if (-> d :type (= :legacy))
+              (f ctx {:args a :signals (s/create-signal-chan)})
+              (f a))]
       (log/trace workflow-id "result:" r)
       (nippy/freeze r))
     (catch Exception e

src/temporal/signals.clj

@@ -9,46 +9,77 @@
   (:import [io.temporal.workflow Workflow]))
 
 (defn is-empty?
-  "Returns 'true' if 'signal-name' either doesn't exist or exists but has no pending messages"
-  [state signal-name]
+  "Returns 'true' if 'signal-name' either doesn't exist within signal-chan or exists but has no pending messages"
+  [signal-chan signal-name]
   (let [signal-name (u/namify signal-name)
-        ch (s/get-ch @state signal-name)
+        ch (s/get-ch @signal-chan signal-name)
         r (or (nil? ch) (.isEmpty ch))]
-    (log/trace "is-empty?:" @state signal-name r)
+    (log/trace "is-empty?:" @signal-chan signal-name r)
     r))
 
 (defn- rx
-  [state signal-name]
+  [signal-chan signal-name]
   (let [signal-name (u/namify signal-name)
-        ch (s/get-ch @state signal-name)
+        ch (s/get-ch @signal-chan signal-name)
         m (.poll ch)]
     (log/trace "rx:" signal-name m)
     m))
 
 (defn poll
-  "Non-blocking check of the signal.  Consumes and returns a message if found, otherwise returns 'nil'"
-  [state signal-name]
-  (when-not (is-empty? state signal-name)
-    (rx state signal-name)))
+  "Non-blocking check of the signal via signal-chan.  Consumes and returns a message if found, otherwise returns 'nil'"
+  [signal-chan signal-name]
+  (when-not (is-empty? signal-chan signal-name)
+    (rx signal-chan signal-name)))
 
 (defn <!
-  "Light-weight/parking receive of a single message with an optional timeout"
-  ([state] (<! state ::default))
-  ([state signal-name] (<! state signal-name nil))
-  ([state signal-name timeout]
+  "Light-weight/parking receive of a single message from signal-chan with an optional timeout"
+  ([signal-chan] (<! signal-chan ::default))
+  ([signal-chan signal-name] (<! signal-chan signal-name nil))
+  ([signal-chan signal-name timeout]
    (log/trace "waiting on:" signal-name "with timeout" timeout)
-   (let [pred #(not (is-empty? state signal-name))]
+   (let [pred #(not (is-empty? signal-chan signal-name))]
      (if (some? timeout)
        (do
          (when (w/await timeout pred)
-           (rx state signal-name)))
+           (rx signal-chan signal-name)))
        (do
          (w/await pred)
-         (rx state signal-name))))))
+         (rx signal-chan signal-name))))))
 
 (defn >!
   "Sends `payload` to `workflow-id` via signal `signal-name`."
   [^String workflow-id signal-name payload]
   (let [signal-name (u/namify signal-name)
         stub (Workflow/newUntypedExternalWorkflowStub workflow-id)]
-    (.signal stub signal-name (u/->objarray payload))))
+    (.signal stub signal-name (u/->objarray payload))))
+
+(def register-signal-handler!
+  "
+Registers a DynamicSignalHandler listener that handles signals sent to the workflow such as with [[>!]].
+
+Use inside a workflow definition with 'f' closing over any desired workflow state (e.g. atom) to mutate
+the workflow state.
+
+Arguments:
+- `f`: a 2-arity function, expecting 2 arguments.
+
+`f` arguments:
+- `signal-name`: string
+- `args`: params value or data structure
+
+```clojure
+(defworkflow signalled-workflow
+  [{:keys [init] :as args}]
+  (let [state (atom init)]
+    (register-signal-handler! (fn [signal-name args]
+                                (when (= signal-name \"mysignal\")
+                                  (update state #(conj % args)))))
+    ;; workflow implementation
+    ))
+```"
+  s/register-signal-handler!)
+
+(def create-signal-chan
+  "Registers the calling workflow to receive signals and returns a 'signal-channel' context for use with functions such as [[<!]] amd [[poll]]"
+  s/create-signal-chan)
+

src/temporal/workflow.clj

@@ -50,7 +50,7 @@ Arguments:
 
 ```clojure
 (defworkflow stateful-workflow
-  [ctx {:keys [signals] {:keys [init] :as args} :args :as params}]
+  [{:keys [init] :as args}]
   (let [state (atom init)]
     (register-query-handler! (fn [query-type args]
                                (when (= query-type :my-query)
@@ -71,19 +71,16 @@ Arguments:
 
 (defmacro defworkflow
   "
-Defines a new workflow, similar to defn, expecting a 2-arity parameter list and body.  Should evaluate to something
+Defines a new workflow, similar to defn, expecting a 1-arity parameter list and body.  Should evaluate to something
 serializable, which will become available for [[temporal.client.core/get-result]].
 
 Arguments:
 
-- `ctx`: Context passed through from [[temporal.client.worker/start]]
-- `params`: A map containing the following
-    - `args`: Passed from 'params' to [[temporal.client.core/start]] or [[temporal.client.core/signal-with-start]]
-    - `signals`: Signal context for use with signal calls such as [[temporal.signals/<!]] and [[temporal.signals/poll]]
+- `args`: Passed from 'params' to [[temporal.client.core/start]] or [[temporal.client.core/signal-with-start]]
 
 ```clojure
 (defworkflow my-workflow
-    [ctx {{:keys [foo]} :args}]
+    [{:keys [foo]}]
     ...)
 
 (let [w (create-workflow client my-workflow {:task-queue ::my-task-queue})]
@@ -93,8 +90,17 @@ Arguments:
   [name params* & body]
   (let [fqn (u/get-fq-classname name)
         sname (str name)]
-    `(def ~name ^{::w/def {:name ~sname :fqn ~fqn}}
-       (fn [ctx# args#]
-         (log/trace (str ~fqn ": ") args#)
-         (let [f# (fn ~params* (do ~@body))]
-           (f# ctx# args#))))))
+    (if (= (count params*) 2)                               ;; legacy
+      (do
+        (println (str *file* ":" (:line (meta &form)) " WARN: (defworkflow " name ") with [ctx {:keys [signals args]}] is deprecated.  Use [args] form."))
+        `(def ~name ^{::w/def {:name ~sname :fqn ~fqn :type :legacy}}
+           (fn [ctx# args#]
+             (log/trace (str ~fqn ": ") args#)
+             (let [f# (fn ~params* (do ~@body))]
+               (f# ctx# args#)))))
+      (do
+        `(def ~name ^{::w/def {:name ~sname :fqn ~fqn}}
+           (fn [args#]
+             (log/trace (str ~fqn ": ") args#)
+             (let [f# (fn ~params* (do ~@body))]
+               (f# args#))))))))

test/temporal/test/async.clj

@@ -20,7 +20,7 @@
       (str "Hi, " name))))
 
 (defworkflow async-greeter-workflow
-  [ctx {:keys [args]}]
+  [args]
   (log/info "greeter-workflow:" args)
   @(a/invoke async-greet-activity args {:retry-options {:maximum-attempts 1}}))