feat: allow selecting payload encoding per value

src/temporal/converter/default.clj

@@ -1,19 +1,77 @@
 (ns temporal.converter.default
   (:require
+   [temporal.converter.encoding :as encoding]
    [temporal.converter.json :as json]
    [temporal.converter.nippy :as nippy])
   (:import
-   [io.temporal.common.converter DefaultDataConverter NullPayloadConverter PayloadConverter]))
+   [io.temporal.common.converter
+    DataConverterException
+    DefaultDataConverter
+    NullPayloadConverter
+    PayloadAndFailureDataConverter
+    PayloadConverter]
+   [java.lang.reflect Field]
+   [java.util Map]))
+
+(set! *warn-on-reflection* true)
 
 (def standard-converters
   [(NullPayloadConverter.)
    (nippy/create)
    (json/create)])
 
+(defn get-field ^Field [field]
+  (doto (.getDeclaredField PayloadAndFailureDataConverter field)
+    (.setAccessible true)))
+
+(def ^Field converters-field (get-field "converters"))
+
+(def ^Field converters-map-field (get-field "convertersMap"))
+
+(def ^Field failure-converter-field (get-field "failureConverter"))
+
+(def ^Field serialization-context-field (get-field "serializationContext"))
+
+(defn get-converter [^PayloadAndFailureDataConverter data-converter encoding]
+  (let [converters-map (.get converters-map-field data-converter)
+        converter (.get ^Map converters-map encoding)]
+    (when converter
+      (if-let [serialization-context (get serialization-context-field data-converter)]
+        (.withContext ^PayloadConverter converter serialization-context)
+        converter))))
+
+(defn encoded->payload
+  "This is a custom re-implementation of the PayloadAndFailureDataConverter/toPayload"
+  [^PayloadAndFailureDataConverter data-converter {:keys [value encoding]}]
+  (if-let [converter (get-converter data-converter encoding)]
+    (let [result (.toData ^PayloadConverter converter value)]
+      (if (.isPresent result)
+        result
+        (throw (DataConverterException.
+                (str "Cannot encode value with the selected converter: " value)))))
+    (throw (DataConverterException.
+            (str "No PayloadConverter is registered for this encoding: " encoding)))))
+
 (defn create
   (^DefaultDataConverter []
    (create standard-converters))
 
+  (^DefaultDataConverter [converters source-data-converter]
+   (let [data-converter (create converters)]
+     (doseq [^Field field [converters-field converters-map-field failure-converter-field]]
+       (->> (.get field source-data-converter)
+            (.set field data-converter)))
+     data-converter))
+
   (^DefaultDataConverter [converters]
-   (DefaultDataConverter.
-    (into-array PayloadConverter converters))))
+   (proxy [DefaultDataConverter] [(into-array PayloadConverter converters)]
+     (withContext [context]
+       (let [data-converter (create converters this)]
+         (.set serialization-context-field data-converter context)
+         data-converter))
+
+     (toPayload [value]
+       (let [^PayloadAndFailureDataConverter this this]
+         (if (encoding/tagged? value)
+           (encoded->payload this value)
+           (proxy-super toPayload value)))))))

src/temporal/converter/encoding.clj

@@ -0,0 +1,14 @@
+(ns temporal.converter.encoding)
+
+(defrecord Tagged [value encoding])
+
+(def mapping
+  {:null "binary/null"
+   :nippy "binary/plain"
+   :json "json/plain"})
+
+(defn tagged? [x]
+  (instance? Tagged x))
+
+(defn with [value encoding]
+  (->Tagged value (get mapping encoding encoding)))

test/temporal/converter/default_test.clj

@@ -1,13 +1,22 @@
 (ns temporal.converter.default-test
   (:require
-   [clojure.test :refer [are deftest is]]
+   [clojure.template :refer [do-template]]
+   [clojure.test :refer [are deftest is testing]]
+   [jsonista.core :as json]
    [taoensso.nippy :as nippy]
    [temporal.converter.default :as sut]
+   [temporal.converter.encoding :as encoding]
    [temporal.converter.metadata :refer [->metadata]]
    [temporal.converter.payload :refer [->payload]])
   (:import
    [io.temporal.api.common.v1 Payload]
-   [io.temporal.common.converter DataConverterException EncodingKeys NullPayloadConverter]))
+   [io.temporal.common.converter
+    DataConverterException
+    EncodingKeys
+    NullPayloadConverter
+    PayloadAndFailureDataConverter]
+   [io.temporal.payload.context WorkflowSerializationContext]
+   [java.lang.reflect Field]))
 
 (defn payload->encoding [^Payload payload]
   (.. payload
@@ -23,31 +32,90 @@
     (and (= "binary/plain" (payload->encoding payload))
          (= expected (-> payload
                          (.. (getData) (toByteArray))
-                         nippy/thaw)))))
+                         (nippy/thaw))))))
+
+(defn json-payload? [expected]
+  (fn [^Payload payload]
+    (and (= "json/plain" (payload->encoding payload))
+         (= expected (-> payload
+                         (.. (getData) (toByteArray))
+                         (json/read-value json/keyword-keys-object-mapper))))))
+
+(def context
+  (WorkflowSerializationContext. "test" "test"))
 
 (deftest to-payload-with-default-converters
-  (let [data-converter (sut/create)]
-    (are [predicate input-value] (-> data-converter
-                                     (.toPayload input-value)
-                                     (.get)
-                                     (predicate))
-      null-payload? nil
-      (nippy-payload? :something) :something
-      (nippy-payload? "Something") "Something"
-      (nippy-payload? [1 2 3 4]) [1 2 3 4]
-      (nippy-payload? {:foo "bar"}) {:foo "bar"})))
+  (do-template
+   [predicate input-value]
+   (let [data-converter (sut/create)]
+     (testing "withhout-context"
+       (is (predicate (-> data-converter
+                          (.toPayload input-value)
+                          (.get)))))
+
+     (testing "with-context"
+       (let [data-converter (.withContext data-converter context)]
+         (is (= context (.get sut/serialization-context-field data-converter)))
+         (is (predicate (-> data-converter
+                            (.toPayload input-value)
+                            (.get)))))))
+   null-payload? nil
+   null-payload? (encoding/with nil :null)
+   null-payload? (encoding/with nil "binary/null")
+
+   (nippy-payload? :something) :something
+   (nippy-payload? "Something") "Something"
+   (nippy-payload? [1 2 3 4]) [1 2 3 4]
+   (nippy-payload? {:foo "bar"}) {:foo "bar"}
+
+   (nippy-payload? :something) (encoding/with :something :nippy)
+   (nippy-payload? "Something") (encoding/with "Something" :nippy)
+   (nippy-payload? [1 2 3 4]) (encoding/with [1 2 3 4] :nippy)
+   (nippy-payload? {:foo "bar"}) (encoding/with {:foo "bar"} :nippy)
+
+   (nippy-payload? :something) (encoding/with :something "binary/plain")
+   (nippy-payload? "Something") (encoding/with "Something" "binary/plain")
+   (nippy-payload? [1 2 3 4]) (encoding/with [1 2 3 4] "binary/plain")
+   (nippy-payload? {:foo "bar"}) (encoding/with {:foo "bar"} "binary/plain")
+
+   (json-payload? nil) (encoding/with nil :json)
+   (json-payload? "Something") (encoding/with "Something" :json)
+   (json-payload? [1 2 3 4]) (encoding/with [1 2 3 4] :json)
+   (json-payload? {:foo "bar"}) (encoding/with {:foo "bar"} :json)
+
+   (json-payload? nil) (encoding/with nil "json/plain")
+   (json-payload? "Something") (encoding/with "Something" "json/plain")
+   (json-payload? [1 2 3 4]) (encoding/with [1 2 3 4] "json/plain")
+   (json-payload? {:foo "bar"}) (encoding/with {:foo "bar"} "json/plain")))
 
 (deftest to-payload-with-custom-converters
   (let [data-converter (sut/create [(NullPayloadConverter.)])]
-    (is (null-payload? (.. data-converter (toPayload nil) (get))))
+    (are [input-value] (null-payload? (.. data-converter (toPayload input-value) (get)))
+      nil
+      (encoding/with nil :null))
 
     (are [input-value] (thrown-with-msg? DataConverterException
                                          #"No PayloadConverter is registered with this DataConverter"
                                          (.toPayload data-converter input-value))
       :something
       "Something"
       [1 2 3 4]
-      {:foo "bar"})))
+      {:foo "bar"})
+
+    (are [input-value] (thrown-with-msg? DataConverterException
+                                         #"Cannot encode value with the selected converter"
+                                         (.toPayload data-converter input-value))
+      (encoding/with :something :null)
+      (encoding/with :something "binary/null"))
+
+    (are [input-value] (thrown-with-msg? DataConverterException
+                                         #"No PayloadConverter is registered for this encoding"
+                                         (.toPayload data-converter input-value))
+      (encoding/with nil :nippy)
+      (encoding/with nil "binary/plain")
+
+      (encoding/with nil :json)
+      (encoding/with nil "json/plain"))))
 
 (def metadata-null
   (->metadata "binary/null"))

test/temporal/converter/metadata_test.clj

@@ -13,6 +13,7 @@
 (deftest metadata?
   (are [input] (is (sut/metadata? input))
     {}
+    sut/empty
     {"custom" (->byte-string "data")})
 
   (are [input] (is (not (sut/metadata? input)))

test/temporal/converter/optional_test.clj

@@ -1,7 +1,7 @@
 (ns temporal.converter.optional-test
   (:require
    [clojure.template :refer [do-template]]
-   [clojure.test :refer [are deftest is testing]]
+   [clojure.test :refer [deftest is testing]]
    [temporal.converter.optional :as sut])
   (:import
    [java.util Optional]))