 
       Publisher
       (^void subscribe [_ ^Subscriber subscriber]
-        (let [subscription (chan->subscription subscriber source xform)]
+        (let [subscription (chan->subscription subscriber source)]
           (.onSubscribe subscriber subscription))))))
 
 (extend-protocol IPublisher
   nil
   (publisher* [source xform]
+    (assert (nil? xform) "not supported operation.")
     (empty->publisher source xform))
 
   java.lang.Long
   (publisher* [source xform]
-    (empty->publisher source xform))
+    (assert (nil? xform) "not supported operation.")
+    (empty->publisher source))
 
   java.lang.Iterable
   (publisher* [source xform]
-    (iterable->publisher source xform))
+    (assert (nil? xform) "not supported operation.")
+    (iterable->publisher source))
 
   manifold.stream.default.Stream
   (publisher* [source xform]
-    (stream->publisher source xform))
+    (assert (nil? xform) "not supported operation.")
+    (stream->publisher source))
 
   clojure.core.async.impl.channels.ManyToManyChannel
   (publisher* [source xform]
-    (channel->publisher source xform))
+    (assert (nil? xform) "not supported operation.")
+    (channel->publisher source))
 
   manifold.deferred.IDeferred
   (publisher* [source xform]

  "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]
   (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)))))))

(defn- promise->subscription
  "Coerce a promise instance to an instance


(defn- channel->publisher
  "Create a publisher with core.async channel as source."
  [source]
  (reify
    Seqable
    (seq [p]
      (publisher->seq p))

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

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

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