                                  spec
                                  ch-in)
                       [m 0 ch-in]))
-                target-step (wrap-on-close target
-                                           #(cancel this target))]
+                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])))

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

      (subscribe [this spec target]
        (let [[spec id] spec
              [endpoint direct-hooks deep-hooks :as spec] (compile-spec spec)
              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
