RxJava Documentation

RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.

48,210 stars Java Java 1 file · ~36,034 tokens #flow#java#reactive-streams#rxjava
RAW Doc

1. Project Overview & Quickstart (ReactiveX/RxJava)

File: README.md

RxJava: Reactive Extensions for the JVM

:information_source: 4.0.0 release: 2026.11.30. Monday. [Track](https://github.com/ReactiveX/RxJava/milestone/30)




RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.

It extends the observer pattern to support sequences of data/events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety and concurrent data structures.

Version 4.x ([Javadoc](http://reactivex.io/RxJava/4.x/javadoc/))

  • :+1: Native Java 26 implementation1.
  • :+1: No 3rd party library required at runtime.
  • :+1: JPMS and :question: OSGi support still intact.
  • :+1: java.util.concurrent.Flow-based implementation.
  • :+1: Virtual Thread support; virtualCreate(), virtualTransform(), :eye: Schedulers.virtual().
  • :+1: New Streamable<T> built around Virtual Threads & virtual blocking. Think IAsyncEnumerable for Java. :satellite: in progress.
  • :+1: Using Java Cleaner API to detect resource leaks and using it for adaptive cleanups.
  • :information_source: Reactive Streams Test Compatibility Kit usage; Reactive-Streams.
  • :satellite: Reduce overload bloat by using record-based configurations.
  • :satellite: Internal optimizations now that I have the master :key:.
  • :eye: Possible usages for Scoped variables for context and per-item resource management.
  • :eye: Possible inclusion of 2nd and 3rd party operators.
  • :question: Android compatibility depends on your API level and what desugaring is available.
  • :warning: RxJava 3.x support will be toned down in the coming months, will be offered for +1 year after 4.x official release.

1 :stop: Unfortunately Java 27 has no language or API enhancements we can use, because Structured Concurrency remained preview for one.

Getting started

Setting up the dependency

The first step is to include RxJava 4 into your project, for example, as a Gradle compile dependency:

groovy
implementation "io.reactivex.rxjava4:rxjava:4.x.y"

(Please replace x and y with the latest version numbers:
)

Hello World

The second is to write the Hello World program:

java
package rxjava.examples;

import io.reactivex.rxjava4.core.*;

public class HelloWorld {
    public static void main(String[] args) {
        Flowable.just("Hello world").subscribe(System.out::println);
    }
}

Note that RxJava 4 components now live under io.reactivex.rxjava4 and the base classes and interfaces live under io.reactivex.rxjava4.core.

Base classes

RxJava 4 features several base classes you can discover operators on:

Some terminology

Upstream, downstream

The dataflows in RxJava consist of a source, zero or more intermediate steps followed by a data consumer or combinator step (where the step is responsible to consume the dataflow by some means):

java
source.operator1().operator2().operator3().subscribe(consumer);

source.flatMap(value -> source.operator1().operator2().operator3());

Here, if we imagine ourselves on operator2, looking to the left towards the source is called the upstream. Looking to the right towards the subscriber/consumer is called the downstream. This is often more apparent when each element is written on a separate line:

java
source
  .operator1()
  .operator2()
  .operator3()
  .subscribe(consumer)

Objects in motion

In RxJava's documentation, emission, emits, item, event, signal, data and message are considered synonyms and represent the object traveling along the dataflow.

Backpressure

When the dataflow runs through asynchronous steps, each step may perform different things with different speed. To avoid overwhelming such steps, which usually would manifest itself as increased memory usage due to temporary buffering or the need for skipping/dropping data, so-called backpressure is applied, which is a form of flow control where the steps can express how many items are they ready to process. This allows constraining the memory usage of the dataflows in situations where there is generally no way for a step to know how many items the upstream will send to it.

In RxJava, the dedicated Flowable class is designated to support backpressure and Observable is dedicated to the non-backpressured operations (short sequences, GUI interactions, etc.). The other types, Single, Maybe and Completable don't support backpressure nor should they; there is always room to store one item temporarily.

Since 4.0.0, the Streamable type gives natural backpressure because producers and consumers have to wait for
each other to hand over data. Since waiting is blocking, the type natively works with Virtual Threaded ExecutorServices
and the new Schedulers.virtual() Scheduler.

Assembly time

The preparation of dataflows by applying various intermediate operators happens in the so-called assembly time:

java
Flowable flow = Flowable.range(1, 5)
.map(v -> v * v)
.filter(v -> v % 3 == 0)
;

At this point, the data is not flowing yet and no side effects are happening.

Subscription time

This is a temporary state when subscribe() is called on a flow that establishes the chain of processing steps internally:

java
flow.subscribe(System.out::println)

This is when the subscription side effects are triggered (see doOnSubscribe). Some sources block or start emitting items right away in this state.

Runtime

This is the state when the flows are actively emitting items, errors or completion signals:

java
Observable.create(emitter -> {
     while (!emitter.isDisposed()) {
         long time = System.currentTimeMillis();
         emitter.onNext(time);
         if (time % 2 != 0) {
             emitter.onError(new IllegalStateException("Odd millisecond!"));
             break;
         }
     }
})
.subscribe(System.out::println, Throwable::printStackTrace);

Practically, this is when the body of the given example above executes.

Simple background computation

One of the common use cases for RxJava is to run some computation, network request on a background thread and show the results (or error) on the UI thread:

java
import io.reactivex.rxjava4.schedulers.Schedulers;

Flowable.fromCallable(() -> {
    Thread.sleep(1000); //  imitate expensive computation
    return "Done";
})
  .subscribeOn(Schedulers.io())
  .observeOn(Schedulers.single())
  .subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(2000); // <--- wait for the flow to finish

This style of chaining methods is called a fluent API which resembles the builder pattern. However, RxJava's reactive types are immutable; each of the method calls returns a new Flowable with added behavior. To illustrate, the example can be rewritten as follows:

java
Flowable<String> source = Flowable.fromCallable(() -> {
    Thread.sleep(1000); //  imitate expensive computation
    return "Done";
});

Flowable<String> runBackground = source.subscribeOn(Schedulers.io());

Flowable<String> showForeground = runBackground.observeOn(Schedulers.single());

showForeground.subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(2000);

Typically, you can move computations or blocking IO to some other thread via subscribeOn. Once the data is ready, you can make sure they get processed on the foreground or GUI thread via observeOn.

Schedulers

RxJava operators don't work with Threads or ExecutorServices directly but with so-called Schedulers that abstract away sources of concurrency behind a uniform API. RxJava 4 features several standard schedulers accessible via Schedulers utility class.

  • Schedulers.computation(): Run computation intensive work on a fixed number of dedicated threads in the background. Most asynchronous operators use this as their default Scheduler.
  • Schedulers.cached(): Run I/O-like or blocking operations on a dynamically changing set of threads backed by native OS threads. :warning: Can exhaust system resources!
  • Schedulers.virtual(): Run I/O-like or blocking scatter-gather operations in a sequential manner on threads with virtualized stacks attached and detached to native OS threads on demand. :information_source: Helps with the issues around unboundedness of Schedulers.cached().
  • Schedulers.single(): Run work on a single thread in a sequential and FIFO manner.
  • Schedulers.trampoline(): Run work in a sequential and FIFO manner in one of the participating threads, usually for testing purposes.
  • Schedulers.createParallel(): Allows creating a Scheduler with an user-configurable worker pool size and other parameters to contrast computation() which is always set to availableProcessors()/configured amount globally.
  • Schedulers.createBlocking(): Allows creating an event-loop style Scheduler which runs tasks and blocks on the thread calling execute(). Can be used to pull tasks onto a specific thread or have it itself run in a virtual threaded executor for maximum efficiency.
  • Scheduler.shared(): Every Scheduler or Worker can now be shared and act like its own full Scheduler with lifecycle tracking and dispose support. I.e., share one worker of cached() like it is some kind of Schedulers.single().

In 4.x, the traditional I/O scheduler Schedulers.io() has been API deprecated and delegates to Schedulers.cached() for compatibility reasons. It is recommended you decide at these deprecated code locations which standard (or custom) scheduler
to use: cached() like before or the new virtual() for more efficient system resource usages.

These are available on all JVM platforms but some specific platforms, such as Android, have their own typical Schedulers defined: AndroidSchedulers.mainThread(), SwingScheduler.instance() or JavaFXScheduler.platform().

In addition, there is an option to wrap an existing Executor (and its subtypes such as ExecutorService) into a Scheduler via Schedulers.from(Executor). This can be used, for example, to have a larger but still fixed pool of threads (unlike computation() and io() respectively).

The Thread.sleep(2000); at the end is no accident. In RxJava the default Schedulers run on daemon threads, which means once the Java main thread exits, they all get stopped and background computations may never happen. Sleeping for some time in this example situations lets you see the output of the flow on the console with time to spare.

Concurrency within a flow

Flows in RxJava are sequential in nature split into processing stages that may run concurrently with each other:

java
Flowable.range(1, 10)
  .observeOn(Schedulers.computation())
  .map(v -> v * v)
  .blockingSubscribe(System.out::println);

This example flow squares the numbers from 1 to 10 on the computation Scheduler and consumes the results on the "main" thread (more precisely, the caller thread of blockingSubscribe). However, the lambda v -> v * v doesn't run in parallel for this flow; it receives the values 1 to 10 on the same computation thread one after the other.

Parallel processing

Processing the numbers 1 to 10 in parallel is a bit more involved:

java
Flowable.range(1, 10)
  .flatMap(v ->
      Flowable.just(v)
        .subscribeOn(Schedulers.computation())
        .map(w -> w * w)
  )
  .blockingSubscribe(System.out::println);

Practically, parallelism in RxJava means running independent flows and merging their results back into a single flow. The operator flatMap does this by first mapping each number from 1 to 10 into its own individual Flowable, runs them and merges the computed squares.

Note, however, that flatMap doesn't guarantee any order and the items from the inner flows may end up interleaved. There are alternative operators:

  • concatMap that maps and runs one inner flow at a time and
  • concatMapEager which runs all inner flows "at once" but the output flow will be in the order those inner flows were created.

Alternatively, the Flowable.parallel() operator and the ParallelFlowable type help achieve the same parallel processing pattern:

java
Flowable.range(1, 10)
  .parallel()
  .runOn(Schedulers.computation())
  .map(v -> v * v)
  .sequential()
  .blockingSubscribe(System.out::println);

Dependent sub-flows

flatMap is a powerful operator and helps in a lot of situations. For example, given a service that returns a Flowable, we'd like to call another service with values emitted by the first service:

java
Flowable inventorySource = warehouse.getInventoryAsync();

inventorySource
    .flatMap(inventoryItem -> erp.getDemandAsync(inventoryItem.getId())
            .map(demand -> "Item " + inventoryItem.getName() + " has demand " + demand))
    .subscribe(System.out::println);

Continuations

Sometimes, when an item has become available, one would like to perform some dependent computations on it. This is sometimes called continuations and, depending on what should happen and what types are involved, may involve various operators to accomplish.

Dependent

The most typical scenario is given a value, invoke another service, await and continue with its result:

java
service.apiCall()
.flatMap(value -> service.anotherApiCall(value))
.flatMap(next -> service.finalCall(next))

It is often the case also that later sequences would require values from earlier mappings. This can be achieved by moving the outer flatMap into the inner parts of the previous flatMap for example:

java
service.apiCall()
.flatMap(value ->
    service.anotherApiCall(value)
    .flatMap(next -> service.finalCallBoth(value, next))
)

Here, the original value will be available inside the inner flatMap, courtesy of lambda variable capture.

Non-dependent

In other scenarios, the result(s) of the first source/dataflow is irrelevant and one would like to continue with a quasi independent another source. Here, flatMap works as well:

java
Observable continued = sourceObservable.flatMapSingle(ignored -> someSingleSource)
continued.map(v -> v.toString())
  .subscribe(System.out::println, Throwable::printStackTrace);

however, the continuation in this case stays Observable instead of the likely more appropriate Single. (This is understandable because
from the perspective of flatMapSingle, sourceObservable is a multivalued source and thus the mapping may result in multiple values as well).

Often though there is a way that is somewhat more expressive (and also lower overhead) by using Completable as the mediator and its operator andThen to resume with something else:

java
sourceObservable
  .ignoreElements()           // returns Completable
  .andThen(someSingleSource)
  .map(v -> v.toString())

The only dependency between the sourceObservable and the someSingleSource is that the former should complete normally in order for the latter to be consumed.

Deferred-dependent

Sometimes, there is an implicit data dependency between the previous sequence and the new sequence that, for some reason, was not flowing through the "regular channels". One would be inclined to write such continuations as follows:

java
AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
  .doOnNext(ignored -> count.incrementAndGet())
  .ignoreElements()
  .andThen(Single.just(count.get()))
  .subscribe(System.out::println);

Unfortunately, this prints 0 because Single.just(count.get()) is evaluated at assembly time when the dataflow hasn't even run yet. We need something that defers the evaluation of this Single source until runtime when the main source completes:

java
AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
  .doOnNext(ignored -> count.incrementAndGet())
  .ignoreElements()
  .andThen(Single.defer(() -> Single.just(count.get())))
  .subscribe(System.out::println);

or

java
AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
  .doOnNext(ignored -> count.incrementAndGet())
  .ignoreElements()
  .andThen(Single.fromCallable(() -> count.get()))
  .subscribe(System.out::println);

Type conversions

Sometimes, a source or service returns a different type than the flow that is supposed to work with it. For example, in the inventory example above, getDemandAsync could return a Single<DemandRecord>. If the code example is left unchanged, this will result in a compile-time error (however, often with a misleading error message about lack of overload).

In such situations, there are usually two options to fix the transformation: 1) convert to the desired type or 2) find and use an overload of the specific operator supporting the different type.

Converting to the desired type

Each reactive base class features operators that can perform such conversions, including the protocol conversions, to match some other type. The following matrix shows the available conversion options:

Flowable Observable Single Maybe Completable Streamable
Flowable toObservable first, firstOrError, single, singleOrError, last, lastOrError1 firstElement, singleElement, lastElement ignoreElements toStreamable
Observable toFlowable2 first, firstOrError, single, singleOrError, last, lastOrError1 firstElement, singleElement, lastElement ignoreElements toStreamable
Single toFlowable3 toObservable toMaybe ignoreElement toStreamable
Maybe toFlowable3 toObservable toSingle ignoreElement toStreamable
Completable toFlowable toObservable toSingle toMaybe toStreamable
Streamable toFlowable toObservable TBD TBD TBD

1: When turning a multivalued source into a single-valued source, one should decide which of the many source values should be considered as the result.

2: Turning an Observable into Flowable requires an additional decision: what to do with the potential unconstrained flow
of the source Observable? There are several strategies available (such as buffering, dropping, keeping the latest) via the BackpressureStrategy parameter or via standard Flowable operators such as onBackpressureBuffer, onBackpressureDrop, onBackpressureLatest which also
allow further customization of the backpressure behavior.

3: When there is only (at most) one source item, there is no problem with backpressure as it can be always stored until the downstream is ready to consume.

Using an overload with the desired type

Many frequently used operator has overloads that can deal with the other types. These are usually named with the suffix of the target type:

Operator Overloads
flatMap flatMapSingle, flatMapMaybe, flatMapCompletable, flatMapIterable
concatMap concatMapSingle, concatMapMaybe, concatMapCompletable, concatMapIterable
switchMap switchMapSingle, switchMapMaybe, switchMapCompletable

The reason these operators have a suffix instead of simply having the same name with different signature is type erasure. Java doesn't consider signatures such as operator(Function<T, Single<R>>) and operator(Function<T, Maybe<R>>) different (unlike C#) and due to erasure, the two operators would end up as duplicate methods with the same signature.

Operator naming conventions

Naming in programming is one of the hardest things as names are expected to be not long, expressive, capturing and easily memorable. Unfortunately, the target language (and pre-existing conventions) may not give too much help in this regard (unusable keywords, type erasure, type ambiguities, etc.).

Unusable keywords

In the original Rx.NET, the operator that emits a single item and then completes is called Return(T). Since the Java convention is to have a lowercase letter start a method name, this would have been return(T) which is a keyword in Java and thus not available. Therefore, RxJava chose to name this operator just(T). The same limitation exists for the operator Switch, which had to be named switchOnNext. Yet another example is Catch which was named onErrorResumeNext.

Type erasure

Many operators that expect the user to provide some function returning a reactive type can't be overloaded because the type erasure around a Function<T, X> turns such method signatures into duplicates. RxJava chose to name such operators by appending the type as suffix as well:

java
Flowable<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper)

Flowable<R> flatMapMaybe(Function<? super T, ? extends MaybeSource<? extends R>> mapper)

Type ambiguities

Even though certain operators have no problems from type erasure, their signature may turn up being ambiguous, especially if one uses Java 8 and lambdas. For example, there are several overloads of concatWith taking the various other reactive base types as arguments (for providing convenience and performance benefits in the underlying implementation):

java
Flowable<T> concatWith(Publisher<? extends T> other);

Flowable<T> concatWith(SingleSource<? extends T> other);

Both Publisher and SingleSource appear as functional interfaces (types with one abstract method) and may encourage users to try to provide a lambda expression:

java
someSource.concatWith(s -> Single.just(2))
.subscribe(System.out::println, Throwable::printStackTrace);

Unfortunately, this approach doesn't work and the example does not print 2 at all. In fact, since version 2.1.10, it doesn't
even compile because at least 4 concatWith overloads exist and the compiler finds the code above ambiguous.

The user in such situations probably wanted to defer some computation until the someSource has completed, thus the correct
unambiguous operator should have been defer:

java
someSource.concatWith(Single.defer(() -> Single.just(2)))
.subscribe(System.out::println, Throwable::printStackTrace);

Sometimes, a suffix is added to avoid logical ambiguities that may compile but produce the wrong type in a flow:

java
Flowable<T> merge(Publisher<? extends Publisher<? extends T>> sources);

Flowable<T> mergeArray(Publisher<? extends T>... sources);

This can get also ambiguous when functional interface types get involved as the type argument T.

Error handling

Dataflows can fail, at which point the error is emitted to the consumer(s). Sometimes though, multiple sources may fail at which point there is a choice whether or not wait for all of them to complete or fail.

:warning: With 4.x, the number of overloads were reduced and all DelayError methods have been folded into
configuration record parameter(s).

To indicate this opportunity, operators now use configuration records, such as StandardBufferedConfig, StandardConcurrentConfig and StandardConcurrentBufferedConfig to pass along an error handling settings:

java
Flowable<T> concat(Publisher<? extends Publisher<? extends T>> sources);

Flowable<T> concat(Publisher<? extends Publisher<? extends T>> sources, StandardBufferedConfig config);

var config = new StandardBufferedConfig(ErrorMode.BOUNDARY);

var config = new StandardBufferedConfig(true); // equivalent to ErrorMode.END

The ErrorMode enum has 3 modes:

  • IMMEDIATE - emit the error as soon as possible
  • BOUNDARY - wait until a boundary or inner source change, may not make sense for certain operators and will act as END
  • END - wait until all sources (outer, inner) have terminated and present an aggregated error (usually via CompositeException) to the consumer.

With varargs-style operators, the configuration record has to precede the varargs parameter:

java
Flowable<T> concatArrayEager(StandardConcurrentBufferedConfig config, Publisher<? extends T>... sources);

Base class vs base type

The base classes can be considered heavy due to the sheer number of static and instance methods on them. RxJava 4's design was heavily influenced by the Reactive Streams specification, therefore, the library features a class and an interface per each reactive type:

Type Class Interface Consumer
0..N backpressured Flowable Flow.Publisher1 Flow.Subscriber
0..N unbounded Observable ObservableSource2 Observer
1 element or error Single SingleSource SingleObserver
0..1 element or error Maybe MaybeSource MaybeObserver
0 element or error Completable CompletableSource CompletableObserver
0..N coordinated Streamable 3 Streamer, StreamSink

1The java.util.concurrent.Flow.Publisher is part of the Java internal Flow library. It is the main type to interact with other reactive libraries through a standardized mechanism governed by the Reactive Streams specification.

2The naming convention of the interface was to append Source to the semi-traditional class name. There is no FlowableSource since Publisher is provided by the Reactive Streams library (and subtyping it wouldn't have helped with interoperation either). These interfaces are, however, not standard in the sense of the Reactive Streams specification and are currently RxJava specific only.

3 2026.07.09 we haven't decided yet to have a StreamableSource becaue we control Streamable, unlike Flow.Publisher, so we can add defaulted operators there and still have Streamable a functional interface.

R8 and ProGuard settings

By default, RxJava itself doesn't require any ProGuard/R8 settings and should work without problems. Unfortunately, the Reactive Streams dependency since version 1.0.3 has embedded Java 9 class files in its JAR that can cause warnings with the plain ProGuard:

text
Warning: org.reactivestreams.FlowAdapters$FlowPublisherFromReactive: can't find superclass or interface java.util.concurrent.Flow$Publisher
Warning: org.reactivestreams.FlowAdapters$FlowToReactiveProcessor: can't find superclass or interface java.util.concurrent.Flow$Processor
Warning: org.reactivestreams.FlowAdapters$FlowToReactiveSubscriber: can't find superclass or interface java.util.concurrent.Flow$Subscriber
Warning: org.reactivestreams.FlowAdapters$FlowToReactiveSubscription: can't find superclass or interface java.util.concurrent.Flow$Subscription
Warning: org.reactivestreams.FlowAdapters: can't find referenced class java.util.concurrent.Flow$Publisher

It is recommended one sets up the following -dontwarn entry in the application's proguard-ruleset file:

text
-dontwarn java.util.concurrent.Flow*

For R8, the RxJava jar includes the META-INF/proguard/rxjava4.pro with the same no-warning clause and should apply automatically.

Further reading

For further details, consult the wiki.

Communication

Versioning

Version 4.x is in development. Bugfixes will be applied to both 3.x and 4.x branches if possible, but new features will only be added to 4.x.

Minor 4.x increments (such as 4.1, 4.2, etc) will occur when non-trivial new functionality is added or significant enhancements or bug fixes occur that may have behavioral changes that may affect some edge cases (such as dependence on behavior resulting from a bug). An example of an enhancement that would classify as this is adding reactive pull backpressure support to an operator that previously did not support it. This should be backwards compatible but does behave differently.

Patch 4.x.y increments (such as 4.0.0 -> 4.0.1, 4.3.1 -> 4.3.2, etc) will occur for bug fixes and trivial functionality (like adding a method overload). New functionality marked with an @Beta or @Experimental annotation can also be added in the patch releases to allow rapid exploration and iteration of unstable new functionality.

@Beta

APIs marked with the @Beta annotation at the class or method level are subject to change. They can be modified in any way, or even removed, at any time. If your code is a library itself (i.e. it is used on the CLASSPATH of users outside your control), you should not use beta APIs, unless you repackage them (e.g. using ProGuard, shading, etc).

@Experimental

APIs marked with the @Experimental annotation at the class or method level will almost certainly change. They can be modified in any way, or even removed, at any time. You should not use or rely on them in any production code. They are purely to allow broad testing and feedback.

@Deprecated

APIs marked with the @Deprecated annotation at the class or method level will remain supported until the next major release, but it is recommended to stop using them.

io.reactivex.rxjava4.internal.*

All code inside the io.reactivex.rxjava4.internal.* packages are considered private API and should not be relied upon at all. It can change at any time.

Full Documentation

Binaries

Binaries and dependency information for Maven, Ivy, Gradle and others can be found at http://search.maven.org.

Example for Gradle:

groovy
implementation 'io.reactivex.rxjava4:rxjava:x.y.z'

and for Maven:

xml
<dependency>
    <groupId>io.reactivex.rxjava4</groupId>
    <artifactId>rxjava</artifactId>
    <version>x.y.z</version>
</dependency>

and for Ivy:

xml
<dependency org="io.reactivex.rxjava4" name="rxjava" rev="x.y.z" />

Snapshots

Snapshots after May 19th, 2025 are available via https://central.sonatype.com/repository/maven-snapshots/io/reactivex/rxjava4/rxjava/

groovy
repositories {
  maven { url 'https://central.sonatype.com/repository/maven-snapshots' }
}

dependencies {
  implementation 'io.reactivex.rxjava4:rxjava:4.0.0-SNAPSHOT'
}

Javadoc snapshots are available at https://reactivex.io/RxJava/4.x/javadoc/snapshot

Build

To build:

text
$ git clone [email protected]:ReactiveX/RxJava.git
$ cd RxJava/
$ ./gradlew build

Further details on building can be found on the Getting Started page of the wiki.

Bugs and Feedback

For bugs, questions and discussions please use the GitHub Issues.

Prior versions

Version 3.x ([Javadoc](http://reactivex.io/RxJava/3.x/javadoc/))

  • Single dependency: Reactive-Streams.
  • Java 8+ or Android API 21+ required.
  • Java 8 lambda-friendly API.
  • Android desugar friendly.
  • Fixed API mistakes and many limits of RxJava 2.
  • Intended to be a replacement for RxJava 2 with relatively few binary incompatible changes.
  • Non-opinionated about the source of concurrency (threads, pools, event loops, fibers, actors, etc.).
  • Async or synchronous execution.
  • Virtual time and schedulers for parameterized concurrency.
  • Test and diagnostic support via test schedulers, test consumers and plugin hooks.
  • Interop with newer JDK versions via 3rd party libraries, such as

Learn more about RxJava in general on the Wiki Home.

:information_source: Please read the What's different in 3.0 for details on the changes and migration information when upgrading from 2.x.

Version 2.x

The 2.x version is end-of-life as of February 28, 2021. No further development, support, maintenance, PRs and updates will happen. The Javadoc of the very last version, 2.2.21, will remain accessible.

Version 1.x

The 1.x version is end-of-life as of March 31, 2018. No further development, support, maintenance, PRs and updates will happen. The Javadoc of the very last version, 1.3.8, will remain accessible.

LICENSE

text
Copyright (c) 2016-present, RxJava Contributors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

Copyright (c) 2016-present, RxJava Contributors.
Twitter @RxJava | Gitter @RxJava

File: docs/_Sidebar.md

File: docs/_Sidebar.md.md

File: docs/Additional-Reading.md

A more complete and up-to-date list of resources can be found at the reactivex.io site

Introducing Reactive Programming

How Netflix Is Using RxJava

RxScala

Rx.NET

RxJS

RxAndroid

Miscellany

File: docs/Alphabetical-List-of-Observable-Operators.md

  • aggregate( ) — see reduce( )
  • all( ) — determine whether all items emitted by an Observable meet some criteria
  • amb( ) — given two or more source Observables, emits all of the items from the first of these Observables to emit an item
  • ambWith( ) — instance version of amb( )
  • and( ) — combine the emissions from two or more source Observables into a Pattern (rxjava-joins)
  • apply( ) (scala) — see create( )
  • asObservable( ) (kotlin) — see from( ) (et al.)
  • asyncAction( ) — convert an Action into an Observable that executes the Action and emits its return value (rxjava-async)
  • asyncFunc( ) — convert a function into an Observable that executes the function and emits its return value (rxjava-async)
  • averageDouble( ) — calculates the average of Doubles emitted by an Observable and emits this average (rxjava-math)
  • averageFloat( ) — calculates the average of Floats emitted by an Observable and emits this average (rxjava-math)
  • averageInteger( ) — calculates the average of Integers emitted by an Observable and emits this average (rxjava-math)
  • averageLong( ) — calculates the average of Longs emitted by an Observable and emits this average (rxjava-math)
  • blocking( ) (clojure) — see toBlocking( )
  • buffer( ) — periodically gather items from an Observable into bundles and emit these bundles rather than emitting the items one at a time
  • byLine( ) (StringObservable) — converts an Observable of Strings into an Observable of Lines by treating the source sequence as a stream and splitting it on line-endings
  • cache( ) — remember the sequence of items emitted by the Observable and emit the same sequence to future Subscribers
  • cast( ) — cast all items from the source Observable into a particular type before reemitting them
  • catch( ) (clojure) — see onErrorResumeNext( )
  • chunkify( ) — returns an iterable that periodically returns a list of items emitted by the source Observable since the last list (⁇)
  • collect( ) — collects items emitted by the source Observable into a single mutable data structure and returns an Observable that emits this structure
  • 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
  • combineLatestWith( ) (scala) — instance version of combineLatest( )
  • concat( ) — concatenate two or more Observables sequentially
  • concatMap( ) — transform the items emitted by an Observable into Observables, then flatten this into a single Observable, without interleaving
  • concatWith( ) — instance version of concat( )
  • connect( ) — instructs a Connectable Observable to begin emitting items
  • cons( ) (clojure) — see concat( )
  • contains( ) — determine whether an Observable emits a particular item or not
  • count( ) — counts the number of items emitted by an Observable and emits this count
  • countLong( ) — counts the number of items emitted by an Observable and emits this count
  • create( ) — create an Observable from scratch by means of a function
  • cycle( ) (clojure) — see repeat( )
  • debounce( ) — only emit an item from the source Observable after a particular timespan has passed without the Observable emitting any other items
  • decode( ) (StringObservable) — convert a stream of multibyte characters into an Observable that emits byte arrays that respect character boundaries
  • defaultIfEmpty( ) — emit items from the source Observable, or emit a default item if the source Observable completes after emitting no items
  • defer( ) — do not create the Observable until a Subscriber subscribes; create a fresh Observable on each subscription
  • deferFuture( ) — convert a Future that returns an Observable into an Observable, but do not attempt to get the Observable that the Future returns until a Subscriber subscribes (rxjava-async)
  • deferCancellableFuture( ) — convert a Future that returns an Observable into an Observable in a way that monitors the subscription status of the Observable to determine whether to halt work on the Future, but do not attempt to get the returned Observable until a Subscriber subscribes (⁇)(rxjava-async)
  • delay( ) — shift the emissions from an Observable forward in time by a specified amount
  • dematerialize( ) — convert a materialized Observable back into its non-materialized form
  • distinct( ) — suppress duplicate items emitted by the source Observable
  • distinctUntilChanged( ) — suppress duplicate consecutive items emitted by the source Observable
  • do( ) (clojure) — see doOnEach( )
  • doOnCompleted( ) — register an action to take when an Observable completes successfully
  • doOnEach( ) — register an action to take whenever an Observable emits an item
  • doOnError( ) — register an action to take when an Observable completes with an error
  • doOnNext( ) — see doOnEach( )
  • doOnRequest( ) — register an action to take when items are requested from an Observable via reactive-pull backpressure (⁇)
  • doOnSubscribe( ) — register an action to take when an observer subscribes to an Observable
  • doOnTerminate( ) — register an action to take when an Observable completes, either successfully or with an error
  • doOnUnsubscribe( ) — register an action to take when an observer unsubscribes from an Observable
  • doWhile( ) — emit the source Observable's sequence, and then repeat the sequence as long as a condition remains true (contrib-computation-expressions)
  • drop( ) (scala/clojure) — see skip( )
  • dropRight( ) (scala) — see skipLast( )
  • dropUntil( ) (scala) — see skipUntil( )
  • dropWhile( ) (scala) — see skipWhile( )
  • drop-while( ) (clojure) — see skipWhile( )
  • elementAt( ) — emit item n emitted by the source Observable
  • elementAtOrDefault( ) — emit item n emitted by the source Observable, or a default item if the source Observable emits fewer than n items
  • empty( ) — create an Observable that emits nothing and then completes
  • encode( ) (StringObservable) — transform an Observable that emits strings into an Observable that emits byte arrays that respect character boundaries of multibyte characters in the original strings
  • error( ) — create an Observable that emits nothing and then signals an error
  • every( ) (clojure) — see all( )
  • exists( ) — determine whether an Observable emits any items or not
  • filter( ) — filter items emitted by an Observable
  • finally( ) (clojure) — see finallyDo( )
  • filterNot( ) (scala) — see filter( )
  • finallyDo( ) — register an action to take when an Observable completes
  • first( ) (Observable) — emit only the first item emitted by an Observable, or the first item that meets some condition
  • first( ) (BlockingObservable) — emit only the first item emitted by an Observable, or the first item that meets some condition
  • firstOrDefault( ) (Observable) — emit only the first item emitted by an Observable, or the first item that meets some condition, or a default value if the source Observable is empty
  • firstOrDefault( ) (BlockingObservable) — emit only the first item emitted by an Observable, or the first item that meets some condition, or a default value if the source Observable is empty
  • firstOrElse( ) (scala) — see firstOrDefault( ) or firstOrDefault( ) (BlockingObservable)
  • flatMap( ) — transform the items emitted by an Observable into Observables, then flatten this into a single Observable
  • flatMapIterable( ) — create Iterables corresponding to each emission from a source Observable and merge the results into a single Observable
  • flatMapIterableWith( ) (scala) — instance version of flatMapIterable( )
  • flatMapWith( ) (scala) — instance version of flatmap( )
  • flatten( ) (scala) — see merge( )
  • flattenDelayError( ) (scala) — see mergeDelayError( )
  • foldLeft( ) (scala) — see reduce( )
  • forall( ) (scala) — see all( )
  • forEach( ) (Observable) — see subscribe( )
  • forEach( ) (BlockingObservable) — invoke a function on each item emitted by the Observable; block until the Observable completes
  • forEachFuture( ) (Async) — pass Subscriber methods to an Observable but also have it behave like a Future that blocks until it completes (rxjava-async)
  • forEachFuture( ) (BlockingObservable)— create a futureTask that will invoke a specified function on each item emitted by an Observable (⁇)
  • forIterable( ) — apply a function to the elements of an Iterable to create Observables which are then concatenated (⁇)
  • from( ) — convert an Iterable, a Future, or an Array into an Observable
  • from( ) (StringObservable) — convert a stream of characters or a Reader into an Observable that emits byte arrays or Strings
  • fromAction( ) — convert an Action into an Observable that invokes the action and emits its result when a Subscriber subscribes (rxjava-async)
  • fromCallable( ) — convert a Callable into an Observable that invokes the callable and emits its result or exception when a Subscriber subscribes (rxjava-async)
  • fromCancellableFuture( ) — convert a Future into an Observable in a way that monitors the subscription status of the Observable to determine whether to halt work on the Future, but do not attempt to get the Future's value until a Subscriber subscribes (⁇)(rxjava-async)
  • fromFunc0( ) — see fromCallable( ) (rxjava-async)
  • fromFuture( ) — convert a Future into an Observable, but do not attempt to get the Future's value until a Subscriber subscribes (⁇)
  • fromRunnable( ) — convert a Runnable into an Observable that invokes the runable and emits its result when a Subscriber subscribes (rxjava-async)
  • generate( ) — create an Observable that emits a sequence of items as generated by a function of your choosing (⁇)
  • generateAbsoluteTime( ) — create an Observable that emits a sequence of items as generated by a function of your choosing, with each item emitted at an item-specific time (⁇)
  • generator( ) (clojure) — see generate( )
  • getIterator( ) — convert the sequence emitted by the Observable into an Iterator
  • groupBy( ) — divide an Observable into a set of Observables that emit groups of items from the original Observable, organized by key
  • group-by( ) (clojure) — see groupBy( )
  • groupByUntil( ) — a variant of the groupBy( ) operator that closes any open GroupedObservable upon a signal from another Observable (⁇)
  • groupJoin( ) — combine the items emitted by two Observables whenever one item from one Observable falls within a window of duration specified by an item emitted by the other Observable
  • head( ) (scala) — see first( ) (BlockingObservable)
  • headOption( ) (scala) — see firstOrDefault( ) or firstOrDefault( ) (BlockingObservable)
  • headOrElse( ) (scala) — see firstOrDefault( ) or firstOrDefault( ) (BlockingObservable)
  • ifThen( ) — only emit the source Observable's sequence if a condition is true, otherwise emit an empty or default sequence (contrib-computation-expressions)
  • ignoreElements( ) — discard the items emitted by the source Observable and only pass through the error or completed notification
  • interval( ) — create an Observable that emits a sequence of integers spaced by a given time interval
  • into( ) (clojure) — see reduce( )
  • isEmpty( ) — determine whether an Observable emits any items or not
  • items( ) (scala) — see just( )
  • join( ) — combine the items emitted by two Observables whenever one item from one Observable falls within a window of duration specified by an item emitted by the other Observable
  • join( ) (StringObservable) — converts an Observable that emits a sequence of strings into an Observable that emits a single string that concatenates them all, separating them by a specified string
  • just( ) — convert an object into an Observable that emits that object
  • last( ) (BlockingObservable) — block until the Observable completes, then return the last item emitted by the Observable
  • last( ) (Observable) — emit only the last item emitted by the source Observable
  • lastOption( ) (scala) — see lastOrDefault( ) or lastOrDefault( ) (BlockingObservable)
  • lastOrDefault( ) (BlockingObservable) — block until the Observable completes, then return the last item emitted by the Observable or a default item if there is no last item
  • lastOrDefault( ) (Observable) — emit only the last item emitted by an Observable, or a default value if the source Observable is empty
  • lastOrElse( ) (scala) — see lastOrDefault( ) or lastOrDefault( ) (BlockingObservable)
  • latest( ) — returns an iterable that blocks until or unless the Observable emits an item that has not been returned by the iterable, then returns the latest such item
  • length( ) (scala) — see count( )
  • limit( ) — see take( )
  • longCount( ) (scala) — see countLong( )
  • map( ) — transform the items emitted by an Observable by applying a function to each of them
  • mapcat( ) (clojure) — see concatMap( )
  • mapMany( ) — see: flatMap( )
  • materialize( ) — convert an Observable into a list of Notifications
  • max( ) — emits the maximum value emitted by a source Observable (rxjava-math)
  • maxBy( ) — emits the item emitted by the source Observable that has the maximum key value (rxjava-math)
  • merge( ) — combine multiple Observables into one
  • mergeDelayError( ) — combine multiple Observables into one, allowing error-free Observables to continue before propagating errors
  • merge-delay-error( ) (clojure) — see mergeDelayError( )
  • mergeMap( ) * — see: flatMap( )
  • mergeMapIterable( ) — see: flatMapIterable( )
  • mergeWith( ) — instance version of merge( )
  • min( ) — emits the minimum value emitted by a source Observable (rxjava-math)
  • minBy( ) — emits the item emitted by the source Observable that has the minimum key value (rxjava-math)
  • mostRecent( ) — returns an iterable that always returns the item most recently emitted by the Observable
  • multicast( ) — represents an Observable as a Connectable Observable
  • never( ) — create an Observable that emits nothing at all
  • next( ) — returns an iterable that blocks until the Observable emits another item, then returns that item
  • nonEmpty( ) (scala) — see isEmpty( )
  • nth( ) (clojure) — see elementAt( ) and elementAtOrDefault( )
  • observeOn( ) — specify on which Scheduler a Subscriber should observe the Observable
  • ofType( ) — emit only those items from the source Observable that are of a particular class
  • onBackpressureBlock( ) — block the Observable's thread until the Observer is ready to accept more items from the Observable (⁇)
  • onBackpressureBuffer( ) — maintain a buffer of all emissions from the source Observable and emit them to downstream Subscribers according to the requests they generate
  • onBackpressureDrop( ) — drop emissions from the source Observable unless there is a pending request from a downstream Subscriber, in which case emit enough items to fulfill the request
  • onErrorFlatMap( ) — instructs an Observable to emit a sequence of items whenever it encounters an error (⁇)
  • onErrorResumeNext( ) — instructs an Observable to emit a sequence of items if it encounters an error
  • onErrorReturn( ) — instructs an Observable to emit a particular item when it encounters an error
  • onExceptionResumeNext( ) — instructs an Observable to continue emitting items after it encounters an exception (but not another variety of throwable)
  • orElse( ) (scala) — see defaultIfEmpty( )
  • parallel( ) — split the work done on the emissions from an Observable into multiple Observables each operating on its own parallel thread (⁇)
  • parallelMerge( ) — combine multiple Observables into smaller number of Observables (⁇)
  • pivot( ) — combine multiple sets of grouped observables so that they are arranged primarily by group rather than by set (⁇)
  • publish( ) — represents an Observable as a Connectable Observable
  • publishLast( ) — represent an Observable as a Connectable Observable that emits only the last item emitted by the source Observable (⁇)
  • range( ) — create an Observable that emits a range of sequential integers
  • reduce( ) — apply a function to each emitted item, sequentially, and emit only the final accumulated value
  • reductions( ) (clojure) — see scan( )
  • refCount( ) — makes a Connectable Observable behave like an ordinary Observable
  • repeat( ) — create an Observable that emits a particular item or sequence of items repeatedly
  • repeatWhen( ) — create an Observable that emits a particular item or sequence of items repeatedly, depending on the emissions of a second Observable
  • replay( ) — ensures that all Subscribers see the same sequence of emitted items, even if they subscribe after the Observable begins emitting the items
  • rest( ) (clojure) — see next( )
  • return( ) (clojure) — see just( )
  • retry( ) — if a source Observable emits an error, resubscribe to it in the hopes that it will complete without error
  • retrywhen( ) — if a source Observable emits an error, pass that error to another Observable to determine whether to resubscribe to the source
  • runAsync( ) — returns a StoppableObservable that emits multiple actions as generated by a specified Action on a Scheduler (rxjava-async)
  • sample( ) — emit the most recent items emitted by an Observable within periodic time intervals
  • scan( ) — apply a function to each item emitted by an Observable, sequentially, and emit each successive value
  • seq( ) (clojure) — see getIterator( )
  • sequenceEqual( ) — test the equality of sequences emitted by two Observables
  • sequenceEqualWith( ) (scala) — instance version of sequenceEqual( )
  • serialize( ) — force an Observable to make serialized calls and to be well-behaved
  • share( ) — see refCount( )
  • single( ) (BlockingObservable) — if the source Observable completes after emitting a single item, return that item, otherwise throw an exception
  • single( ) (Observable) — if the source Observable completes after emitting a single item, emit that item, otherwise notify of an exception
  • singleOption( ) (scala) — see singleOrDefault( ) (BlockingObservable)
  • singleOrDefault( ) (BlockingObservable) — if the source Observable completes after emitting a single item, return that item, otherwise return a default item
  • singleOrDefault( ) (Observable) — if the source Observable completes after emitting a single item, emit that item, otherwise emit a default item
  • singleOrElse( ) (scala) — see singleOrDefault( )
  • size( ) (scala) — see count( )
  • skip( ) — ignore the first n items emitted by an Observable
  • skipLast( ) — ignore the last n items emitted by an Observable
  • skipUntil( ) — discard items emitted by a source Observable until a second Observable emits an item, then emit the remainder of the source Observable's items
  • skipWhile( ) — discard items emitted by an Observable until a specified condition is false, then emit the remainder
  • sliding( ) (scala) — see window( )
  • slidingBuffer( ) (scala) — see buffer( )
  • split( ) (StringObservable) — converts an Observable of Strings into an Observable of Strings that treats the source sequence as a stream and splits it on a specified regex boundary
  • start( ) — create an Observable that emits the return value of a function (rxjava-async)
  • startCancellableFuture( ) — convert a function that returns Future into an Observable that emits that Future's return value in a way that monitors the subscription status of the Observable to determine whether to halt work on the Future (⁇)(rxjava-async)
  • startFuture( ) — convert a function that returns Future into an Observable that emits that Future's return value (rxjava-async)
  • startWith( ) — emit a specified sequence of items before beginning to emit the items from the Observable
  • stringConcat( ) (StringObservable) — converts an Observable that emits a sequence of strings into an Observable that emits a single string that concatenates them all
  • subscribeOn( ) — specify which Scheduler an Observable should use when its subscription is invoked
  • sumDouble( ) — adds the Doubles emitted by an Observable and emits this sum (rxjava-math)
  • sumFloat( ) — adds the Floats emitted by an Observable and emits this sum (rxjava-math)
  • sumInt( ) — adds the Integers emitted by an Observable and emits this sum (rxjava-math)
  • sumLong( ) — adds the Longs emitted by an Observable and emits this sum (rxjava-math)
  • switch( ) (scala) — see switchOnNext( )
  • switchCase( ) — emit the sequence from a particular Observable based on the results of an evaluation (contrib-computation-expressions)
  • switchMap( ) — transform the items emitted by an Observable into Observables, and mirror those items emitted by the most-recently transformed Observable
  • switchOnNext( ) — convert an Observable that emits Observables into a single Observable that emits the items emitted by the most-recently emitted of those Observables
  • synchronize( ) — see serialize( )
  • take( ) — emit only the first n items emitted by an Observable
  • takeFirst( ) — emit only the first item emitted by an Observable, or the first item that meets some condition
  • takeLast( ) — only emit the last n items emitted by an Observable
  • takeLastBuffer( ) — emit the last n items emitted by an Observable, as a single list item
  • takeRight( ) (scala) — see last( ) (Observable) or takeLast( )
  • takeUntil( ) — emits the items from the source Observable until a second Observable emits an item
  • takeWhile( ) — emit items emitted by an Observable as long as a specified condition is true, then skip the remainder
  • take-while( ) (clojure) — see takeWhile( )
  • then( ) — transform a series of Pattern objects via a Plan template (rxjava-joins)
  • throttleFirst( ) — emit the first items emitted by an Observable within periodic time intervals
  • throttleLast( ) — emit the most recent items emitted by an Observable within periodic time intervals
  • throttleWithTimeout( ) — only emit an item from the source Observable after a particular timespan has passed without the Observable emitting any other items
  • throw( ) (clojure) — see error( )
  • timeInterval( ) — emit the time lapsed between consecutive emissions of a source Observable
  • timeout( ) — emit items from a source Observable, but issue an exception if no item is emitted in a specified timespan
  • timer( ) — create an Observable that emits a single item after a given delay
  • timestamp( ) — attach a timestamp to every item emitted by an Observable
  • toAsync( ) — convert a function or Action into an Observable that executes the function and emits its return value (rxjava-async)
  • toBlocking( ) — transform an Observable into a BlockingObservable
  • toBlockingObservable( ) - see toBlocking( )
  • toFuture( ) — convert the Observable into a Future
  • toIterable( ) — convert the sequence emitted by the Observable into an Iterable
  • toIterator( ) — see getIterator( )
  • toList( ) — collect all items from an Observable and emit them as a single List
  • toMap( ) — convert the sequence of items emitted by an Observable into a map keyed by a specified key function
  • toMultimap( ) — convert the sequence of items emitted by an Observable into an ArrayList that is also a map keyed by a specified key function
  • toSeq( ) (scala) — see toList( )
  • toSortedList( ) — collect all items from an Observable and emit them as a single, sorted List
  • tumbling( ) (scala) — see window( )
  • tumblingBuffer( ) (scala) — see buffer( )
  • using( ) — create a disposable resource that has the same lifespan as an Observable
  • when( ) — convert a series of Plan objects into an Observable (rxjava-joins)
  • where( ) — see: filter( )
  • whileDo( ) — if a condition is true, emit the source Observable's sequence and then repeat the sequence as long as the condition remains true (contrib-computation-expressions)
  • window( ) — periodically subdivide items from an Observable into Observable windows and emit these windows rather than emitting the items one at a time
  • zip( ) — combine sets of items emitted by two or more Observables together via a specified function and emit items based on the results of this function
  • zipWith( ) — instance version of zip( )
  • zipWithIndex( ) (scala) — see zip( )
  • ++ (scala) — see concat( )
  • +: (scala) — see startWith( )

(⁇) — this proposed operator is not part of RxJava 1.0

File: docs/Async-Operators.md

The following operators are part of the distinct rxjava-async module. They are used to convert synchronous methods into Observables.

  • start( ) — create an Observable that emits the return value of a function
  • toAsync( ) or asyncAction( ) or asyncFunc( ) — convert a function or Action into an Observable that executes the function and emits its return value
  • startFuture( ) — convert a function that returns Future into an Observable that emits that Future's return value
  • deferFuture( ) — convert a Future that returns an Observable into an Observable, but do not attempt to get the Observable that the Future returns until a Subscriber subscribes
  • forEachFuture( ) — pass Subscriber methods to an Observable but also have it behave like a Future that blocks until it completes
  • fromAction( ) — convert an Action into an Observable that invokes the action and emits its result when a Subscriber subscribes
  • fromCallable( ) — convert a Callable into an Observable that invokes the callable and emits its result or exception when a Subscriber subscribes
  • fromRunnable( ) — convert a Runnable into an Observable that invokes the runable and emits its result when a Subscriber subscribes
  • runAsync( ) — returns a StoppableObservable that emits multiple actions as generated by a specified Action on a Scheduler

File: docs/Backpressure.md

Introduction

In RxJava it is not difficult to get into a situation in which an Observable is emitting items more rapidly than an operator or subscriber can consume them. This presents the problem of what to do with such a growing backlog of unconsumed items.

For example, imagine using the zip operator to zip together two infinite Observables, one of which emits items twice as frequently as the other. A naive implementation of the zip operator would have to maintain an ever-expanding buffer of items emitted by the faster Observable to eventually combine with items emitted by the slower one. This could cause RxJava to seize an unwieldy amount of system resources.

There are a variety of strategies with which you can exercise flow control and backpressure in RxJava in order to alleviate the problems caused when a quickly-producing Observable meets a slow-consuming observer. This page explains some of these strategies, and also shows you how you can design your own Observables and Observable operators to respect requests for flow control.

Hot and cold Observables, and multicasted Observables

A cold Observable emits a particular sequence of items, but can begin emitting this sequence when its Observer finds it to be convenient, and at whatever rate the Observer desires, without disrupting the integrity of the sequence. For example if you convert a static Iterable into an Observable, that Observable will emit the same sequence of items no matter when it is later subscribed to or how frequently those items are observed. Examples of items emitted by a cold Observable might include the results of a database query, file retrieval, or web request.

A hot Observable begins generating items to emit immediately when it is created. Subscribers typically begin observing the sequence of items emitted by a hot Observable from somewhere in the middle of the sequence, beginning with the first item emitted by the Observable subsequent to the establishment of the subscription. Such an Observable emits items at its own pace, and it is up to its observers to keep up. Examples of items emitted by a hot Observable might include mouse & keyboard events, system events, or stock prices.

When a cold Observable is multicast (when it is converted into a ConnectableObservable and its connect() method is called), it effectively becomes hot and for the purposes of backpressure and flow-control it should be treated as a hot Observable.

Cold Observables are ideal for the reactive pull model of backpressure described below. Hot Observables typically do not cope well with a reactive pull model, and are better candidates for some of the other flow control strategies discussed on this page, such as the use of the onBackpressureBuffer or onBackpressureDrop operators, throttling, buffers, or windows.

Useful operators that avoid the need for backpressure

Your first line of defense against the problems of over-producing Observables is to use some of the ordinary set of Observable operators to reduce the number of emitted items to a more manageable number. The examples in this section will show how you might use such operators to handle a bursty Observable like the one illustrated in the following marble diagram:

​

By fine-tuning the parameters to these operators you can ensure that a slow-consuming observer is not overwhelmed by a fast-producing Observable.

Throttling

Operators like sample( ) or throttleLast( ), throttleFirst( ), and throttleWithTimeout( ) or debounce( ) allow you to regulate the rate at which an Observable emits items.

The following diagrams show how you could use each of these operators on the bursty Observable shown above.

sample (or throttleLast)

The sample operator periodically "dips" into the sequence and emits only the most recently emitted item during each dip:

​

java
Observable burstySampled = bursty.sample(500, TimeUnit.MILLISECONDS);

throttleFirst

The throttleFirst operator is similar, but emits not the most recently emitted item, but the first item that was emitted after the previous "dip":

​

java
Observable burstyThrottled = bursty.throttleFirst(500, TimeUnit.MILLISECONDS);

debounce (or throttleWithTimeout)

The debounce operator emits only those items from the source Observable that are not followed by another item within a specified duration:

​

java
Observable burstyDebounced = bursty.debounce(10, TimeUnit.MILLISECONDS);

Buffers and windows

You can also use an operator like buffer( ) or window( ) to collect items from the over-producing Observable and then emit them, less-frequently, as collections (or Observables) of items. The slow consumer can then decide whether to process only one particular item from each collection, to process some combination of those items, or to schedule work to be done on each item in the collection, as appropriate.

The following diagrams show how you could use each of these operators on the bursty Observable shown above.

buffer

You could, for example, close and emit a buffer of items from the bursty Observable periodically, at a regular interval of time:

​

java
Observable<List> burstyBuffered = bursty.buffer(500, TimeUnit.MILLISECONDS);

Or you could get fancy, and collect items in buffers during the bursty periods and emit them at the end of each burst, by using the debounce operator to emit a buffer closing indicator to the buffer operator:

​

java
// we have to multicast the original bursty Observable so we can use it
// both as our source and as the source for our buffer closing selector:
Observable burstyMulticast = bursty.publish().refCount();
// burstyDebounced will be our buffer closing selector:
Observable burstyDebounced = burstMulticast.debounce(10, TimeUnit.MILLISECONDS);
// and this, finally, is the Observable of buffers we're interested in:
Observable<List> burstyBuffered = burstyMulticast.buffer(burstyDebounced);

window

window is similar to buffer. One variant of window allows you to periodically emit Observable windows of items at a regular interval of time:

​

java
Observable<Observable> burstyWindowed = bursty.window(500, TimeUnit.MILLISECONDS);

You could also choose to emit a new window each time you have collected a particular number of items from the source Observable:

​

java
Observable<Observable> burstyWindowed = bursty.window(5);

Callstack blocking as a flow-control alternative to backpressure

Another way of handling an overproductive Observable is to block the callstack (parking the thread that governs the overproductive Observable). This has the disadvantage of going against the “reactive” and non-blocking model of Rx. However this can be a viable option if the problematic Observable is on a thread that can be blocked safely. Currently RxJava does not expose any operators to facilitate this.

If the Observable, all of the operators that operate on it, and the observer that is subscribed to it, are all operating in the same thread, this effectively establishes a form of backpressure by means of callstack blocking. But be aware that many Observable operators do operate in distinct threads by default (the javadocs for those operators will indicate this).

How a subscriber establishes “reactive pull” backpressure

When you subscribe to an Observable with a Subscriber, you can request reactive pull backpressure by calling Subscriber.request(n) in the Subscriber’s onStart() method (where n is the maximum number of items you want the Observable to emit before the next request() call).

Then, after handling this item (or these items) in onNext(), you can call request() again to instruct the Observable to emit another item (or items). Here is an example of a Subscriber that requests one item at a time from someObservable:

java
someObservable.subscribe(new Subscriber<T>() {
    @Override
    public void onStart() {
      request(1);
    }

    @Override
    public void onCompleted() {
      // gracefully handle sequence-complete
    }

    @Override
    public void onError(Throwable e) {
      // gracefully handle error
    }

    @Override
    public void onNext(T n) {
      // do something with the emitted item "n"
      // request another item:
      request(1);
    }
});

You can pass a magic number to request, request(Long.MAX_VALUE), to disable reactive pull backpressure and to ask the Observable to emit items at its own pace. request(0) is a legal call, but has no effect. Passing values less than zero to request will cause an exception to be thrown.

Reactive pull backpressure isn’t magic

Backpressure doesn’t make the problem of an overproducing Observable or an underconsuming Subscriber go away. It just moves the problem up the chain of operators to a point where it can be handled better.

Let’s take a closer look at the problem of the uneven zip.

You have two Observables, A and B, where B is inclined to emit items more frequently than A. When you try to zip these two Observables together, the zip operator combines item n from A and item n from B, but meanwhile B has also emitted items n+1 to n+m. The zip operator has to hold on to these items so it can combine them with items n+1 to n+m from A as they are emitted, but meanwhile m keeps growing and so the size of the buffer needed to hold on to these items keeps increasing.

You could attach a throttling operator to B, but this would mean ignoring some of the items B emits, which might not be appropriate. What you’d really like to do is to signal to B that it needs to slow down and then let B decide how to do this in a way that maintains the integrity of its emissions.

The reactive pull backpressure model lets you do this. It creates a sort of active pull from the Subscriber in contrast to the normal passive push Observable behavior.

The zip operator as implemented in RxJava uses this technique. It maintains a small buffer of items for each source Observable, and it requests no more items from each source Observable than would fill its buffer. Each time zip emits an item, it removes the corresponding items from its buffers and requests exactly one more item from each of its source Observables.

(Many RxJava operators exercise reactive pull backpressure. Some operators do not need to use this variety of backpressure, as they operate in the same thread as the Observable they operate on, and so they exert a form of blocking backpressure simply by not giving the Observable the opportunity to emit another item until they have finished processing the previous one. For other operators, backpressure is inappropriate as they have been explicitly designed to deal with flow control in other ways. The RxJava javadocs for those operators that are methods of the Observable class indicate which ones do not use reactive pull backpressure and why.)

For this to work, though, Observables A and B must respond correctly to the request(). If an Observable has not been written to support reactive pull backpressure (such support is not a requirement for Observables), you can apply one of the following operators to it, each of which forces a simple form of backpressure behavior:

onBackpressureBuffer
maintains a buffer of all emissions from the source Observable and emits them to downstream Subscribers according to the requests they generatean experimental version of this operator (not available in RxJava 1.0) allows you to set the capacity of the buffer; applying this operator will cause the resulting Observable to terminate with an error if this buffer is overrun​
onBackpressureDrop
drops emissions from the source Observable unless there is a pending request from a downstream Subscriber, in which case it will emit enough items to fulfill the request​
onBackpressureBlock (experimental, not in RxJava 1.0)
blocks the thread on which the source Observable is operating until such time as a Subscriber issues a request for items, and then unblocks the thread only so long as there are pending requests

If you do not apply any of these operators to an Observable that does not support backpressure, and if either you as the Subscriber or some operator between you and the Observable attempts to apply reactive pull backpressure, you will encounter a MissingBackpressureException which you will be notified of via your onError() callback.

Further reading

If the standard operators aren't providing the expected behavior, one can write custom operators in RxJava.

See also

File: docs/Backpressure-(2.0).md

Originally contributed to StackOverflow Documentation (going defunct) by @akarnokd, revised for version 2.x.

Introduction

Backpressure is when in an Flowable processing pipeline, some asynchronous stages can't process the values fast enough and need a way to tell the upstream producer to slow down.

The classic case of the need for backpressure is when the producer is a hot source:

java
PublishProcessor source = PublishProcessor.create();

    source
    .observeOn(Schedulers.computation())
    .subscribe(v -> compute(v), Throwable::printStackTrace);

    for (int i = 0; i < 1_000_000; i++) {
        source.onNext(i);
    }

    Thread.sleep(10_000);

In this example, the main thread will produce 1 million items to an end consumer which is processing it on a background thread. It is likely the compute(int) method takes some time but the overhead of the Flowable operator chain may also add to the time it takes to process items. However, the producing thread with the for loop can't know this and keeps onNexting.

Internally, asynchronous operators have buffers to hold such elements until they can be processed. In the classical Rx.NET and early RxJava, these buffers were unbounded, meaning that they would likely hold nearly all 1 million elements from the example. The problem starts when there are, for example, 1 billion elements or the same 1 million sequence appears 1000 times in a program, leading to OutOfMemoryError and generally slowdowns due to excessive GC overhead.

Similar to how error-handling became a first-class citizen and received operators to deal with it (via onErrorXXX operators), backpressure is another property of dataflows that the programmer has to think about and handle (via onBackpressureXXX operators).

Beyond the PublishProcessorabove, there are other operators that don't support backpressure, mostly due to functional reasons. For example, the operator interval emits values periodically, backpressuring it would lead to shifting in the period relative to a wall clock.

In modern RxJava, most asynchronous operators now have a bounded internal buffer, like observeOn above and any attempt to overflow this buffer will terminate the whole sequence with MissingBackpressureException. The documentation of each operator has a description about its backpressure behavior.

However, backpressure is present more subtly in regular cold sequences (which don't and shouldn't yield MissingBackpressureException). If the first example is rewritten:

text
Flowable.range(1, 1_000_000)
.observeOn(Schedulers.computation())
.subscribe(v -> compute(v), Throwable::printStackTrace);

Thread.sleep(10_000);

There is no error and everything runs smoothly with small memory usage. The reason for this is that many source operators can "generate" values on demand and thus the operator observeOn can tell the range generate at most so many values the observeOn buffer can hold at once without overflow.

This negotiation is based on the computer science concept of co-routines (I call you, you call me). The operator range sends a callback, in the form of an implementation of the java.util.concurrent.Flow.Subscription interface, to the observeOn by calling its (inner Subscriber's) onSubscribe. In return, the observeOn calls Subscription.request(n) with a value to tell the range it is allowed to produce (i.e., onNext it) that many additional elements. It is then the observeOn's responsibility to call the request method in the right time and with the right value to keep the data flowing but not overflowing.

Expressing backpressure in end-consumers is rarely necessary (because they are synchronous in respect to their immediate upstream and backpressure naturally happens due to call-stack blocking), but it may be easier to understand the workings of it:

java
Flowable.range(1, 1_000_000)
    .subscribe(new DisposableSubscriber() {
        @Override
        public void onStart() {
            request(1);
        }

        public void onNext(Integer v) {
            compute(v);

            request(1);
        }

        @Override
        public void onError(Throwable ex) {
            ex.printStackTrace();
        }

        @Override
        public void onComplete() {
            System.out.println("Done!");
        }
    });

Here the onStart implementation indicates range to produce its first value, which is then received in onNext. Once the compute(int) finishes, the another value is then requested from range. In a naive implementation of range, such call would recursively call onNext, leading to StackOverflowError which is of course undesirable.

To prevent this, operators use so-called trampolining logic that prevents such reentrant calls. In range's terms, it will remember that there was a request(1) call while it called onNext() and once onNext() returns, it will make another round and call onNext() with the next integer value. Therefore, if the two are swapped, the example still works the same:

java
@Override
    public void onNext(Integer v) {
        request(1);

        compute(v);
    }

However, this is not true for onStart. Although the Flowable infrastructure guarantees it will be called at most once on each Subscriber, the call to request(1) may trigger the emission of an element right away. If one has initialization logic after the call to request(1) which is needed by onNext, you may end up with exceptions:

java
Flowable.range(1, 1_000_000)
    .subscribe(new DisposableSubscriber() {

        String name;

        @Override
        public void onStart() {
            request(1);

            name = "RangeExample";
        }

        @Override
        public void onNext(Integer v) {
            compute(name.length + v);

            request(1);
        }

        // ... rest is the same
    });

In this synchronous case, a NullPointerException will be thrown immediately while still executing onStart. A more subtle bug happens if the call to request(1) triggers an asynchronous call to onNext on some other thread and reading name in onNext races writing it in onStart post request.

Therefore, one should do all field initialization in onStart or even before that and call request() last. Implementations of request() in operators ensure proper happens-before relation (or in other terms, memory release or full fence) when necessary.

The onBackpressureXXX operators

Most developers encounter backpressure when their application fails with MissingBackpressureException and the exception usually points to the observeOn operator. The actual cause is usually the non-backpressured use of PublishProcessor, timer() or interval() or custom operators created via create().

There are several ways of dealing with such situations.

Increasing the buffer sizes

Sometimes such overflows happen due to bursty sources. Suddenly, the user taps the screen too quickly and observeOn's default 16-element internal buffer on Android overflows.

Most backpressure-sensitive operators in the recent versions of RxJava now allow programmers to specify the size of their internal buffers. The relevant parameters are usually called bufferSize, prefetch or capacityHint. Given the overflowing example in the introduction, we can just increase the buffer size of observeOn to have enough room for all values.

java
PublishProcessor source = PublishProcessor.create();

    source.observeOn(Schedulers.computation(), 1024 * 1024)
          .subscribe(e -> { }, Throwable::printStackTrace);

    for (int i = 0; i < 1_000_000; i++) {
        source.onNext(i);
    }

Note however that generally, this may be only a temporary fix as the overflow can still happen if the source overproduces the predicted buffer size. In this case, one can use one of the following operators.

Batching/skipping values with standard operators

In case the source data can be processed more efficiently in batch, one can reduce the likelihood of MissingBackpressureException by using one of the standard batching operators (by size and/or by time).

text
PublishProcessor source = PublishProcessor.create();

    source
          .buffer(1024)
          .observeOn(Schedulers.computation(), 1024)
          .subscribe(list -> { 
              list.parallelStream().map(e -> e * e).first();
          }, Throwable::printStackTrace);

    for (int i = 0; i < 1_000_000; i++) {
        source.onNext(i);
    }

If some of the values can be safely ignored, one can use the sampling (with time or another Flowable) and throttling operators (throttleFirst, throttleLast, throttleWithTimeout).

java
PublishProcessor source = PublishProcessor.create();

    source
          .sample(1, TimeUnit.MILLISECONDS)
          .observeOn(Schedulers.computation(), 1024)
          .subscribe(v -> compute(v), Throwable::printStackTrace);

    for (int i = 0; i < 1_000_000; i++) {
        source.onNext(i);
    }

Note however that these operators only reduce the rate of value reception by the downstream and thus they may still lead to MissingBackpressureException.

onBackpressureBuffer()

This operator in its parameterless form reintroduces an unbounded buffer between the upstream source and the downstream operator. Being unbounded means as long as the JVM doesn't run out of memory, it can handle almost any amount coming from a bursty source.

java
Flowable.range(1, 1_000_000)
               .onBackpressureBuffer()
               .observeOn(Schedulers.computation(), 8)
               .subscribe(e -> { }, Throwable::printStackTrace);

In this example, the observeOn goes with a very low buffer size yet there is no MissingBackpressureException as onBackpressureBuffer soaks up all the 1 million values and hands over small batches of it to observeOn.

Note however that onBackpressureBuffer consumes its source in an unbounded manner, that is, without applying any backpressure to it. This has the consequence that even a backpressure-supporting source such as range will be completely realized.

There are 4 additional overloads of onBackpressureBuffer

onBackpressureBuffer(int capacity)

This is a bounded version that signals BufferOverflowErrorin case its buffer reaches the given capacity.

java
Flowable.range(1, 1_000_000)
              .onBackpressureBuffer(16)
              .observeOn(Schedulers.computation())
              .subscribe(e -> { }, Throwable::printStackTrace);

The relevance of this operator is decreasing as more and more operators now allow setting their buffer sizes. For the rest, this gives an opportunity to "extend their internal buffer" by having a larger number with onBackpressureBuffer than their default.

onBackpressureBuffer(int capacity, Action onOverflow)

This overload calls a (shared) action in case an overflow happens. Its usefulness is rather limited as there is no other information provided about the overflow than the current call stack.

onBackpressureBuffer(int capacity, Action onOverflow, BackpressureOverflowStrategy strategy)

This overload is actually more useful as it lets one define what to do in case the capacity has been reached. The BackpressureOverflow.Strategy is an interface actually but the class BackpressureOverflow offers 4 static fields with implementations of it representing typical actions:

  • ON_OVERFLOW_ERROR: this is the default behavior of the previous two overloads, signalling a BufferOverflowException
  • ON_OVERFLOW_DEFAULT: currently it is the same as ON_OVERFLOW_ERROR
  • ON_OVERFLOW_DROP_LATEST : if an overflow would happen, the current value will be simply ignored and only the old values will be delivered once the downstream requests.
  • ON_OVERFLOW_DROP_OLDEST : drops the oldest element in the buffer and adds the current value to it.
java
Flowable.range(1, 1_000_000)
              .onBackpressureBuffer(16, () -> { },
                  BufferOverflowStrategy.ON_OVERFLOW_DROP_OLDEST)
              .observeOn(Schedulers.computation())
              .subscribe(e -> { }, Throwable::printStackTrace);

Note that the last two strategies cause discontinuity in the stream as they drop out elements. In addition, they won't signal BufferOverflowException.

onBackpressureDrop()

Whenever the downstream is not ready to receive values, this operator will drop that element from the sequence. One can think of it as a 0 capacity onBackpressureBuffer with strategy ON_OVERFLOW_DROP_LATEST.

This operator is useful when one can safely ignore values from a source (such as mouse moves or current GPS location signals) as there will be more up-to-date values later on.

java
component.mouseMoves()
     .onBackpressureDrop()
     .observeOn(Schedulers.computation(), 1)
     .subscribe(event -> compute(event.x, event.y));

It may be useful in conjunction with the source operator interval(). For example, if one wants to perform some periodic background task but each iteration may last longer than the period, it is safe to drop the excess interval notification as there will be more later on:

java
Flowable.interval(1, TimeUnit.MINUTES)
     .onBackpressureDrop()
     .observeOn(Schedulers.io())
     .doOnNext(e -> networkCall.doStuff())
     .subscribe(v -> { }, Throwable::printStackTrace);

There exist one overload of this operator: onBackpressureDrop(Consumer<? super T> onDrop) where the (shared) action is called with the value being dropped. This variant allows cleaning up the values themselves (e.g., releasing associated resources).

onBackpressureLatest()

The final operator keeps only the latest value and practically overwrites older, undelivered values. One can think of this as a variant of the onBackpressureBuffer with a capacity of 1 and strategy of ON_OVERFLOW_DROP_OLDEST.

Unlike onBackpressureDrop there is always a value available for consumption if the downstream happened to be lagging behind. This can be useful in some telemetry-like situations where the data may come in some bursty pattern but only the very latest is interesting for processing.

For example, if the user clicks a lot on the screen, we'd still want to react to its latest input.

java
component.mouseClicks()
    .onBackpressureLatest()
    .observeOn(Schedulers.computation())
    .subscribe(event -> compute(event.x, event.y), Throwable::printStackTrace);

The use of onBackpressureDrop in this case would lead to a situation where the very last click gets dropped and leaves the user wondering why the business logic wasn't executed.

Creating backpressured datasources

Creating backpressured data sources is the relatively easier task when dealing with backpressure in general because the library already offers static methods on Flowable that handle backpressure for the developer. We can distinguish two kinds of factory methods: cold "generators" that either return and generate elements based on downstream demand and hot "pushers" that usually bridge non-reactive and/or non-backpressurable data sources and layer some backpressure handling on top of them.

just

The most basic backpressure aware source is created via just:

java
Flowable.just(1).subscribe(new DisposableSubscriber() {
        @Override
        public void onStart() {
            request(0);
        }

        @Override
        public void onNext(Integer v) {
            System.out.println(v);
        }
       
        // the rest is omitted for brevity
    }

Since we explicitly don't request in onStart, this will not print anything. just is great when there is a constant value we'd like to jump-start a sequence.

Unfortunately, just is often mistaken for a way to compute something dynamically to be consumed by Subscribers:

java
int counter;

    int computeValue() {
       return ++counter;
    }
    
    Flowable o = Flowable.just(computeValue());

    o.subscribe(System.out:println);
    o.subscribe(System.out:println);

Surprising to some, this prints 1 twice instead of printing 1 and 2 respectively. If the call is rewritten, it becomes obvious why it works so:

java
int temp = computeValue();

    Flowable o = Flowable.just(temp);

The computeValue is called as part of the main routine and not in response to the subscribers subscribing.

fromCallable

What people actually need is the method fromCallable:

java
Flowable o = Flowable.fromCallable(() -> computeValue());

Here the computeValue is executed only when a subscriber subscribes and for each of them, printing the expected 1 and 2. Naturally, fromCallable also properly supports backpressure and won't emit the computed value unless requested. Note however that the computation does happen anyway. In case the computation itself should be delayed until the downstream actually requests, we can use just with map:

java
Flowable.just("This doesn't matter").map(ignored -> computeValue())...

just won't emit its constant value until requested when it is mapped to the result of the computeValue, still called for each subscriber individually.

fromArray

If the data is already available as an array of objects, a list of objects or any Iterable source, the respective from overloads will handle the backpressure and emission of such sources:

java
Flowable.fromArray(1, 2, 3, 4, 5).subscribe(System.out::println);

For convenience (and avoiding warnings about generic array creation) there are 2 to 10 argument overloads to just that internally delegate to from.

The fromIterable also gives an interesting opportunity. Many value generation can be expressed in a form of a state-machine. Each requested element triggers a state transition and computation of the returned value.

Writing such state machines as Iterables is somewhat complicated (but still easier than writing an Flowable for consuming it) and unlike C#, Java doesn't have any support from the compiler to build such state machines by simply writing classically looking code (with yield return and yield break). Some libraries offer some help, such as Google Guava's AbstractIterable and IxJava's Ix.generate() and Ix.forloop(). These are by themselves worthy of a full series so let's see some very basic Iterable source that repeats some constant value indefinitely:

java
Iterable iterable = () -> new Iterator() {
        @Override
        public boolean hasNext() {
            return true;
        }

        @Override
        public Integer next() {
            return 1;
        }
    };

    Flowable.fromIterable(iterable).take(5).subscribe(System.out::println);

If we'd consume the iterator via classic for-loop, that would result in an infinite loop. Since we build an Flowable out of it, we can express our will to consume only the first 5 of it and then stop requesting anything. This is the true power of lazily evaluating and computing inside Flowables.

generate()

Sometimes, the data source to be converted into the reactive world itself is synchronous (blocking) and pull-like, that is, we have to call some get or read method to get the next piece of data. One could, of course, turn that into an Iterable but when such sources are associated with resources, we may leak those resources if the downstream unsubscribes the sequence before it would end.

To handle such cases, RxJava has the generate factory method family.

java
Flowable o = Flowable.generate(
         () -> new FileInputStream("data.bin"),
         (inputstream, output) -> {
             try {
                 int abyte = inputstream.read();
                 if (abyte < 0) {
                     output.onComplete();
                 } else {
                     output.onNext(abyte);
                 }
             } catch (IOException ex) {
                 output.onError(ex);
             }
             return inputstream;
         },
         inputstream -> {
             try {
                 inputstream.close();
             } catch (IOException ex) {
                 RxJavaPlugins.onError(ex);
             }
         } 
    );

Generally, generate uses 3 callbacks.

The first callbacks allows one to create a per-subscriber state, such as the FileInputStream in the example; the file will be opened independently to each individual subscriber.

The second callback takes this state object and provides an output Observer whose onXXX methods can be called to emit values. This callback is executed as many times as the downstream requested. At each invocation, it has to call onNext at most once optionally followed by either onError or onComplete. In the example we call onComplete() if the read byte is negative, indicating and end of file, and call onError in case the read throws an IOException.

The final callback gets invoked when the downstream unsubscribes (closing the inputstream) or when the previous callback called the terminal methods; it allows freeing up resources. Since not all sources need all these features, the static methods of Flowable.generate lets one create instances without them.

Unfortunately, many method calls across the JVM and other libraries throw checked exceptions and need to be wrapped into try-catches as the functional interfaces used by this class don't allow throwing checked exceptions.

Of course, we can imitate other typical sources, such as an unbounded range with it:

java
Flowable.generate(
         () -> 0,
         (current, output) -> {
             output.onNext(current);
             return current + 1;
         },
         e -> { }
    );

In this setup, the current starts out with 0 and next time the lambda is invoked, the parameter current now holds 1.

(Remark: the 1.x classes SyncOnSubscribe and AsyncOnSubscribe are no longer available.)

create(emitter)

Sometimes, the source to be wrapped into an Flowable is already hot (such as mouse moves) or cold but not backpressurable in its API (such as an asynchronous network callback).

To handle such cases, a recent version of RxJava introduced the create(emitter) factory method. It takes two parameters:

  • a callback that will be called with an instance of the Emitter<T> interface for each incoming subscriber,
  • a BackpressureStrategy enumeration that mandates the developer to specify the backpressure behavior to be applied. It has the usual modes, similar to onBackpressureXXX in addition to signalling a MissingBackpressureException or simply ignoring such overflow inside it altogether.

Note that it currently doesn't support additional parameters to those backpressure modes. If one needs those customization, using NONE as the backpressure mode and applying the relevant onBackpressureXXX on the resulting Flowable is the way to go.

The first typical case for its use when one wants to interact with a push-based source, such as GUI events. Those APIs feature some form of addListener/removeListener calls that one can utilize:

java
Flowable.create(emitter -> {
        ActionListener al = e -> {
            emitter.onNext(e);
        };

        button.addActionListener(al);

        emitter.setCancellation(() -> 
            button.removeListener(al));

    }, BackpressureStrategy.BUFFER);

The Emitter is relatively straightforward to use; one can call onNext, onError and onComplete on it and the operator handles backpressure and unsubscription management on its own. In addition, if the wrapped API supports cancellation (such as the listener removal in the example), one can use the setCancellation (or setSubscription for Subscription-like resources) to register a cancellation callback that gets invoked when the downstream unsubscribes or the onError/onComplete is called on the provided Emitterinstance.

These methods allow only a single resource to be associated with the emitter at a time and setting a new one unsubscribes the old one automatically. If one has to handle multiple resources, create a CompositeSubscription, associate it with the emitter and then add further resources to the CompositeSubscription itself:

java
Flowable.create(emitter -> {
        CompositeSubscription cs = new CompositeSubscription();

        Worker worker = Schedulers.computation().createWorker();

        ActionListener al = e -> {
            emitter.onNext(e);
        };

        button.addActionListener(al);

        cs.add(worker);
        cs.add(Subscriptions.create(() -> 
            button.removeActionListener(al));

        emitter.setSubscription(cs);

    }, BackpressureMode.BUFFER);

The second scenario usually involves some asynchronous, callback-based API that has to be converted into an Flowable.

java
Flowable.create(emitter -> {
        
        someAPI.remoteCall(new Callback<Data>() {
            @Override
            public void onSuccess(Data data) {
                emitter.onNext(data);
                emitter.onComplete();
            }

            @Override
            public void onFailure(Exception error) {
                emitter.onError(error);
            }
        });

    }, BackpressureMode.LATEST);

In this case, the delegation works the same way. Unfortunately, usually, these classical callback-style APIs don't support cancellation, but if they do, one can set up their cancellation just like in the previoius examples (with perhaps a more involved way though). Note the use of the LATEST backpressure mode; if we know there will be only a single value, we don't need the BUFFER strategy as it allocates a default 128 element long buffer (that grows as necessary) that is never going to be fully utilized.

File: docs/Blocking-Observable-Operators.md

This section explains the BlockingObservable subclass. A Blocking Observable extends the ordinary Observable class by providing a set of operators on the items emitted by the Observable that block.

To transform an Observable into a BlockingObservable, use the Observable.toBlocking( ) method or the BlockingObservable.from( ) method.

  • forEach( ) — invoke a function on each item emitted by the Observable; block until the Observable completes
  • first( ) — block until the Observable emits an item, then return the first item emitted by the Observable
  • firstOrDefault( ) — block until the Observable emits an item or completes, then return the first item emitted by the Observable or a default item if the Observable did not emit an item
  • last( ) — block until the Observable completes, then return the last item emitted by the Observable
  • lastOrDefault( ) — block until the Observable completes, then return the last item emitted by the Observable or a default item if there is no last item
  • mostRecent( ) — returns an iterable that always returns the item most recently emitted by the Observable
  • next( ) — returns an iterable that blocks until the Observable emits another item, then returns that item
  • latest( ) — returns an iterable that blocks until or unless the Observable emits an item that has not been returned by the iterable, then returns that item
  • single( ) — if the Observable completes after emitting a single item, return that item, otherwise throw an exception
  • singleOrDefault( ) — if the Observable completes after emitting a single item, return that item, otherwise return a default item
  • toFuture( ) — convert the Observable into a Future
  • toIterable( ) — convert the sequence emitted by the Observable into an Iterable
  • getIterator( ) — convert the sequence emitted by the Observable into an Iterator

This documentation accompanies its explanations with a modified form of "marble diagrams." Here is how these marble diagrams represent Blocking Observables:

see also:

Appendix: similar blocking and non-blocking operators

operator result when it acts on equivalent in Rx.NET
Observable that emits multiple items Observable that emits one item Observable that emits no items
Observable.first the first item the single item NoSuchElement firstAsync
BlockingObservable.first the first item the single item NoSuchElement first
Observable.firstOrDefault the first item the single item the default item firstOrDefaultAsync
BlockingObservable.firstOrDefault the first item the single item the default item firstOrDefault
Observable.last the last item the single item NoSuchElement lastAsync
BlockingObservable.last the last item the single item NoSuchElement last
Observable.lastOrDefault the last item the single item the default item lastOrDefaultAsync
BlockingObservable.lastOrDefault the last item the single item the default item lastOrDefault
Observable.single Illegal Argument the single item NoSuchElement singleAsync
BlockingObservable.single Illegal Argument the single item NoSuchElement single
Observable.singleOrDefault Illegal Argument the single item the default item singleOrDefaultAsync
BlockingObservable.singleOrDefault Illegal Argument the single item the default item singleOrDefault

2. Official Technical Reference & Guides (ReactiveX/ReactiveX.github.io)

reactivex.github.io

ReactiveX Website - Jekyll + Bootstrap.

If you want to contribute, please make sure to make your changes to the develop branch, from which the master branch is generated.

Use Docker to Build & Run (Optional)

If you have Docker installed, you can use the Dockerfile in this repository to build and run the ReactiveX website. This one-liner will perform the build and run:
docker build -t reactivex.io - < Dockerfile && docker run -p 4000:4000 -it --rm -v $PWD:/app -t reactivex.io

Install tools

Jekyll is the static site generator. Uglifier, LESS, and Rake are build tools necessary for JavaScript and CSS compilation prior to Jekyll.

text
$ gem install jekyll uglifier rake
$ npm install less

Run

Rake's default task will compile styles the run jekyll serve.

text
$ rake

Marble diagrams

The marble diagrams are created in OmniGraffle. The source file is found at the RxJava Wiki (https://github.com/Netflix/RxJava.wiki.git) or can be downloaded here: https://raw.githubusercontent.com/wiki/ReactiveX/RxJava/images/rx-operators.graffle

If you want to make changes to this file, please check it out from the wiki, make changes, and check it back in there.

As of this writing, this OmniGraffle source is used to make two separate copies of the .png marble diagrams: one set hosted at the RxJava wiki which also serves as the source for the images in the javadocs, and one set hosted at this reactivex.github.io site. You have to generate these yourself by using OmniGraffle's "export" feature. If you check in a change that you want to be visible in both places, you'll have to check in the images in both repos.

The OmniGraffle file has grown to be cumbersome, and so there are some false-canvases with names that begin and end with lots of equals signs that are used as separators/headings to make individual diagrams easier to find. When you export the .png images based on this OmniGraffle file, it will also spit out empty .png images for these false-canvases. You'll just have to delete those manually or put them in your .gitignore or something. At least they're easy to spot.

Explore Related Technologies & Topics

View All 9,111+ Repositories →

Discover similar high-velocity repositories, agent skills, and OpenAPI specifications across the ecosystem.

#flow Browse flow catalog → #java Browse java catalog → #reactive-streams Browse reactive-streams catalog → #rxjava Browse rxjava catalog → ⚡ MCP Servers Model Context Protocol → 🧠 Agent Skills Prompt & rule files → 🔌 OpenAPI Specs REST & Swagger schemas →