 
   futura.promise.Promise
   (publisher* [source xform]
-    (promise->publisher source xform))
+    (assert (nil? xform) "not supported operation.")
+    (promise->publisher source))
 
   Publisher
   (publisher* [source xform]


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


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

(defn- stream->publisher
  "Create a publisher with manifold stream as source."
  [source]
  (let [source' (async/chan)]
    (ms/connect source source')
    (reify
      IPullStream
      (pull [p]
        (p/promise #(subscribe-once p %)))

      Seqable
      (seq [p]
        (publisher->seq p))

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

(defn- empty->publisher
  "Creates an empty publisher that implements the
  push stream protocol."
  [bufflen]
  (let [source (async/chan bufflen)]
    (reify
      IPushStream
