           (unsubs-fn))))))
 
 (defn- caching-mult
-  [ch accumulate]
+  [ch {:keys [accumulate
+              mode]
+       :or {mode :value}}]
   (let [tap-ch (chan)
         chs (atom #{})
         dctr (atom nil)

              (reset! chs nil)
              (recur cache
                     false))
            (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)
                       true))))))
    m))

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

  "Caches on endpoints.  Sends the current sum to a new subscriber,
  continues with deltas."
  [service]
  (cached-service service {:accumulate
                           (fn [cache v]
                             #_(println ["Hybrid cache processing"
                                         [cache v]])
                             (update (or cache [:is]) 1
                                     delta/add v))}))

(defn compscriber
  [service]
