rx.lang.clojure.core — org.clojars.jconti/rxclojure 1.2.2

cache clj

(cache xs)

caches the observable value so that multiple subscribers don't re-evaluate it.

See: rx.Observable/cache

catch clj macro

(catch p binding & body observable)

Macro version of catch*.

The body of the catch is wrapped in an implicit (do). It must evaluate to an Observable.

Note that the source observable is the last argument so this works with ->> but may look slightly odd when used standalone.

Example:

(->> my-observable ; just emit 0 on IllegalArgumentException (catch IllegalArgumentException e (rx/return 0))

   (catch DependencyException e
     (if (.isMinor e)
       (rx/return 0)
       (rx/throw (WebException. 503)))))

See: catch*

catch* clj

(catch* p f o)

Returns an observable that, when Observable o triggers an error, e, continues with Observable returned by (f e) if (p e) is true. If (p e) returns a Throwable that value is passed as e.

If p is a class object, a normal instance? check is performed rather than calling it as a function. If the value returned by (p e) is not true, the error is propagated.

Examples:

(->> my-observable

   ; On IllegalArgumentException, just emit 1
   (catch* IllegalArgumentException
           (fn [e] (rx/return 1)))

   ; If exception message contains "WAT", emit [\W \A \T]
   (catch* (fn [e] (-> e .getMessage (.contains "WAT")))
           (fn [e] (rx/seq->o [\W \A \T]))))

See: rx.Observable/onErrorResumeNext http://netflix.github.io/RxJava/javadoc/rx/Observable.html#onErrorResumeNext(rx.functions.Func1)

catch-error-value clj macro

(catch-error-value error-observer value & body)

Experimental

TODO: Better name, better abstraction.

Evaluate body and return its value. If an exception e is thrown, inject the given value into the exception's cause and call (on-error error-observer e), returning e.

This is meant to facilitate implementing Observers that call user-supplied code safely. The general pattern is something like:

(fn [o v] (rx/catch-error-value o v (rx/on-next o (some-func v))))

If (some-func v) throws an exception, it is caught, v is injected into the exception's cause (with OnErrorThrowable/addValueAsLastCause) and (rx/on-error o e) is invoked.

See: rx.exceptions.OnErrorThrowable/addValueAsLastCause

concat clj

(concat & xs)

Concatenate the given Observables one after the another.

Note that xs is separate Observables which are concatentated. To concatenate an Observable of Observables, use concat*

See: rx.Observable/concat concat*

concat* clj

(concat* os)

Concatenate the given Observable of Observables one after another.

See: rx.Observable/concat concat

cons clj

(cons x xs)

cons x to the beginning of xs

count clj

(count xs)

Returns an Observable that emits the number of items is xs as a long.

See: rx.Observable/countLong

cycle clj

(cycle xs)

Returns an Observable that emits the items of xs repeatedly, forever.

TODO: Other sigs.

See: rx.Observable/repeat clojure.core/cycle

distinct clj

(distinct xs)

(distinct key-fn xs)

Returns an Observable of the elements of Observable xs with duplicates removed. key-fn, if provided, is a one arg function that determines the key used to determined duplicates. key-fn defaults to identity.

This implementation doesn't use rx.Observable/distinct because it doesn't honor Clojure's equality semantics.

See: clojure.core/distinct

do clj

(do do-fn xs)

Returns a new Observable that, for each x in Observable xs, executes (do-fn x), presumably for its side effects, and then passes x along unchanged.

If do-fn throws an exception, that exception is emitted via onError and the sequence is finished.

Example:

(->> (rx/seq->o [1 2 3]) (rx/do println) ...)

Will print 1, 2, 3.

See: rx.Observable/doOnNext

drop clj

(drop n xs)

drop-while clj

(drop-while p xs)

empty clj

(empty)

Returns an Observable that completes immediately without emitting any values.

See: rx.Observable/empty

every? clj

(every? p xs)

Returns an Observable that emits a single true value if (p x) is true for all x in xs. Otherwise emits false.

See: clojure.core/every? rx.Observable/all

filter clj

(filter p xs)

finally clj macro

(finally & body observable)

Macro version of finally*.

Note that the source observable is the last argument so this works with ->> but may look slightly odd when used standalone.

Example:

(->> my-observable (finally (println "Done")))

See: finally*

finally* clj

(finally* f o)

Returns an Observable that, as a side-effect, executes (f) when the given Observable completes regardless of success or failure.

Example:

(->> my-observable (finally* (fn [] (println "Done"))))

first clj

(first xs)

Returns an Observable that emits the first item emitted by xs, or an empty Observable if xs is empty.

See: rx.Observable/take(1)

flatmap clj

(flatmap f & xs)

Like mapcat, but the Observables produced by f are merged rather than concatenated. This behavior is preferable in asynchronous contexts where order is not important.

See: mapcat rx.Observable/flatMap

flatmap* clj

(flatmap* f xs)

Same as multi-arg flatmap, but input is an Observable of Observables.

See: flatmap

fn->predicate clj

(fn->predicate f)

Turn f into a predicate that returns true/false like Rx predicates should

generator clj macro

(generator bindings & body)

Create an observable that executes body which should emit values with (rx/on-next observer value) where observer comes from bindings.

Automatically calls on-completed on return, or on-error if any exception is thrown.

The body should exit early if (rx/unsubscribed? observable) returns true

Examples:

; make an observer that emits [0 1 2 3 4] (generator [observer] (dotimes [i 5] (on-next observer i)))

generator* clj

(generator* f & args)

Creates an observable that calls (f observer & args) which should emit values with (rx/on-next observer value).

Automatically calls on-completed on return, or on-error if any exception is thrown.

f should exit early if (rx/unsubscribed? observable) returns true

Examples:

; An observable that emits just 99 (rx/generator* on-next 99)

group-by clj

(group-by key-fn xs)

Returns an Observable of clojure.lang.MapEntry where the key is the result of (key-fn x) and the val is an Observable of x for each key.

This returns a clojure.lang.MapEntry rather than rx.observables.GroupedObservable for some vague consistency with clojure.core/group-by and so that clojure.core/key, clojure.core/val and destructuring will work as expected.

See: clojure.core/group-by rx.Observable/groupBy rx.observables.GroupedObservable

interleave clj

(interleave o1 & observables)

Returns an Observable of the first item in each Observable, then the second etc.

Each argument is an individual Observable

See: observable* clojure.core/interleave

interleave* clj

(interleave* observables)

Returns an Observable of the first item in each Observable emitted by observables, then the second etc.

observables is an Observable of Observables

See: interleave clojure.core/interleave

interpose clj

(interpose sep xs)

Returns an Observable of the elements of xs separated by sep

See: clojure.core/interpose

into clj

(into to from)

Returns an observable that emits a single value which is all of the values of from-observable conjoined onto to

See: clojure.core/into rx.Observable/toList

iterate clj

(iterate f x)

Returns an Observable of x, (f x), (f (f x)) etc. f must be free of side-effects

See: clojure.core/iterate

keep clj

(keep f xs)

keep-indexed clj

(keep-indexed f xs)

let-o clj macro

(let-o bindings & result-body)

EXTREMELY EXPERIMENTAL AND SUBJECT TO CHANGE OR DELETION

Similar to clojure.core/let, but bindings are Observables and the result of the body must be an Observable. Binding names must start with ?. Binding order doesn't matter and any binding is visible to all other expressions as long as no cycles are produced in the resulting Observable expression.

The key difference here is that the macro can identify the dependencies between Observables and correctly connect them, protecting from variations in subscribe behavior as well as the idiosyncracies of setting up multiple subscriptions to Observables.

This is only very useful for constructing graphs of Observables where you'd usually have to fiddle around with publish, connect, replay and all that stuff. If you have a linear sequence of operators, just chain them together.

Current limitations:

• All Observables are cache()'d so watch out for large sequences. This will be fixed eventually.
• let-o cannot be nested. Some deep-walking macro-magic will be required for this.

Example:

; Note that both ?c and ?d depend on ?b and the result Observable depends on ; ?c and ?d. (let-o [?a (rx/return 99) ?b (... some observable network request ...) ?c (rx/map vector ?a ?b) ?d (rx/map ... ?b)] (rx/map vector ?c ?d))

See: let-o*

let-o* clj

(let-o* description)

EXTREMELY EXPERIMENTAL AND SUBJECT TO CHANGE OR DELETION

Given a graph description, returns an observable that emits a single map of observables all hooked up and ready for subscription.

A graph is a map from name to a map with keys:

:deps A vector of dependency names :factory A function that takes a map from name to Observable for the names in :deps and returns an Observable

Returns a map from name to Observable. Additionally, there will be a ::non-terminals key in the map with a vector of non-terminal names.

See: let-o

let-realized clj macro

(let-realized bindings & body)

EXTREMELY EXPERIMENTAL AND SUBJECT TO CHANGE OR DELETION

'let' version of realized map.

(let-realized [a (make-observable)] (* 2 a))

is equivalent to:

(->> (realized-map :a (make-observable)) (map (fn [{:keys [a]}] (* 2 a))))

That is, it eliminates the repition of the map keys when you want to do something with the final result.

Evaluates to an Observable that emits the value of the let body.

See: rx.lang.clojure.realized/realized-map rx.lang.clojure.realized/all

lift clj

(lift op xs)

Lift the Operator op over the given Observable xs

Example:

(->> my-observable (rx/lift (rx/operator ...)) ...)

See: rx.Observable/lift operator

map clj

(map f & observables)

Map a function over one or more observable sequences.

Each item from the first Observable is the first arg, each item from the second Observable is the second arg, and so on.

See: clojure.core/map rx.Observable/zip

map* clj

(map* f observable)

Map a function over an Observable of Observables.

Each item from the first emitted Observable is the first arg, each item from the second emitted Observable is the second arg, and so on.

See: map clojure.core/map rx.Observable/zip

map-indexed clj

(map-indexed f xs)

Returns an observable that invokes (f index value) for each value of the input observable. index starts at 0.

See: clojure.core/map-indexed

mapcat clj

(mapcat f & xs)

Returns an observable which, for each value x in xs, calls (f x), which must return an Observable. The resulting observables are concatentated together into one observable.

WARNING: This operator, like clojure.core/mapcat, preserves ordering of the generated Observables. In an asynchronous context, this may cause unintended blocking. Try flatmap instead.

If multiple Observables are given, the arguments to f are the first item from each observable, then the second item, etc.

See: clojure.core/mapcat flatmap rx.Observable/concatMap

mapcat* clj

(mapcat* f xs)

Same as multi-arg mapcat, but input is an Observable of Observables.

See: mapcat clojure.core/mapcat

merge clj

(merge & os)

Merge one or more Observables into a single observable.

If you want clojure.core/merge, it's just this:

(rx/reduce clojure.core/merge {} maps)

See: merge* merge-delay-error rx.Observable/merge

merge* clj

(merge* xs)

Merge an Observable of Observables into a single Observable

If you want clojure.core/merge, it's just this:

(rx/reduce clojure.core/merge {} maps)

See: merge merge-delay-error* rx.Observable/merge

merge-delay-error clj

(merge-delay-error & os)

Same as merge, but all values are emitted before errors are propagated

merge-delay-error* clj

(merge-delay-error* xs)

Same as merge*, but all values are emitted before errors are propagated

never clj

(never)

Returns an Observable that never emits any values and never completes.

See: rx.Observable/never

next clj

Returns an observable that emits all but the first element of the input observable.

See: clojure.core/next

nth clj

(nth xs index)

(nth xs index not-found)

Returns an Observable that emits the value at the index in the given Observable. nth throws an IndexOutOfBoundsException unless not-found is supplied.

Note that the Observable is the first arg!

observable* clj

(observable* f)

Create an Observable from the given function.

When subscribed to, (f subscriber) is called at which point, f can start emitting values, etc. The passed subscriber is of type rx.Subscriber.

See: rx.Subscriber rx.Observable/create

observable? clj

(observable? o)

Returns true if o is an rx.Observable

on-completed clj

(on-completed o)

Call onCompleted on the given observer and return o.

on-error clj

(on-error o e)

Call onError on the given observer and return o.

on-next clj

(on-next o value)

Call onNext on the given observer and return o.

operator* clj

(operator* f)

Experimental, subject to change or deletion.

Returns a new implementation of rx.Observable$Operator that calls the given function with a rx.Subscriber. The function should return a rx.Subscriber.

See: lift rx.Observable$Operator

partition-all clj

(partition-all n xs)

(partition-all n step xs)

Returns an Observable of Observables of n items each, at offsets step apart. If step is not supplied, defaults to n, i.e. the partitions do not overlap. May include partitions with fewer than n items at the end.

See: clojure.core/partition-all rx.Observable/window

range clj

(range)

(range end)

(range start end)

(range start end step)

Returns an Observable nums from start (inclusive) to end (exclusive), by step, where start defaults to 0, step to 1, and end to infinity.

Note: this is not implemented on rx.Observable/range

See: clojure.core/range

reduce clj

(reduce f xs)

(reduce f val xs)

reductions clj

(reductions f xs)

(reductions f val xs)

rest clj

Same as rx/next

return clj

(return value)

Returns an observable that emits a single value.

See: rx.Observable/just

seq->o clj

(seq->o xs)

Make an observable out of some seq-able thing. The rx equivalent of clojure.core/seq.

serialize clj

(serialize xs)

Serialize execution.

See: rx.Observable/serialize

some clj

(some p xs)

Returns an observable that emits the first logical true value of (pred x) for any x in xs, else completes immediately.

See: clojure.core/some

sort clj

(sort xs)

(sort comp xs)

Returns an observable that emits the items in xs, where the sort order is determined by comparing items. If no comparator is supplied, uses compare. comparator must implement java.util.Comparator.

See: clojure.core/sort

sort-by clj

(sort-by keyfn xs)

(sort-by keyfn comp xs)

Returns an observable that emits the items in xs, where the sort order is determined by comparing (keyfn item). If no comparator is supplied, uses compare. comparator must implement java.util.Comparator.

See: clojure.core/sort-by

split-with clj

(split-with p xs)

Returns an observable that emits a pair of observables

[(take-while p xs) (drop-while p xs)]

See: rx.lang.clojure/take-while rx.lang.clojure/drop-while clojure.core/split-with

subscribe clj

(subscribe o on-next-action)

(subscribe o on-next-action on-error-action)

(subscribe o on-next-action on-error-action on-completed-action)

Subscribe to the given observable.

on-X-action is a normal clojure function.

See: rx.Observable/subscribe

subscribe-on clj

(subscribe-on s xs)

Cause subscriptions to the given observable to happen on the given scheduler.

Returns a new Observable.

See: rx.Observable/subscribeOn

subscriber clj

(subscriber o on-next-action)

(subscriber o on-next-action on-error-action)

(subscriber o on-next-action on-error-action on-completed-action)

Experimental, subject to change or deletion.

subscription clj

(subscription handler)

Create a new subscription that calls the given no-arg handler function when unsubscribe is called

See: rx.subscriptions.Subscriptions/create

take clj

(take n xs)

Returns an observable that emits the first n elements of xs.

See: clojure.core/take

take-while clj

(take-while p xs)

Returns an Observable that emits xs until the first x such that (p x) is falsey.

See: clojure.core/take-while rx.Observable/takeWhile

throw clj

(throw e)

Returns an Observable the simply emits the given exception with on-error

See: rx.Observable/error

unsubscribe clj

(unsubscribe s)

Unsubscribe from Subscription s and return it.

unsubscribe-on clj

(unsubscribe-on s xs)

Cause unsubscriptions from the given observable to happen on the given scheduler.

Returns a new Observable.

See: rx.Observable/unsubscribeOn

unsubscribed? clj

(unsubscribed? s)

Returns true if the given Subscription (or Subscriber) is unsubscribed.

See: rx.Observable/create observable*

wrap-on-completed clj

(wrap-on-completed handler)

Wrap handler with code that automaticaly calls rx.Observable.onCompleted.

wrap-on-error clj

(wrap-on-error handler)

Wrap handler with code that automaticaly calls (on-error) if an exception is thrown