                                                alts!
                                                pipe
                                                #?@(:clj [go go-loop])
-                                               mult tap untap] :as a]
+                                               tap untap] :as a]
             [clojure.walk :refer [prewalk postwalk]]
             [commos.shared.core :refer [flatten-keys]])
   #?(:cljs (:require-macros [cljs.core.async.macros :refer [go-loop go]])))

  "Caches on endpoints.  Sends the current sum to a new subscriber,
  continues with deltas."
  [service]
  (cached-service service sum-and-delta first-then-second))

(defn compscriber
  [service]

          #_(println "unsubscribing" (::identifier (meta unsubs-fn)))
          (unsubs-fn))))))

(defn- caching-mult
  [ch accumulate]
  (let [tap-ch (chan)
        chs (atom #{})
        dctr (atom nil)
        dchan (chan 1)
        done (fn [_] (when (zero? (swap! dctr dec))
                       (put! dchan true)))
        m (reify
            a/Mux
            (muxch* [_] ch)
            a/Mult
            (tap* [_ ch close?]
              (if close?
                (put! tap-ch ch)
                (throw (throw (UnsupportedOperationException.
                               "close?=false not supported")))))
            (untap* [_ ch]
              (swap! chs disj ch))
            (untap-all* [_] (throw (throw (UnsupportedOperationException.)))))]
    (go-loop [cache nil]
      (let [[v port] (alts! [tap-ch ch] :priority true)]
        (condp identical? port
          tap-ch
          (do (when (or (nil? cache)
                        (put! v cache))
                (swap! chs conj v))
              (recur cache))
          ch
          (if (nil? v)
            (run! close! @chs)
            (do (when-let [chs (seq @chs)]
                  (reset! dctr (count chs))
                  (doseq [ch chs]
                    (when-not (put! ch v done)
                      (a/untap* m ch)))
                  (<! dchan))
                (recur (accumulate cache v)))))))
    m))

(defn- cached-service
  [service accumulate]
  (let [subscribe-ch (chan)
        cancel-ch (chan)]
    (go-loop [subs {}
