# RxJava 4 User Guide

RxJava is a Java VM implementation of [Reactive Extensions (ReactiveX)](http://reactivex.io): a library for composing asynchronous and event-based programs using **observable sequences**. It extends the observer pattern to support streams of data and events, and adds a large catalogue of **operators** that let you compose those streams declaratively — while the library handles low-level threading, synchronization, thread-safety, and concurrent data structures for you.

This guide covers the **4.x** line of RxJava, whose classes live under the `io.reactivex.rxjava4` package and whose Maven coordinates are `io.reactivex.rxjava4:rxjava`. RxJava 4 is a native Java 26 implementation built on `java.util.concurrent.Flow`, with first-class support for **virtual threads**.

> [!NOTE]
> RxJava 4 is under active development (current build version `4.0.0-SNAPSHOT`). The classic reactive types (`Flowable`, `Observable`, `Single`, `Maybe`, `Completable`) and their operators are stable and battle-tested across 2.x/3.x. The newer virtual-thread APIs (`Streamable`, `Schedulers.virtual()`, `virtualCreate`) are new in 4.x and still evolving — see [Virtual threads and `Streamable`](#9-virtual-threads-and-streamable-new-in-4x).

---

## 1. Quick Start

### 1.1 Add the dependency

RxJava 4 ships as a single JAR with **no third-party runtime dependencies**.

=== "Gradle (Kotlin/Groovy DSL)"

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

=== "Maven"

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

=== "Ivy"

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

Replace `x`, `y`, `z` with the latest published version. Released binaries are on Maven Central; pre-release snapshots are published separately (see [Installation](#2-installation)).

### 1.2 Hello World

```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);
    }
}
```

```text
Hello world
```

> [!IMPORTANT]
> In RxJava 4 the base classes and interfaces live under `io.reactivex.rxjava4.core`. If you are migrating from RxJava 3, the package prefix changed from `io.reactivex.rxjava3` to `io.reactivex.rxjava4`.

### 1.3 A more realistic example

Run an expensive computation on a background thread and observe the result on another thread:

```java
import io.reactivex.rxjava4.core.Flowable;
import io.reactivex.rxjava4.schedulers.Schedulers;

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

Thread.sleep(2000);             // wait for the async flow to finish
```

```text
Done
```

> [!WARNING]
> RxJava's standard schedulers run on **daemon threads**. Once the Java `main` thread exits, those threads stop and pending background work may never run. In short examples like the one above, the trailing `Thread.sleep(...)` keeps the JVM alive long enough to see the output. In a real application (a server, an Android app, a GUI), the application lifecycle keeps the JVM running, so the sleep is unnecessary.

---

## 2. Installation

### 2.1 Requirements

| Requirement | Detail |
|-------------|--------|
| Java runtime | **Java 26** (RxJava 4 is a native Java 26 build; the build toolchain targets JDK 26) |
| Runtime dependencies | **None** — RxJava 4 is self-contained in a single JAR |
| Module system | JPMS (`module-info`) and OSGi metadata are provided in the JAR |
| Android | Supported subject to your API level and available desugaring |

> [!NOTE]
> RxJava 3 required Java 8+ (or Android API 21+). RxJava 4 raises the baseline to Java 26 to take advantage of virtual threads, the `java.util.concurrent.Flow` types, records, and sealed interfaces. If you cannot run Java 26 yet, stay on RxJava 3 (`io.reactivex.rxjava3:rxjava`), which is supported for at least one year after the 4.x official release.

### 2.2 Released binaries

Released versions are available from Maven Central. Use the dependency snippets in [Quick Start](#11-add-the-dependency).

### 2.3 Snapshots

Pre-release snapshots are published to the Sonatype snapshots repository:

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

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

### 2.4 R8 / ProGuard (Android)

By default RxJava itself needs no ProGuard/R8 rules. However, the Reactive Streams metadata embedded in dependencies can trigger ProGuard warnings about `java.util.concurrent.Flow`. If you see them, add:

```text
-dontwarn java.util.concurrent.Flow*
```

For **R8**, the RxJava JAR bundles `META-INF/proguard/rxjava4.pro` with the same no-warning clause, which is applied automatically.

---

## 3. Core Concepts

RxJava revolves around a small number of ideas. Understanding them makes every operator in the library easier to reason about.

### 3.1 The five base reactive types

RxJava 4 offers five base classes. You pick one based on **how many items** the source can produce and whether you need **backpressure** (flow control).

| Type | Cardinality | Backpressure | Use it for |
|------|-------------|--------------|------------|
| `Flowable<T>` | 0..N items | **Yes** | Large or unbounded streams where the producer can outpace the consumer (file/DB reads, network streams) |
| `Observable<T>` | 0..N items | No | Short sequences, GUI/UI events, where overwhelming the consumer isn't a concern |
| `Single<T>` | exactly 1 item **or** error | n/a | A single async result (the lazy equivalent of a `Future`) |
| `Maybe<T>` | 0 or 1 item **or** error | n/a | An optional async result (may complete empty) |
| `Completable` | 0 items — completion **or** error | n/a | Async work with no result, only success/failure (e.g. "write to DB") |

All five live in `io.reactivex.rxjava4.core`:

```java
import io.reactivex.rxjava4.core.Flowable;
import io.reactivex.rxjava4.core.Observable;
import io.reactivex.rxjava4.core.Single;
import io.reactivex.rxjava4.core.Maybe;
import io.reactivex.rxjava4.core.Completable;
```

> [!TIP]
> When in doubt between `Flowable` and `Observable`: use `Flowable` if the source could emit a large or unknown number of items, or if any step in the pipeline is asynchronous and could fall behind. Use `Observable` for bounded, in-memory, or UI-event streams. `Single`, `Maybe`, and `Completable` never need backpressure — there is always room to hold at most one item.

### 3.2 Upstream and downstream

A flow is a **source**, zero or more **intermediate operators**, and a **consumer**:

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

Standing on any operator: everything toward the **source** (left/up) is **upstream**; everything toward the **consumer** (right/down) is **downstream**.

### 3.3 Objects in motion

In RxJava documentation, the words **emission**, **emits**, **item**, **event**, **signal**, **data**, and **message** are synonyms — they all refer to the object travelling along the dataflow.

### 3.4 Assembly time vs. subscription time vs. runtime

This distinction explains a large class of bugs (see [the deferred-evaluation pitfall](#52-defer-evaluation-until-runtime)).

- **Assembly time** — when you chain operators together. No data flows yet; no side effects happen.

  ```java
  Flowable<Integer> flow = Flowable.range(1, 5)
      .map(v -> v * v)
      .filter(v -> v % 3 == 0);
  // nothing has run yet
  ```

- **Subscription time** — when you call `subscribe(...)`. The processing chain is wired up internally; `doOnSubscribe` side effects fire; some sources begin emitting.

- **Runtime** — when the flow is actively emitting items, errors, or a completion signal to the consumer.

### 3.5 Hot vs. cold sources

| Term | Meaning |
|------|---------|
| **Cold** | Has side effects when subscribed. Each subscription starts fresh (e.g. an HTTP call runs again per subscriber). |
| **Hot** | Does not cause side effects on subscription — it is already producing (e.g. mouse events, a `Subject`). Late subscribers may miss earlier emissions. |

### 3.6 Immutability and the fluent API

Operator chaining looks like the builder pattern, but RxJava's reactive types are **immutable**: each operator call returns a **new** instance with the added behavior. The original is unchanged.

```java
Flowable<String> source       = Flowable.fromCallable(() -> compute());
Flowable<String> runBackground = source.subscribeOn(Schedulers.computation());
Flowable<String> showResult    = runBackground.observeOn(Schedulers.single());
showResult.subscribe(System.out::println);
```

---

## 4. Creating Flows

Every base type exposes a family of **static factory methods**. The names below are the same across `Flowable`, `Observable`, `Single`, `Maybe`, and `Completable` where they make sense.

### 4.1 From known values

| Factory | Emits |
|---------|-------|
| `just(item)` / `just(a, b, ...)` | The given constant item(s), then completes |
| `fromArray(T...)` | Each element of the array |
| `fromIterable(Iterable<T>)` | Each element of the iterable |
| `range(start, count)` | A sequence of `count` integers starting at `start` |
| `empty()` | No items, immediate completion |
| `never()` | No items and never terminates |
| `error(Throwable)` / `error(Supplier)` | Terminates immediately with the given error |

```java
Flowable.just(1, 2, 3).subscribe(System.out::println);
Flowable.range(1, 5).subscribe(System.out::println);
Flowable.fromIterable(List.of("a", "b", "c")).subscribe(System.out::println);
```

### 4.2 From a computation or external call

| Factory | Behavior |
|---------|----------|
| `fromCallable(Callable<T>)` | Runs the callable lazily on each subscription; emits its result |
| `fromSupplier(Supplier<T>)` | Like `fromCallable` but for a non-throwing supplier |
| `fromFuture(Future<T>)` | Bridges a `java.util.concurrent.Future` into the flow |
| `defer(Supplier<Source>)` | Defers building the source until subscription time — a fresh source per subscriber |

```java
Flowable<String> lazy = Flowable.fromCallable(() -> expensiveCall());
// expensiveCall() runs only when someone subscribes — and once per subscription
```

### 4.3 From custom producers

`create(...)` lets you bridge an arbitrary push-based API (callbacks, listeners) into a flow. Because `Flowable` supports backpressure, its emitter requires a `BackpressureStrategy`:

```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);
```

`generate(...)` is a safe, synchronous, **backpressure-aware** generator that produces items one at a time on demand — preferable to `create` when you can produce values lazily.

### 4.4 From time

| Factory | Behavior |
|---------|----------|
| `interval(period, unit)` | Emits an incrementing `Long` periodically |
| `timer(delay, unit)` | Emits a single `0L` after a delay, then completes |

```java
Flowable.interval(1, java.util.concurrent.TimeUnit.SECONDS)
    .take(5)
    .subscribe(System.out::println);
```

### 4.5 From other reactive libraries

`Flowable.fromPublisher(Publisher)` wraps any Reactive Streams / `java.util.concurrent.Flow.Publisher` into a `Flowable`, enabling interop with other reactive libraries.

---

## 5. Transforming and Combining Flows

This is the heart of RxJava. Operators are instance methods that return a new flow.

### 5.1 The everyday operators

| Operator | What it does |
|----------|--------------|
| `map(Function)` | Transform each item 1:1 |
| `filter(Predicate)` | Keep only items matching the predicate |
| `flatMap(Function)` | Map each item to an inner flow and **merge** the results (order not guaranteed) |
| `concatMap(Function)` | Like `flatMap` but runs inner flows **one at a time, in order** |
| `concatMapEager(Function)` | Runs inner flows concurrently but emits results in source order |
| `switchMap(Function)` | Map to an inner flow, cancelling the previous inner flow on each new item |
| `scan(BiFunction)` | Emit running accumulations (each intermediate fold) |
| `reduce(BiFunction)` | Emit a single folded result (returns a `Maybe`/`Single`) |
| `buffer(...)` | Group items into `List`s by count or time |
| `window(...)` | Like `buffer` but emits inner flows instead of lists |
| `groupBy(Function)` | Split one flow into keyed sub-flows |
| `distinct()` / `distinctUntilChanged()` | Drop duplicates / consecutive duplicates |
| `take(n)` / `skip(n)` | Keep the first `n` / drop the first `n` |
| `debounce(...)` / `throttleLatest(...)` | Rate-limit bursty sources |

```java
Flowable.range(1, 10)
    .map(v -> v * v)
    .filter(v -> v % 2 == 0)
    .subscribe(System.out::println);
```

```text
4
16
36
64
100
```

### 5.2 Defer evaluation until runtime

A common pitfall: code captured at **assembly time** runs too early.

```java
AtomicInteger count = new AtomicInteger();

Observable.range(1, 10)
    .doOnNext(ignored -> count.incrementAndGet())
    .ignoreElements()                       // -> Completable
    .andThen(Single.just(count.get()))      // BUG: count.get() read at assembly time
    .subscribe(System.out::println);
```

```text
0
```

`Single.just(count.get())` evaluates `count.get()` immediately when the chain is **assembled**, before any data flows. Defer it to **runtime** with `defer` or `fromCallable`:

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

```text
10
```

> [!CAUTION]
> Any value you compute and pass **into** a factory method (`just`, `error(new ...)`, etc.) is evaluated at assembly time. If the value must reflect state that only exists once the flow runs, wrap it in `defer(...)` or `fromCallable(...)`.

### 5.3 Continuations (chaining dependent calls)

Call one service, then another with the first result:

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

When a later step needs an **earlier** value, nest the mapping to capture it via the lambda:

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

When the next step is **independent** of the previous result, use `Completable.andThen(...)`:

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

### 5.4 Dependent sub-flows

`flatMap` shines when each item fans out into another asynchronous call:

```java
Flowable<Inventory> inventory = warehouse.getInventoryAsync();

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

---

## 6. Concurrency and Schedulers

RxJava operators never touch `Thread` or `ExecutorService` directly. Instead they work through **`Scheduler`s**, which abstract a source of concurrency behind one uniform API.

### 6.1 The standard schedulers

Access them via the `io.reactivex.rxjava4.schedulers.Schedulers` utility class.

| Scheduler | Backed by | Use for |
|-----------|-----------|---------|
| `Schedulers.computation()` | Fixed pool sized to available CPUs | CPU-bound work, event loops, processing callbacks (the default for most async operators) |
| `Schedulers.cached()` | Dynamically growing pool of reusable single-thread executors | Blocking I/O-bound work *(new in 4.x; the replacement for `io()`)* |
| `Schedulers.virtual()` | One `Executors.newVirtualThreadPerTaskExecutor()` per worker | Virtual-blocking and many-IO workloads on virtual threads *(new in 4.x)* |
| `Schedulers.single()` | One shared background thread (FIFO, sequential) | Strongly-sequential work; returning to a single dedicated thread |
| `Schedulers.newThread()` | A brand-new thread per unit of work | Rare; prefer the pooled schedulers |
| `Schedulers.trampoline()` | The current thread, queueing work FIFO | Testing; recursive scheduling without new threads |

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

> [!WARNING]
> `Schedulers.io()` is **deprecated since 4.0.0** and now simply delegates to `Schedulers.cached()`. Migrate I/O-bound work to `Schedulers.cached()` (classic blocking pool) or `Schedulers.virtual()` (virtual threads). Both `cached()` and `newThread()` can create an unbounded number of threads under load — bound your concurrency (e.g. with `flatMap`'s `maxConcurrency`) to avoid `OutOfMemoryError`.

### 6.2 `subscribeOn` vs. `observeOn`

These two operators are the levers you use to move work between threads.

| Operator | Controls | Notes |
|----------|----------|-------|
| `subscribeOn(scheduler)` | The thread where the **source** starts (subscription side effects + emission origin) | Only the **first** `subscribeOn` in a chain takes effect |
| `observeOn(scheduler)` | The thread used for everything **downstream** of the call | Can appear multiple times to hop threads repeatedly |

```java
Flowable.fromCallable(() -> blockingRead())
    .subscribeOn(Schedulers.cached())     // do the blocking read on a cached I/O thread
    .observeOn(Schedulers.computation())  // process results on a computation thread
    .map(this::parse)
    .subscribe(this::render);
```

### 6.3 Wrapping your own executor

```java
ExecutorService exec = Executors.newSingleThreadExecutor();
try {
    Scheduler scheduler = Schedulers.from(exec);
    Flowable.just(1)
        .subscribeOn(scheduler)
        .map(v -> v + 1)
        .observeOn(scheduler)
        .blockingSubscribe(System.out::println);
} finally {
    exec.shutdown();
}
```

Overloads let you opt into task interruption and fairness:

- `Schedulers.from(Executor)` — basic wrapper; tasks are not interruptible.
- `Schedulers.from(Executor, interruptibleWorker)` — disposing a task interrupts it.
- `Schedulers.from(Executor, interruptibleWorker, fair)` — `fair=true` reschedules between tasks so other users of the executor can interleave.
- `Schedulers.fromSupplier(Supplier<Executor>, interruptible, fair)` — each worker gets its own executor.

> [!IMPORTANT]
> The wrapped `Executor` should not throw `RejectedExecutionException` (don't shut it down prematurely or use a bounded-queue executor). If it does, the exception is routed to the global error handler via `RxJavaPlugins.onError(...)`. Cancel/dispose all flows **before** shutting down the underlying executor.

### 6.4 Concurrency within a flow

A flow is sequential by nature, but its stages can run concurrently:

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

Here squaring happens on a computation thread and consumption on the caller thread — but `v -> v * v` still processes items **one at a time**, not in parallel.

### 6.5 True parallel processing

To process items in parallel, run independent flows and merge them. Two idioms:

**`flatMap` fan-out** (results may interleave):

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

**`ParallelFlowable`** via the `parallel()` operator:

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

---

## 7. Backpressure (Flowable only)

**Backpressure** is flow control: a consumer signals how many items it is ready to receive, so a fast producer can't overwhelm it (which would otherwise show up as unbounded buffering, memory growth, or dropped data). Only `Flowable` supports backpressure; `Observable` does not.

### 7.1 Converting an Observable into a Flowable

When you turn an unbounded `Observable` into a backpressured `Flowable`, you must decide what to do with overflow by supplying a `BackpressureStrategy`:

```java
observable.toFlowable(BackpressureStrategy.BUFFER);
```

| `BackpressureStrategy` | Behavior when downstream can't keep up |
|------------------------|----------------------------------------|
| `MISSING` | Emit without buffering/dropping; downstream must handle overflow (often paired with an `onBackpressureXXX` operator) |
| `ERROR` | Signal `MissingBackpressureException` |
| `BUFFER` | Buffer **all** items until the downstream consumes them |
| `DROP` | Drop the most recent item that can't be delivered |
| `LATEST` | Keep only the latest item, overwriting earlier undelivered ones |

### 7.2 The `onBackpressureXXX` operators

For finer control on an existing `Flowable`:

- `onBackpressureBuffer(...)` — buffer overflow (many overloads, including bounded capacity + a `BackpressureOverflowStrategy`).
- `onBackpressureDrop(...)` — drop items the downstream isn't ready for.
- `onBackpressureLatest()` — keep only the latest undelivered item.

When using a **bounded** buffer, choose what happens when the buffer fills via `BackpressureOverflowStrategy`:

| `BackpressureOverflowStrategy` | Behavior |
|--------------------------------|----------|
| `ERROR` | Signal `MissingBackpressureException` and terminate |
| `DROP_OLDEST` | Discard the oldest buffered value |
| `DROP_LATEST` | Discard the newest value |

> [!TIP]
> The default internal buffer size is **128** items. You can override it globally before the `Flowable` class loads via the system property `rx4.buffer-size`, or read it with `Flowable.bufferSize()`.

---

## 8. Consuming Flows

### 8.1 `subscribe` overloads

```java
flow.subscribe();                                  // fire-and-forget
flow.subscribe(onNext);                            // values only
flow.subscribe(onNext, onError);                   // values + error
flow.subscribe(onNext, onError, onComplete);       // values + error + completion
```

> [!CAUTION]
> If you subscribe **without** an `onError` handler and the flow errors, RxJava wraps the error in an `OnErrorNotImplementedException` and routes it to the global error handler. Always provide an `onError` handler in production code.

### 8.2 Blocking consumption

For tests, `main` methods, and CLI tools, the `blocking*` family runs the flow on the caller thread until it terminates:

```java
flow.blockingSubscribe(System.out::println);
List<T> all = flow.toList().blockingGet();
T first    = flow.blockingFirst();
```

> [!NOTE]
> Avoid blocking operators inside reactive pipelines or on the `computation()`/`single()` schedulers. If `RxJavaPlugins.setFailOnNonBlockingScheduler(true)` is set, attempting to block on those schedulers throws `IllegalStateException`.

### 8.3 Converting between types

Each base type can convert to the others. The most common conversions:

| From → To | Operator(s) |
|-----------|-------------|
| `Flowable`/`Observable` → `Single` | `first`, `firstOrError`, `single`, `singleOrError`, `last`, `lastOrError` |
| `Flowable`/`Observable` → `Maybe` | `firstElement`, `singleElement`, `lastElement` |
| `Flowable`/`Observable` → `Completable` | `ignoreElements` |
| `Observable` → `Flowable` | `toFlowable(BackpressureStrategy)` |
| `Single`/`Maybe`/`Completable` → `Flowable`/`Observable` | `toFlowable`, `toObservable` |
| `Single` ↔ `Maybe` | `toMaybe`, `toSingle` |

When an operator must accept a *different* reactive type, RxJava uses **type-suffixed** overloads (because Java type erasure forbids identically-named overloads):

| Base operator | Suffixed variants |
|---------------|-------------------|
| `flatMap` | `flatMapSingle`, `flatMapMaybe`, `flatMapCompletable`, `flatMapIterable` |
| `concatMap` | `concatMapSingle`, `concatMapMaybe`, `concatMapCompletable`, `concatMapIterable` |
| `switchMap` | `switchMapSingle`, `switchMapMaybe`, `switchMapCompletable` |

---

## 9. Virtual threads and `Streamable` (new in 4.x)

RxJava 4 introduces a **virtual-thread-native** way of building flows. Where classic operators avoid blocking at all costs, the virtual-thread APIs let your generator code **block freely** — virtual threads make blocking cheap — while still participating in backpressure.

> [!NOTE]
> These APIs are new in 4.x and still being finalized (several carry "TODO proper docs" notes in their source). Treat them as evolving. The classic reactive types remain the recommended foundation for most code today.

### 9.1 `Flowable.virtualCreate`

Generate a backpressured `Flowable` from a callback that is free to block:

```java
Flowable<Integer> flow = Flowable.virtualCreate(emitter -> {
    for (int i = 1; i <= 5; i++) {
        emitter.emit(i);        // blocks until the downstream requests more
    }
    // returning normally signals onComplete
});
```

- Return normally from the generator to signal `onComplete`.
- Throw any exception to signal `onError`.
- By default the generator runs on `Schedulers.virtual()`. Overloads accept a specific `Scheduler` or `ExecutorService`:
  - `virtualCreate(VirtualGenerator)`
  - `virtualCreate(VirtualGenerator, Scheduler)`
  - `virtualCreate(VirtualGenerator, ExecutorService)`

### 9.2 `virtualTransform`

A virtual-thread analogue of `flatMap`/`concatMap` — for each upstream item your callback may emit zero or more downstream items and may block:

```java
flow.virtualTransform((value, emitter, stopper) -> {
    emitter.emit(value);
    emitter.emit(value * 10);
    // call stopper.dispose() to stop the upstream early
});
```

### 9.3 `Streamable` — the async-enumerable of the Java world

`Streamable<T>` (with its realized counterpart `Streamer<T>`) is a pull-style async sequence — conceptually like C#'s `IAsyncEnumerable`/`IAsyncEnumerator`, built to run on virtual threads.

```java
import io.reactivex.rxjava4.core.Streamable;

Streamable<Integer> s = Streamable.create(emitter -> {
    emitter.emit(1);
    emitter.emit(2);
    emitter.emit(3);
});

s.forEach(System.out::println);          // consume on a virtual-thread executor
```

Useful entry points:

| Method | Purpose |
|--------|---------|
| `Streamable.empty()` / `Streamable.just(item)` | Trivial sources |
| `Streamable.create(VirtualGenerator [, Scheduler / ExecutorService])` | Generate values from a (blocking-friendly) callback |
| `Streamable.fromPublisher(Flow.Publisher [, ExecutorService])` | Bridge any Reactive Streams publisher |
| `streamable.transform(VirtualTransformer)` | Per-item virtual transform |
| `streamable.toFlowable([ExecutorService])` | Convert to a backpressured `Flowable` |
| `flowable.toStreamable([ExecutorService])` | Convert a `Flowable` to a `Streamable` |
| `streamable.forEach(consumer)` / `subscribe(Flow.Subscriber)` / `test()` | Consume the sequence |

---

## 10. Error Handling

Errors travel downstream as `onError` signals. RxJava provides operators to recover, retry, or transform them.

| Operator | Behavior |
|----------|----------|
| `onErrorReturn(Function)` / `onErrorReturnItem(item)` | Replace an error with a fallback value, then complete |
| `onErrorResumeNext(Function)` | Switch to a fallback flow when an error occurs |
| `onErrorComplete()` | Swallow the error and complete normally |
| `retry()` / `retry(n)` / `retry(Predicate)` | Re-subscribe on error (optionally limited or conditional) |
| `retryWhen(Function)` | Drive retries from a secondary signal flow (e.g. exponential backoff) |
| `doOnError(Consumer)` | Side effect on error without consuming it |
| `timeout(...)` | Emit a `TimeoutException` (or switch to a fallback) if the source is too slow |

```java
service.apiCall()
    .timeout(5, TimeUnit.SECONDS)
    .retry(3)
    .onErrorReturnItem(Collections.emptyList())
    .subscribe(this::render, this::logError);
```

### 10.1 Delaying errors across multiple sources

When combining several sources, many operators have a `DelayError` variant (or a `delayError`/`delayErrors` boolean) that waits for all sources to finish before propagating any error, instead of failing fast:

```java
Flowable<T> result = Flowable.concatDelayError(sources);
```

Suffixes can stack, e.g. `concatArrayEagerDelayError(...)`.

### 10.2 RxJava's exception types

| Exception | Meaning |
|-----------|---------|
| `MissingBackpressureException` | The downstream wasn't ready and didn't `request(...)` in time. Default message: *"Could not emit value due to lack of requests"* |
| `QueueOverflowException` | Upstream ignored backpressure or called `onNext` concurrently without synchronization. Message: *"Queue overflow due to illegal concurrent onNext calls or a bug in an operator"* |
| `OnErrorNotImplementedException` | You subscribed without an `onError` handler and the flow errored |
| `UndeliverableException` | An error had nowhere to go (the flow was already disposed/terminated) and was routed to the global handler |
| `CompositeException` | A container of multiple errors (e.g. from `mergeDelayError`) |
| `ProtocolViolationException` | An internal Reactive Streams protocol violation |

The `io.reactivex.rxjava4.exceptions.Exceptions` utility offers `Exceptions.propagate(throwable)` (rethrow as unchecked) and `Exceptions.throwIfFatal(throwable)` (rethrow only truly fatal errors such as `VirtualMachineError`/`LinkageError`).

---

## 11. Best Practices

- [ ] **Pick the narrowest base type.** Use `Single`/`Maybe`/`Completable` when the cardinality is known — they're simpler and need no backpressure.
- [ ] **Always supply an `onError` handler** in `subscribe(...)`. Unhandled errors become `OnErrorNotImplementedException`.
- [ ] **Use `Flowable` (not `Observable`) for large or asynchronous streams** so backpressure protects you from `OutOfMemoryError`.
- [ ] **Bound concurrency.** `flatMap` accepts a `maxConcurrency`; the `FlatMapConfig` record bundles it with `delayErrors` and `bufferSize`.
- [ ] **Defer state-dependent values** with `defer(...)`/`fromCallable(...)` so they evaluate at runtime, not assembly time.
- [ ] **Match the scheduler to the work:** `computation()` for CPU work, `cached()` (or `virtual()`) for blocking I/O, `single()` for sequential work. Never block on `computation()`/`single()`.
- [ ] **Dispose subscriptions** tied to a lifecycle (UI screens, request scopes) to release resources — keep the returned `Disposable` and call `dispose()` (or collect them in a `CompositeDisposable`).
- [ ] **Treat `io.reactivex.rxjava4.internal.*` as off-limits** — it is private API and can change at any time.

> [!TIP]
> Operator names follow consistent conventions. A suffix tells you the variant: a target **type** (`flatMapSingle`), error handling (`...DelayError`), or array input (`mergeArray`). Once you learn the conventions, you can usually guess the operator you need.

---

## 12. Troubleshooting

### Error: `Could not emit value due to lack of requests`

**Cause:** A `MissingBackpressureException` — your `Flowable` produced items faster than the downstream requested them, with no buffering strategy in place (or a bounded buffer using `BackpressureStrategy.ERROR`).

**Solution:** Apply a backpressure strategy: `onBackpressureBuffer()`, `onBackpressureLatest()`, or `onBackpressureDrop()`. If converting from `Observable`, pass `BackpressureStrategy.BUFFER`/`LATEST`/`DROP` to `toFlowable(...)`.

**Prevention:** Prefer backpressure-aware sources (`range`, `fromIterable`, `generate`) over `create(...)` with `MISSING`, and use `Flowable` rather than `Observable` for fast or unbounded producers.

---

### Error: `Queue overflow due to illegal concurrent onNext calls or a bug in an operator`

**Cause:** A `QueueOverflowException` — an upstream emitted `onNext` concurrently from multiple threads without serialization, or completely ignored backpressure.

**Solution:** Serialize emissions. If you write to an emitter from multiple threads, use a `Subject`/`Processor`'s `toSerialized()` form, or restructure so a single thread emits.

**Prevention:** Never call `onNext` from multiple threads simultaneously. Follow the Reactive Streams contract: `onSubscribe onNext* (onError | onComplete)?`.

---

### Error: `OnErrorNotImplementedException`

**Cause:** You called a `subscribe(...)` overload without an `onError` handler, and the flow emitted an error.

**Solution:** Add an `onError` consumer: `subscribe(onNext, onError)`.

**Prevention:** Make providing `onError` a code-review rule; consider a global handler via `RxJavaPlugins.setErrorHandler(...)` as a safety net (not a substitute).

---

### Problem: The program exits before any output appears

**Cause:** RxJava's standard schedulers run on **daemon threads**. When `main` returns, the JVM exits and kills them before async work completes.

**Solution:** Keep the JVM alive — `blockingSubscribe(...)`, `blockingGet()`, a `CountDownLatch`, or (in examples) a trailing `Thread.sleep(...)`.

**Prevention:** In long-running apps (servers, Android, GUIs) the application lifecycle keeps the JVM alive; this only bites in short `main` methods and tests.

---

### Problem: A computed value is wrong / shows a stale value (e.g. prints `0`)

**Cause:** The value was read at **assembly time** instead of runtime — e.g. `Single.just(counter.get())` captures the counter when the chain is built.

**Solution:** Wrap it in `Single.defer(() -> Single.just(counter.get()))` or `Single.fromCallable(() -> counter.get())`.

**Prevention:** Remember: arguments passed into factory methods are evaluated immediately. Defer anything that depends on the flow having already run.

---

### Error: `IllegalStateException` when blocking on a scheduler

**Cause:** `RxJavaPlugins.setFailOnNonBlockingScheduler(true)` is enabled and you attempted a blocking operation on `computation()` or `single()`.

**Solution:** Move blocking work to `cached()`/`virtual()` via `subscribeOn(...)`, or restructure to avoid blocking.

**Prevention:** Keep blocking calls off the computation/single schedulers as a matter of course.

---

### Warning: ProGuard `can't find ... java.util.concurrent.Flow$Publisher`

**Cause:** Reactive Streams metadata references `java.util.concurrent.Flow` types that plain ProGuard can't resolve.

**Solution:** Add `-dontwarn java.util.concurrent.Flow*` to your ProGuard rules. R8 picks up the bundled `rxjava4.pro` automatically.

**Prevention:** Use R8 (default on modern Android), which applies the consumer rules shipped in the JAR.

---

## 13. FAQ

**Q: What's the difference between `Flowable` and `Observable`?**
Both model 0..N item streams. `Flowable` supports **backpressure** (the consumer can throttle the producer) and implements Reactive Streams `Publisher`; `Observable` does not. Use `Flowable` for large/unbounded/async streams and `Observable` for short or UI-event streams.

**Q: When should I use `Single`, `Maybe`, or `Completable`?**
`Single` for exactly one result-or-error, `Maybe` for zero-or-one result-or-error, `Completable` for work that only succeeds or fails with no value. They're simpler than `Flowable`/`Observable` and never need backpressure.

**Q: How do I run work in the background and get results on another thread?**
`subscribeOn(scheduler)` chooses where the source runs; `observeOn(scheduler)` chooses where downstream operators run. Typical pattern: `.subscribeOn(Schedulers.cached()).observeOn(Schedulers.computation())`.

**Q: `Schedulers.io()` shows as deprecated — what do I use?**
`io()` is deprecated since 4.0.0 and delegates to `cached()`. Use `Schedulers.cached()` for classic blocking I/O, or `Schedulers.virtual()` to run blocking work on virtual threads.

**Q: What are virtual threads good for here?**
`Schedulers.virtual()` and the `Streamable`/`virtualCreate` APIs let your generator/consumer code **block freely** (e.g. blocking I/O) cheaply, because each task gets a lightweight virtual thread. This is new in 4.x and well-suited to many-IO workloads.

**Q: Why are some operators named with suffixes like `flatMapSingle` or `mergeArray`?**
Java type erasure makes `flatMap(Function<T, Single<R>>)` and `flatMap(Function<T, Maybe<R>>)` indistinguishable, so RxJava appends the target type as a suffix. Array-input variants get an `Array` suffix to avoid varargs ambiguity.

**Q: My lambda passed to `concatWith(...)` doesn't work — why?**
Several `concatWith` overloads accept functional-interface types (`Publisher`, `SingleSource`), so a lambda is ambiguous and won't compile. You almost certainly wanted `defer(...)`: `someSource.concatWith(Single.defer(() -> Single.just(2)))`.

**Q: Is RxJava 4 a drop-in upgrade from RxJava 3?**
No. RxJava 4 changes the package prefix to `io.reactivex.rxjava4`, raises the minimum JVM to Java 26, deprecates `io()`, renames system properties to the `rx4.*` prefix, and adds new virtual-thread APIs. Plan a migration rather than a version bump. RxJava 3 remains supported for at least a year after the 4.x release.

**Q: Can RxJava interoperate with other reactive libraries?**
Yes. `Flowable` implements Reactive Streams `Publisher`, and `Flowable.fromPublisher(...)` / `Streamable.fromPublisher(...)` bridge any `java.util.concurrent.Flow.Publisher` into RxJava.

**Q: What stability guarantees do APIs carry?**
APIs marked `@Beta` may change or be removed; `@Experimental` APIs almost certainly will and shouldn't be used in production; `@Deprecated` APIs remain until the next major release. Anything in `io.reactivex.rxjava4.internal.*` is private and unsupported.

**Q: Where can I find the complete operator reference?**
The Javadoc at `http://reactivex.io/RxJava/4.x/javadoc/` documents every operator and overload on each base class. The project [Wiki](https://github.com/ReactiveX/RxJava/wiki) covers concepts and migration.

---

_Generated June 2026 from ReactiveX/RxJava at 6beb0e3. AI-generated documentation. The library is checked by sampling (three repositories so far); this guide was not individually reviewed. Verify against the project before relying on it. Unofficial; not affiliated with or reviewed by the project._