 (defn dedupe
   "Returns an observable sequence that contains only
   distinct contiguous elements."
-  ([ob]
+  ([^observable ob]
+   {:pre [(observable? ob)]}
    (.distinctUntilChanged ob))
-  ([f ob]
+  ([f ^observable ob]
+   {:pre [(observable? ob)]}
    (.distinctUntilChanged ob f)))
 
 (defn dedupe'

  repeats the source observable the specified number of
  times or until it terminates. If no number of retries
  is given, it will be retried indefinitely."
  ([ob]
   (.retry ob))
  ([n ob]
   (.retry ob n)))

(defn to-observable
  "Hides the identity of an observable sequence."
  [b]
  {:pre [(or (observable? b) (bus? b))]}
  (.asObservable b))

(defn transform

(defn on-error
  "Subscribes a function to invoke upon exceptional termination
  of the observable sequence."
  [^observable ob f]
  {:pre [(observable? ob)]}
  (let [disposable (.subscribeOnError ob #(f %))]
    #(.dispose disposable)))
