   "Coerce a core.async channel to an instance
   of subscription interface."
   {:no-doc true}
-  ([subscriber source xform]
-   (chan->subscription subscriber source xform false))
-  ([subscriber source xform close]
+  ([subscriber source]
+   (chan->subscription subscriber source false))
+  ([subscriber source close]
    (let [completed (atomic/boolean false)
          requests (async/chan)
-         xform (or xform (map identity))
-         rf (xform (fn
-                     ([s] s)
-                     ([s v] (.onNext s v))))]
-     (letfn [(pipeline [n]
-               (go-loop [n n]
-                 (when (and (pos? n) (not @completed))
-                   (if-let [value (<! source)]
-                     (let [res (rf subscriber value)]
-                       (if (reduced? res)
-                         (do
-                           (.onComplete (rf subscriber))
-                           (async/close! requests)
-                           (when close (async/close! source))
-                           (atomic/compare-and-set! completed false true))
-                         (recur (dec n))))
-                     (do
-                       (.onComplete subscriber)
-                       (async/close! requests)
-                       (atomic/compare-and-set! completed false true))))))]
-       (go-loop []
-         (when-let [n (<! requests)]
-           (when-not @completed
-             (<! (pipeline n))
-             (recur))))
-       (reify Subscription
-         (^void request [_ ^long n]
-           (when-not @completed
-             (async/put! requests n)))
-         (^void cancel [_]
-           (atomic/set! completed true)
-           (when close (async/close! source))
-           (async/close! requests)))))))
+         pipeline (fn [n]
+                    (go-loop [n n]
+                      (when (and (pos? n) (not @completed))
+                        (if-let [value (<! source)]
+                          (do
+                            (.onNext subscriber value)
+                            (recur (dec n)))
+                          (do
+                            (.onComplete subscriber)
+                            (async/close! requests)
+                            (when close (async/close! source))
+                            (atomic/compare-and-set! completed false true))))))]
+     (go-loop []
+       (when-let [n (<! requests)]
+         (when-not @completed
+           (<! (pipeline n))
+           (recur))))
+     (reify Subscription
+       (^void request [_ ^long n]
+         (when-not @completed
+           (async/put! requests n)))
+       (^void cancel [_]
+         (atomic/set! completed true)
+         (when close (async/close! source))
+         (async/close! requests))))))
 
 (defn- promise->subscription
   "Coerce a promise instance to an instance


    Publisher
    (^void subscribe [_ ^Subscriber subscriber]
      (let [subscription (chan->subscription subscriber source xform)]
        (.onSubscribe subscriber subscription)))))

(defn- publisher->publisher
  "Create a publisher from an other publisher."
  [source xform]
  (reify Publisher
    Seqable
    (seq [p]
      (publisher->seq p))

    IPullStream
    (pull [p]
      (p/promise #(subscribe-once p %)))

    (^void subscribe [_ ^Subscriber subscriber]
      (let [subscriber (proxy-subscriber source xform subscriber)]
        (.subscribe source subscriber)))))

(defn- iterable->publisher
  "Create a publisher with iterable as source."
  [source xform]
  (let [source' (async/chan)
        _       (async/onto-chan source' (seq source))]
    (reify
      IPullStream
      (pull [p]


(defn- promise->publisher
  "Create a publisher with promise as source."
  [source]
  (reify Publisher
    (^void subscribe [_ ^Subscriber subscriber]
      (let [subscription (promise->subscription subscriber source)]
