         ([] (rf))
         ([result] (rf result))
         ([result input]
-         (rf result [(vswap! sum delta/add input) input]))))))
+         (rf result [[:is (vswap! sum delta/add input)] input]))))))
 
 (def ^:private first-then-second
   (fn [rf]


(defn compscriber
  [service]
  (let [service (compscribe-service service)
        service (reify
                  IStream
                  (subscribe [this identifier target]
                    (-> service
                        (caching this)
                        (subscribe identifier target)))
                  (cancel [this target]
                    (cancel service target)))
        compile-spec (memoize compile-spec)]
    (reify
      IStream

              chs {}]
      (let [[msg port] (alts! [subscribe-ch
                               cancel-ch])]
        #_(println [msg port subscribe-ch cancel-ch])
        (condp identical? port
          subscribe-ch
          (let [[this identifier target] msg
                [m :as cache]
