RxJava. Функциональное реактивное программирование (FRP)

Функциональное реактивное программирование (FRP)
Функциональное. Основной элемент – функции.
Реактивное. Программирование с асинхронными потоками данных.
Функциональное реактивное программирование – программирование с асинхронными потоками данных, которые манипулируются с помощью различных функций.

Что такое поток данных?
Любые объекты и примитивы
Любая последовательность объектов и примитивов
Бесконечные последовательности
Любое событие (ввод текста, клик на кнопку и так далее)

Ввод текста как поток данных

RxJava
Обеспечение многопоточности
Управление потоками данных
Обработка ошибок
Красивый и компактный код (при использовании лямбда-выражений)

Паттерн Observer

Поток данных

Создаем поток данных

Подписываемся на поток данных

Подписываемся на поток данных

Почему Observable, а не for?

Rx vs for

Rx vs for

Observable.from

Observable.create

Observable.create

Observable.create

Observable.fromCallable
Observable.fromCallable(() -> getUserIdsFromDatabase());

Observable. Создание последовательности
Observable.empty вернет только onCompleted
Observable.error вернет только onError
Observable.never ничего не вернет
values.subscribe(
val -> System.out.println("Received: " + val),
error -> System.out.println("Error: " + error),
() -> System.out.println("Completed")
);

subscribeOn и observeOn
Код подписчика (observer) выполняется в потоке, переданном в observeOn
Код потока данных выполняется в потоке, переданном в subscribeOn

Schedulers
Schedulers.io() – выполнение задач, которые не сильно нагружают процессор, но являются долгими: сетевые запросы, база данных
Schedulers.computation() – тяжелые вычислительные задачи, нагружающие CPU
Schedulers.newThread – новый поток для каждой новой задачи
Schedulers.immediate() – выполнение задачи в том же потоке
AndroidScheduler.mainThread() (RxAndroid) – главный поток Android-приложения

Transformer
Одинаковый код для каждого сетевого запроса

Transformer

Transformer

Transformer

Observable.map

Observable.map

Observable.filter

Observable.skip

Observable.take

Observable.concat

Observable.concat

Observable.merge

Observable.merge

Observable.zip

Observable.zip

Observable.zip

Observable.flatMap

Observable.flatMap

RxJava в Android
Использование RxJava в Android не ограничено
Поддержка RxJava в Retrofit
Реализация стандартных для Android подходов в реактивном стиле: RxBindings, RxLifecycle и множество других

RxJava в Retrofit

RxJava в Retrofit

Реактивный запрос на сервер

Отображение процесса загрузки

Other Utility Operators
doOnEach – register an action to take whenever an Observable emits an item
doOnCompleted – register an action to take when an Observable completes successfully
doOnError – register an action to take when an Observable completes with an error
doAfterTerminate – register an action to call just after an Observable terminated, either successfully or with an error
doFinally - register an action to call when an Observable terminates or it gets disposed

Кэшируем данные

Возвращаем закэшированнные данные в случае ошибки

Кэшируем запросы

Проблема Backpressure

Observable.sample

Observable.sample

Observable.debounce

Observable.buffer

Observable.buffer

RxJava2

Single.just

Single.concat

Single.merge

Single.map

Single.flatMap

Single.zip

Single

Completable

Maybe

Maybe.fromAction
Maybe.fromAction(()-> System.out.println(“Hello”));
Maybe.fromRunnable(()-> System.out.println(“Hello”));
Completable.fromAction(()-> System.out.println(“Hello”));
Completable.fromRunnable(()-> System.out.println(“Hello”));

Dispose
.subscribe – возвращает объект Disposable, который хранит состояние подписки на текущий момент.
Содержит два метода:
isDisposed() – освобождены ли ресурсы
dispose() – освободить ресурсы(отписаться)
В методе onDestroy() всегда стоит вызывать метод dispose()

Dispose