       (subscribe [this spec target]
         (let [[spec id] spec
               [endpoint direct-hooks deep-hooks :as spec] (compile-spec spec)
-              subs-target (wrap-on-close target #(cancel this target))]
+              subs-target (on-close-source target #(cancel this target))]
           (if (and (empty? direct-hooks)
                    (empty? deep-hooks))
             ;; OPT: If there is nothing to compscribe, directly reach

    (pipe source watch-ch)
    (pipe watch-ch target)))

(defn- wrap-on-close
  "Pipes a channel to target and returns it, on-close is invoked when
  it is closed."
  [target on-close]

                                 spec
                                 ch-in)
                      [m 0 ch-in]))
                target-step (on-close-source target
                                             #(cancel this target))]
            (tap m target-step)
            (recur (assoc subs spec (update cache 1 inc))
                   (assoc chs target [spec target-step])))
