Misc updates and bug fixes

- Convert to nippy for serialization
- Add poll and >! for signals
- Fix bug in is-empty
- Clean up unit-tests

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

Author: ghaskins

project.clj

@@ -14,7 +14,9 @@
   :dependencies [[org.clojure/clojure "1.11.1"]
                  [io.temporal/temporal-sdk "1.16.0"]
                  [io.temporal/temporal-testing "1.16.0"]
+                 [com.taoensso/encore "3.24.0"]
                  [com.taoensso/timbre "5.2.1"]
+                 [com.taoensso/nippy "3.2.0"]
                  [funcool/promesa "8.0.450"]]
   :repl-options {:init-ns user}
   :aot [temporal.internal.dispatcher]
@@ -27,5 +29,5 @@
   :cloverage {:runner :eftest
               :runner-opts {:multithread? false
                             :fail-fast? true}
-              :fail-threshold 91
+              :fail-threshold 92
               :ns-exclude-regex [#"temporal.client.worker"]})

src/temporal/activity.clj

@@ -2,8 +2,8 @@
 
 (ns temporal.activity
   "Methods for defining and invoking activity tasks"
-  (:require [clojure.edn :as edn]
-            [taoensso.timbre :as log]
+  (:require [taoensso.timbre :as log]
+            [taoensso.nippy :as nippy]
             [promesa.core :as p]
             [temporal.internal.activity :as a]
             [temporal.internal.utils :refer [->promise] :as u])
@@ -38,13 +38,13 @@ the evaluation of the defactivity once the activity concludes.
    (let [act-name (a/get-annotation activity)
          stub (Workflow/newUntypedActivityStub (invoke-options-> options))]
      (-> (->promise
-          (.executeAsync stub act-name String (u/->objarray params)))
-         (p/then edn/read-string)))))
+          (.executeAsync stub act-name u/bytes-type (u/->objarray params)))
+         (p/then nippy/thaw)))))
 
 (defmacro defactivity
   "
 Defines a new activity, similar to defn, expecting a 2-arity parameter list and body.  Should evaluate to something
-serializable with [prn-str](https://clojuredocs.org/clojure.core/prn-str), which will be available to the [[invoke]] caller.
+serializable, which will be available to the [[invoke]] caller.
 
 Arguments:
 

src/temporal/client/core.clj

@@ -2,7 +2,7 @@
 
 (ns temporal.client.core
   "Methods for client interaction with Temporal"
-  (:require [clojure.edn :as edn]
+  (:require [taoensso.nippy :as nippy]
             [promesa.core :as p]
             [temporal.internal.workflow :as w]
             [temporal.internal.utils :as u])
@@ -87,8 +87,8 @@ defworkflow once the workflow concludes.
 ```
 "
   [{:keys [^WorkflowStub stub] :as workflow}]
-  (-> (.getResultAsync stub String)
-      (p/then edn/read-string)))
+  (-> (.getResultAsync stub u/bytes-type)
+      (p/then nippy/thaw)))
 
 (defn cancel
   "

src/temporal/core.clj

@@ -16,3 +16,7 @@
    (Workflow/await (->supplier pred)))
   ([duration pred]
    (Workflow/await duration (->supplier pred))))
+
+(defn gen-uuid
+  []
+  (str (Workflow/randomUUID)))

src/temporal/internal/signals.clj

@@ -17,6 +17,7 @@
          (fn [s]
            (let [ch (or (get-ch s signal-name)
                         (ConcurrentLinkedQueue.))]
+             (log/trace "saving signal:" signal-name payload)
              (.add ch payload)
              (assoc s signal-name ch)))))
 

src/temporal/internal/utils.clj

@@ -2,8 +2,8 @@
 
 (ns ^:no-doc temporal.internal.utils
   (:require [clojure.string :as string]
-            [clojure.edn :as edn]
             [taoensso.timbre :as log]
+            [taoensso.nippy :as nippy]
             [promesa.core :as p])
   (:import [io.temporal.common.converter EncodedValues]
            [io.temporal.workflow Promise
@@ -15,6 +15,8 @@
             Functions$Func5
             Functions$Func6]))
 
+(def ^Class bytes-type (Class/forName "[B"))
+
 (defn build [builder spec params]
   (doseq [[key value] params]
     (log/trace "building" builder "->" key "=" value)
@@ -52,22 +54,22 @@
       (str "." sym)))
 
 (defn ->objarray
-  "Converts a collection of serializable elements to an array of Objects, suitable for many Temporal APIs"
-  [coll]
-  (into-array Object [(prn-str coll)]))
+  "Serializes x to an array of Objects, suitable for many Temporal APIs"
+  [x]
+  (into-array Object [(nippy/freeze x)]))
 
 (defn ->args
   "Decodes EncodedValues to native clojure data type.  Assumes all data is in the first element"
   [^EncodedValues args]
-  (edn/read-string (.get args (int 0) ^Class String)))
+  (nippy/thaw (.get args (int 0) bytes-type)))
 
 (defn wrap-encoded
-  "Wraps 'f' in a codec, EncodedValues args in, and edn out"
+  "Wraps 'f' in a codec, EncodedValues args in, and nippy-encoded bytes out"
   [f args]
   (-> args
       (->args)
       (f)
-      (prn-str)))
+      (nippy/freeze)))
 
 (def namify
   "Converts strings or keywords to strings, preserving fully qualified keywords when applicable"

src/temporal/internal/workflow.clj

@@ -26,7 +26,6 @@
 (defn execute
   [ctx args]
   (let [{:keys [workflow-type] :as info} (get-info)
-        f (u/find-annotated-fn ::def workflow-type)
-        signals (s/create)]
+        f (u/find-annotated-fn ::def workflow-type)]
     (log/trace "execute:" info)
-    (u/wrap-encoded (fn [args] (f ctx {:args args :signals signals})) args)))
+    (u/wrap-encoded (fn [args] (f ctx {:args args :signals (s/create)})) args)))

src/temporal/signals.clj

@@ -5,22 +5,43 @@
   (:require [taoensso.timbre :as log]
             [temporal.core :as core]
             [temporal.internal.utils :as u]
-            [temporal.internal.signals :as s]))
+            [temporal.internal.signals :as s])
+  (: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]
-  (let [ch (s/get-ch @state signal-name)]
-    (or (nil? ch) (not (.isEmpty ch)))))
+  (let [signal-name (u/namify signal-name)
+        ch (s/get-ch @state signal-name)
+        r (or (nil? ch) (.isEmpty ch))]
+    (log/trace "is-empty?:" @state signal-name r)
+    r))
+
+(defn- rx
+  [state signal-name]
+  (let [signal-name (u/namify signal-name)
+        ch (s/get-ch @state 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)))
 
 (defn <!
   "Light-weight/parking receive of a single message"
   ([state] (<! state ::default))
   ([state signal-name]
-   (let [signal-name (u/namify signal-name)]
-     (log/trace "waiting on:" signal-name)
-     (core/await (partial is-empty? state signal-name))
-     (let [ch (s/get-ch @state signal-name)
-           m (.poll ch)]
-       (log/trace "rx:" m)
-       m))))
+   (log/trace "waiting on:" signal-name)
+   (core/await #(not (is-empty? state signal-name)))
+   (rx state 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))))

src/temporal/workflow.clj

@@ -6,11 +6,17 @@
             [temporal.internal.utils :as u]
             [temporal.internal.workflow :as w]))
 
+(defn get-info
+  "
+Return info about the current workflow
+"
+  []
+  (w/get-info))
+
 (defmacro defworkflow
   "
 Defines a new workflow, similar to defn, expecting a 2-arity parameter list and body.  Should evaluate to something
-serializable with [prn-str](https://clojuredocs.org/clojure.core/prn-str), which will become available for
-[[temporal.client.workflow/get-result]].
+serializable, which will become available for [[temporal.client.workflow/get-result]].
 
 Arguments:
 

test/temporal/test/client_signal.clj

@@ -0,0 +1,32 @@
+;; Copyright © 2022 Manetu, Inc.  All rights reserved
+
+(ns temporal.test.client-signal
+  (: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]))
+
+(use-fixtures :once t/wrap-service)
+
+(def signal-name ::signal)
+
+(defn lazy-signals [signals]
+  (lazy-seq (when-let [m (<! signals signal-name)]
+              (cons m (lazy-signals signals)))))
+
+(defworkflow test-workflow
+  [ctx {:keys [signals] {:keys [nr] :as args} :args}]
+  (log/info "test-workflow:" args)
+  (doall (take nr (lazy-signals signals))))
+
+(def expected 3)
+
+(deftest the-test
+  (testing "Verifies that we can send signals from a client"
+    (let [workflow (t/create-workflow test-workflow)]
+      (c/start workflow {:nr expected})
+      (dotimes [n expected]
+        (>! workflow signal-name n))
+      (is (-> workflow c/get-result deref count (= expected))))))