       #_(println "unsubscribing" (::identifier (meta unsubs-fn)))
       (unsubs-fn))))
 
-(def ^:private sum-xf
+(def ^:private sum-and-delta
   "Transducer that sums deltas and passes on [sum delta] for each
   delta."
   (fn [rf]

        ([result input]
         (rf result [(vswap! sum delta/add input) input]))))))

(defn- hybrid-cache
  "Caches on endpoints.  Sends the current sum to a new subscriber,
  continues with deltas."
  [service]
  (let [subscribe-ch (chan)
        cancel-ch (chan)]
    (go-loop [subs {}
              ch->identifier {}]
      (let [[msg port] (alts! [subscribe-ch
                               cancel-ch])]
        (case port

          (let [[this identifier target] msg
                [m :as cache]
                (or (get subs identifier)
                    (let [ch-in (chan 1 source-xf)
                          m (mult ch-in)]
                      (subscribe (caching service this)
                                 identifier
                                 ch-in)
                      [m #{} ch-in]))
                target-step (chan 1 target-xf)]
            (pipe target-step target)
            (tap m target-step)
            (recur (assoc subs identifier (update cache 1 conj target))
                   (assoc chs target [identifier target-step])))
          cancel-ch
          (let [[this target] msg]
            (if-let [[identifier target-step] (get chs target)]
              (let [[m m-chs ch-in :as cache] (get subs identifier)
                    m-chs (disj m-chs target)
                    chs (dissoc chs target)]
                (untap m target-step)
                (close! target-step)
                (if (empty? m-chs)
                  (do
                    (cancel service ch-in)
                    (recur (dissoc subs identifier)
                           chs))
                  (recur (assoc subs identifier (assoc cache 1 m-chs))
                         chs)))
              (recur subs
                     chs))))))
    (reify
      IStream
      (subscribe [this identifier target]
        (put! subscribe-ch [this identifier target]))
      (cancel [this target]
        (put! cancel-ch [this target])))))

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

(defn compscriber
  [service]
