    (fn [ob]
      (letfn [(callback [v]
                (cond
-                 (next? v)
+                 (-next? v)
                  (.onNext ob)
-                 (end? v)
-                 (if (instance? EndValue v)
-                   (.onComplete ob (.-v v))
-                   (.onComplete ob v))
-
-                 (error? v)
+
+                 (-end? v)
+                 (.onCompleted ob)
+
+                 (-error? v)
                  (.onError ob v)))]
        (try
          (sf callback)
          (catch js/Error e
-           (.onError ob e))))))))
+           (.onError ob e)))))))
+
+(defn observable?
+  "Return true if `ob` is a instance
+  of Rx.Observable."
+  [ob]
+  (instance? js/Rx.Observable ob))
+
+;; (defn from-poll
+;;   "Creates an observable sequence polling given
+;;   function with given interval."
+;;   [ms f]
+;;   (create (fn [sick]
+
+;;             (let [sem (js/setInterval
+;;                        (fn []
+;;                          (let [v (f)]
+;;                            (when (or (-end? v) (-error? v))
+;;                              (js/clearInterval sem))
+;;                            (sick v))))]
+;;                    (.onNext ob)
+
+;;                    (-end? v)
+;;                  (if (instance? EndValue v)
+;;                    (.onComplete ob (.-v v))
+;;                    (.onComplete ob v))
+
+;;                  (-error? v)
+;;                  (.onError ob v))))))
+
+
+
+
+
+
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+;; Observable Subscription
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+
+(defn on-value
+  "Subscribes a function to invoke for each element
+  in the observable sequence."
+  [ob f]
+  {:pre [(observable? ob) (fn? f)]}
+  (.subscribeOnNext ob f))
+
+(defn on-error
+  "Subscribes a function to invoke upon exceptional termination
+  of the observable sequence."
+  [ob f]
+  {:pre [(observable? ob) (fn? f)]}
+  (.subscribeOnError ob f))
+
+(defn on-end
+  "Subscribes a function to invoke upon graceful termination
+  of the observable sequence."
+  [ob f]
+  {:pre [(observable? ob) (fn? f)]}
+  (.subscribeOnCompleted ob f))
+
+(defn subscribe
+  "Subscribes an observer to the observable sequence."
+  [ob nf ef cf]
+  {:pre [(observable? ob)]}
+  (.subscribe ob nf ef cf))
+
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+;; Observable Transformations
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
 (defn choice
   "Create an observable that surfaces any of the given

  (-error? [_] false)

  IObservableEndValue
  (-end? [_] false)e

(extend-protocol IObservableErrorValue
  js/Error
  (-error? [_] true)

  cljs.core.ExceptionInfo
  (-error? [_] true))

(deftype EndValue [v]
  (-end? [_] true))

(defn next
  [v]
  (NextValue. v))

(defn next?
  [v]
  (instance? NextValue v))

(defn end
  [v]
  (EndValue. v))

(defn end?
  [v]
  (instance? EndValue v))

(defn create
  "Creates an observable sequence from a specified

          (observable? b)]}
   (.merge a b))
  ([a b & more]
   (cljs.core/reduce merge (merge a b) more)))

(defn filter
  "Filters the elements of an observable sequence
