Support providing custom transit write handlers

Author: JarrodCTaylor

README.md

@@ -40,6 +40,38 @@ Add the following dependency to your project:
 ;; http://localhost:9876/index.html#/?port=9876
 ```
 
+## Custom Transit Handlers
+
+You can provide custom Transit write handlers to properly serialize types that aren't natively supported by Transit (used for visualizing state). These handlers should follow the format expected by cognitect.transit/writer :handlers. If not provided, the default handler will be used, which converts objects to strings.
+
+```clojure
+(:require
+  [clojure.core.async.flow-monitor :as monitor]
+  [clojure.core.async.flow :as flow]
+  [cognitect.transit :as transit]
+  [java.time :as time])
+(:import
+  [java.time Instant ZoneId]
+  [java.time.format DateTimeFormatter])
+
+(def my-flow (flow/create-flow ...))
+
+(defn format-instant [instant]
+  (.format
+    (.withZone
+      (DateTimeFormatter/ofPattern "MMMM d, yyyy")
+      (ZoneId/systemDefault))
+    instant))
+
+(def instant-write-handler
+  (transit/write-handler "instant-long"
+                         (fn [instant] (format-instant instant))))
+
+(def server-state (monitor/start-server {:flow my-flow
+                                         :port 9876
+                                         :handlers {Instant instant-write-handler}}))
+```
+
 ### Stopping the Server
 
 ```clojure

resources/public/assets/js/compiled/main.js

@@ -20,13 +20,16 @@
                     :loop-ping? false
                     :flow nil})
 
-(defn clj->transit-str [arg]
+(def default-write-handler (transit/write-handler "default" (fn [obj] (str obj))))
+
+(defn transit-str-writer [data user-handlers]
   (let [out (ByteArrayOutputStream. 4096)
-        writer (transit/writer out :json)]
-    (transit/write writer arg)
+        writer (transit/writer out :json {:handlers user-handlers
+                                          :default-handler default-write-handler})]
+    (transit/write writer data)
     (.toString out)))
 
-(defn transit-str->clj [arg]
+(defn transit-str-reader [arg]
   (let [arg-json (json/read-str arg)
         arg-bytes (.getBytes arg-json "UTF-8")
         in (ByteArrayInputStream. arg-bytes)
@@ -42,17 +45,17 @@
 
 (defn send-message [state message]
   (doall (for [channel (:channels @state)]
-           (httpkit/send! channel (clj->transit-str message)))))
+           (httpkit/send! channel (transit-str-writer message (:handlers @state))))))
 
 (defn loop-ping [state]
   (async/thread
     (loop [s state]
       (if (:loop-ping? @s)
-        (do (send-message state {:action :ping :data (reduce-kv
-                                                       (fn [res k v]
-                                                         (assoc res k (mainline-chan-meta v)))
-                                                       {}
-                                                       (flow/ping (:flow @state)))})
+        (do (send-message state {:action :ping :data (d/datafy (reduce-kv
+                                                                 (fn [res k v]
+                                                                   (assoc res k (mainline-chan-meta v)))
+                                                                 {}
+                                                                 (flow/ping (:flow @state))))})
             (Thread/sleep 1000)
             (recur s))
         (println "Ping loop stopped")))))
@@ -67,7 +70,7 @@
                 (swap! state assoc-in [:loop-ping?] true)
                 (loop-ping state))
      :on-receive (fn [ch data]
-                   (let [clj-data (transit-str->clj data)
+                   (let [clj-data (transit-str-reader data)
                          action (:action clj-data)]
                      (case action
                        :inject (flow/inject (:flow @state) (:target clj-data) (edn/read-string (:data clj-data)))
@@ -115,14 +118,17 @@
   - opts: A map with the following keys:
     - :flow (required) - The return value from clojure.core.async.flow/create-flow
     - :port (optional) - The port to run the server on (default: 9998)
+    - :handlers (optional) - A map of custom Transit write handlers to use when serializing state
+                             data to send to the frontend. These handlers should follow the format
+                             expected by cognitect.transit/writer :handlers
 
   Returns:
   An atom containing the server's state, and prints a local url where the frontend can be reached"
-  [{:keys [flow port] :or {port 9998}}]
+  [{:keys [flow port handlers] :or {port 9998}}]
   (let [state (atom default-state)
         error-chan (:clojure.datafy/obj (meta (:error (:chans (d/datafy flow)))))
         report-chan (:clojure.datafy/obj (meta (:report (:chans (d/datafy flow)))))]
-    (swap! state assoc :flow flow)
+    (swap! state assoc :flow flow :handlers handlers)
     (report-monitoring state report-chan error-chan)
     (let [server (httpkit/run-server (app state) {:port port
                                                   :max-body 100000000

src/clojure/core/async/flow_monitor.clj

@@ -116,8 +116,21 @@
 (defn send-socket-data [data]
   (.send (:ws-chan @global-state) (.stringify js/JSON (t/write writer data))))
 
+(defn tagged-value? [^js v]
+  (try
+    (and (.-tag v) (.-rep v))
+    (catch :default _ false)))
+
+(defn transform-tagged-values [data]
+  (cond
+    (tagged-value? data) [(.-tag ^js data) (transform-tagged-values (.-rep ^js data))]
+    (map? data) (reduce-kv (fn [m k v] (assoc m k (transform-tagged-values v))) {} data)
+    (vector? data) (mapv transform-tagged-values data)
+    (set? data) (into #{} (map transform-tagged-values data))
+    :else data))
+
 (defn process-ws-message [msg]
-  (let [data (->> msg .-data (t/read reader))
+  (let [data (->> msg .-data (t/read reader) transform-tagged-values)
         action (:action data)]
     (case action
       :datafy (do