   and close target-ch."
   {:arglists '([target-ch service spec id])} 
   ([target-ch service spec id]
+   (let [service (compscriber service)]
+     (subscribe service [spec id] target-ch)))
+  ([target-ch subs-fn unsubs-fn spec id]
    (compscribe target-ch
-               (fn [endpoint id]
-                 (subscribe service [endpoint id]))
-               (fn [ch]
-                 (cancel service ch))))
-  ([target-ch subs-fn unsubs-fn spec id]
-   (let [conformed-spec (conform-spec spec)] ;; could allow compiled
-                                             ;; spec by checking for
-                                             ;; ::spec metadata
-     (compscribe* target-ch subs-fn unsubs-fn conformed-spec id))))
+               (let [chs (atom {})]
+                 (reify
+                   IStream
+                   (subscribe [this [endpoint id] target]
+                     (let [ch-in (subs-fn endpoint id)]
+                       (swap! chs assoc target ch-in)
+                       (on-close-pipe ch-in target
+                                      #(cancel this target))))
+                   (cancel [_ ch]
+                     (some-> (swap-out! chs ch)
+                             (unsubs-fn)))))
+               spec id)))

  (doto (chan)
    (on-close-pipe target on-close)))


(defn compscribe
  "Asynchronously subscribes via subs-fn and unsubs-fn at one or more

                                         (some-> (find @subs ks)
                                                 (vector)))))
                       distinct)]
            (cancel compscribe-service xch)
            (mix-out target-mix xch)
            (vswap! subs dissoc ks))

