adds print-object idea, and initial flow post

Author: timothypratley

content/ideas/clojure/plus/print/objects/objects_and_protocols.clj

@@ -0,0 +1,144 @@
+(ns clojure.plus.print.objects.objects-and-protocols
+  (:require [clojure.core.async :as async]
+            [clojure.datafy :as datafy]
+            [clojure.string :as str]
+            [core.async.flow.example.stats :as stats]))
+
+;; Clojure plus includes print-methods to improve printing objects
+
+(require 'clojure+.print)
+^:kind/hidden
+(clojure+.print/install!)
+
+;; Once activated, we can print functions, atoms, namespaces, and more sensibly
+
+inc
+(atom {})
+*ns*
+
+;; Clojure Plus only adds printers for pre-defined types.
+;; No printer is provided for Object,
+;; so it falls back to Clojure's default printing method:
+
+(defmethod print-method Object [x ^java.io.Writer w]
+  (#'clojure.core/print-object x w))
+
+;; There are plenty of objects left over that print messily
+
+(def my-chan
+  (async/chan))
+
+my-chan
+
+;; Clojure's Object print-method calls print-object,
+;; which prints the ugly [type-str hash-str type-str].
+
+;; Clojure Plus could clean that up like this:
+
+(defmethod print-method Object [x ^java.io.Writer w]
+  (.write w "#object ")
+  (.write w (.getName (class x))))
+
+my-chan
+
+;; Much nicer!
+;; This solves the majority of "left over" situations.
+;; However, there are still some ugly ones left...
+
+(def stats-flow
+  (stats/create-flow))
+
+stats-flow
+
+;; What is this? It's a reified object that implements protocols.
+;; We can see this by the $reify part at the end.
+;; The description is not terrible, at least we know where it was made,
+;; which hints that it must be a flow.
+;; Can we do better?
+
+;; AFAIK the only way to check what protocols an object satisfies
+;; is to call `satisfies?` for every possible protocol:
+
+(defn all-protocol-vars [x]
+  (->> (all-ns)
+       (mapcat ns-publics)
+       (vals)
+       (keep #(-> % meta :protocol))
+       (distinct)
+       (filter #(satisfies? @% x))))
+
+;; On the one hand, this is concerning for performance.
+;; On the other hand, at my REPL I don't care about that, it's faster than I can notice.
+;; Leaving aside those concerns, it returns quite a long list...
+
+(all-protocol-vars stats-flow)
+
+;; But notice that one of them; `#'clojure.core.async.flow.impl.graph/Graph`
+;; just feels like it is the one we care about most.
+;; Furthermore, it shares a similar namespace with the classname.
+;; Let's try matching by the namespace...
+
+(defn var-ns-name [v]
+  (-> (meta v) (:ns) (ns-name)))
+
+(defn ns-match? [p x]
+  (-> (var-ns-name p)
+      (str/starts-with? (.getPackageName (class x)))))
+
+(defn protocol-ns-matches [x]
+  (filter #(ns-match? % x) (all-protocol-vars x)))
+
+(protocol-ns-matches stats-flow)
+
+;; Nice.
+;; In my opinion this is more representative of the object.
+;; The `#'` out front is unnecessary and can be removed...
+
+(defn var-sym [v]
+  (let [m (meta v)]
+    (symbol (str (ns-name (:ns m))) (str (:name m)))))
+
+(defn protocol-ns-match-names [x]
+  (->> (protocol-ns-matches x)
+       (map var-sym)))
+
+(protocol-ns-match-names stats-flow)
+
+;; The other protocol of interest is Datafiable,
+;; because it indicates I can get a data representation if I would like to.
+
+(datafy/datafy stats-flow)
+
+;; I think this one is so helpful that it should always be shown on objects,
+;; regardless of their type of other protocols,
+;; as a hint that it is possible to get more information.
+;; I wouldn't want to print them as data by default, because it would be too spammy.
+;; And checking Datafiable is much less of a performance concern.
+
+(satisfies? clojure.core.protocols/Datafiable stats-flow)
+
+;; But there is a big problem... everything is Datafiable...
+
+(satisfies? clojure.core.protocols/Datafiable (Object.))
+
+;; So there is no way for us to know whether `datafy/datafy` will do anything useful or not.
+;; Sad.
+;; But we can improve the print-method to show protocols,
+;; bearing in mind it is a performance concern.
+
+(defmethod print-method Object [x ^java.io.Writer w]
+  (let [class-name (.getName (class x))
+        r (str/includes? class-name "$reify")
+        p (when r (first (protocol-ns-match-names x)))]
+    (.write w (if p
+                "#reify "
+                "#object "))
+    (.write w (if p
+                (str p)
+                class-name))))
+
+my-chan
+stats-flow
+
+;; Even if we don't care to improve reify (due to performance),
+;; I think the Object printer should still be improved to align with the other printers.

content/posts/core/async/flow/example/stats.clj

@@ -0,0 +1,200 @@
+(ns core.async.flow.example.stats)
+
+;; # Core async flow stats example
+;;
+;; Adapted from https://github.com/puredanger/flow-example
+
+^:kind/video
+{:youtube-id   "lXFwf3O4BVY"
+ :iframe-width "100%"}
+
+;; Necessary dependencies
+
+(clojure.repl.deps/add-libs '{org.clojure/core.async {:mvn/version "1.9.808-alpha1"}})
+(clojure.repl.deps/add-libs '{io.github.clojure/core.async.flow-monitor {:git/tag "v0.1.1" :git/sha "61e8d31"}})
+
+(require
+  '[clojure.core.async :as a]
+  '[clojure.core.async.flow :as flow]
+  '[clojure.core.async.flow-monitor :as mon])
+
+(defn stat-gen
+  "Generates a random value between min (inclusive) and max (exclusive)
+  and writes it to out chan, waiting wait ms between until stop-atom is flagged."
+  ([out min max wait stop-atom]
+   (loop []
+     (let [val (+ min (rand-int (- max min)))
+           put (a/>!! out val)]
+       ;(println "stat-gen" (System/identityHashCode stop-atom) val put (not @stop-atom))
+       (when (and put (not @stop-atom))
+         (^[long] Thread/sleep wait)
+         (recur))))))
+
+(defn source
+  "Source proc for random stats"
+  ;; describe
+  ([] {:params {:min "Min value to generate"
+                :max "Max value to generate"
+                :wait "Time in ms to wait between generating"}
+       :outs {:out "Output channel for stats"}})
+
+  ;; init
+  ([args]
+   (assoc args
+     ::flow/in-ports {:stat (a/chan 100)}
+     :stop (atom false)))
+
+  ;; transition
+  ([{:keys [min max wait ::flow/in-ports] :as state} transition]
+   ;(println "transition" transition)
+   (case transition
+     ::flow/resume
+     (let [stop-atom (atom false)]
+       (future (stat-gen (:stat in-ports) min max wait stop-atom))
+       (assoc state :stop stop-atom))
+
+     (::flow/pause ::flow/stop)
+     (do
+       (reset! (:stop state) true)
+       state)))
+
+  ;; transform
+  ([state in msg]
+   ;(println "source transform" in msg)
+   [state (when (= in :stat) {:out [msg]})]))
+
+(defn aggregator
+  ;; describe
+  ([] {:params {:min "Min value, alert if lower"
+                :max "Max value, alert if higher"}
+       :ins {:stat "Channel to receive stat values"
+             :poke "Channel to poke when it is time to report a window of data to the log"}
+       :outs {:alert "Notify of value out of range {:val value, :error :high|:low"}
+       :workload :compute
+       })
+
+  ;; init
+  ([args] (assoc args :vals []))
+
+  ;; transition
+  ([state transition] state)
+
+  ;; transform
+  ([{:keys [min max vals] :as state} input-id msg]
+   (case input-id
+     :stat (let [state' (assoc state :vals (conj vals msg))
+                 msgs (cond
+                        (< msg min) {:alert [{:val msg, :error :low}]}
+                        (< max msg) {:alert [{:val msg, :error :high}]}
+                        :else nil)]
+             [state' msgs])
+     :poke [(assoc state :vals [])
+            {::flow/report (if (empty? vals)
+                             [{:count 0}]
+                             [{:avg (/ (double (reduce + vals)) (count vals))
+                               :count (count vals)}])}]
+     [state nil])))
+
+(comment
+  ;; test aggregator alert case - no channels involved
+  (let [state {:min 1 :max 5 :vals []}
+        [state' msgs'] (aggregator state :stat 100)]
+    (assert (= msgs' {:alert [{:val 100, :error :high}]})))
+  )
+
+
+(defn scheduler
+  ;; describe
+  ([] {:params {:wait "Time to wait between pokes"}
+       :outs {:out "Poke channel, will send true when the alarm goes off"}})
+
+  ;; init
+  ([args]
+   (assoc args
+     ::flow/in-ports {:alarm (a/chan 10)}
+     :stop (atom false)))
+
+  ;; transition
+  ([{:keys [wait ::flow/in-ports] :as state} transition]
+   ;(println "scheduler transition" transition state transition)
+   (case transition
+     ::flow/resume
+     (let [stop-atom (atom false)]
+       (future (loop []
+                 (let [put (a/>!! (:alarm in-ports) true)]
+                   (when (and put (not @stop-atom))
+                     (^[long] Thread/sleep wait)
+                     (recur)))))
+       (assoc state :stop stop-atom))
+
+     (::flow/pause ::flow/stop)
+     (do
+       (reset! (:stop state) true)
+       state)))
+
+  ;; transform
+  ([state in msg]
+   [state (when (= in :alarm) {:out [true]})]))
+
+(defn printer
+  ;; describe
+  ([] {:params {:prefix "Log message prefix"}
+       :ins {:in "Channel to receive messages"}})
+
+  ;; init
+  ([state] state)
+
+  ;; transition
+  ([state _transition] state)
+
+  ;; transform
+  ([{:keys [prefix] :as state} _in msg]
+   (println prefix msg)
+   [state nil]))
+
+(def config
+  {:procs {:generator  {:args {:min 0 :max 12 :wait 500} :proc (flow/process #'source)}
+           :aggregator {:args {:min 1 :max 10} :proc (flow/process #'aggregator)}
+           :scheduler  {:args {:wait 3000} :proc (flow/process #'scheduler)}
+           :notifier   {:args      {:prefix "Alert: "} :proc (flow/process #'printer)
+                        :chan-opts {:in {:buf-or-n (a/sliding-buffer 3)}}}}
+   :conns [[[:generator :out] [:aggregator :stat]]
+           [[:scheduler :out] [:aggregator :poke]]
+           [[:aggregator :alert] [:notifier :in]]]})
+
+(defn create-flow
+  []
+  (flow/create-flow config))
+
+(comment
+  (def f (create-flow))
+  (def chs (flow/start f))
+  (flow/resume f)
+  (flow/pause f)
+  (flow/stop f)
+
+  (def server (mon/start-server {:flow f}))
+  (mon/stop-server server)
+
+  @(flow/inject f [:aggregator :poke] [true])
+  @(flow/inject f [:aggregator :stat] ["abc1000"]) ;; trigger an alert
+  @(flow/inject f [:notifier :in] [:sandwich])
+
+  (def report-chan (:report-chan chs))
+  (flow/ping f)
+  (a/poll! report-chan)
+  (def error-chan (:error-chan chs))
+  (a/poll! error-chan)
+
+  (flow/stop f)
+  (a/close! stat-chan)
+
+  @(flow/inject f [:aggregator :poke] [true])
+
+  (require '[clojure.datafy :as datafy])
+  (datafy/datafy f)
+
+  (require '[clojure.core.async.flow-static :refer [graph]])
+  (graph f)
+
+  )