DOCSEARCHbyAlgolia
Submit Docs

RxJava

repository·4.x·Indexed 13 days ago

https://github.com/reactivex/rxjava

A JVM implementation of ReactiveX (Reactive Extensions) for composing asynchronous and event-based programs using observable sequences. RxJava 4.x features native Java 26 implementation, zero runtime dependencies, virtual thread support via Schedulers.virtual(), and integration with java.util.concurrent.Flow. It provides reactive types including Flowable, Observable, Single, Maybe, Completable, and Streamable to abstract concurrency, threading, and backpressure.

Tokens
78.4K
Snippets
191
Records
379
Agent score
99%

What's inside RxJava

  1. Overview of RxJava 4.x features

    4.x

    RxJava 4.x is a modern implementation with the following characteristics:

    • Native Java 26 implementation: Optimized for modern Java environments.
    • Zero runtime dependencies: No 3rd party libraries are required at runtime.
    • Module support: Maintains JPMS and OSGi support.
    • Reactive Streams integration: Built on java.util.concurrent.Flow and compatible with the Reactive Streams Test Compatibility Kit.
    • Virtual Thread support: Includes specialized methods like virtualCreate(), virtualTransform(), and Schedulers.virtual().
    • Streamable<T>: A new feature (in progress) built around Virtual Threads and virtual blocking, providing a pattern similar to IAsyncEnumerable for Java.
    • Resource Management: Uses the Java Cleaner API to detect resource leaks and perform adaptive cleanups.
    • Android Compatibility: Depends on the specific API level and available desugaring.

  2. Overview of RxJava

    4.x
    RxJava is a Java VM implementation of ReactiveX (Reactive Extensions). It is a library designed for composing asynchronous and event-based programs by using observable sequences. It is implemented as a lightweight, single JAR focused on the Observable abstraction and its related higher-order functions.

  3. Use Async Operators from the rxjava-async module

    4.x
    The rxjava-async module provides a set of operators designed to bridge synchronous code and Future objects into the Reactive paradigm by converting them into Observable instances. These operators allow you to wrap functions, actions, or futures so they can be composed within an RxJava stream.

  4. What is RxJava

    4.x

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

    It extends the observer pattern to support sequences of data/events and provides operators that allow for declarative composition of sequences. This abstracts away low-level concerns such as threading, synchronization, thread-safety, and the management of concurrent data structures.

  5. What is a Subject and how does it work

    4.x

    A Subject acts as a bridge or proxy that functions as both a Subscriber and an Observable.

    • As a Subscriber: It can subscribe to one or more Observable sources.
    • As an Observable: It can reemit items it observes and can also emit new items directly.

    This dual nature allows you to feed data from an existing stream into a Subject and then have multiple observers listen to that Subject.

  6. Handle empty Maybe sources in flatMapSingle

    4.x

    When using Maybe::flatMapSingle, if the source Maybe is empty, the resulting Single will signal an error notification instead of completing. If you prefer a Maybe that completes when the source is empty, use Maybe::flatMapSingleElement instead.

    Maybe<Object> emptySource = Maybe.empty();
Single<Object> result = emptySource.flatMapSingle(x -> Single.just(x));
result.subscribe(
    x -> System.out.println("onSuccess will not be printed!"),
    error -> System.out.println("onError: Source was empty!"));

  7. Use buffer to group elements in Flowable or Observable

    4.x

    The buffer operator is available for Flowable and Observable. It is used to group elements into collections.

    Note: For Maybe and Single, buffer is not available. Instead, use map() to transform the single element into a list or collection.

  8. Transform Observables with Operators

    4.x

    RxJava allows you to chain operators to transform and compose Observables. Common patterns include:

    • Filtering and Slicing: Use skip(n) to bypass the first $n$ items and take(n) to only take the next $n$ items.
    • Transformation: Use map(function) to transform each item emitted by an Observable into a new form.
    • Combining Observables:
      • zip: Combines multiple Observables by pairing their emitted items (e.g., taking the 1st item from each, then the 2nd, etc.) into a single object.
      • merge: Combines multiple Observables into a single stream by interleaving their emissions.
      • reduce: Aggregates all emitted items into a single value using a seed value and an accumulator function.
    • Dependency Management: Use mapMany (or similar flat-mapping operators) to trigger a new asynchronous Observable based on the result of a previous one, allowing for dependent sub-flows.
    // Example: skip, take, and map
customObservableNonBlocking()
    .skip(10)
    .take(5)
    .map({ stringValue -> return stringValue + "_xform"})
    .subscribe({ println "onNext => " + it})

  9. Concepts: Assembly, Subscription, and Runtime

    4.x

    RxJava lifecycles consist of three distinct stages:

    1. Assembly time: The stage where you define the dataflow by applying operators (e.g., .map(), .filter()). No data is flowing and no side effects occur yet.
    2. Subscription time: The stage when .subscribe() is called. This establishes the internal processing chain and triggers subscription side effects (e.g., doOnSubscribe).
    3. Runtime: The stage when the flow is actively emitting items, errors, or completion signals.

  11. How reactive pull backpressure works

    4.x

    Reactive pull backpressure is a mechanism where a Subscriber actively requests items from an Observable, rather than the Observable passively pushing items. This moves the problem of an overproducing Observable or an underconsuming Subscriber up the operator chain to a point where it can be managed.

    For example, the zip operator uses this technique. It maintains a small buffer for each source Observable and only requests enough items to fill that buffer. When zip emits an item, it removes the used items and requests exactly one more item from each source. This prevents the buffer from growing indefinitely when one source emits faster than the other.

  12. Transform an Observable into a BlockingObservable

    4.x

    A BlockingObservable extends the standard Observable class by providing operators that block the current thread until the items are emitted or the Observable completes.

    You can convert a standard Observable into a BlockingObservable using one of two methods:

    1. Use Observable.toBlocking() on an existing Observable instance.
    2. Use BlockingObservable.from(Observable) to wrap an existing Observable.
    // Example conversion
BlockingObservable<String> blocking = myObservable.toBlocking();
// OR
BlockingObservable<String> blocking = BlockingObservable.from(myObservable);