observe Missionary task and flow events

mobserve.cljc

(ns mobserve
  (:require
   [missionary.core :as m]))

;; Encapsulate differences between Clojure and ClojureScript
;; on how IFn and IDeref are implemented.

#?(:clj
   (deftype FlowIterator [transfer cancel]
     clojure.lang.IDeref
     (deref [_this]
       (transfer))

     clojure.lang.IFn
     (invoke [_this]
       (cancel)))

   :cljs
   (deftype FlowIterator [transfer cancel]
     IDeref
     (-deref [_this]
       (transfer))

     IFn
     (-invoke [_this]
       (cancel))))

(defn println-reporter [description msg & args]
  ;; Lazy evaluation interacts badly with dynamic binding;
  ;; evaluate the pr-str before println changes the print mode.
  (let [pr-args (mapv pr-str args)]
    (apply println description msg pr-args)))

(defn observe-task
  ([description task]
   (observe-task println-reporter description task))

  ([report description task]
   (fn [s f]
     (let [cancel! (task
                    (fn [v]
                      (report description :succeeded v)
                      (s v))
                    (fn [e]
                      (report description :failed e)
                      (f e)))]
       (report description :instantiated)
       (fn []
         (report description :canceled)
         (cancel!))))))

(defn hook-call [thunk successful-call call-threw]
  (try
    (let [v (thunk)]
      ;; This isn't quite right: we catch errors thrown by
      ;; `successful-call`, not just `thunk`.
      (successful-call v)
      v)
    (catch #?(:clj Throwable, :cljs :default) e
      (call-threw e)
      (throw e))))

(defn observe-flow
  ([description flow]
   (observe-flow println-reporter description flow))

  ([report description flow]
   (fn [notifier terminator]
     (let [child-iterator
           (flow
            (fn notify []
              (report description :notified)
              (notifier))
            (fn terminate []
              (report description :terminated)
              (terminator)))]
       (report description :instantiated)
       (FlowIterator.
        (fn transfer []
          (hook-call
           (fn [] @child-iterator)
           (fn [v]
             (report description :transferred-value v))
           (fn [e]
             (report description :transferred-error e))))
       (fn cancel []
         (report description :cancel-signaled)
         (child-iterator)))))))