           (async/close! requests))))))
 
 (defn- promise->subscription
+  "Coerce a promise instance to an instance
+  of subscription interface."
   [subscriber source]
   (let [canceled (atomic/boolean false)
         completed (atomic/boolean false)]

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(defn- chan->subscription
  [subscriber source xform]
  (let [completed (atomic/boolean false)
        requests (async/chan)

          (.onComplete subscriber))))))

(defn- subscribe-once
  "Create a subscription instance that only
  subscribes for one value."
  [^Publisher p callback]
  (let [subscription (atomic/ref nil)
        completed (atomic/boolean false)
