RxJava Tutorials

Tutorial 1: Creating Observables

In this tutorial, we'll learn how to create Observables using different factory methods and understand the core concepts behind Observable creation.

Basic Observable Creation

There are several ways to create Observables in RxJava:

• Observable.create() - Create an Observable from scratch
• Observable.just() - Create an Observable that emits a specified item
• Observable.fromIterable() - Convert an Iterable into an Observable
• Observable.fromArray() - Convert an array into an Observable
• Observable.fromCallable() - Create an Observable from a Callable
• Observable.interval() - Create an Observable that emits a sequence of integers spaced by a given time interval

1import io.reactivex.rxjava3.core.Observable;
2
3// Creating and subscribing to a simple Observable
4Observable<String> observable = Observable.create(emitter -> {
5    try {
6        emitter.onNext("Hello");
7        emitter.onNext("World");
8        emitter.onComplete();
9    } catch (Exception e) {
10        emitter.onError(e);
11    }
12});
13
14// Subscribe with separate action handlers
15observable.subscribe(
16    item -> System.out.println(item),             // onNext handler
17    error -> error.printStackTrace(),             // onError handler
18    () -> System.out.println("Completed")         // onComplete handler
19);

Creating Observables from Common Sources

RxJava provides factory methods to create Observables from common data sources:

List[object Object]

Future[object Object]

Tutorial 2: Working with Schedulers

In this tutorial, we'll explore how to use Schedulers to control threading in RxJava applications, allowing you to perform operations on different threads.

Understanding Schedulers

Schedulers in RxJava determine which threads are used for executing operations and delivering results. They're crucial for managing concurrency and ensuring that operations don't block critical threads like the UI thread.

Key Scheduler Types

Controlling the Execution Thread

RxJava provides two main operators for controlling which threads are used:

• subscribeOn() - Specifies the Scheduler on which the Observable will operate
• observeOn() - Specifies the Scheduler on which the Observer will receive notifications

1import io.reactivex.rxjava3.core.Observable;
2import io.reactivex.rxjava3.schedulers.Schedulers;
3
4Observable.just("Long running operation")
5    .map(s -> {
6        System.out.println("Processing on thread: " + Thread.currentThread().getName());
7        // Simulate long-running operation
8        Thread.sleep(1000);
9        return s + " completed";
10    })
11    .subscribeOn(Schedulers.io())           // Specify thread for the source Observable
12    .observeOn(Schedulers.computation())    // Specify thread for downstream operations
13    .subscribe(
14        result -> System.out.println("Received on thread: " + 
15                    Thread.currentThread().getName() + ", Result: " + result),
16        Throwable::printStackTrace
17    );