Mastering RxJava: A Comprehensive Guide to Reactive Programming in Java

Many Java developers feel intimidated when they hear the term “RxJava” due to preconceived notions. I won’t claim that RxJava is very simple right from the start; it does have a learning curve. However, trust me, it’s worth every second you invest in learning it. In this article, I will break down Java’s Reactive Extensions (Rx) to make them more accessible and understandable.

There’s a common belief that as developers begin writing code in RxJava, everything shifts to a parallel mode, with numerous threads created, causing different parts of the code to run in separate threads. This can lead to a cognitive disconnect and make it challenging to visualize. This misconception often causes fear when working with RxJava. Most of the Java developers are more familiar with traditional coding methods and may feel a bit uneasy with the functional approach, which is one of the core focuses of RXJava.

First things first, RxJava is a tool for handling asynchronous code. However, by default, it operates synchronously, meaning it doesn’t automatically create separate threads for concurrent execution. To introduce concurrency and make it asynchronous, you must configure RxJava explicitly by using operators like subscribeOn and observeOn to specify when and where concurrency should occur. More on this in the later part of this blog.

Now, you might be wondering, “Why invest time in learning RxJava? Can’t we use traditional Java packages to handle asynchronous execution?” The short answer is yes, you can. However, the real question is, are you willing to deal with the complexities of callback hell, error handling, and thread management? RxJava isn’t a magical solution, but rather a well-crafted Java library designed to address these challenges and more.

Before we start digging into RXJava and its constructs, Let’s first understand the Imperative and functional programming paradigms.

1 Imperative programming :- Imperative programming focuses on describing a sequence of steps or commands for the computer to follow to achieve a specific task. It often relies on statements that change the program’s state or modify data directly. It emphasizes how to achieve a goal by specifying the exact steps to take.

import java.util.Arrays;
import java.util.List;

public class ImperativeExample {
    public static void main(String[] args) {
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        int sum = 0;
        for (int number : numbers) {
            if (number % 2 == 0) {
                sum += number;
            }
        }
        System.out.println("Sum of even numbers: " + sum);
    }
}

2 Functional programming :- Functional programming, on the other hand, focuses on expressing computations as the evaluation of mathematical functions. It emphasizes what should be done rather than how to do it. In functional programming, functions are first-class citizens, which means they can be assigned to variables, passed as arguments and returned as values.

import java.util.Arrays;
import java.util.List;

public class FunctionalExample {
    public static void main(String[] args) {
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        int sum = numbers.stream()
                .filter(number -> number % 2 == 0)
                .mapToInt(Integer::intValue)
                .sum();
        System.out.println("Sum of even numbers: " + sum);
    }
}

Now that you understand the difference between imperative and functional programming, you might wonder about reactive programming. Is it the same as functional programming? The answer is that they are not the same, but reactive programming strongly promotes functional programming. That’s why you often see that many of the constructs in reactive programming are written in a functional way.

In other words, functional programming revolves around independent functions that are immutable and consistently produce the same result for the same input. On the other hand, reactive programming focuses on asynchronous operations, treating the entire code as a sequence of events and listening to and reacting to those events. One is function-driven, and the other is event-driven. In practice, we combine both to make the coding easier and less error-prone.

Is imperative programming obsolete? No, ultimately, whatever we write and execute is translated into a sequence of operations, which is in essence, imperative. So, whether it’s functional or reactive or both, under the hood, everything becomes imperative when it comes to the hardware. While we can have an unlimited stream of objects, we have only limited hardware for processing them. Therefore, all this processing should be sequential and follow a specific order; in other words, it has to be imperative.

RxJava Deep Dive

In this section, let’s delve into RxJava in-depth and explore how it enables us to build a scalable and asynchronous system. I will break down this discussion into the following sections, allowing anyone looking for specific information to skip others and find precisely what you need

• Concurrency and Parallelism :- How RxJava handles multiple tasks concurrently, including parallel execution when applicable.
  
• Reactive Programming Principles :- Exploring observables, observers, and operators at the core of RxJava’s reactive approach.
  
• Specialized Observables :- In RxJava, you have specialized types of Observables called Single, Maybe, and Completable. They’re built for specific tasks, making your code easier to read and manage.
  
• Asynchronous Operations and Schedulers :- Dealing with non-blocking operations, such as network requests and file I/O, using RxJava observables.Understanding and effectively utilizing RxJava’s schedulers to manage thread execution and context.
  
• Flow Control and Backpressure Handling :- Techniques for controlling data flow and preventing overload when working with data streams.
  
• Error Handling :- Performing robust error management, including exception handling and error propagation in RxJava.
  
• RxJava Through the Ages: From 1.x to 3.x :- A chill run-through of how RxJava evolved. Each version refined what was already there and brought its own unique flair to the table.

Let’s review each point step by step. It might be a bit detailed, so if you’re interested in a specific topic, please click on the respective section heading above to navigate directly to it

1. Concurrency and Parallelism

People often use the terms “concurrency” and “parallelism” interchangeably, but there is a subtle distinction between them. Concurrency is handling multiple tasks at the same time but parallelism is executing those tasks at the same time.

For example, Imagine a call center with one customer service representative and he/she manages multiple customer inquiries at the same time. He/she might be on the phone with one customer while responding to emails or chat messages from others. However, he/she can only assist one customer at a time; this is referred to as concurrency. On the other hand, parallelism is like having an additional representative. In this scenario, while one person is handling calls, another can manage emails.

As mentioned earlier, by default, RxJava operates in a blocking manner, meaning that all operations are executed one after the other, and the executing thread can get stuck, especially during IO operations. However, RxJava provides a sophisticated way to handle concurrency and parallelism. It offers operators like subscribeOn and observeOn that allows you to specify the Schedulers on which operations should run. This enables efficient thread management and helps achieve concurrency, making your code non-blocking and responsive.

I would use the word non-blocking a bit carefully here. In the context of RxJava, it means that the calling thread is not blocked; instead, RxJava executes the IO operation on a different thread, allowing the main thread to continue with the next operation in the sequence. So, non-blocking refers to the fact that the calling thread remains unblocked. It’s worth noting that the separate thread handling the IO operation can still be blocked if needed.

While Java did introduce NIO libraries in Java 7 to perform IO operations without blocking any thread, RxJava takes a different approach. It doesn’t use these libraries and, as a result, doesn’t fully leverage the advantage of true non-blocking operations. Instead, RxJava has its own set of tools that make asynchronous programming simpler and more streamlined.

For now, we don’t need to dive deeper into Concurrency and Parallelism; we’ll cover those topics later in this blog. In summary, if you don’t see subscribeOn and observeOn in your RxJava chain, the entire chain runs sequentially. There is one special case to note: if you’re working with a parallel stream in your code, and the source you’re using is also a parallel stream, then the operations will run concurrently and in parallel. The level of parallelism is determined by the number of CPU cores available

2. Reactive Programming Principles

Now, let’s explore the important RxJava building blocks and how to use them effectively. Don’t worry; I’ll break it down step by step, and by the end of this section, you’ll feel comfortable writing RxJava code with confidence and a clear understanding.

In RxJava, we deal with streams of data, where everything is treated as a continuous flow. We pay attention to and respond to each piece of data as it comes along in the stream. It’s like watching a river and reacting to each passing wave.

Fundamentally, the building blocks of RxJava can be classified into four main components: Observable , Observer , Operators and Schedulers . We will talk about the first three in this section and Schedulers we will cover them in the later part of this blog along with concurrency and multithreading.

In essence, an Observable serves as the source of a data or event stream, acting as the producer, while the Observer plays the role of the consumer, receiving and reacting to this data. Operators bridge the gap by transforming the raw data into the desired format required by the Observer.

If we Look at the example above, the car body on a conveyor belt represents the Observable. The conveyor belt continuously emits car bodies as they move along the assembly line. Think of the car body as a stream of events or data that we want to process further.

The tire-fixing machine and the painting machine act as Operators in our RxJava analogy. These machines are responsible for transforming or modifying the car bodies emitted by the conveyor belt Observable.

In this analogy, the container that carries the assembled car serves as the Observer. Without the presence of this container at the end of the assembly line, the conveyor belt comes to a halt, and further emissions cease. This highlights a crucial point in RxJava: the entire operation only takes place if someone or something is there to consume and collect the output.

Okay, enough gibber-gabber. Let’s begin with a real-world example and break it down to understand how it works. Let’s take an example of a burger machine which gives us the raw material to make the burger

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;

public class BurgerMaker {
    public static void main(String[] args) {
        // Creating an Observable to represent the burger-making process
        Observable<String> burgerMachine = Observable.create(emitter -> {
            emitter.onNext("Bun");
            emitter.onNext("Lettuce");
            emitter.onNext("Tomato");
            emitter.onNext("Cheese");
            emitter.onNext("Patty");
            emitter.onNext("Ketchup");
            emitter.onNext("Bun");
            emitter.onComplete(); // Completion event
        });

        // Creating an Observer (Burger Chef)
        Observer<String> burgerChef = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                // runs only once, at the moment of subscription
                System.out.println("Let's make a burger");
            }

            @Override
            public void onNext(String ingredient) {
                // Handle each ingredient as it comes
                System.out.println("Adding " + ingredient);
            }

            @Override
            public void onError(Throwable e) {
                // Handle errors, if any
                System.err.println("Something went wrong: " + e.getMessage());
            }

            @Override
            public void onComplete() {
                // Burger is ready
                System.out.println("Burger is ready to serve!");
            }
        };

        /*Actual subscrioption happens here. without this burger machine itself will
        not start*/
        burgerMachine.subscribe(burgerChef);
    }
}

And the out put would look like below.

Let's make a burger
Adding Bun
Adding Lettuce
Adding Tomato
Adding Cheese
Adding Patty
Adding Ketchup
Adding Bun
Burger is ready to serve!

Process finished with exit code 0

In the above example, you can see how the burger machine emits each event and how the burger Chef consumes it. Just to clarify, the Observer must explicitly consume any event produced by the Observable. If the handling of an event is not specified, it will be missed even though the Observable emits it. Let’s go through each operation briefly.

onSubscribe: This event automatically occurs when an Observer starts observing the Observable. It happens only exactly once per subscription. If the Observer has a callback function written to handle this event, it will execute the corresponding action. In the above example we handle the event and print ‘Let’s make a burger‘. Observable passes Disposable object as a parameter, we can use this to unsubscribe from the stream either within the subscription logic or at a later point in our code by storing it as a global or accessible variable. This gives us the flexibility to manage our subscriptions and resources effectively based on our application’s requirements.

onNext: This method is called by the Observable to emit the next successful element. If the Observer has a corresponding function to handle it, that function gets called. In the above example, we print each ingredient as it comes. This event can be emitted zero or more times. This method accepts an object as a parameter, and the type of that parameter matches the type of the Observable. In the example, the Observable is of type String, so this method accepts a String-type argument.

onError: This method is invoked by the Observable to report errors. If the Observer has a corresponding function to handle it, that function is called. In the example above, we print ‘Something went wrong: ‘ followed by the error message. This event can occur either zero times or at most once, and it also marks the end of the Observable. The method accepts a Throwable as a parameter, allowing us to efficiently handle exceptions and log errors in a meaningful way.

For instance, in the example above, if the Observable had emitted an error like this

Observable<String> burgerMachine = Observable.create(emitter -> {
            emitter.onNext("Bun");
            emitter.onNext("Lettuce");
            emitter.onNext("Tomato");
            emitter.onNext("Cheese");
            emitter.onNext("Patty");
            emitter.onError(new RuntimeException("Machine power failure"));
            emitter.onNext("Ketchup");
            emitter.onNext("Bun");
            emitter.onComplete(); // Completion event
        });

the output would have looked like below.

Let's make a burger
Adding Bun
Adding Lettuce
Adding Tomato
Adding Cheese
Adding Patty
Something went wrong: Machine power failure

Process finished with exit code 0

onComplete: This method is called by the Observable to signal the completion of the current stream. If the Observer has a corresponding function to handle it, and then that function is executed. In the example above, we print ‘Burger is ready to serve!’. This event can happen zero times or at most once, and it also marks the end of the Observable. Typically, this function is used to log the successful completion of a specific operation.

Either the onError or onComplete event happens in an Observable stream, but not both. When an error occurs, onError triggers and ends the Observable. For successful completion, onComplete triggers and also ends the Observable.

Now let’s look at the different kinds of Observables and Observers. This is just to create a mental map of observables and observers. Each of these will be revisited and explained deeply in the later parts of this blog

Different types of Observables

1. Observable: The basic type of Observable that emits a sequence of items. It’s used for synchronous and asynchronous data sources.
2. Flowable: A specialized Observable for handling potentially large streams of data or data sources that can produce data at a high rate. It provides backpressure support to handle situations where the consumer can’t keep up with the producer.
3. Single: Represents a single value or an error. It’s used when you expect only one item to be emitted or an error to occur. Commonly used for network requests or database queries that return a single result.
4. Maybe: Represents either a single item, no item at all, or an error. It’s used when the presence of an item is optional. Commonly used for optional configuration or settings.
5. Completable: Represents a completion or error event without emitting any data. It’s used when you only care about the completion or failure of an operation, such as saving data to a database.
6. Subject: A special type of Observable that can act as both an Observable and an Observer. Subjects can be used to multicast events to multiple subscribers and are often used in event bus implementations.
7. BehaviorSubject: A type of Subject that emits the most recent item it has observed and all subsequent items to new subscribers.
8. PublishSubject: A type of Subject that only emits items that are observed after the subscription.
9. ReplaySubject: A type of Subject that stores all the events it’s ever processed. When a new subscriber comes along, it first gets a rundown of the archived events before getting the live updates. It’s set up to remember everything from the get-go, which could be risky if we’re dealing with a never-ending stream of events.

Different types of Observers

1. Observer: The basic interface for observing emissions from an Observable. It has methods like onNext, onError, and onComplete for handling emissions, errors, and completion events.
2. DisposableObserver: An extension of the Observer interface that includes a dispose method for explicit disposal of the subscription. It’s useful for managing and canceling subscriptions.
3. SingleObserver: An interface specifically designed for observing emissions from a Single Observable. It includes methods like onSuccess and onError to handle the single value or error emitted by the Single.
4. MaybeObserver: An interface for observing emissions from a Maybe Observable. It includes methods like onSuccess, onComplete, and onError to handle one item, no item, or an error.
5. CompletableObserver: An interface for observing emissions from a Completable Observable. It includes methods like onComplete and onError to handle the completion or error event.
6. FlowableSubscriber: An interface for observing emissions from a Flowable Observable, which is used for handling potentially large streams of data with backpressure support. It includes methods like onNext, onError, and onComplete.
7. DisposableSubscriber: An extension of the FlowableSubscriber interface that includes a dispose method for explicit disposal of the subscription.
8. ResourceSubscriber: An extension of the DisposableSubscriber interface that provides additional resource management features for handling subscriptions.

Cold and Hot Observables

Ah, the world of Reactive Programming! In RxJava, the concepts of “Cold” and “Hot” Observables are pivotal. They help you manage how data streams are created and consumed, and understanding the distinction can make a big difference in how your application behaves.

Think of a Cold observable (Pull Model) like a movie on a streaming service. Every time you hit “Play,” the movie starts from the beginning, just for you. Similarly, a cold observable doesn’t start emitting items until an observer subscribes to it. Each subscriber gets its own independent set of emitted items. So if two observers subscribe to the same cold observable, the data stream starts anew for each.

import io.reactivex.Observable;

public class ColdObservableExample {
    public static void main(String[] args) {
        // Create a cold observable from a list of numbers
        Observable<Integer> numbersObservable = Observable.fromArray(1, 2, 3, 4, 5);

        // Subscribe multiple times to the same observable
        numbersObservable.subscribe(n -> System.out.println("Subscriber 1: " + n));
        numbersObservable.subscribe(n -> System.out.println("Subscriber 2: " + n));
    }
}

and the output would look like the below

Subscriber 1: 1
Subscriber 1: 2
Subscriber 1: 3
Subscriber 1: 4
Subscriber 1: 5
Subscriber 2: 1
Subscriber 2: 2
Subscriber 2: 3
Subscriber 2: 4
Subscriber 2: 5

Process finished with exit code 0

In the above example, each subscriber starts receiving their own set of items from the start as soon as they subscribe.

A Hot Observable is like a film rolling in a theater. The movie starts at a specific time and keeps playing, regardless of whether the theater is empty or packed. When you walk into the theater and the movie is already 20 minutes in, you’ve missed the first 20 minutes.

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.observables.ConnectableObservable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class HotObservableExample {

  public static void main(String[] args) throws InterruptedException {
    ConnectableObservable<Object> movieTheater =
        Observable.create(
                emitter -> {
                  Thread.sleep(1000);
                  emitter.onNext(emitVals("Scene 1"));
                  Thread.sleep(1000);
                  emitter.onNext(emitVals("Scene 2"));
                  Thread.sleep(1000);
                  emitter.onNext(emitVals("Scene 3"));
                })
            .subscribeOn(Schedulers.io()) // do not want the main thread to be 
            .publish();                  // for the emission

    // Viewer1 walks in
    movieTheater
        .subscribe(scene -> System.out.println("Viewer1: " + scene));

    // Start the movie
    movieTheater.connect();

    Thread.sleep(2000);
    // Viewer2 walks in during Scene 2
    movieTheater.subscribe(scene -> System.out.println("Viewer2: " + scene));

    Thread.sleep(1000);
    // Viewer3 walks in during Scene 3
    movieTheater.subscribe(scene -> System.out.println("Viewer3: " + scene));

    Thread.sleep(10000);
  }

  private static String emitVals(String val) {
    System.out.println("________" + val + "________");
    return val;
  }
}

and the output would look like the below

________Scene 1________
Viewer1: Scene 1
________Scene 2________
Viewer1: Scene 2
Viewer2: Scene 2
________Scene 3________
Viewer1: Scene 3
Viewer2: Scene 3
Viewer3: Scene 3

Process finished with exit code 0

The above code simulates the movie theater that I mentioned earlier, where the movie (scenes) plays continuously. New viewers (subscribers) only see scenes from the point they enter. So, Viewer1 sees all scenes, Viewer2 misses the first one, and Viewer3 only catches the last. All of this happens in real-time, making it a “hot” experience for the viewers.

Sensor data analysis, mouse click tracking, real-time stock price updates, and smartphone accelerometer readings are just a few practical examples of hot observables.

Operators

Operators are the bread and butter of reactive programming in Java. They help you manipulate and transform data streams, like turning dough into all sorts of pastries—some sweet, some savory. You’ve got operators to filter events, to merge multiple streams, to map one type of data to another, and so on. In short, they’re your toolkit for making the reactive magic happen! 😊

Alright, it’s time to dive into some key RxJava operators. I won’t cover every single one, as the list is quite extensive. However, I’ll highlight a handful of the most useful operators that you’ll likely use on a day-to-day basis.

map() This Operator in RxJava is like a little magician in your data pipeline. It takes each item from the source Observable, performs a transformation on it, and then emits the transformed item to the subscriber. Imagine having a conveyor belt of apples and you need them sliced; the map() operator is that super-efficient slicer making sure every apple that comes out is just how you want it—sliced to perfection! It’s incredibly handy when you want to modify the data in-flight before it reaches its final destination.

import io.reactivex.rxjava3.core.Observable;

public class MapExample {
    public static void main(String[] args) {
        Observable<String> fruitsObservable = Observable.just("Apple", "Banana", "Orange");

        fruitsObservable.map(fruit -> getColor(fruit))
                        .subscribe(color -> System.out.println("Received: " + color));
    }

    private static String getColor(String fruit) {
        switch (fruit) {
            case "Apple":
                return "Red";
            case "Banana":
                return "Yellow";
            case "Orange":
                return "Orange";
            default:
                return "Unknown";
        }
    }
}

and the output would look like the below

Received: Red
Received: Yellow
Received: Orange

Process finished with exit code 0

flatMap()Unlike map(), which transforms one item into one other item, flatMap() can transform one item into multiple items, possibly even of a different type. The flatMap() operator in RxJava is your multitasking buddy, it can take one input and it can call multiple downstream services that too concurrently on different threads. The concurrency part we will cover later in this blog, for now, let’s look at the very common use cases where we use flatMap quite frequently. One such case is that where we call a function which itself is returning the observable. In that case, the map will not work as it will then output Observable<Observable<Object>>> rather than Observable<Object>.

For example

import io.reactivex.rxjava3.core.Observable;

public class FlatMapSingleObservableExample {
    public static void main(String[] args) {
        Observable<String> fruitsObservable = Observable.just("Apple", "Banana", "Orange");

        fruitsObservable.flatMap(fruit -> getColor(fruit))
                        .subscribe(color -> System.out.println("Received: " + color));
    }

    private static Observable<String> getColor(String fruit) {
        switch (fruit) {
            case "Apple":
                return Observable.just("Red");
            case "Banana":
                return Observable.just("Yellow");
            case "Orange":
                return Observable.just("Orange");
            default:
                return Observable.just("Unknown");
        }
    }
}

and the output would look like the below

Received: Red
Received: Yellow
Received: Orange

Process finished with exit code 0

Another example is, where it transforms one item into more than one elements

import io.reactivex.rxjava3.core.Observable;

public class FlatMapExample {
    public static void main(String[] args) {
        Observable<String> fruitsObservable = Observable.just("Apple", "Banana", "Orange");

        fruitsObservable.flatMap(fruit -> {
            String color = getColor(fruit);
            String taste = getTaste(fruit);
            return Observable.just(color, taste);
        }).subscribe(item -> System.out.println("Received: " + item));
    }

    private static String getColor(String fruit) {
        switch (fruit) {
            case "Apple":
                return "Red";
            case "Banana":
                return "Yellow";
            case "Orange":
                return "Orange";
            default:
                return "Unknown";
        }
    }

    private static String getTaste(String fruit) {
        switch (fruit) {
            case "Apple":
                return "Sweet";
            case "Banana":
                return "Mellow";
            case "Orange":
                return "Tangy";
            default:
                return "Unknown";
        }
    }
}

and the output would look like the below

Received: Red
Received: Sweet
Received: Yellow
Received: Mellow
Received: Orange
Received: Tangy

Process finished with exit code 0

Another example of flatMap is where the operator accepts there different functional parameters

< R > Observable < R > flatMap(
    Func1 < T, Observable < R >> onNext,
    Func1 < Throwable, Observable < R >> onError,
    Func0 < Observable < R >> onCompleted)

Let’s look at an example where these kinds of signature is helpful, In the following example, the focus is on signaling that the shop is closed after all the fruits have been sent out. We’re not concerned about the individual fruits, just the “Close the shop” event. But here the flatmap gives us a provision to process all the fruit events where we can also use multiple threads for concurrency and parallelism to expedite the process and at the same time send a “Close the shop” event right after processing the last fruit. Just to keep things simple, Multi-threaded fruit processing is not included here, more about that in the later part of this article.

import io.reactivex.rxjava3.core.Observable;

public class FlatMapExample {
    public static void main(String[] args) {
        Observable<String> observable = Observable.create(emitter -> {
            emitter.onNext("Apple");
            emitter.onNext("Banana");
            emitter.onNext("Orange");
            emitter.onComplete();
        });

        observable.flatMap(
                fruit -> Observable.empty(),  // onNext-- fruit processing end here
                throwable -> Observable.just("Close the shop"),  // onError
                () -> Observable.just("Close the shop")  // onCompleted
        ).subscribe(
                item -> System.out.println("Received: " + item),
                throwable -> System.out.println("Error: " + throwable),
                () -> System.out.println("Done!")
        );
    }
}

and the output would look like the below

Received: Close the shop
Done!

Process finished with exit code 0

Concurrency in flatMap() :- Due to flatMap creating individual observables for each item and running them in parallel, you might run into concurrency issues. In the following example, applying flatMap on a large list could overwhelm and crash the system. This could get even trickier if the inner observable is making calls to an external API.

Observable < Profile > profiles = Observable
    .from(veryLargeList)
    .flatMap(Fruit::prosess);

To tackle this problem, there exists an overloaded flatMap with an additional parameter maxConcurrency to control the concurrency, the signature would look like below

Observable < Profile > profiles = Observable
    .from(veryLargeList)
    .flatMap(Fruit::prosess,10);// maxConcurrency control parameter

concatMap() This operator is like a more organized version of flatMap(). It avoids any parallel execution and maintains the original order of events thus preventing any overlaps. flatMap() lacks a sense of order; it essentially relies on the merge() operator internally. This causes it to subscribe to all the inner observables simultaneously, without differentiating between them. In other words, the concatMap() is equivalent to flatMap() with maxConcurrency one.

Observable < Profile > profiles = Observable
    .just("Apple", "Banana", "Orange")
    .concatMap(Fruit::prosess);

The above code is equivalent to the below one

Observable < Profile > profiles = Observable
    .from(veryLargeList)
    .flatMap(Fruit::prosess,1);// maxConcurrency =1 ie: no concurrency

flatMapIterable()This operator is very similar to flatMap(). However, as the name suggests, it works with iterables instead of single elements. This operator is useful when you have an Observable that emits lists. It takes an Observable of a list as input and emits an Observable for each item in that list.

import io.reactivex.rxjava3.core.Observable;

import java.util.Arrays;
import java.util.List;

public class FlatMapIterableExample {
  public static void main(String[] args) throws InterruptedException {
    Observable<List<String>> fruitBaskets =
        Observable.just(
            Arrays.asList("Apple", "Banana"), Arrays.asList("Mango", "Orange", "Papaya"));

    // With flatMapIterable
    fruitBaskets
        .flatMapIterable(basket -> basket)
        .subscribe(fruit -> System.out.println("Got fruit: " + fruit));
  }
}

and the output would look like below

Got fruit: Apple
Got fruit: Banana
Got fruit: Mango
Got fruit: Orange
Got fruit: Papaya

Process finished with exit code 0

In the above example, fruitBaskets is a stream of list of fruits and the flatMapIterable() flattens the Observable<List<String>> to stream of Observable<String> of each items in the lists. Now, if flatMapIterable() wasn’t there, you’d have to manually unpack each basket, likely using flatMap() and creating a new observable for each item like below

fruitBaskets.flatMap(basket -> Observable.fromIterable(basket))
              .subscribe(fruit -> System.out.println("Got fruit: " + fruit));

merge() and mergeWith()This operator is like a traffic cop at a busy intersection. Imagine you have multiple lanes (Observables) of cars (data items), each lane moving at its own pace. The merge() operator directs these cars into a single lane (a new Observable), maintaining the original speed and order of each incoming lane. In simple terms, it combines multiple Observables into one, emitting items as they arrive from any of the source Observables. It’s a great way to manage data from multiple sources without worrying about sequencing.

import io.reactivex.rxjava3.core.Observable;

public class MergeExample {
  public static void main(String[] args) throws InterruptedException {
    Observable<String> vendorA = Observable.just("Apple", "Orange");
    Observable<String> vendorB = Observable.just("Banana", "Grape");

    Observable<String> mergedFruits = Observable.merge(vendorA, vendorB);
    mergedFruits.subscribe(fruit -> System.out.println("Received: " + fruit));
  }
}

and the output would look like below

Received: Apple
Received: Orange
Received: Banana
Received: Grape

Process finished with exit code 0

The mergeWith() operator is basically a shorthand for using merge() when you already have one Observable and want to merge it with another. Instead of having to wrap both Observables in a merge() function, you can just use mergeWith() on one of the Observables and pass the other as an argument.

In the above code, you can rewrite it to use mergeWith() like below, which is the shorthand form of merge().

Observable<String> mergedFruits = vendorA.mergeWith(vendorB);

When using merge(), if any of the merged Observables encounter an error, the merged Observable stream will also terminate immediately and call its onError() method. This means if you’re merging multiple Observables, and even just one fails, you won’t receive any more items from the other Observables.

import io.reactivex.rxjava3.core.Observable;

public class MergeExample {
  public static void main(String[] args) throws InterruptedException {
    Observable<String> vendorA = Observable.error(new RuntimeException("Oops, something went wrong!"));
    Observable<String> vendorB = Observable.just("Apple", "Orange");

    Observable<String> mergedFruits = Observable.merge(vendorA, vendorB);
    mergedFruits.subscribe(
            fruit -> System.out.println("Received: " + fruit),
            error -> System.out.println("Error: " + error.getMessage())
    );
  }
}

and the output would look like below

Error: Oops, something went wrong!

Process finished with exit code 0

mergeDelayError()There is a way to delay any error notifications until all the Observables have completed their emissions during a merge operation. mergeDelayError() is a nifty operator that comes in handy when you want to merge multiple Observables but also want to ensure that all emissions are captured, even if one of the Observables throws an error.

import io.reactivex.rxjava3.core.Observable;

public class MergeDelayErrorExample {
  public static void main(String[] args) throws InterruptedException {
    Observable<String> vendorA = Observable.error(new RuntimeException("Oops, something went wrong!"));
    Observable<String> vendorB = Observable.just("Apple", "Orange");

    Observable<String> mergedFruits = Observable.mergeDelayError(vendorA, vendorB);
    mergedFruits.subscribe(
            fruit -> System.out.println("Received: " + fruit),
            error -> System.out.println("Error: " + error.getMessage())
    );
  }
}

Output

Received: Apple
Received: Orange
Error: Oops, something went wrong!

Process finished with exit code 0

There is an interesting thing to note here is that the error object. Since there is only one error the error message throws the exact same error

what if an error occurred in multiple streams, for example

import io.reactivex.rxjava3.core.Observable;

public class MergeDelayErrorExample {
  public static void main(String[] args) throws InterruptedException {
    Observable<String> vendorA = Observable.error(new RuntimeException("Oops, something went wrong!"));
    Observable<String> vendorB = Observable.just("Apple", "Orange");
    Observable<String> vendorC = Observable.error(new RuntimeException("Oops, something went wrong again!"));

    Observable<String> mergedFruits = Observable.mergeDelayError(vendorA, vendorB, vendorC);
    mergedFruits.subscribe(
            fruit -> System.out.println("Received: " + fruit),
            error -> System.out.println("Error: " + error.getMessage())
    );
  }
}

Output

Received: Apple
Received: Orange
Error: 2 exceptions occurred. 

Process finished with exit code 0

The error message shows there are two exceptions that happened, and there is an additional field in the error object called Throwable.cause. and it has an array of exceptions that happened during the merge.

zip() and zipWith()They take emissions from multiple Observables and combine them based on a function you provide. In simple terms, zip() waits for each of the source Observables to emit an item and then applies a function to all those items, emitting the result.

Imagine you have three Observables in the context of a fruit market:

• One emits the types of fruits (Apple, Banana, Orange, Cherry).
• The second emits their corresponding prices ($2, $1, $3, $4).
• The third emits their availability status (In stock, Out of stock, In stock, In stock).

You could use zip() to combine these three Observables so that each emitted tuple contains all the information you’d need for each type of fruit. For instance, the resulting Observable might emit tuples like:

• ("Apple", $2, "In stock")
• ("Banana", $1, "Out of stock")
• ("Orange", $3, "In stock")
• ("Cherry", $4, "In stock")

So, for each type of fruit, you get its price and its availability status, all neatly packed into one data point. This is a perfect example of how zip() can offer a one-on-one mapping, correlating each fruit with its corresponding price and stock status.

Let’s check how it is done programmatically

import io.reactivex.rxjava3.core.Observable;

public class ZipExample {
  public static void main(String[] args) throws InterruptedException {
    // Observable for fruit types
    Observable<String> fruits = Observable.just("Apple", "Banana", "Orange", "Cherry");

    // Observable for fruit prices
    Observable<Integer> prices = Observable.just(2, 1, 3, 4);

    // Observable for availability
    Observable<String> availability =
        Observable.just("In stock", "Out of stock", "In stock", "In stock");

    // Using zip to combine these Observables
    Observable.zip(
            fruits,
            prices,
            availability,
            (fruit, price, avail) -> "(" + fruit + ", $" + price + ", " + avail + ")")
        .subscribe(
            zipped -> System.out.println("Zipped: " + zipped),
            throwable -> System.out.println("Error: " + throwable),
            () -> System.out.println("Done!"));
  }
}

Output

Zipped: (Apple, $2, In stock)
Zipped: (Banana, $1, Out of stock)
Zipped: (Orange, $3, In stock)
Zipped: (Cherry, $4, In stock)
Done!

Process finished with exit code 0

In the above example zip() waited for fruits, prices and availability to emit one element and perform the operation on emitted items (fruit, price, avail) -> "(" + fruit + ", $" + price + ", " + avail + ")").

In the zip operation the entire resultant stream gets terminated when one of the source streams gets completed, In the above example all the source streams emit exactly 4 elements, what if one of the streams gets terminated a bit early? For example, in the above code if the fruits Observable emits only 2 elements then the entire stream gets terminated once the fruits Observable finishes.

import io.reactivex.rxjava3.core.Observable;

public class ZipExample {
  public static void main(String[] args) throws InterruptedException {
    // Observable for fruit types
    Observable<String> fruits = Observable.just("Apple", "Banana", "Orange", "Cherry");

    // Observable for fruit prices
    Observable<Integer> prices = Observable.just(2, 1);

    // Observable for availability
    Observable<String> availability =
        Observable.just("In stock", "Out of stock", "In stock", "In stock");

    // Using zip to combine these Observables
    Observable.zip(
            fruits,
            prices,
            availability,
            (fruit, price, avail) -> "(" + fruit + ", $" + price + ", " + avail + ")")
        .subscribe(
            zipped -> System.out.println("Zipped: " + zipped),
            throwable -> System.out.println("Error: " + throwable),
            () -> System.out.println("Done!"));
  }
}

Output

Zipped: (Apple, $2, In stock)
Zipped: (Banana, $1, Out of stock)
Done!

Process finished with exit code 0

zipWith() is just a shorthand form of zip() but at a time it can handle only two. So for handling three, it has to chain with one more, the above code can be rewritten to use zipWith() like below and it produces the exact same result.

fruits.zipWith(prices, (fruit, price) -> "(" + fruit + ", $" + price )
            .zipWith(availability,(fruitPrice,avail)-> fruitPrice+", "+avail+ ")")
        .subscribe(
            zipped -> System.out.println("Zipped: " + zipped),
            throwable -> System.out.println("Error: " + throwable),
            () -> System.out.println("Done!"));

combineLatest()In zip() the stream ends when one of the source Observables terminates, but combineLatest() wait for all the source observables to finish emission. In other words,It starts emitting items as soon as each of the source Observables has emitted at least one item. After that, whenever any source Observable emits a new item, combineLatest() combines the latest item from each Observable using the provided function and emits the result.

import io.reactivex.rxjava3.core.Observable;

public class CombineLatestExample {
  public static void main(String[] args) {
    // Observable for fruit types
    Observable<String> fruits =
        Observable.just("Apple", "Banana", "Cherry", "Orange")
            .concatMap(
                fruit ->
                    Observable.just(fruit).delay(100, java.util.concurrent.TimeUnit.MILLISECONDS));

    // Observable for fruit prices
    Observable<Integer> prices =
        Observable.just(2, 1)
            .concatMap(
                price ->
                    Observable.just(price).delay(150, java.util.concurrent.TimeUnit.MILLISECONDS));

    // Observable for fruit availability
    Observable<String> availability =
        Observable.just("In Stock", "Out of Stock", "In Stock", "Limited")
            .concatMap(
                avail ->
                    Observable.just(avail).delay(200, java.util.concurrent.TimeUnit.MILLISECONDS));

    // Using combineLatest to get the latest of each Observable
    Observable.combineLatest(
            fruits,
            prices,
            availability,
            (fruit, price, avail) ->
                "Fruit: " + fruit + ", Price: $" + price + ", Availability: " + avail)
        .subscribe(
            combined -> System.out.println("Combined: " + combined),
            throwable -> System.out.println("Error: " + throwable),
            () -> System.out.println("Done!"));

    // Sleep for a while to let the observables emit values
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

Output

Combined: Fruit: Banana, Price: $2, Availability: In Stock
Combined: Fruit: Banana, Price: $1, Availability: In Stock
Combined: Fruit: Cherry, Price: $1, Availability: In Stock
Combined: Fruit: Orange, Price: $1, Availability: In Stock
Combined: Fruit: Orange, Price: $1, Availability: Out of Stock
Combined: Fruit: Orange, Price: $1, Availability: In Stock
Combined: Fruit: Orange, Price: $1, Availability: Limited
Done!

Process finished with exit code 0

In the above fruits emitted 4 times (“Apple”, “Banana”, “Cherry”, “Orange”), prices emitted 2 times(2,1) from the source observable, and availability emitted 4 times (“In Stock”, “Out of Stock”, “In Stock”, “Limited”) so total 10 emissions from source observables. But there were only 7 emissions that happened with combineLatest() why? Because the very initial emission requires all the source observables to emit at least one. So the firs emission happens after it receives one element from all the source observable. Since there were three observables, the final number of emissions matched up to 10 – 3, which is 7.

Note:- One thing to note here is the Thread.sleep(1000) at the end. Why do we need sleep to block the main thread here? Where are we delegating the operation to a different scheduler or thread here? By default the delay() operators get executed on their own schedulers that is ComputationalThreadPool. So even without mentioning the schedulers explicitly here the work is being transferred to a different thread for its execution. It’s one of those things that makes RxJava powerful but also requires you to be aware of these default behaviors.

withLatestFrom()In combineLatest() it emits an item whenever any of the source observables emits but here there is a clear distinction between the two observables. It only emits when one of the source observables emits and it combines this value with the latest value of the other observable to perform the action mentioned in the transformation function.

import io.reactivex.rxjava3.core.Observable;

public class WithLatestFromExample {
  public static void main(String[] args) {
    Observable<String> fruitNames = Observable.just("Apple", "Banana", "Cherry");
    Observable<Integer> fruitPrices = Observable.just(1, 2, 3);
    fruitNames.withLatestFrom(fruitPrices, (name, price) -> "Fruit: " + name + ", Latest Price: $" + price)
            .subscribe(System.out::println);

  }
}

Output

Fruit: Apple, Latest Price: $3
Fruit: Banana, Latest Price: $3
Fruit: Cherry, Latest Price: $3

Process finished with exit code 0

In the above example, combineLatest() emits a value whenever the fruitNames Observable emits, using the latest values from the fruitPrices Observable at that specific time. By the time fruitNames emits its first element “Apple”, the fruitPrices Observable has already emitted its third element. That’s why you see only “3” from fruitPrices in the result.

stratWith()This operator in RxJava lets you prepend an initial item or sequence of items to an Observable. Basically, it makes sure your subscriber receives these items before it receives the items from the source Observable. Think of it like getting a free sample at a fruit market before you start shopping for fruits!

import io.reactivex.rxjava3.core.Observable;

public class StartWithExample {
  public static void main(String[] args) {
    Observable<String> fruits = Observable.just("Apple", "Banana", "Cherry");
    Observable<String> withInitialFruit = fruits.startWith( Observable.just("Grapes"));
    withInitialFruit.subscribe(fruit -> System.out.println("Received: " + fruit));
  }
}

Output

Received: Grapes
Received: Apple
Received: Banana
Received: Cherry

Process finished with exit code 0

Here the Grapes get printed before the actual emission happens.

Where do we use this? A use case could be in a UI where you want to display a list of fruits available in the inventory. You could use startWith() to insert a message like “Fetching available fruits…” before the actual list gets populated.

amb()This operator takes multiple Observables and subscribes to the one that emits an item first, ignoring the others. It’s like having multiple baskets of fruits in front of you but picking the one that first catches your eye and ignoring the rest.

import io.reactivex.rxjava3.core.Observable;

import java.util.List;

public class AmbExample {
  public static void main(String[] args) {
    // First Observable
    Observable<String> apples = Observable.just("Red Apple", "Green Apple")
            .delay(2, java.util.concurrent.TimeUnit.SECONDS);

    // Second Observable
    Observable<String> bananas = Observable.just("Yellow Banana")
            .delay(1, java.util.concurrent.TimeUnit.SECONDS);

    // Using amb() to pick the Observable that emits first
    
    Observable.amb(List.of(apples, bananas))
            .subscribe(
                    fruit -> System.out.println("Received: " + fruit),
                    throwable -> System.out.println("Error: " + throwable),
                    () -> System.out.println("Done!")
            );

    // Sleep to allow observables time to emit
    try {
      Thread.sleep(5000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

Output

Received: Yellow Banana
Done!

Process finished with exit code 0

In this example, bananas Observable emits first due to its shorter delay, so amb() will pick it, and you will only see “Yellow Banana” printed to the console. The apples Observable will be ignored.

we can also use ambArray() to avoid an explicit array declaration in the signature

Observable.ambArray(apples, bananas)
            .subscribe(
                    fruit -> System.out.println("Received: " + fruit),
                    throwable -> System.out.println("Error: " + throwable),
                    () -> System.out.println("Done!")
            );

scan()This operator is like a rolling accumulator. It’s really useful when you want to keep a running total or ongoing transformation of the data. The scan() operator takes an initial value and a function as parameters. This function is applied to each item emitted by the source Observable.

import io.reactivex.rxjava3.core.Observable;

public class ScanExample {
  public static void main(String[] args) {
    Observable<String> fruits = Observable.just("Apple", "Banana", "Apple", "Orange", "Banana", "Apple");

    fruits.scan(new StringBuilder(), (acc, fruit) -> acc.append(fruit).append(", "))
            .subscribe(System.out::println);
  }
}

Output

Apple, 
Apple, Banana, 
Apple, Banana, Apple, 
Apple, Banana, Apple, Orange, 
Apple, Banana, Apple, Orange, Banana, 
Apple, Banana, Apple, Orange, Banana, Apple, 

Process finished with exit code 0

In the code above, the accumulator (acc) is initialized with a new StringBuilder() during the first emission. For subsequent emissions, it carries over the previous result. The code also emits all intermediate results.

reduce()This operator is also similar to scan() but the only difference is that it emits only the final value. This can be considered as the

import io.reactivex.rxjava3.core.Observable;

public class ReduceExample {
  public static void main(String[] args) {
    Observable<String> fruits = Observable.just("Apple", "Banana", "Apple", "Orange", "Banana", "Apple");

    fruits.reduce(new StringBuilder(), (acc, fruit) -> acc.append(fruit).append(", "))
            .subscribe(System.out::println);
  }
}

Output

Apple, Banana, Apple, Orange, Banana, Apple, 

Process finished with exit code 0

this is equivalent to

fruits.scan(new StringBuilder(), (acc, fruit) -> acc.append(fruit).append(", ")).takeLast(1)
            .subscribe(System.out::println);

if your sequence is infinite, scan() keeps emitting events for each upstream event, whereas reduce() will never emit any event. So due diligence needs to be taken when using these kinds of operations.

collect()This operator in RxJava is useful when you want to collect items from an Observable into a single mutable data structure. This is particularly handy if you need to transform an Observable stream into a collection like a list, set, or map. This is also similar to reduce() but more intuitive and has fewer lines of code.

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Single;
import java.util.ArrayList;
import java.util.List;

public class CollectExample {
  public static void main(String[] args) {
    Observable<String> fruitObservable = Observable.just("Apple", "Banana", "Orange", "Grapes");
    Single<List<String>> collectedFruits = fruitObservable.collect(ArrayList::new, List::add);

    collectedFruits.subscribe(
            result -> System.out.println("Collected Fruits: " + result),
            error -> System.err.println("Error: " + error)
    );
  }
}

Output

Collected Fruits: [Apple, Banana, Orange, Grapes]

Process finished with exit code 0

In this example, collect() starts with an empty ArrayList. As fruits are emitted from the fruitObservable, they are added to this list using the List::add method. Finally, when the Observable completes, a Single is emitted containing all the collected fruits.

Without collect(), you’d probably use subscribe() to collect these elements manually, complicating your code. So, collect() simplifies things quite a bit.

single()This operator is used when you expect the Observable to emit exactly one item. If the Observable emits more than one item, it will trigger an error. If it emits no items, it will also trigger an error.

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Single;

public class SingleExample {
  public static void main(String[] args) {
    // Example where Observable emits exactly one item
    Observable<String> singleFruit = Observable.just("Apple");
    Single<String> result = singleFruit.single("Default");
    result.subscribe(
        item -> System.out.println("Received: " + item),
        error -> System.err.println("Error: " + error));

    // Example where Observable emits more than one item (This will cause an error)
    Observable<String> multipleFruits = Observable.just("Apple", "Banana");
    Single<String> resultMultiple = multipleFruits.single("Default");
    resultMultiple.subscribe(
        item -> System.out.println("Received: " + item),
        error -> System.err.println("Error: " + error));

    // Example where Observable emits no items (This will take the default value)
    Observable<String> emptyFruits = Observable.empty();
    Single<String> resultEmpty = emptyFruits.single("Default");
    resultEmpty.subscribe(
        item -> System.out.println("Received: " + item),
        error -> System.err.println("Error: " + error));
  }
}

Output

Received: Apple
Received: Default
Error: java.lang.IllegalArgumentException: Sequence contains more than one element!

Process finished with exit code 0

• In the first case, the Observable emits exactly one item “Apple”, and everything goes as expected.
• In the second case, the Observable emits more than one item, so an error will be triggered.
• In the third case, no items are emitted. The single() operator will then take the default value specified (“Default”).

distinct()This operator filters out duplicate items based on their value, ensuring that only unique items are emitted downstream.

import io.reactivex.rxjava3.core.Observable;

public class DistinctExample {
  public static void main(String[] args) {
    Observable<String> fruits = Observable.just("Apple", "Banana", "Apple", "Orange", "Banana", "Cherry");

    fruits.distinct()
            .subscribe(fruit -> System.out.println("Received: " + fruit));
  }
}

Output

Received: Apple
Received: Banana
Received: Orange
Received: Cherry

Process finished with exit code 0

So, let’s say you have an Observable emitting a very large set of items, and a significant number of them are unique. In this scenario, the internal data structure used by distinct() to track unique items could grow large, consuming more memory. So due diligence needs to be taken when using these kinds of operations.

distinctUntilChanged()This operator is a sibling of distinct(), but with a more narrow focus. Instead of ensuring that all emitted items are unique throughout the entire observable chain, distinctUntilChanged() only suppresses consecutively duplicated items. In other words, it allows an item to be emitted if it’s different from the one immediately before it.

import io.reactivex.rxjava3.core.Observable;

public class DistinctUntilChangedExample {
  public static void main(String[] args) {
    Observable<String> fruits = Observable.just("Apple","Apple", "Banana", "Apple", "Orange", "Banana", "Cherry");

    fruits.distinctUntilChanged()
            .subscribe(fruit -> System.out.println("Received: " + fruit));
  }
}

Output

Received: Apple
Received: Banana
Received: Apple
Received: Orange
Received: Banana
Received: Cherry

Process finished with exit code 0

In the above example, Apple and Banana are repeating as it eliminates only the consecutive duplicates.

concat() and concatWith()This operator takes multiple observables and concatenates their emissions. It subscribes to the first observable, emits all its items, and once it’s complete, subscribes to the next observable, and so on.

import io.reactivex.rxjava3.core.Observable;

public class ConcatExample {
  public static void main(String[] args) {
    Observable<String> summerFruits = Observable.just("Mango", "Watermelon");
    Observable<String> winterFruits = Observable.just("Orange", "Apple");

    Observable.concat(summerFruits, winterFruits)
            .subscribe(fruit -> System.out.println("Received: " + fruit));
  }
}

Output

Received: Mango
Received: Watermelon
Received: Orange
Received: Apple

Process finished with exit code 0

The concatWith() operator is a simplified form of concat(). It concatenates the current observable with another observable. The above code can be rewritten to the following using concatWith() .

summerFruits.concatWith(winterFruits)
                    .subscribe(fruit -> System.out.println("Received: " + fruit));

Slicing and Dicing OperatorsFollowing are some of the operators used for slicing and dicing the emission.

Observable.range(1, 5).take(3); // [1, 2, 3] 
Observable.range(1, 5).skip(3); // [4, 5] 
Observable.range(1, 5).skip(5); // []
Observable.range(1, 5).takeLast(2); // [4, 5] 
Observable.range(1, 5).skipLast(2); // [1, 2, 3]
Observable.range(1, 5).first(); // [1]
Observable.range(1, 5).last(); // [5]
Observable.range(1, 5).first(x -> x % 2 == 0); // [2]
Observable.range(1, 5).last(x -> x % 2 == 0); // [4]
​​Observable.range(1, 5).takeUntil(x -> x == 3); // [1, 2, 3] 
Observable.range(1, 5).takeWhile(x -> x != 3); // [1, 2]

Predicate OperatorsFollowing are some of the predicate operators used for emitting true/false based on the condition.

numbers.all(x -> x != 4);    // [false] 
numbers.exists(x -> x == 4); // [true] 
numbers.contains(4);         // [true]

switchOnNext()Operator is a bit unique; it’s used with an Observable that emits other Observables. When it receives a new Observable, it unsubscribes from the previous Observable and starts emitting items from the most recently emitted Observable.

To put it simply, it “switches” to the latest Observable.

Let’s imagine you have different fruit seasons, and each season has its own list of fruits. You could use switchOnNext() to only focus on the fruits of the current season.

import io.reactivex.rxjava3.core.Observable;
import java.util.concurrent.TimeUnit;

public class SwitchOnNextExample {
  public static void main(String[] args) {
    Observable<Observable<String>> seasonalFruits =
        Observable.create(
            emitter -> {
              Observable<String> summerFruits =
                  Observable.interval(1, TimeUnit.SECONDS)
                      .map(tick -> tick == 0 ? "Mango" : "Watermelon");
              Observable<String> winterFruits =
                  Observable.interval(1, TimeUnit.SECONDS)
                      .map(tick -> tick == 0 ? "Orange" : "Apple");

              emitter.onNext(summerFruits);
              try {
                Thread.sleep(3000);
              } catch (InterruptedException ignored) {
              }
              emitter.onNext(winterFruits);
            });

    Observable.switchOnNext(seasonalFruits)
        .subscribe(fruit -> System.out.println("Received: " + fruit));

    // Let it run for some time to see the output
    try {
      Thread.sleep(6000);
    } catch (InterruptedException ignored) {
    }
  }
}

In this example, the seasonalFruits observable emits two inner observables: summerFruits and winterFruits. The switchOnNext() operator unsubscribes from summerFruits and starts emitting from winterFruits as soon as winterFruits is emitted by seasonalFruits.

The output would look something like this:

Received: Mango
Received: Watermelon
Received: Watermelon
Received: Orange
Received: Apple
Received: Apple
Received: Apple
Received: Apple
Received: Apple

Process finished with exit code 0

groupBy()This operator sorts the items emitted by an Observable into separate groups based on a specified key. Each group is itself an Observable that emits those items from the original Observable that share the same key.

Here’s an example using fruits, where we’ll categorize fruits by their color:

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.observables.GroupedObservable;

public class GroupByExample {
  public static void main(String[] args) {
    Observable<String> fruits = Observable.just("Apple", "Banana", "Cherry", "Blueberry", "Orange");

    Observable<GroupedObservable<String, String>> groupedFruits =
        fruits.groupBy(fruit -> getColor(fruit));

    groupedFruits
        .flatMapSingle(group -> group.toList().map(fruitList -> group.getKey() + ": " + fruitList))
        .subscribe(
            result -> System.out.println("Received: " + result),
            error -> System.err.println("Error: " + error),
            () -> System.out.println("Done!"));
  }

  private static String getColor(String fruit) {
    switch (fruit) {
      case "Apple":
        return "Red";
      case "Banana":
        return "Yellow";
      case "Cherry":
        return "Red";
      case "Blueberry":
        return "Blue";
      case "Orange":
        return "Orange";
      default:
        return "Unknown";
    }
  }
}