RxAndroid Tutorials
Lesson1: Introduction
SubscribeOn
specify the Scheduler on which an Observable will operate
Many implementations of ReactiveX use “Schedulers” to govern an Observable’s
transitions between threads in a multi-threaded environment. You can instruct an
Observable to do its work on a particular Scheduler by calling the
Observable’s SubscribeOn operator.
The ObserveOn operator is similar, but more limited. It instructs the Observable to send
notifications to observers on a specified Scheduler.
In some implementations there is also an UnsubscribeOn operator.
By default, an Observable and the chain of operators that you apply to it will do its work,
and will notify its observers, on the same thread on which its Subscribe method is
called. The SubscribeOn operator changes this behavior by specifying a different
Scheduler on which the Observable should operate. The ObserveOnoperator specifies a
different Scheduler that the Observable will use to send notifications to its observers.
As shown in this illustration, the SubscribeOn operator designates which thread the
Observable will begin operating on, no matter at what point in the chain of operators
that operator is called. ObserveOn, on the other hand, affects the thread that the
Observable will use below where that operator appears. For this reason, you may
call ObserveOn multiple times at various points during the chain of Observable operators
in order to change on which threads certain of those operators operate.
See Also
Scheduler
ObserveOn
Rx Workshop: Schedulers
RxJava Threading Examples by Graham Lea
Introduction to Rx: SubscribeOn and ObserveOn
Async Abstractions using rx-java by Biju Kunjummen, DZone
RxJava: Understanding observeOn() and subscribeOn() by Thomas Nield
Advanced Reactive Java: SubscribeOn and ObserveOn by Dávid Karnok
Language-Specific Information:
RxClojure subscribe-on unsubscribe-on
RxCpp
RxGroovy subscribeOn unsubscribeOn
RxJava 1․x subscribeOn unsubscribeOn
RxJava 2․x subscribeOn unsubscribeOn
RxJS subscribeOn
RxKotlin subscribeOn unsubscribeOn
RxNET SubscribeOn SubscribeOnDispatcher
RxPY subscribe_on
Rxrb subscribe_on
RxScala subscribeOn unsubscribeOn
RxSwift subscribeOn
Lesson 2: Custom Operators
Introduction
Each language-specific implementation of ReactiveX implements a set of operators.
Although there is much overlap between implementations, there are also some
operators that are only implemented in certain implementations. Also, each
implementation tends to name its operators to resemble those of similar methods that
are already familiar from other contexts in that language.
Chaining Operators
Most operators operate on an Observable and return an Observable. This allows you to
apply these operators one after the other, in a chain. Each operator in the chain
modifies the Observable that results from the operation of the previous operator.
There are other patterns, like the Builder Pattern, in which a variety of methods of a
particular class operate on an item of that same class by modifying that object through
the operation of the method. These patterns also allow you to chain the methods in a
similar way. But while in the Builder Pattern, the order in which the methods appear in
the chain does not usually matter, with the Observable operators order matters.
A chain of Observable operators do not operate independently on the original
Observable that originates the chain, but they operate in turn, each one operating on
the Observable generated by the operator immediately previous in the chain.
The Operators of ReactiveX
This page first lists what could be considered the “core” operators in ReactiveX, and
links to pages that have more in-depth information on how these operators work and
how particular language-specific ReactiveX versions have implemented these
operators.
Next is a “decision tree” that may help you choose the operator that is most appropriate
to your use case.
Finally, there is an alphabetical list of most of the operators available in the many
language-specific implementations of ReactiveX. These link to the page that documents
the core operator that most closely resembles the language-specific operator (so, for
instance, the Rx.NET “SelectMany” operator links to the documentation of
the FlatMap ReactiveX operator, of which “SelectMany” is the Rx.NET
implementation).
If you want to implement your own operator, see Implementing Your Own Operators.
Contents
1. Operators By Category
2. A Decision Tree of Observable Operators
3. An Alphabetical List of Observable Operators
Operators By Category
Creating Observables
Operators that originate new Observables.
Create — create an Observable from scratch by calling observer methods
programmatically
Defer — do not create the Observable until the observer subscribes, and create
a fresh Observable for each observer
Empty/Never/Throw — create Observables that have very precise and limited
behavior
From — convert some other object or data structure into an Observable
Interval — create an Observable that emits a sequence of integers spaced by a
particular time interval
Just — convert an object or a set of objects into an Observable that emits that or
those objects
Range — create an Observable that emits a range of sequential integers
Repeat — create an Observable that emits a particular item or sequence of items
repeatedly
Start — create an Observable that emits the return value of a function
Timer — create an Observable that emits a single item after a given delay
Transforming Observables
Operators that transform items that are emitted by an Observable.
Buffer — periodically gather items from an Observable into bundles and emit
these bundles rather than emitting the items one at a time
FlatMap — transform the items emitted by an Observable into Observables, then
flatten the emissions from those into a single Observable
GroupBy — divide an Observable into a set of Observables that each emit a
different group of items from the original Observable, organized by key
Map — transform the items emitted by an Observable by applying a function to
each item
Scan — apply a function to each item emitted by an Observable, sequentially,
and emit each successive value
Window — periodically subdivide items from an Observable into Observable
windows and emit these windows rather than emitting the items one at a time
Filtering Observables
Operators that selectively emit items from a source Observable.
Debounce — only emit an item from an Observable if a particular timespan has
passed without it emitting another item
Distinct — suppress duplicate items emitted by an Observable
ElementAt — emit only item n emitted by an Observable
Filter — emit only those items from an Observable that pass a predicate test
First — emit only the first item, or the first item that meets a condition, from an
Observable
IgnoreElements — do not emit any items from an Observable but mirror its
termination notification
Last — emit only the last item emitted by an Observable
Sample — emit the most recent item emitted by an Observable within periodic
time intervals
Skip — suppress the first n items emitted by an Observable
SkipLast — suppress the last n items emitted by an Observable
Take — emit only the first n items emitted by an Observable
TakeLast — emit only the last n items emitted by an Observable
Combining Observables
Operators that work with multiple source Observables to create a single Observable
And/Then/When — combine sets of items emitted by two or more Observables by
means of Patternand Plan intermediaries
CombineLatest — when an item is emitted by either of two Observables, combine
the latest item emitted by each Observable via a specified function and emit
items based on the results of this function
Join — combine items emitted by two Observables whenever an item from one
Observable is emitted during a time window defined according to an item emitted
by the other Observable
Merge — combine multiple Observables into one by merging their emissions
StartWith — emit a specified sequence of items before beginning to emit the
items from the source Observable
Switch — convert an Observable that emits Observables into a single
Observable that emits the items emitted by the most-recently-emitted of those
Observables
Zip — combine the emissions of multiple Observables together via a specified
function and emit single items for each combination based on the results of this
function
Error Handling Operators
Operators that help to recover from error notifications from an Observable
Catch — recover from an onError notification by continuing the sequence without
error
Retry — if a source Observable sends an onError notification, resubscribe to it in
the hopes that it will complete without error
Observable Utility Operators
A toolbox of useful Operators for working with Observables
Delay — shift the emissions from an Observable forward in time by a particular
amount
Do — register an action to take upon a variety of Observable lifecycle events
Materialize/Dematerialize — represent both the items emitted and the
notifications sent as emitted items, or reverse this process
ObserveOn — specify the scheduler on which an observer will observe this
Observable
Serialize — force an Observable to make serialized calls and to be well-
behaved
Subscribe — operate upon the emissions and notifications from an Observable
SubscribeOn — specify the scheduler an Observable should use when it is
subscribed to
TimeInterval — convert an Observable that emits items into one that emits
indications of the amount of time elapsed between those emissions
Timeout — mirror the source Observable, but issue an error notification if a
particular period of time elapses without any emitted items
Timestamp — attach a timestamp to each item emitted by an Observable
Using — create a disposable resource that has the same lifespan as the
Observable
Conditional and Boolean Operators
Operators that evaluate one or more Observables or items emitted by Observables
All — determine whether all items emitted by an Observable meet some criteria
Amb — given two or more source Observables, emit all of the items from only the
first of these Observables to emit an item
Contains — determine whether an Observable emits a particular item or not
DefaultIfEmpty — emit items from the source Observable, or a default item if the
source Observable emits nothing
SequenceEqual — determine whether two Observables emit the same sequence
of items
SkipUntil — discard items emitted by an Observable until a second Observable
emits an item
SkipWhile — discard items emitted by an Observable until a specified condition
becomes false
TakeUntil — discard items emitted by an Observable after a second Observable
emits an item or terminates
TakeWhile — discard items emitted by an Observable after a specified condition
becomes false
Mathematical and Aggregate Operators
Operators that operate on the entire sequence of items emitted by an Observable
Average — calculates the average of numbers emitted by an Observable and
emits this average
Concat — emit the emissions from two or more Observables without interleaving
them
Count — count the number of items emitted by the source Observable and emit
only this value
Max — determine, and emit, the maximum-valued item emitted by an Observable
Min — determine, and emit, the minimum-valued item emitted by an Observable
Reduce — apply a function to each item emitted by an Observable, sequentially,
and emit the final value
Sum — calculate the sum of numbers emitted by an Observable and emit this sum
Backpressure Operators
backpressure operators — strategies for coping with Observables that produce
items more rapidly than their observers consume them
Connectable Observable Operators
Specialty Observables that have more precisely-controlled subscription dynamics
Connect — instruct a connectable Observable to begin emitting items to its
subscribers
Publish — convert an ordinary Observable into a connectable Observable
RefCount — make a Connectable Observable behave like an ordinary
Observable
Replay — ensure that all observers see the same sequence of emitted items,
even if they subscribe after the Observable has begun emitting items
Operators to Convert Observables
To — convert an Observable into another object or data structure
A Decision Tree of Observable Operators
This tree can help you find the ReactiveX Observable operator you’re looking for.
I want to create a new Observable
that emits a particular item
Just
that was returned from a function called at subscribe-time
Start
that was returned from an Action, Callable, Runnable, or something of that
sort, called at subscribe-time
From
after a specified delay
Timer
that pulls its emissions from a particular Array, Iterable, or something like
that
From
by retrieving it from a Future
Start
that obtains its sequence from a Future
From
that emits a sequence of items repeatedly
Repeat
from scratch, with custom logic
Create
for each observer that subscribes
Defer
that emits a sequence of integers
Range
at particular intervals of time
Interval
after a specified delay
Timer
that completes without emitting items
Empty
that does nothing at all
Never
I want to create an Observable by combining other Observables
and emitting all of the items from all of the Observables in whatever order they
are received
Merge
and emitting all of the items from all of the Observables, one Observable at a
time
Concat
by combining the items from two or more Observables sequentially to
come up with new items to emit
whenever each of the Observables has emitted a new item
Zip
whenever any of the Observables has emitted a new item
CombineLatest
whenever an item is emitted by one Observable in a window defined by an
item emitted by another
Join
by means of Pattern and Plan intermediaries
And/Then/When
and emitting the items from only the most-recently emitted of those
Observables
Switch
I want to emit the items from an Observable after transforming them
one at a time with a function
Map
by emitting all of the items emitted by corresponding Observables
FlatMap
one Observable at a time, in the order they are emitted
ConcatMap
based on all of the items that preceded them
Scan
by attaching a timestamp to them
Timestamp
into an indicator of the amount of time that lapsed before the
emission of the item
TimeInterval
I want to shift the items emitted by an Observable forward in time before
reemitting them
Delay
I want to transform items and notifications from an Observable into items and
reemit them
by wrapping them in Notification objects
Materialize
which I can then unwrap again with
Dematerialize
I want to ignore all items emitted by an Observable and only pass along its
completed/error notification
IgnoreElements
I want to mirror an Observable but prefix items to its sequence
StartWith
only if its sequence is empty
DefaultIfEmpty
I want to collect items from an Observable and reemit them as buffers of
items
Buffer
containing only the last items emitted
TakeLastBuffer
I want to split one Observable into multiple Observables
Window
so that similar items end up on the same Observable
GroupBy
I want to retrieve a particular item emitted by an Observable:
the last item emitted before it completed
Last
the sole item it emitted
Single
the first item it emitted
First
I want to reemit only certain items from an Observable
by filtering out those that do not match some predicate
Filter
that is, only the first item
First
that is, only the first items
Take
that is, only the last item
Last
that is, only item n
ElementAt
that is, only those items after the first items
that is, after the first n items
Skip
that is, until one of those items matches a predicate
SkipWhile
that is, after an initial period of time
Skip
that is, after a second Observable emits an item
SkipUntil
that is, those items except the last items
that is, except the last n items
SkipLast
that is, until one of those items matches a predicate
TakeWhile
that is, except items emitted during a period of time before the source
completes
SkipLast
that is, except items emitted after a second Observable emits an item
TakeUntil
by sampling the Observable periodically
Sample
by only emitting items that are not followed by other items
within some duration
Debounce
by suppressing items that are duplicates of already-
emitted items
Distinct
if they immediately follow the item they are duplicates of
DistinctUntilChanged
by delaying my subscription to it for some time after it
begins emitting items
DelaySubscription
I want to reemit items from an Observable only on condition that
it was the first of a collection of Observables to emit an item
Amb
I want to evaluate the entire sequence of items emitted by an
Observable
and emit a single boolean indicating if all of the items pass some test
All
and emit a single boolean indicating if the Observable emitted any item (that
passes some test)
Contains
and emit a single boolean indicating if the Observable emitted no items
IsEmpty
and emit a single boolean indicating if the sequence is identical to one
emitted by a second Observable
SequenceEqual
and emit the average of all of their values
Average
and emit the sum of all of their values
Sum
and emit a number indicating how many items were in the
sequence
Count
and emit the item with the maximum value
Max
and emit the item with the minimum value
Min
by applying an aggregation function to each item
in turn and emitting the result
Scan
I want to convert the entire sequence of items emitted by an
Observable into some other data structure
To
I want an operator to operate on a particular Scheduler
SubscribeOn
when it notifies observers
ObserveOn
I want an Observable to invoke a particular action when
certain events occur
Do
I want an Observable that will notify observers of an
error
Throw
if a specified period of time elapses without it emitting an item
Timeout
I want an Observable to recover gracefully
from a timeout by switching to a backup Observable
Timeout
from an upstream error notification
Catch
by attempting to resubscribe to the upstream Observable
Retry
I want to create a resource that has the same
lifespan as the Observable
Using
I want to subscribe to an Observable and
receive a Future that blocks until the
Observable completes
Start
I want an Observable that does not start
emitting items to subscribers until asked
Publish
and then only emits the last item in its sequence
PublishLast
and then emits the complete sequence, even to those who subscribe after the
sequence has begun
Replay
but I want it to go away once all of its subscribers unsubscribe
RefCount
and then I want to ask it to start
Connect
See Also
Which Operator do I use? by Dennis Stoyanov (a similar decision tree, specific to
RxJS operators)
An Alphabetical List of Observable Operators
Canonical, core operator names are in boldface. Other entries represent language-
specific variants of these operators or specialty operators outside of the main ReactiveX
core set of operators.
Aggregate
All
Amb
ambArray
ambWith
and_
And
Any
apply
as_blocking
asObservable
AssertEqual
asyncAction
asyncFunc
Average
averageDouble
averageFloat
averageInteger
averageLong
blocking
blockingFirst
blockingForEach
blockingIterable
blockingLast
blockingLatest
blockingMostRecent
blockingNext
blockingSingle
blockingSubscribe
Buffer
bufferWithCount
bufferWithTime
bufferWithTimeOrCount
byLine
cache
cacheWithInitialCapacity
case
Cast
Catch
catchError
catchException
collect
collect (RxScala version of Filter)
collectInto
CombineLatest
combineLatestDelayError
combineLatestWith
Concat
concat_all
concatAll
concatArray
concatArrayDelayError
concatArrayEager
concatDelayError
concatEager
concatMap
concatMapDelayError
concatMapEager
concatMapEagerDelayError
concatMapIterable
concatMapObserver
concatMapTo
concatWith
Connect
connect_forever
cons
Contains
controlled
Count
countLong
Create
cycle
Debounce
decode
DefaultIfEmpty
Defer
deferFuture
Delay
delaySubscription
delayWithSelector
Dematerialize
Distinct
distinctKey
distinctUntilChanged
distinctUntilKeyChanged
Do
doAction
doAfterTerminate
doOnComplete
doOnCompleted
doOnDispose
doOnEach
doOnError
doOnLifecycle
doOnNext
doOnRequest
doOnSubscribe
doOnTerminate
doOnUnsubscribe
doseq
doWhile
drop
dropRight
dropUntil
dropWhile
ElementAt
ElementAtOrDefault
Empty
emptyObservable
empty?
encode
ensures
error
every
exclusive
exists
expand
failWith
Filter
filterNot
Finally
finallyAction
finallyDo
find
findIndex
First
firstElement
FirstOrDefault
firstOrElse
FlatMap
flatMapFirst
flatMapIterable
flatMapIterableWith
flatMapLatest
flatMapObserver
flatMapWith
flatMapWithMaxConcurrent
flat_map_with_index
flatten
flattenDelayError
foldl
foldLeft
for
forall
ForEach
forEachFuture
forEachWhile
forIn
forkJoin
From
fromAction
fromArray
FromAsyncPattern
fromCallable
fromCallback
FromEvent
FromEventPattern
fromFunc0
fromFuture
fromIterable
fromIterator
from_list
fromNodeCallback
fromPromise
fromPublisher
fromRunnable
Generate
generateWithAbsoluteTime
generateWithRelativeTime
generator
GetEnumerator
getIterator
GroupBy
GroupByUntil
GroupJoin
head
headOption
headOrElse
if
ifThen
IgnoreElements
indexOf
interleave
interpose
Interval
intervalRange
into
isEmpty
items
Join
join (string)
jortSort
jortSortUntil
Just
keep
keep-indexed
Last
lastElement
lastOption
LastOrDefault
lastOrElse
Latest
latest (Rx.rb version of Switch)
length
let
letBind
lift
limit
LongCount
ManySelect
Map
map (RxClojure version of Zip)
MapCat
mapCat (RxClojure version of Zip)
map-indexed
mapTo
mapWithIndex
Materialize
Max
MaxBy
Merge
mergeAll
mergeArray
mergeArrayDelayError
merge_concurrent
mergeDelayError
mergeObservable
mergeWith
Min
MinBy
MostRecent
Multicast
multicastWithSelector
nest
Never
Next
Next (BlockingObservable version)
none
nonEmpty
nth
ObserveOn
ObserveOnDispatcher
observeSingleOn
of
of_array
ofArrayChanges
of_enumerable
of_enumerator
ofObjectChanges
OfType
ofWithScheduler
onBackpressureBlock
onBackpressureBuffer
onBackpressureDrop
OnErrorResumeNext
onErrorReturn
onErrorReturnItem
onExceptionResumeNext
onTerminateDetach
orElse
pairs
pairwise
partition
partition-all
pausable
pausableBuffered
pluck
product
Publish
PublishLast
publish_synchronized
publishValue
raise_error
Range
Reduce
reduceWith
reductions
RefCount
Repeat
repeat_infinitely
repeatUntil
repeatWhen
Replay
rescue_error
rest
Retry
retry_infinitely
retryUntil
retryWhen
Return
returnElement
returnValue
runAsync
safeSubscribe
Sample
Scan
scanWith
scope
Select (alternate name of Map)
select (alternate name of Filter)
selectConcat
selectConcatObserver
SelectMany
selectManyObserver
select_switch
selectSwitch
selectSwitchFirst
selectWithMaxConcurrent
select_with_index
seq
SequenceEqual
sequence_eql?
SequenceEqualWith
Serialize
share
shareReplay
shareValue
Single
singleElement
SingleOrDefault
singleOption
singleOrElse
size
Skip
SkipLast
skipLastWithTime
SkipUntil
skipUntilWithTime
SkipWhile
skipWhileWithIndex
skip_with_time
slice
sliding
slidingBuffer
some
sort
sorted
sort-by
sorted-list-by
split
split-with
Start
startAsync
startFuture
StartWith
startWithArray
stringConcat
stopAndWait
subscribe
subscribeActual
SubscribeOn
SubscribeOnDispatcher
subscribeOnCompleted
subscribeOnError
subscribeOnNext
subscribeWith
Sum
sumDouble
sumFloat
sumInteger
sumLong
Switch
switchCase
switchIfEmpty
switchLatest
switchMap
switchMapDelayError
switchOnNext
switchOnNextDelayError
Synchronize
Take
take_with_time
takeFirst
TakeLast
takeLastBuffer
takeLastBufferWithTime
takeLastWithTime
takeRight (see also: TakeLast)
TakeUntil
takeUntilWithTime
TakeWhile
takeWhileWithIndex
tail
tap
tapOnCompleted
tapOnError
tapOnNext
Then
thenDo
Throttle
throttleFirst
throttleLast
throttleWithSelector
throttleWithTimeout
Throw
throwError
throwException
TimeInterval
Timeout
timeoutWithSelector
Timer
Timestamp
To
to_a
ToArray
ToAsync
toBlocking
toBuffer
to_dict
ToDictionary
ToEnumerable
ToEvent
ToEventPattern
ToFlowable
ToFuture
to_h
toIndexedSeq
toIterable
toIterator
ToList
ToLookup
toMap
toMultiMap
ToObservable
toSet
toSortedList
toStream
ToTask
toTraversable
toVector
tumbling
tumblingBuffer
unsafeCreate
unsubscribeOn
Using
When
Where
while
whileDo
Window
windowWithCount
windowWithTime
windowWithTimeOrCount
windowed
withFilter
withLatestFrom
Zip
zipArray
zipIterable
zipWith
zipWithIndex
++
+:
:+
Lesson 3: Implementing Your Own Operators
You can implement your own Observable operators. This page shows you how.
If your operator is designed to originate an Observable, rather than to transform or react
to a source Observable, use the create( ) method rather than trying to
implement Observable manually. Otherwise, follow the instructions below.
Chaining Your Custom Operators with Standard RxJava Operators
The following example shows how you can chain a custom operator (in this
example: myOperator) along with standard RxJava operators by using
the lift( ) operator:
Observable foo = barObservable.ofType(Integer).map({it*2}).lift(new
myOperator<T>()).map({"transformed by myOperator: " + it});
The following section will show how to form the scaffolding of your operator so that it will
work correctly with lift( ).
Implementing Your Operator
Define your operator as a public class that implements the Operator interface, like so:
public class myOperator<T> implements Operator<T> {
public myOperator( /* any necessary params here */ ) {
/* any necessary initialization here */
}
@Override
public Subscriber<? super T> call(final Subscriber<? super T> s) {
return new Subscriber<t>(s) {
@Override
public void onCompleted() {
/* add your own onCompleted behavior here, or just pass the completed
notification through: */
if(!s.isUnsubscribed()) {
s.onCompleted();
}
}
@Override
public void onError(Throwable t) {
/* add your own onError behavior here, or just pass the error notification
through: */
if(!s.isUnsubscribed()) {
s.onError(t);
}
}
@Override
public void onNext(T item) {
/* this example performs some sort of simple transformation on each incoming
item and then passes it along */
if(!s.isUnsubscribed()) {
transformedItem = myOperatorTransformOperation(item);
s.onNext(transformedItem);
}
}
};
}
}
Other Considerations
Your operator should check its Subscriber's isUnsubscribed( ) status before it
emits any item to (or sends any notification to) the Subscriber. Do not waste time
generating items that no Subscriber is interested in seeing.
Your operator should obey the core tenets of the Observable contract:
o It may call a Subscriber's onNext( ) method any number of times, but
these calls must be non-overlapping.
o It may call either a Subscriber's onCompleted( ) or onError( ) method,
but not both, exactly once, and it may not subsequently call a
Subscriber's onNext( ) method.
o If you are unable to guarantee that your operator conforms to the above
two tenets, you can add the serialize( ) operator to it to force the
correct behavior.
Do not block within your operator.
It is usually best that you compose new operators by combining existing ones, to
the extent that this is possible, rather than reinventing the wheel. RxJava itself
does this with some of its standard operators, for example:
o first( ) is defined as take(1).single( )
o ignoreElements( ) is defined as filter(alwaysFalse( ))
o reduce(a) is defined as scan(a).last( )
If your operator uses functions or lambdas that are passed in as parameters
(predicates, for instance), note that these may be sources of exceptions, and be
prepared to catch these and notify subscribers via onError( ) calls.
In general, notify subscribers of error conditions immediately, rather than making
an effort to emit more items first.
In some ReactiveX implementations, your operator may need to be sensitive to
that implementation’s “backpressure” strategies. (See, for example: Pitfalls of
Operator Implementations (part 2) by Dávid Karnok.)
See Also
Pitfalls of Operator Implementations (part 1) and (part 2) by Dávid Karnok.
Implementing Your Own Observable Operators (in RxJS) by Dennis Stoyanov
