   (doto (chan)
     (on-close-pipe target on-close)))
 
-(defrecord CompscribeService [cached-service source-service subscriptions]
+(defn- caching
+  [service cached-service]
+  (vary-meta service assoc ::cached-service cached-service))
+
+(defn- this-cached
+  [service]
+  (::cached-service (meta service) service))
+
+(defrecord CompscribeService [source-service subscriptions]
   IStream
   (subscribe [this identifier target]
     (let [[spec id] identifier

          (swap! subscriptions assoc target
                 (vary-meta (compscribe* subs-target
                                         source-service
                                         cached-service
                                         spec
                                         id)
                            assoc ::identifier identifier))))))

                  IStream
                  (subscribe [this identifier target]
                    (-> service
                        (caching this)
                        (subscribe identifier target)))
                  (cancel [this target]
                    (cancel service target)))
