       #_(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]

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

(defn compscriber
  [service]

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

(def ^:private sum-then-deltas
  (fn [rf]
    (let [done (volatile! false)]
      (fn
        ([] (rf))
        ([result] (rf result))
        ([result input]
         (if @done
           (rf result (second input))
           (do (vreset! done true)
               (rf result (first input)))))))))

(defn- cached-service
  [service source-xf target-xf]
  (let [subscribe-ch (chan)
        cancel-ch (chan)]
    (go-loop [subs {}
              chs {}]
      (let [[msg port] (alts! [subscribe-ch
                               cancel-ch])]
        (case port
