RxAndroid

import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.android.MainThreadDisposable;
import io.reactivex.android.plugins.RxAndroidPlugins;
import io.reactivex.Scheduler;
import android.os.Looper;

import io.reactivex.Observable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.schedulers.Schedulers;

// Schedule work on background thread and observe results on main thread
Observable.fromCallable(() -> {
    // Background work (e.g., network call, database operation)
    return processData();
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(result -> {
    // Update UI on main thread
    updateUI(result);
});

// Custom Looper scheduling
HandlerThread handlerThread = new HandlerThread("Background");
handlerThread.start();
Scheduler customScheduler = AndroidSchedulers.from(handlerThread.getLooper());

Observable.just("data")
    .observeOn(customScheduler)
    .subscribe(data -> {
        // Process on custom thread
    });

/**
 * Returns a Scheduler which executes actions on the Android main thread.
 * Thread-safe and can be called from any thread.
 */
public static Scheduler mainThread();

/**
 * Returns a Scheduler which executes actions on the specified Looper.
 * @param looper the Looper to schedule actions on
 * @throws NullPointerException if looper is null
 */
public static Scheduler from(Looper looper);

/**
 * Returns a Scheduler which executes actions on the specified Looper with async messaging option.
 * @param looper the Looper to schedule actions on
 * @param async if true, uses async messaging on API >= 16 to avoid VSYNC locking.
 *              Automatically set to false on API < 16. On API 16-21, validates that
 *              Message.setAsynchronous() is available before enabling async messaging.
 * @throws NullPointerException if looper is null
 */
public static Scheduler from(Looper looper, boolean async);

// Create background thread with custom Looper
HandlerThread backgroundThread = new HandlerThread("MyBackground");
backgroundThread.start();

// Schedule work on custom Looper
Scheduler customScheduler = AndroidSchedulers.from(backgroundThread.getLooper());
Observable.just("work")
    .observeOn(customScheduler)
    .subscribe(item -> {
        // Executes on custom background thread
    });

// Using async messaging (API 16+)
Scheduler asyncScheduler = AndroidSchedulers.from(backgroundThread.getLooper(), true);

// Note: AndroidSchedulers uses HandlerScheduler internally for all Looper-based scheduling

/**
 * Abstract class implementing Disposable with main thread disposal guarantee.
 * Useful for UI-related subscriptions that need cleanup on the main thread.
 */
public abstract class MainThreadDisposable implements Disposable {
    
    /**
     * Verify that the calling thread is the Android main thread.
     * Commonly used as precondition check in custom observables.
     * @throws IllegalStateException when called from any thread other than main thread
     */
    public static void verifyMainThread();
    
    /**
     * Returns whether this Disposable is disposed.
     */
    public final boolean isDisposed();
    
    /**
     * Disposes this Disposable, ensuring onDispose() runs on main thread.
     * Thread-safe - can be called from any thread.
     */
    public final void dispose();
    
    /**
     * Called when disposal occurs, guaranteed to execute on main thread.
     * Subclasses must implement this method to perform cleanup.
     */
    protected abstract void onDispose();
}

public class CustomObservable extends Observable<String> {
    @Override
    protected void subscribeActual(Observer<? super String> observer) {
        // Verify we're on main thread for UI access
        MainThreadDisposable.verifyMainThread();
        
        // Set up some UI listener or resource
        setupUIListener();
        
        observer.onSubscribe(new MainThreadDisposable() {
            @Override
            protected void onDispose() {
                // Cleanup UI resources - guaranteed to run on main thread
                cleanupUIListener();
            }
        });
        
        observer.onNext("data");
        observer.onComplete();
    }
}

/**
 * Sets a handler to customize main thread scheduler initialization.
 * @param handler function to transform scheduler initialization
 */
public static void setInitMainThreadSchedulerHandler(Function<Callable<Scheduler>, Scheduler> handler);

/**
 * Initializes main thread scheduler with optional handler transformation.
 * @param scheduler callable that provides default scheduler
 * @throws NullPointerException if scheduler is null
 * @return transformed scheduler or default if no handler set
 */
public static Scheduler initMainThreadScheduler(Callable<Scheduler> scheduler);

/**
 * Sets a handler to customize main thread scheduler instances.
 * @param handler function to transform scheduler instances
 */
public static void setMainThreadSchedulerHandler(Function<Scheduler, Scheduler> handler);

/**
 * Applies main thread scheduler handler if set.
 * @param scheduler scheduler to potentially transform
 * @throws NullPointerException if scheduler is null
 * @return transformed scheduler or original if no handler set
 */
public static Scheduler onMainThreadScheduler(Scheduler scheduler);

/**
 * Returns current init handler.
 * @return handler function or null if not set
 */
public static Function<Callable<Scheduler>, Scheduler> getInitMainThreadSchedulerHandler();

/**
 * Returns current main thread handler.
 * @return handler function or null if not set
 */
public static Function<Scheduler, Scheduler> getOnMainThreadSchedulerHandler();

/**
 * Removes all handlers and resets to default behavior.
 */
public static void reset();

// Replace main thread scheduler for testing
RxAndroidPlugins.setMainThreadSchedulerHandler(scheduler -> Schedulers.trampoline());

// Custom initialization
RxAndroidPlugins.setInitMainThreadSchedulerHandler(schedulerCallable -> {
    return new CustomTestScheduler();
});

// Reset to defaults (typically in test teardown)
RxAndroidPlugins.reset();

// From RxJava 2 - key types used by RxAndroid
interface Scheduler {
    Disposable scheduleDirect(Runnable run);
    Worker createWorker();
}

interface Disposable {
    void dispose();
    boolean isDisposed();
}

interface Function<T, R> {
    R apply(T t) throws Exception;
}

// From Android SDK
class Looper {
    static Looper getMainLooper();
    static Looper myLooper();
}