   (pipe ob (.skipWhile rxop #(boolean (f %)))))
 
 (defn skip-until
-  "Returns the values from the source observable sequence
-  only after the other observable sequence produces a value."
+  "Returns the values from the source observable sequence only after the
+  other observable sequence produces a value."
   [pob ob]
   (pipe ob (.skipUntil rxop pob)))
 
 (defn take
-  "Bypasses a specified number of elements in an
-  observable sequence and then returns the remaining
-  elements."
+  "Bypasses a specified number of elements in an observable sequence and
+  then returns the remaining elements."
   [n ob]
   (pipe ob (.take rxop n)))
 
 (defn first
-  "Return an observable that only has the first value
-  of the provided observable."
-  [ob]
-  (take 1 ob))
+  "Return an observable that only has the first value of the provided
+  observable. You can optionally pass a predicate and default value."
+  ([ob] (pipe ob (.first rxop)))
+  ([f ob] (pipe ob (.first rxop #(boolean (f %)))))
+  ([f default ob] (pipe ob (.first rxop #(boolean (f %)) default))))
+
+(defn last
+  "Return an observable that only has the last value of the provided
+  observable. You can optionally pass a predicate and default value."
+  [ob]
+  ([ob] (pipe ob (.last rxop)))
+  ([f ob] (pipe ob (.last rxop #(boolean (f %)))))
+  ([f default ob] (pipe ob (.last rxop #(boolean (f %)) default))))
 
 (defn take-while
-  "Returns elements from an observable sequence as long as a
-  specified predicate returns true."
+  "Returns elements from an observable sequence as long as a specified
+  predicate returns true."
   [f ob]
   (pipe ob (.takeWhile rxop #(boolean (f %)))))
 
 (defn take-until
-  "Returns the values from the source observable sequence until
-  the other observable sequence or Promise produces a value."
+  "Returns the values from the source observable sequence until the
+  other observable sequence or Promise produces a value."
   [other ob]
   (pipe ob (.takeUntil rxop other)))
 
 (defn reduce
-  "Applies an accumulator function over an observable
-  sequence, returning the result of the aggregation as a
-  single element in the result sequence."
+  "Applies an accumulator function over an observable sequence,
+  returning the result of the aggregation as a single element in the
+  result sequence."
   ([f ob] (pipe ob (.reduce rxop #(f %1 %2))))
   ([f seed ob] (pipe ob (.reduce rxop #(f %1 %2) seed))))
 
 (defn scan
-  "Applies an accumulator function over an observable
-  sequence and returns each intermediate result.
-  Same as reduce but with intermediate results"
+  "Applies an accumulator function over an observable sequence and
+  returns each intermediate result.  Same as reduce but with
+  intermediate results"
   ([f ob] (pipe ob (.scan rxop #(f %1 %2))))
   ([f seed ob] (pipe ob (.scan rxop #(f %1 %2) seed))))
 
 (defn with-latest
-  "Merges the specified observable sequences into
-  one observable sequence by using the selector
-  function only when the source observable sequence
-  (the instance) produces an element."
+  "Merges the specified observable sequences into one observable
+  sequence by using the selector function only when the source
+  observable sequence (the instance) produces an element."
   [f other source]
   (pipe source (.withLatestFrom rxop other f)))
 
 (defn combine-latest
-  "Combines multiple Observables to create an Observable
-  whose values are calculated from the latest values of
-  each of its input Observables."
+  "Combines multiple Observables to create an Observable whose values
+  are calculated from the latest values of each of its input
+  Observables."
   ([other ob]
    (combine-latest vector other ob))
   ([f other ob]

  (-on-error [_ error] "Subscription error notification hook.")
  (-on-end [_] "Subscription termination notification hook."))

(defprotocol ICancellable
  (-cancel [_] "dispose resources."))

(defn cancel!
  "Dispose resources acquired by the subscription."
  [v]
  (-cancel v))

(defn- wrap-disposable
  [disposable]
  (specify! disposable
    ICancellable
    (-cancel [this]
      (.unsubscribe this))))

(defn subscribe-with

  or array/iterable into one observable sequence.

  In other languages is called: flatMap or mergeMap."
  [f ob]
  (pipe ob (.flatMap rxop #(f %))))

;; Aliases
(def fmap merge-map)
