       service/IService
       (request [this spec target]
         (let [[spec id] spec
-              [endpoint direct-hooks deep-hooks :as spec] (compile-spec spec)
+              [endpoint spec-map] spec
               subs-target (on-close-source target
                                            #(service/cancel this target))]
-          (if (and (empty? direct-hooks)
-                   (empty? deep-hooks))
+          (if (empty? spec-map)
             ;; OPT: If there is nothing to compscribe, directly reach
             ;; through to the source-service:
             (do
               (service/request source-service [endpoint id] subs-target)
-              (swap! subscriptions assoc target
-                     #(service/cancel source-service subs-target)))
-            (do
-              (swap! subscriptions assoc target
-                     (compscribe* subs-target
-                                  source-service
-                                  (service/cached this)
-                                  spec
-                                  id))))))
+              (swap! subscriptions assoc target subs-target))
+            (let [ch-in (chan)]
+              (compscribe* subs-target
+                           ch-in
+                           (service/cached this)
+                           spec-map)
+              (service/request source-service [endpoint id] ch-in)
+              (swap! subscriptions assoc target ch-in)))))
       (cancel [this target]
-        (when-let [unsubs-fn (swap-out! subscriptions target)]
-          (unsubs-fn))))))
+        (when-let [ch (swap-out! subscriptions target)]
+          (service/cancel source-service ch))))))
 
 (defn compscriber
   "Return a service for requesting composite streams of multiple

             :or {cache dc/sum-cache}}]
  (let [service (-> source
                    (compscribe-service)
                    (cache))
        compile-spec (memoize compile-spec)]
    (reify
      service/IService
      (request [_ spec target]
        (service/request service 
                         (update spec 0 compile-spec)
                         target))
      (cancel [_ target]
        (service/cancel service target)))))

                                               pipe
                                               #?@(:clj [go go-loop])
                                               tap untap] :as a]
            [clojure.walk :refer [prewalk postwalk]])
  #?(:cljs (:require-macros [cljs.core.async.macros :refer [go-loop go]])))

(defprotocol ClosingMix
  "Implementation detail, subject to change."
