Управление несколькими Observable в RxJava

Условие

Необходимо построить реактивную цепочку, которая при нажатии кнопки считывает строку, преобразует её в объект Car и отправляет запрос на сервер. При появлении нового запроса предыдущий должен быть отменён, чтобы не выполнять лишние сетевые операции.

val clickStream: Observable<String>? = null
val clickStream1: Observable<String>? = null
val clickStream2: Observable<String>? = null

data class Car(
    val name: String
)

Ответ

Объедините потоки кликов и используйте switchMap: новый элемент снимает подписку с предыдущего сетевого источника. Реальная отмена HTTP требует поддержки отмены в сетевом адаптере.

Если sendCar возвращает Observable результата, ядро цепочки такое:

val clicks = Observable.merge(listOfNotNull(clickStream, clickStream1, clickStream2))
val disposable = clicks
    .map { Car(it.trim()) }
    .switchMap { car -> sendCar(car) }
    .subscribe(::renderResult, ::showError)

sendCar, renderResult и showError здесь — функции приложения. Валидацию добавьте до запроса, а обработку восстанавливаемых ошибок внутрь switchMap, если после ошибки нужно продолжать принимать клики. Disposable освобождайте по lifecycle. Снятие подписки не отменяет уже выполненный сервером побочный эффект.

Источник: reactivex.io.