RxJava for Reactive Android in Java

Introduction to RxJava
RxJava Basics and Setup
Observables and Observers
Operators and Transformation
Schedulers and Threading
Android-specific RxJava
Networking with RxJava
Database Operations
Error Handling
Testing RxJava
Best Practices

Introduction to RxJava

RxJava is a Java implementation of Reactive Extensions for composing asynchronous and event-based programs using observable sequences. It greatly simplifies asynchronous programming in Android applications.

Key Benefits for Android:

Simplified async operations: Handle background tasks elegantly
Functional programming: Use operators to transform, combine, and manipulate data
Error handling: Comprehensive error handling mechanisms
Thread management: Easy switching between threads
Memory leak prevention: Built-in lifecycle awareness with RxAndroid

RxJava Basics and Setup

1. Gradle Dependencies

// build.gradle (Module: app) dependencies { // RxJava 3 implementation 'io.reactivex.rxjava3:rxjava:3.1.8' implementation 'io.reactivex.rxjava3:rxandroid:3.0.2' // RxBinding for Android Views implementation 'com.jakewharton.rxbinding4:rxbinding:4.0.0' implementation 'com.jakewharton.rxbinding4:rxbinding-core:4.0.0' implementation 'com.jakewharton.rxbinding4:rxbinding-appcompat:4.0.0' // RxJava Adapter for Retrofit implementation 'com.squareup.retrofit2:adapter-rxjava3:2.9.0' // RxPermissions for runtime permissions implementation 'com.github.tbruyelle:rxpermissions:0.12' // Testing testImplementation 'io.reactivex.rxjava3:rxjava:3.1.8' testImplementation 'junit:junit:4.13.2' androidTestImplementation 'androidx.test.ext:junit:1.1.5' }

2. Basic RxJava Components

// RxJavaBasicExample.java public class RxJavaBasicExample { // 1. Observable - The data source that emits items // 2. Observer - The consumer that receives items // 3. Operators - Methods that transform, filter, or combine Observables // 4. Schedulers - Control which thread operations occur on public static void main(String[] args) { basicObservableExample(); operatorsExample(); threadingExample(); } private static void basicObservableExample() { System.out.println("=== Basic Observable Example ==="); // Create an Observable that emits a sequence of integers Observable<Integer> observable = Observable.create(emitter -> { try { emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); } catch (Exception e) { emitter.onError(e); } }); // Create an Observer to consume the emitted items Observer<Integer> observer = new Observer<Integer>() { @Override public void onSubscribe(@NonNull Disposable d) { System.out.println("Subscribed"); } @Override public void onNext(@NonNull Integer integer) { System.out.println("Received: " + integer); } @Override public void onError(@NonNull Throwable e) { System.out.println("Error: " + e.getMessage()); } @Override public void onComplete() { System.out.println("Completed"); } }; // Subscribe the observer to the observable observable.subscribe(observer); } private static void operatorsExample() { System.out.println("\n=== Operators Example ==="); Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) .filter(number -> number % 2 == 0) // Filter even numbers .map(number -> number * number) // Square each number .take(3) // Take only first 3 .subscribe( result -> System.out.println("Processed: " + result), error -> System.out.println("Error: " + error), () -> System.out.println("Operation completed") ); } private static void threadingExample() { System.out.println("\n=== Threading Example ==="); Observable.just("Hello", "World") .subscribeOn(Schedulers.io()) // Execute on IO thread .observeOn(Schedulers.computation()) // Observe on computation thread .doOnNext(item -> System.out.println("Processing on: " + Thread.currentThread().getName())) .observeOn(Schedulers.newThread()) // Switch to new thread .subscribe( item -> System.out.println("Received on: " + Thread.currentThread().getName() + " - " + item) ); // Sleep to allow async operations to complete try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } }

Observables and Observers

1. Different Types of Observables

// ObservableTypesExample.java public class ObservableTypesExample { public void demonstrateObservableTypes() { createObservable(); justObservable(); fromObservable(); rangeObservable(); intervalObservable(); deferObservable(); } private void createObservable() { System.out.println("=== Observable.create() ==="); Observable<String> observable = Observable.create(emitter -> { emitter.onNext("First item"); emitter.onNext("Second item"); emitter.onNext("Third item"); emitter.onComplete(); }); observable.subscribe( item -> System.out.println("Received: " + item), error -> System.out.println("Error: " + error), () -> System.out.println("Completed") ); } private void justObservable() { System.out.println("\n=== Observable.just() ==="); // Emits fixed set of items Observable.just("Apple", "Banana", "Cherry") .subscribe(item -> System.out.println("Fruit: " + item)); } private void fromObservable() { System.out.println("\n=== Observable.fromXXX() ==="); // From Iterable List<String> list = Arrays.asList("One", "Two", "Three"); Observable.fromIterable(list) .subscribe(item -> System.out.println("From list: " + item)); // From Array String[] array = {"A", "B", "C"}; Observable.fromArray(array) .subscribe(item -> System.out.println("From array: " + item)); // From Callable Observable.fromCallable(() -> "Result from callable") .subscribe(item -> System.out.println("From callable: " + item)); } private void rangeObservable() { System.out.println("\n=== Observable.range() ==="); Observable.range(1, 5) // Start from 1, emit 5 numbers .subscribe(number -> System.out.println("Number: " + number)); } private void intervalObservable() { System.out.println("\n=== Observable.interval() ==="); // Emits numbers at fixed time intervals Disposable disposable = Observable.interval(1, TimeUnit.SECONDS) .take(5) // Take only 5 emissions .subscribe( tick -> System.out.println("Tick: " + tick), error -> System.out.println("Error: " + error), () -> System.out.println("Interval completed") ); // In real Android app, you'd dispose in onDestroy() try { Thread.sleep(6000); // Wait for completion } catch (InterruptedException e) { e.printStackTrace(); } disposable.dispose(); } private void deferObservable() { System.out.println("\n=== Observable.defer() ==="); // Defer creation until subscription Observable<String> deferred = Observable.defer(() -> Observable.just("Current time: " + System.currentTimeMillis()) ); // Each subscription gets fresh data deferred.subscribe(time -> System.out.println("First: " + time)); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } deferred.subscribe(time -> System.out.println("Second: " + time)); } // Special Observable types public void specialObservables() { System.out.println("\n=== Special Observables ==="); // Single - emits exactly one item or error Single.just("Single result") .subscribe( result -> System.out.println("Single: " + result), error -> System.out.println("Single error: " + error) ); // Maybe - emits 0 or 1 item, optionally followed by completion Maybe.just("Maybe result") .subscribe( result -> System.out.println("Maybe: " + result), error -> System.out.println("Maybe error: " + error), () -> System.out.println("Maybe completed") ); // Completable - emits only completion or error signal Completable.complete() .subscribe( () -> System.out.println("Completable completed"), error -> System.out.println("Completable error: " + error) ); } }

2. Observer Patterns and Disposables

// ObserverPatternsExample.java public class ObserverPatternsExample { private CompositeDisposable compositeDisposable = new CompositeDisposable(); public void demonstrateObserverPatterns() { basicObserver(); resourceManagement(); hotVsColdObservables(); backpressureExample(); } private void basicObserver() { System.out.println("=== Basic Observer Patterns ==="); // Using lambda expressions for cleaner code Disposable disposable = Observable.range(1, 3) .subscribe( item -> System.out.println("Item: " + item), error -> System.err.println("Error: " + error), () -> System.out.println("Completed") ); compositeDisposable.add(disposable); } private void resourceManagement() { System.out.println("\n=== Resource Management ==="); // Using Disposable for resource cleanup Disposable disposable1 = Observable.interval(1, TimeUnit.SECONDS) .subscribe(tick -> System.out.println("Tick 1: " + tick)); Disposable disposable2 = Observable.interval(500, TimeUnit.MILLISECONDS) .subscribe(tick -> System.out.println("Tick 2: " + tick)); // Add to composite disposable for easy cleanup compositeDisposable.addAll(disposable1, disposable2); // Later, dispose all at once try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("Disposing all subscriptions..."); compositeDisposable.clear(); } private void hotVsColdObservables() { System.out.println("\n=== Hot vs Cold Observables ==="); // Cold Observable - data produced for each subscriber Observable<String> coldObservable = Observable.defer(() -> Observable.just("Data created at: " + System.currentTimeMillis()) ); System.out.println("Cold Observable:"); coldObservable.subscribe(data -> System.out.println("Subscriber 1: " + data)); coldObservable.subscribe(data -> System.out.println("Subscriber 2: " + data)); // Hot Observable - data produced independently of subscribers PublishSubject<String> hotObservable = PublishSubject.create(); System.out.println("\nHot Observable:"); hotObservable.subscribe(data -> System.out.println("Subscriber A: " + data)); // Emit data hotObservable.onNext("First emission"); hotObservable.onNext("Second emission"); // New subscriber joins late hotObservable.subscribe(data -> System.out.println("Subscriber B: " + data)); hotObservable.onNext("Third emission"); } private void backpressureExample() { System.out.println("\n=== Backpressure Handling ==="); // Using Flowable for backpressure support Flowable.range(1, 1000) .onBackpressureBuffer() // Handle backpressure by buffering .observeOn(Schedulers.computation()) .subscribe( item -> { try { Thread.sleep(10); // Simulate slow consumer } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("Processed: " + item); }, error -> System.err.println("Error: " + error) ); try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } } public void cleanup() { compositeDisposable.dispose(); } }

Operators and Transformation

1. Common RxJava Operators

// RxJavaOperatorsExample.java public class RxJavaOperatorsExample { public void demonstrateOperators() { transformationOperators(); filteringOperators(); combiningOperators(); utilityOperators(); errorHandlingOperators(); conditionalOperators(); } private void transformationOperators() { System.out.println("=== Transformation Operators ==="); // map - transform each item Observable.just(1, 2, 3) .map(number -> number * 2) .subscribe(result -> System.out.println("Map: " + result)); // flatMap - transform each item into an Observable Observable.just("Hello", "World") .flatMap(word -> Observable.fromArray(word.split(""))) .subscribe(letter -> System.out.println("FlatMap: " + letter)); // concatMap - like flatMap but preserves order Observable.just(1, 2, 3) .concatMap(number -> Observable.just(number * 10, number * 100)) .subscribe(result -> System.out.println("ConcatMap: " + result)); // switchMap - cancels previous emissions Observable.just(1, 2, 3) .switchMap(number -> Observable.timer(100 * number, TimeUnit.MILLISECONDS) .map(tick -> "SwitchMap: " + number)) .subscribe(result -> System.out.println(result)); // buffer - group items into lists Observable.range(1, 10) .buffer(3) .subscribe(batch -> System.out.println("Buffer: " + batch)); } private void filteringOperators() { System.out.println("\n=== Filtering Operators ==="); // filter - emit only items that pass a predicate Observable.range(1, 10) .filter(number -> number % 2 == 0) .subscribe(even -> System.out.println("Filter: " + even)); // take - take only first n items Observable.interval(1, TimeUnit.SECONDS) .take(3) .subscribe(item -> System.out.println("Take: " + item)); // skip - skip first n items Observable.range(1, 5) .skip(2) .subscribe(item -> System.out.println("Skip: " + item)); // distinct - remove duplicates Observable.just(1, 2, 2, 3, 3, 3, 4) .distinct() .subscribe(unique -> System.out.println("Distinct: " + unique)); // debounce - emit only after a quiet period Observable.create(emitter -> { emitter.onNext("A"); Thread.sleep(100); emitter.onNext("B"); Thread.sleep(400); emitter.onNext("C"); Thread.sleep(100); emitter.onNext("D"); emitter.onComplete(); }) .debounce(200, TimeUnit.MILLISECONDS) // Wait for 200ms of silence .subscribe(item -> System.out.println("Debounce: " + item)); } private void combiningOperators() { System.out.println("\n=== Combining Operators ==="); Observable<String> first = Observable.just("A", "B", "C"); Observable<String> second = Observable.just("1", "2", "3"); // merge - combine multiple Observables Observable.merge(first, second) .subscribe(item -> System.out.println("Merge: " + item)); // concat - combine sequentially Observable.concat(first, second) .subscribe(item -> System.out.println("Concat: " + item)); // zip - combine item-by-item Observable.zip(first, second, (f, s) -> f + s) .subscribe(combined -> System.out.println("Zip: " + combined)); // combineLatest - combine latest from each Observable.combineLatest( Observable.interval(1, TimeUnit.SECONDS).map(tick -> "Tick_" + tick), Observable.interval(2, TimeUnit.SECONDS).map(tick -> "Tock_" + tick), (tick, tock) -> tick + " | " + tock ) .take(5) .subscribe(combined -> System.out.println("CombineLatest: " + combined)); } private void utilityOperators() { System.out.println("\n=== Utility Operators ==="); // doOnNext - side effect for each emission Observable.just(1, 2, 3) .doOnNext(item -> System.out.println("About to emit: " + item)) .doOnSubscribe(d -> System.out.println("Subscribed")) .doOnComplete(() -> System.out.println("Completed")) .subscribe(item -> System.out.println("Emitted: " + item)); // delay - delay emissions Observable.just("Delayed") .delay(1, TimeUnit.SECONDS) .subscribe(item -> System.out.println("Delay: " + item)); // timeout - error if no emission in time Observable.timer(2, TimeUnit.SECONDS) .timeout(1, TimeUnit.SECONDS) .subscribe( item -> System.out.println("Timeout success"), error -> System.out.println("Timeout error: " + error) ); // retry - resubscribe on error AtomicInteger attempts = new AtomicInteger(); Observable.create(emitter -> { if (attempts.incrementAndGet() < 3) { emitter.onError(new RuntimeException("Simulated error")); } else { emitter.onNext("Success after retries"); emitter.onComplete(); } }) .retry(2) // Retry up to 2 times .subscribe( result -> System.out.println("Retry: " + result), error -> System.out.println("Retry failed: " + error) ); } private void errorHandlingOperators() { System.out.println("\n=== Error Handling Operators ==="); // onErrorReturn - return default value on error Observable.error(new RuntimeException("Error!")) .onErrorReturn(throwable -> "Default Value") .subscribe( item -> System.out.println("onErrorReturn: " + item), error -> System.out.println("This should not be called") ); // onErrorResumeNext - switch to another Observable on error Observable.error(new RuntimeException("Error!")) .onErrorResumeNext(throwable -> Observable.just("Recovered", "Values")) .subscribe(item -> System.out.println("onErrorResumeNext: " + item)); // onErrorReturnItem - simpler version Observable.error(new RuntimeException("Error!")) .onErrorReturnItem("Fallback Item") .subscribe(item -> System.out.println("onErrorReturnItem: " + item)); } private void conditionalOperators() { System.out.println("\n=== Conditional Operators ==="); // all - check if all items meet condition Observable.just(2, 4, 6, 8) .all(number -> number % 2 == 0) .subscribe(result -> System.out.println("All even: " + result)); // any - check if any item meets condition Observable.just(1, 3, 5, 8) .any(number -> number % 2 == 0) .subscribe(result -> System.out.println("Any even: " + result)); // contains - check if sequence contains item Observable.just("Apple", "Banana", "Cherry") .contains("Banana") .subscribe(result -> System.out.println("Contains Banana: " + result)); // defaultIfEmpty - emit default if empty Observable.empty() .defaultIfEmpty("Default Value") .subscribe(item -> System.out.println("defaultIfEmpty: " + item)); // sequenceEqual - check if two Observables emit same sequence Observable<Integer> first = Observable.just(1, 2, 3); Observable<Integer> second = Observable.just(1, 2, 3); Observable.sequenceEqual(first, second) .subscribe(equal -> System.out.println("Sequences equal: " + equal)); } }

Schedulers and Threading

1. Scheduler Management

// SchedulerExample.java public class SchedulerExample { private CompositeDisposable disposables = new CompositeDisposable(); public void demonstrateSchedulers() { basicSchedulers(); observeOnVsSubscribeOn(); androidSchedulers(); customSchedulers(); } private void basicSchedulers() { System.out.println("=== Basic Schedulers ==="); // Schedulers.io() - For I/O bound work Disposable ioWork = Observable.fromCallable(() -> { System.out.println("IO work on: " + Thread.currentThread().getName()); Thread.sleep(1000); // Simulate I/O operation return "IO Result"; }) .subscribeOn(Schedulers.io()) .observeOn(Schedulers.single()) .subscribe(result -> System.out.println("IO result on: " + Thread.currentThread().getName() + " - " + result) ); // Schedulers.computation() - For CPU-intensive work Disposable computationWork = Observable.range(1, 5) .subscribeOn(Schedulers.computation()) .map(number -> { System.out.println("Computation on: " + Thread.currentThread().getName()); return number * number; }) .observeOn(Schedulers.newThread()) .subscribe(result -> System.out.println("Computation result on: " + Thread.currentThread().getName() + " - " + result) ); disposables.addAll(ioWork, computationWork); } private void observeOnVsSubscribeOn() { System.out.println("\n=== observeOn vs subscribeOn ==="); Disposable example = Observable.create(emitter -> { System.out.println("Emit on: " + Thread.currentThread().getName()); emitter.onNext(1); emitter.onNext(2); emitter.onNext(3); emitter.onComplete(); }) .subscribeOn(Schedulers.io()) // Affects upstream operations .observeOn(Schedulers.computation()) // Affects downstream operations .map(item -> { System.out.println("Map on: " + Thread.currentThread().getName()); return item * 10; }) .observeOn(Schedulers.single()) .subscribe(item -> System.out.println("Receive on: " + Thread.currentThread().getName() + " - " + item) ); disposables.add(example); } private void androidSchedulers() { System.out.println("\n=== Android Schedulers ==="); // Note: In Android, you'd use AndroidSchedulers.mainThread() // from RxAndroid library // Simulating Android main thread with Schedulers.single() Disposable androidExample = Observable.just("Network Result") .subscribeOn(Schedulers.io()) // Network call on IO thread .observeOn(Schedulers.single()) // Simulate Android main thread .doOnNext(result -> System.out.println("Updating UI on: " + Thread.currentThread().getName()) ) .subscribe(result -> System.out.println("UI updated with: " + result) ); disposables.add(androidExample); } private void customSchedulers() { System.out.println("\n=== Custom Schedulers ==="); // Create custom scheduler from executor ExecutorService customExecutor = Executors.newFixedThreadPool(3); Scheduler customScheduler = Schedulers.from(customExecutor); Disposable customWork = Observable.range(1, 6) .subscribeOn(customScheduler) .map(item -> { System.out.println("Custom scheduler: " + Thread.currentThread().getName()); return item; }) .subscribe(); disposables.add(customWork); // Shutdown executor when done disposables.add(Disposable.fromRunnable(() -> { customExecutor.shutdown(); System.out.println("Custom executor shutdown"); })); } public void cleanup() { disposables.dispose(); } }

2. Threading Best Practices

// ThreadingBestPractices.java public class ThreadingBestPractices { private CompositeDisposable disposables = new CompositeDisposable(); public void demonstrateBestPractices() { avoidBlockingOperations(); properSchedulerUsage(); memoryLeakPrevention(); errorHandlingInThreads(); } private void avoidBlockingOperations() { System.out.println("=== Avoid Blocking Operations ==="); // ❌ Bad: Blocking operation on computation scheduler Disposable badExample = Observable.just("data") .subscribeOn(Schedulers.computation()) .map(data -> { try { Thread.sleep(2000); // Blocking operation! } catch (InterruptedException e) { e.printStackTrace(); } return data.toUpperCase(); }) .subscribe(result -> System.out.println("Bad result: " + result)); // ✅ Good: Use proper scheduler for blocking operations Disposable goodExample = Observable.just("data") .subscribeOn(Schedulers.io()) // Use IO scheduler for blocking ops .map(data -> { try { Thread.sleep(2000); // This is okay on IO scheduler } catch (InterruptedException e) { e.printStackTrace(); } return data.toUpperCase(); }) .observeOn(Schedulers.computation()) // Switch for computation .subscribe(result -> System.out.println("Good result: " + result)); disposables.addAll(badExample, goodExample); } private void properSchedulerUsage() { System.out.println("\n=== Proper Scheduler Usage ==="); Disposable example = Observable.range(1, 100) .subscribeOn(Schedulers.io()) // Heavy initial work on IO .filter(number -> number % 2 == 0) .observeOn(Schedulers.computation()) // Computation for CPU work .map(number -> number * number) .buffer(10) .observeOn(Schedulers.single()) // Single thread for final processing .subscribe( batch -> System.out.println("Batch on " + Thread.currentThread().getName() + ": " + batch), error -> System.err.println("Error: " + error), () -> System.out.println("Processing completed") ); disposables.add(example); } private void memoryLeakPrevention() { System.out.println("\n=== Memory Leak Prevention ==="); // Using CompositeDisposable for easy cleanup CompositeDisposable localDisposables = new CompositeDisposable(); // Multiple subscriptions Disposable sub1 = Observable.interval(1, TimeUnit.SECONDS) .subscribe(tick -> System.out.println("Sub1: " + tick)); Disposable sub2 = Observable.interval(500, TimeUnit.MILLISECONDS) .subscribe(tick -> System.out.println("Sub2: " + tick)); localDisposables.addAll(sub1, sub2); // Simulate activity/fragment destruction System.out.println("Cleaning up local disposables..."); localDisposables.dispose(); } private void errorHandlingInThreads() { System.out.println("\n=== Error Handling in Threads ==="); Disposable example = Observable.create(emitter -> { // This runs on IO scheduler if (Math.random() > 0.5) { emitter.onError(new RuntimeException("Random error!")); } else { emitter.onNext("Success"); emitter.onComplete(); } }) .subscribeOn(Schedulers.io()) .observeOn(Schedulers.single()) .subscribe( result -> System.out.println("Success: " + result), error -> System.err.println("Error handled on: " + Thread.currentThread().getName() + " - " + error.getMessage()) ); disposables.add(example); } public void cleanup() { disposables.dispose(); } }

Android-specific RxJava

1. RxJava in Android Activities

// MainActivity.java public class MainActivity extends AppCompatActivity { private CompositeDisposable disposables = new CompositeDisposable(); private TextView resultTextView; private ProgressBar progressBar; private EditText searchEditText; @Override protected void onCreate(Bundle savedInstanceState) { super.onCreate(savedInstanceState); setContentView(R.layout.activity_main); initializeViews(); setupClickListeners(); setupSearchListener(); loadInitialData(); } private void initializeViews() { resultTextView = findViewById(R.id.resultTextView); progressBar = findViewById(R.id.progressBar); searchEditText = findViewById(R.id.searchEditText); } private void setupClickListeners() { Button loadButton = findViewById(R.id.loadButton); Button parallelButton = findViewById(R.id.parallelButton); // Single network call disposables.add( RxView.clicks(loadButton) .throttleFirst(1, TimeUnit.SECONDS) // Prevent multiple clicks .observeOn(AndroidSchedulers.mainThread()) .doOnNext(click -> showLoading(true)) .observeOn(Schedulers.io()) .flatMapSingle(click -> performNetworkCall()) .observeOn(AndroidSchedulers.mainThread()) .subscribe( result -> { showLoading(false); resultTextView.setText("Result: " + result); }, error -> { showLoading(false); resultTextView.setText("Error: " + error.getMessage()); } ) ); // Parallel network calls disposables.add( RxView.clicks(parallelButton) .throttleFirst(1, TimeUnit.SECONDS) .observeOn(AndroidSchedulers.mainThread()) .doOnNext(click -> showLoading(true)) .observeOn(Schedulers.io()) .flatMap(click -> Observable.merge( performNetworkCall("Call 1").toObservable(), performNetworkCall("Call 2").toObservable(), performNetworkCall("Call 3").toObservable() )) .observeOn(AndroidSchedulers.mainThread()) .subscribe( result -> { resultTextView.append("\nParallel: " + result); }, error -> { showLoading(false); resultTextView.setText("Parallel Error: " + error.getMessage()); }, () -> { showLoading(false); resultTextView.append("\nAll parallel calls completed"); } ) ); } private void setupSearchListener() { // Real-time search with debounce disposables.add( RxTextView.textChanges(searchEditText) .debounce(300, TimeUnit.MILLISECONDS) // Wait for user to stop typing .filter(text -> text.length() > 2) // Only search if more than 2 chars .distinctUntilChanged() // Only if text changed .switchMapSingle(query -> performSearch(query.toString())) .observeOn(AndroidSchedulers.mainThread()) .subscribe( results -> resultTextView.setText("Search results: " + results), error -> resultTextView.setText("Search error: " + error.getMessage()) ) ); } private void loadInitialData() { // Load data when activity starts disposables.add( performNetworkCall("Initial Data") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe( result -> resultTextView.setText("Initial: " + result), error -> resultTextView.setText("Initial load error: " + error.getMessage()) ) ); } private Single<String> performNetworkCall() { return performNetworkCall("Default"); } private Single<String> performNetworkCall(String callName) { return Single.fromCallable(() -> { // Simulate network call Thread.sleep(2000); if (Math.random() > 0.8) { throw new RuntimeException(callName + " failed!"); } return callName + " completed at " + System.currentTimeMillis(); }); } private Single<String> performSearch(String query) { return Single.fromCallable(() -> { // Simulate search API call Thread.sleep(500); return "Results for '" + query + "': " + (int)(Math.random() * 100); }).subscribeOn(Schedulers.io()); } private void showLoading(boolean show) { progressBar.setVisibility(show ? View.VISIBLE : View.GONE); } @Override protected void onDestroy() { super.onDestroy(); disposables.dispose(); // Prevent memory leaks } }

2. RxBinding for Android Views

// RxBindingExample.java public class RxBindingExample { private CompositeDisposable disposables = new CompositeDisposable(); public void setupViewBindings(Activity activity) { setupButtonClicks(activity); setupTextChanges(activity); setupCheckboxChanges(activity); setupRecyclerViewScroll(activity); setupMultipleViews(activity); } private void setupButtonClicks(Activity activity) { Button button = activity.findViewById(R.id.someButton); disposables.add( RxView.clicks(button) .throttleFirst(1, TimeUnit.SECONDS) // Prevent double clicks .observeOn(AndroidSchedulers.mainThread()) .subscribe(click -> { // Handle button click showToast(activity, "Button clicked!"); }) ); // Long clicks disposables.add( RxView.longClicks(button) .observeOn(AndroidSchedulers.mainThread()) .subscribe(click -> { showToast(activity, "Button long clicked!"); }) ); } private void setupTextChanges(Activity activity) { EditText editText = activity.findViewById(R.id.someEditText); disposables.add( RxTextView.textChanges(editText) .debounce(300, TimeUnit.MILLISECONDS) // Wait for typing to stop .filter(text -> text.length() > 0) // Ignore empty text .distinctUntilChanged() // Only emit when text actually changes .observeOn(AndroidSchedulers.mainThread()) .subscribe(text -> { // Handle text changes updateSearchResults(activity, text.toString()); }) ); // Focus changes disposables.add( RxView.focusChanges(editText) .skip(1) // Skip initial emission .observeOn(AndroidSchedulers.mainThread()) .subscribe(hasFocus -> { if (hasFocus) { // EditText gained focus showToast(activity, "EditText focused"); } else { // EditText lost focus showToast(activity, "EditText lost focus"); } }) ); } private void setupCheckboxChanges(Activity activity) { CheckBox checkBox = activity.findViewById(R.id.someCheckBox); disposables.add( RxCompoundButton.checkedChanges(checkBox) .skip(1) // Skip initial value .observeOn(AndroidSchedulers.mainThread()) .subscribe(isChecked -> { // Handle checkbox state change String state = isChecked ? "checked" : "unchecked"; showToast(activity, "Checkbox " + state); }) ); } private void setupRecyclerViewScroll(Activity activity) { RecyclerView recyclerView = activity.findViewById(R.id.someRecyclerView); disposables.add( RxRecyclerView.scrollStateChanges(recyclerView) .observeOn(AndroidSchedulers.mainThread()) .subscribe(state -> { switch (state) { case RecyclerView.SCROLL_STATE_DRAGGING: // User is dragging break; case RecyclerView.SCROLL_STATE_SETTLING: // Scroll is settling break; case RecyclerView.SCROLL_STATE_IDLE: // Scroll is idle checkLoadMore(activity); break; } }) ); // Scroll events with more details disposables.add( RxRecyclerView.scrollEvents(recyclerView) .observeOn(AndroidSchedulers.mainThread()) .subscribe(event -> { // Handle scroll events with more details int dx = event.dx(); int dy = event.dy(); // Process scroll data... }) ); } private void setupMultipleViews(Activity activity) { Button button1 = activity.findViewById(R.id.button1); Button button2 = activity.findViewById(R.id.button2); Button button3 = activity.findViewById(R.id.button3); // Combine multiple button clicks Observable<Object> button1Clicks = RxView.clicks(button1); Observable<Object> button2Clicks = RxView.clicks(button2); Observable<Object> button3Clicks = RxView.clicks(button3); disposables.add( Observable.merge(button1Clicks, button2Clicks, button3Clicks) .throttleFirst(500, TimeUnit.MILLISECONDS) .observeOn(AndroidSchedulers.mainThread()) .subscribe(click -> { // Handle any button click showToast(activity, "A button was clicked!"); }) ); // Combine specific conditions EditText emailEditText = activity.findViewById(R.id.emailEditText); EditText passwordEditText = activity.findViewById(R.id.passwordEditText); Button submitButton = activity.findViewById(R.id.submitButton); disposables.add( Observable.combineLatest( RxTextView.textChanges(emailEditText), RxTextView.textChanges(passwordEditText), (email, password) -> { // Validate form boolean emailValid = isValidEmail(email.toString()); boolean passwordValid = password.length() >= 6; return emailValid && passwordValid; } ) .observeOn(AndroidSchedulers.mainThread()) .subscribe(isValid -> { // Enable/disable submit button based on validation submitButton.setEnabled(isValid); }) ); } private void updateSearchResults(Activity activity, String query) { // Implement search logic TextView resultsView = activity.findViewById(R.id.resultsTextView); resultsView.setText("Searching for: " + query); // You would typically make an API call here } private void checkLoadMore(Activity activity) { // Implement load more logic for RecyclerView RecyclerView recyclerView = activity.findViewById(R.id.someRecyclerView); LinearLayoutManager layoutManager = (LinearLayoutManager) recyclerView.getLayoutManager(); if (layoutManager != null) { int visibleItemCount = layoutManager.getChildCount(); int totalItemCount = layoutManager.getItemCount(); int firstVisibleItemPosition = layoutManager.findFirstVisibleItemPosition(); if ((visibleItemCount + firstVisibleItemPosition) >= totalItemCount && firstVisibleItemPosition >= 0) { // Load more data loadMoreData(activity); } } } private void loadMoreData(Activity activity) { // Implement load more data logic showToast(activity, "Loading more data..."); } private boolean isValidEmail(String email) { return email != null && android.util.Patterns.EMAIL_ADDRESS.matcher(email).matches(); } private void showToast(Activity activity, String message) { Toast.makeText(activity, message, Toast.LENGTH_SHORT).show(); } public void cleanup() { disposables.dispose(); } }

Networking with RxJava

1. Retrofit with RxJava

// ApiService.java public interface ApiService { @GET("users/{userId}") Single<User> getUser(@Path("userId") String userId); @GET("users") Single<List<User>> getUsers(); @POST("users") Single<User> createUser(@Body User user); @GET("posts") Single<List<Post>> getUserPosts(@Query("userId") String userId); @GET("users/{userId}/posts") Single<List<Post>> getUserPostsWithPath(@Path("userId") String userId); @Multipart @POST("upload") Single<UploadResponse> uploadFile( @Part MultipartBody.Part file, @Part("description") RequestBody description ); @Streaming @GET("files/{fileName}") Single<ResponseBody> downloadFile(@Path("fileName") String fileName); } // User.java public class User { private String id; private String name; private String email; private String avatar; // Constructors, getters, setters public User() {} public User(String id, String name, String email) { this.id = id; this.name = name; this.email = email; } // Getters and setters... public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getEmail() { return email; } public void setEmail(String email) { this.email = email; } public String getAvatar() { return avatar; } public void setAvatar(String avatar) { this.avatar = avatar; } } // Post.java public class Post { private String id; private String userId; private String title; private String body; // Constructors, getters, setters... public Post() {} public Post(String id, String userId, String title, String body) { this.id = id; this.userId = userId; this.title = title; this.body = body; } // Getters and setters... public String getId() { return id; } public void setId(String id) { this.id = id; } public String getUserId() { return userId; } public void setUserId(String userId) { this.userId = userId; } public String getTitle() { return title; } public void setTitle(String title) { this.title = title; } public String getBody() { return body; } public void setBody(String body) { this.body = body; } }

2. Network Repository Pattern

// UserRepository.java public class UserRepository { private final ApiService apiService; private final CompositeDisposable disposables = new CompositeDisposable(); public UserRepository(ApiService apiService) { this.apiService = apiService; } // Basic CRUD operations public Single<User> getUser(String userId) { return apiService.getUser(userId) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } public Single<List<User>> getUsers() { return apiService.getUsers() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } public Single<User> createUser(User user) { return apiService.createUser(user) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } // Complex operations public Single<User> getUserWithPosts(String userId) { return apiService.getUser(userId) .flatMap(user -> apiService.getUserPosts(userId) .map(posts -> { user.setPosts(posts); // Assuming User has posts field return user; }) ) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } // Parallel operations public Single<List<User>> getUsersWithPosts() { return apiService.getUsers() .flatMapObservable(Observable::fromIterable) .flatMapSingle(user -> apiService.getUserPosts(user.getId()) .map(posts -> { user.setPosts(posts); return user; }) .onErrorReturnItem(user) // Return user even if posts fail ) .toList() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } // Batch operations public Completable updateMultipleUsers(List<User> users) { return Observable.fromIterable(users) .flatMapCompletable(user -> apiService.createUser(user) .ignoreElement() // Convert Single to Completable .onErrorComplete() // Continue on error ) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } // Cached network calls public Observable<User> getUserWithCache(String userId) { return Observable.concat( getCachedUser(userId).toObservable(), apiService.getUser(userId) .doOnSuccess(this::cacheUser) .toObservable() ) .firstElement() // Take first non-empty emission .toObservable() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } private Single<User> getCachedUser(String userId) { // Implement cache retrieval return Single.error(new RuntimeException("Not in cache")); // Simplified } private void cacheUser(User user) { // Implement cache storage } // Error handling with retry public Single<User> getUserWithRetry(String userId) { return apiService.getUser(userId) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .retryWhen(errors -> errors.zipWith(Observable.range(1, 3), (error, retryCount) -> { if (retryCount < 3 && isNetworkError(error)) { return retryCount; } throw Exceptions.propagate(error); }) .flatMap(retryCount -> Observable.timer((long) Math.pow(2, retryCount), TimeUnit.SECONDS) ) ); } private boolean isNetworkError(Throwable error) { return error instanceof IOException || error instanceof SocketTimeoutException || error instanceof ConnectException; } public void cleanup() { disposables.dispose(); } }

3. Network ViewModel

// UserViewModel.java public class UserViewModel extends ViewModel { private final UserRepository userRepository; private final CompositeDisposable disposables = new CompositeDisposable(); // LiveData for UI observation private final MutableLiveData<List<User>> usersLiveData = new MutableLiveData<>(); private final MutableLiveData<User> userLiveData = new MutableLiveData<>(); private final MutableLiveData<Boolean> loadingLiveData = new MutableLiveData<>(); private final MutableLiveData<String> errorLiveData = new MutableLiveData<>(); public UserViewModel(UserRepository userRepository) { this.userRepository = userRepository; } public void loadUsers() { loadingLiveData.setValue(true); disposables.add( userRepository.getUsers() .subscribe( users -> { loadingLiveData.setValue(false); usersLiveData.setValue(users); errorLiveData.setValue(null); }, error -> { loadingLiveData.setValue(false); errorLiveData.setValue("Failed to load users: " + error.getMessage()); } ) ); } public void loadUser(String userId) { loadingLiveData.setValue(true); disposables.add( userRepository.getUserWithPosts(userId) .subscribe( user -> { loadingLiveData.setValue(false); userLiveData.setValue(user); errorLiveData.setValue(null); }, error -> { loadingLiveData.setValue(false); errorLiveData.setValue("Failed to load user: " + error.getMessage()); } ) ); } public void createUser(User user) { loadingLiveData.setValue(true); disposables.add( userRepository.createUser(user) .subscribe( createdUser -> { loadingLiveData.setValue(false); // Update users list or navigate loadUsers(); // Refresh the list }, error -> { loadingLiveData.setValue(false); errorLiveData.setValue("Failed to create user: " + error.getMessage()); } ) ); } public void searchUsers(String query) { if (query == null || query.trim().isEmpty()) { loadUsers(); return; } loadingLiveData.setValue(true); disposables.add( userRepository.getUsers() .flatMapObservable(Observable::fromIterable) .filter(user -> user.getName().toLowerCase().contains(query.toLowerCase()) || user.getEmail().toLowerCase().contains(query.toLowerCase()) ) .toList() .subscribe( filteredUsers -> { loadingLiveData.setValue(false); usersLiveData.setValue(filteredUsers); }, error -> { loadingLiveData.setValue(false); errorLiveData.setValue("Search failed: " + error.getMessage()); } ) ); } // LiveData getters public LiveData<List<User>> getUsersLiveData() { return usersLiveData; } public LiveData<User> getUserLiveData() { return userLiveData; } public LiveData<Boolean> getLoadingLiveData() { return loadingLiveData; } public LiveData<String> getErrorLiveData() { return errorLiveData; } @Override protected void onCleared() { super.onCleared(); disposables.dispose(); } }

Database Operations

1. Room Database with RxJava

// UserDao.java @Dao public interface UserDao { @Insert Completable insert(User user); @Insert Completable insertAll(List<User> users); @Update Completable update(User user); @Delete Completable delete(User user); @Query("SELECT * FROM users") Flowable<List<User>> getAllUsers(); @Query("SELECT * FROM users WHERE id = :userId") Maybe<User> getUserById(String userId); @Query("SELECT * FROM users WHERE name LIKE :searchQuery") Flowable<List<User>> searchUsers(String searchQuery); @Query("SELECT * FROM users ORDER BY name ASC") Flowable<List<User>> getUsersSortedByName(); @Query("DELETE FROM users") Completable deleteAllUsers(); } // AppDatabase.java @Database(entities = {User.class, Post.class}, version = 1) public abstract class AppDatabase extends RoomDatabase { public abstract UserDao userDao(); public abstract PostDao postDao(); } // LocalUserRepository.java public class LocalUserRepository { private final UserDao userDao; private final CompositeDisposable disposables = new CompositeDisposable(); public LocalUserRepository(UserDao userDao) { this.userDao = userDao; } // Basic operations public Completable insertUser(User user) { return userDao.insert(user) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } public Completable insertUsers(List<User> users) { return userDao.insertAll(users) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } public Flowable<List<User>> getAllUsers() { return userDao.getAllUsers() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } public Maybe<User> getUserById(String userId) { return userDao.getUserById(userId) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } // Complex operations public Completable saveUserWithPosts(User user, List<Post> posts) { return Completable.mergeArray( userDao.insert(user), // Assuming you have a PostDao with insertAll method Completable.fromAction(() -> { // Save posts logic }) ) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } public Flowable<List<User>> searchUsers(String query) { return userDao.searchUsers("%" + query + "%") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .onErrorReturnItem(Collections.emptyList()); // Return empty list on error } // Sync operations public Completable syncUsers(List<User> users) { return userDao.deleteAllUsers() .andThen(userDao.insertAll(users)) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } public void cleanup() { disposables.dispose(); } }

Error Handling

1. Comprehensive Error Handling

// ErrorHandlingExample.java public class ErrorHandlingExample { private CompositeDisposable disposables = new CompositeDisposable(); public void demonstrateErrorHandling() { basicErrorHandling(); retryStrategies(); errorRecovery(); globalErrorHandler(); } private void basicErrorHandling() { System.out.println("=== Basic Error Handling ==="); Observable<String> errorObservable = Observable.create(emitter -> { emitter.onNext("First item"); emitter.onError(new RuntimeException("Something went wrong!")); }); disposables.add( errorObservable .subscribe( item -> System.out.println("Received: " + item), error -> System.err.println("Error handled: " + error.getMessage()), () -> System.out.println("Completed") // This won't be called ) ); } private void retryStrategies() { System.out.println("\n=== Retry Strategies ==="); AtomicInteger attempts = new AtomicInteger(); Observable<String> flakyObservable = Observable.create(emitter -> { int attempt = attempts.incrementAndGet(); if (attempt < 3) { emitter.onError(new RuntimeException("Failed on attempt " + attempt)); } else { emitter.onNext("Success on attempt " + attempt); emitter.onComplete(); } }); // Simple retry disposables.add( flakyObservable .retry(2) // Retry up to 2 times .subscribe( result -> System.out.println("Simple retry: " + result), error -> System.err.println("Simple retry failed: " + error.getMessage()) ) ); // Retry with delay disposables.add( flakyObservable .retryWhen(errors -> errors.zipWith(Observable.range(1, 3), (error, retryCount) -> { if (retryCount > 3) { throw Exceptions.propagate(error); } return retryCount; }) .flatMap(retryCount -> Observable.timer(retryCount, TimeUnit.SECONDS) ) ) .subscribe( result -> System.out.println("Delayed retry: " + result), error -> System.err.println("Delayed retry failed: " + error.getMessage()) ) ); } private void errorRecovery() { System.out.println("\n=== Error Recovery ==="); Observable<String> unreliableObservable = Observable.create(emitter -> { if (Math.random() > 0.5) { emitter.onNext("Success"); emitter.onComplete(); } else { emitter.onError(new RuntimeException("Random failure")); } }); // Return default value on error disposables.add( unreliableObservable .onErrorReturnItem("Default Value") .subscribe(result -> System.out.println("Error return: " + result)) ); // Switch to fallback Observable disposables.add( unreliableObservable .onErrorResumeNext(throwable -> Observable.just("Fallback", "Values") ) .subscribe(result -> System.out.println("Error resume: " + result)) ); // Continue with next item Observable<String> source = Observable.just("First", "Second", "Third") .flatMap(item -> { if ("Second".equals(item)) { return Observable.error(new RuntimeException("Failed on second")); } return Observable.just(item); }); disposables.add( source .onErrorResumeNext(Observable.empty()) // Continue with empty .subscribe( item -> System.out.println("Continue: " + item), error -> System.err.println("This shouldn't be called"), () -> System.out.println("Completed despite errors") ) ); } private void globalErrorHandler() { System.out.println("\n=== Global Error Handler ==="); RxJavaPlugins.setErrorHandler(throwable -> { // Global error handler for uncaught exceptions System.err.println("Global error handler: " + throwable.getMessage()); // You could log to Crashlytics, show notification, etc. }); // This error won't crash the app due to global handler Observable.error(new RuntimeException("Unhandled error")) .subscribe(); } public void cleanup() { disposables.dispose(); } }

Testing RxJava

1. RxJava Testing

// RxJavaTestExample.java public class RxJavaTestExample { private TestScheduler testScheduler; @Before public void setup() { testScheduler = new TestScheduler(); } @Test public void testBasicObservable() { // Given Observable<String> observable = Observable.just("Hello", "World"); TestObserver<String> testObserver = new TestObserver<>(); // When observable.subscribe(testObserver); // Then testObserver.assertValues("Hello", "World"); testObserver.assertComplete(); testObserver.assertNoErrors(); } @Test public void testErrorHandling() { // Given Observable<String> observable = Observable.error(new RuntimeException("Test error")); TestObserver<String> testObserver = new TestObserver<>(); // When observable.subscribe(testObserver); // Then testObserver.assertError(RuntimeException.class); testObserver.assertErrorMessage("Test error"); } @Test public void testTimeBasedObservable() { // Given Observable<Long> observable = Observable.interval(1, TimeUnit.SECONDS, testScheduler) .take(3); TestObserver<Long> testObserver = new TestObserver<>(); // When observable.subscribe(testObserver); // Then - no values yet testObserver.assertNoValues(); // Advance time testScheduler.advanceTimeBy(2, TimeUnit.SECONDS); testObserver.assertValues(0L, 1L); testScheduler.advanceTimeBy(1, TimeUnit.SECONDS); testObserver.assertValues(0L, 1L, 2L); testObserver.assertComplete(); } @Test public void testAsyncOperation() { // Given Observable<String> observable = Observable.fromCallable(() -> { Thread.sleep(1000); // Simulate async work return "Result"; }).subscribeOn(Schedulers.trampoline()); // Use trampoline for testing TestObserver<String> testObserver = new TestObserver<>(); // When observable.subscribe(testObserver); // Then testObserver.assertValue("Result"); testObserver.assertComplete(); } @Test public void testOperatorChain() { // Given Observable<Integer> observable = Observable.range(1, 5) .filter(number -> number % 2 == 0) .map(number -> number * 10); TestObserver<Integer> testObserver = new TestObserver<>(); // When observable.subscribe(testObserver); // Then testObserver.assertValues(20, 40); testObserver.assertComplete(); } }

Best Practices

1. RxJava Best Practices

// RxJavaBestPractices.java public class RxJavaBestPractices { private CompositeDisposable disposables = new CompositeDisposable(); public void demonstrateBestPractices() { useCompositeDisposable(); avoidMemoryLeaks(); properErrorHandling(); useAppropriateSchedulers(); avoidOverusingRxJava(); useDisposableMaybe(); } private void useCompositeDisposable() { System.out.println("=== Use CompositeDisposable ==="); CompositeDisposable localDisposables = new CompositeDisposable(); // Add multiple disposables localDisposables.add( Observable.interval(1, TimeUnit.SECONDS) .subscribe(tick -> System.out.println("Tick: " + tick)) ); localDisposables.add( Observable.just("Data") .subscribe(data -> System.out.println("Data: " + data)) ); // Clean up all at once localDisposables.dispose(); } private void avoidMemoryLeaks() { System.out.println("\n=== Avoid Memory Leaks ==="); // ❌ Bad: Never disposing subscriptions Observable.interval(1, TimeUnit.SECONDS) .subscribe(tick -> System.out.println("Leaky: " + tick)); // ✅ Good: Proper disposal Disposable disposable = Observable.interval(1, TimeUnit.SECONDS) .subscribe(tick -> System.out.println("Safe: " + tick)); disposables.add(disposable); } private void properErrorHandling() { System.out.println("\n=== Proper Error Handling ==="); // ❌ Bad: Ignoring errors Observable.error(new RuntimeException("Error")) .subscribe( item -> System.out.println("Received: " + item) // Missing error handler! ); // ✅ Good: Always handle errors Observable.error(new RuntimeException("Error")) .subscribe( item -> System.out.println("Received: " + item), error -> System.err.println("Error handled: " + error.getMessage()) ); // Even better: Use onErrorReturn or onErrorResumeNext Observable.error(new RuntimeException("Error")) .onErrorReturnItem("Fallback") .subscribe(item -> System.out.println("Result: " + item)); } private void useAppropriateSchedulers() { System.out.println("\n=== Use Appropriate Schedulers ==="); // I/O operations disposables.add( Observable.fromCallable(() -> { // Database or file operation return "I/O Result"; }) .subscribeOn(Schedulers.io()) .subscribe() ); // Computation operations disposables.add( Observable.range(1, 1000) .subscribeOn(Schedulers.computation()) .map(number -> number * number) // CPU-intensive .subscribe() ); // UI updates (in Android) disposables.add( Observable.just("Data") .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(data -> { // Update UI System.out.println("UI update: " + data); }) ); } private void avoidOverusingRxJava() { System.out.println("\n=== Avoid Overusing RxJava ==="); // ❌ Bad: Using RxJava for simple synchronous operations Observable.just("Hello") .map(String::toUpperCase) .subscribe(result -> System.out.println("Overkill: " + result)); // ✅ Good: Use regular Java for simple operations String result = "Hello".toUpperCase(); System.out.println("Simple: " + result); // Use RxJava when you need its benefits: // - Asynchronous operations // - Complex data transformations // - Combining multiple data sources // - Error handling across async boundaries } private void useDisposableMaybe() { System.out.println("\n=== Use Disposable Maybe ==="); // For operations that might not complete immediately Maybe<String> maybeOperation = Maybe.create(emitter -> { if (Math.random() > 0.5) { emitter.onSuccess("Success"); } else { emitter.onComplete(); // No result, but no error either } }); disposables.add( maybeOperation .subscribe( result -> System.out.println("Maybe success: " + result), error -> System.err.println("Maybe error: " + error), () -> System.out.println("Maybe completed without result") ) ); } // Additional best practices public void additionalBestPractices() { useMeaningfulVariableNames(); avoidNestedSubscriptions(); useOperatorsEffectively(); testYourRxCode(); } private void useMeaningfulVariableNames() { // ❌ Bad: Unclear variable names Observable<String> o = Observable.just("data"); Disposable d = o.subscribe(); // ✅ Good: Descriptive names Observable<String> userDataStream = Observable.just("user data"); Disposable userSubscription = userDataStream.subscribe(); disposables.add(userSubscription); } private void avoidNestedSubscriptions() { // ❌ Bad: Nested subscriptions (callback hell in RxJava) Observable.just("outer") .subscribe(outer -> { Observable.just("inner") .subscribe(inner -> System.out.println(outer + " - " + inner)); }); // ✅ Good: Use flatMap or other operators Observable.just("outer") .flatMap(outer -> Observable.just("inner") .map(inner -> outer + " - " + inner) ) .subscribe(result -> System.out.println(result)); } private void useOperatorsEffectively() { // Chain operators for better readability disposables.add( Observable.range(1, 100) .filter(number -> number % 2 == 0) // Filter even numbers .map(number -> number * 2) // Double them .take(10) // Take first 10 .reduce(0, Integer::sum) // Sum them up .subscribe(total -> System.out.println("Total: " + total)) ); } private void testYourRxCode() { // Always test your RxJava code, especially: // - Async operations // - Error handling // - Operator chains // - Threading behavior } public void cleanup() { disposables.dispose(); } }

Conclusion

RxJava is a powerful tool for Android development that can greatly simplify asynchronous programming and data stream handling. Key takeaways:

Use appropriate schedulers for different types of work
Always handle errors and use proper error recovery strategies
Prevent memory leaks with CompositeDisposable and proper lifecycle management
Choose the right Observable type (Single, Maybe, Completable) for your use case
Use operators effectively to transform, filter, and combine data streams
Test your RxJava code thoroughly, especially async operations
Avoid overusing RxJava for simple synchronous operations
Follow reactive programming principles for clean, maintainable code

By following these best practices and patterns, you can build robust, responsive Android applications that handle complex asynchronous operations with ease.

Table of Contents