Инициализация Rx цепочки

Всем привет, меня зовут Иван, я Android-разработчик. Сегодня хочу поделиться своим опытом работы с RxJava2 и рассказать, как происходит инициализация цепочки. Почему я вообще решил поднять эту тему? Пообщавшись со знакомыми разработчиками, я понял, что не каждый кто использует этот инструмент понимает, как он работает. И тогда я решил разобраться как же устроены подписки в RxJava2 и в какой последовательности вся работа инициализируется. Я не нашел ни одной статьи, поясняющей это. В свете этого я полез в исходники, чтобы посмотреть, как же все работает и набросал для себя небольшую шпаргалку, которая выросла в данную статью.

В этой статье я не буду описывать что такое Observable, Observer и все остальные сущности, которые используются в RxJava2. Если вы решили прочитать данную статью, то я предполагаю, что вы уже знакомы с этой информацией. А если вы всё же не знакомы с этими понятия, то я рекомендую перед началом чтения ознакомиться с ними.

Вот с чего можно начать:

Грокаем* RxJava

Исследуем RxJava 2 для Android

Давайте посмотрим, как работает простейшая цепочка:

Observable.just (1, 2, 3, 4, 5) .map {…} .filter {…} .subscribe();

По верхам

Сначала опишу вкратце каждый шаг, через который пройдем в этой цепочке (шаги начинаются сверху вниз):

• Создается объект в операторе just ObservableFromArray.

• Создается объект в операторе map ObservableMap, который принимает в конструктор ссылку на ранее созданный объект в операторе just.

• Создается объект в операторе filter ObservableFilter, который принимает в конструктор ссылку на ранее созданный объект в map, в котором уже хранится ссылка на just.

• После создания всех Observable’ов у последнего Observable в цепочки вызывается метод subscribe() (в нашем случае это ObservableFilter созданный в операторе filter) в котором создается новый Observer, который и будет обрабатывать все полученные события.

• В методе ObservableFilter.subscribe() вызывается следующий метод ObservableFilter.subscribeActual(), в котором создается внутренний Observer, в случае с оператором filter, это FilterObserver. В этот внутренний Observer передается ссылка на первый созданный Observer в ObservableFilter.subscribe().

• Вызывается ObservableMap.subscribe() в котором так же вызывается ObservableMap.subscribeActual()и создается внутренний Observer, в случае с оператором map, это MapObserver, в который передается ссылка на FilterObserver.

• Вызывается ObservableFromArray.subscribe() и после ObservableFromArray.subscribeActual(), и уже там вызывается метод onSubscribe()у переданного в ObservableFromArray.subscribeActual() Observer’а.

• onSubscribe() вызывается у каждого нижележащего Observer’а в цепочке.

• ObservableFromArray начинает излучать все события в метод onNext() нижележащего Observer’а.

Создание источников данных

Теперь давайте рассмотрим описанные выше шаги подробнее, сначала попадаем в метод just() где происходит проверка каждого значения на null, далее идет вызов метода fromArray(), который возвращает Observable.

public static <T> Observable<T> just(T item1, T item2, T item3, T item4, T item5) {    ObjectHelper.requireNonNull(item1, "item1 is null");    ObjectHelper.requireNonNull(item2, "item2 is null");    ObjectHelper.requireNonNull(item3, "item3 is null");    ObjectHelper.requireNonNull(item4, "item4 is null");    ObjectHelper.requireNonNull(item5, "item5 is null");     return fromArray(item1, item2, item3, item4, item5); }

В fromArray() проверяется, что метод принимает в себя не пустой массив и имеет больше одного элемента.

public static <T> Observable<T> fromArray(T... items) {    ObjectHelper.requireNonNull(items, "items is null");    if (items.length == 0) {        return empty();    }    if (items.length == 1) {        return just(items[0]);    }    return RxJavaPlugins.onAssembly(new ObservableFromArray<T>(items)); }

После прохода через все эти шаги создается новый экземпляр класса ObservableFromArray, который на вход принимает массив с данными.

После этого новый объект передается в метод onAssembly(), так как нет необходимости как-то еще модифицировать Observable, то возвращается тот же источник, который поступал на вход.

public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {    Function<? super Observable, ? extends Observable> f = onObservableAssembly;    if (f != null) {        return apply(f, source);    }    return source; }

onAssembly() проверяет хотим ли перехватить текущий Observable и как-то модифицировать его, например таким образом:

RxJavaPlugins.setOnObservableAssembly(o -> { 	if (o instanceof ObservableFromArray) {     	return new ObservableFromArray<>(new Integer[] { 4, 5, 6 }); 	} 	return o; });   Observable.just(1, 2, 3) .filter(v -> v > 3) .test() .assertResult(4, 5, 6);

Вызывается следующий оператор в цепочке map(). Оператор проходит почти через те же шаги, что были описаны выше. Сначала делается проверка на null, потом создается новый экземпляр класса ObservableMap.

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {    ObjectHelper.requireNonNull(mapper, "mapper is null");    return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper)); }

Вот тут появляется небольшое отличие, в конструктор ObservableMap передается не только mapper, который будет преобразовывать одно значение в другое, но также принимает в конструктор this (source). В данном случае this это ссылка на экземпляр класса ObservableFromArray созданный на предыдущем шаге. ObservableMap расширяет абстрактный класс AbstractObservableWithUpstream, в котором и храниться source.

AbstractObservableWithUpstream абстрактный класс, который реализуют Observable операторы и в котором хранится ссылка на вышележащий источник данных.

Далее происходит вызов метода onAssembly() и возвращение созданного Observable. 

Переходим к следующему оператору в цепочки filter(). В нем не происходит ничего нового, за исключением того, что создается объект ObservableFilter и в его конструктор в this передается ссылка на экземпляр ObservableMap (у которого уже есть ссылка на ObservableFromArray, как показано на схеме выше) созданный на предыдущем шаге.

public final Observable<T> filter(Predicate<? super T> predicate) {    ObjectHelper.requireNonNull(predicate, "predicate is null");    return RxJavaPlugins.onAssembly(new ObservableFilter<T>(this, predicate)); }

Начало подписки

Последний оператор в цепочке subscribe(), который вызывает перегруженную версию метода. В нашем случае обрабатывается только onNext(). Метод subscribe() вызывается у ObservableFilter, который был последним созданным Observable в цепочке.

public final Disposable subscribe(Consumer<? super T> onNext) {    return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer()); }

В перегруженном методе сначала проверяются все переданные параметры на null, далее создается объект класса LambdaObserver и происходит подписка.

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,        Action onComplete, Consumer<? super Disposable> onSubscribe) {    ObjectHelper.requireNonNull(onNext, "onNext is null");    ObjectHelper.requireNonNull(onError, "onError is null");    ObjectHelper.requireNonNull(onComplete, "onComplete is null");    ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");     LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);     subscribe(ls);     return ls; }

А вот и сам метод, в котором и происходит подписка.

public final void subscribe(Observer<? super T> observer) {    ObjectHelper.requireNonNull(observer, "observer is null");    try {        observer = RxJavaPlugins.onSubscribe(this, observer);         ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");         subscribeActual(observer);    } catch (NullPointerException e) {       ......  } }

В методе subscribeActual() производится подписка на источник данных и в него же передается созданный ранее LambdaObserver. subscribeActual() вызывается в классе ObservableFilter. И вот что там происходит.

public void subscribeActual(Observer<? super T> observer) {    source.subscribe(new FilterObserver<T>(observer, predicate)); }

Создается новый объект класса FilterObserver, который принимает в конструктор LambdaObserver созданный ранее и предикат для фильтрации, которые хранится в ObservableFilter в виде поля класса.

Класс FilterObserver расширяет класс BasicFuseableObserver, в котором уже реализован метод onSubscribe(). BasicFuseableObserver это абстрактный класс, который реализуют промежуточные Observer’ы. Если посмотреть исходники, то его реализуют только 6 классов, два из которых это FilterObserver и MapObserver. В методе BasicFuseableObserver.onSubscribe() также вызывается метод onSubscribe() у нижележащего Observer’а, который передавался в конструктор этого класса. А выглядит это вот так:

public final void onSubscribe(Disposable d) {    if (DisposableHelper.validate(this.upstream, d)) {        this.upstream = d;        if (d instanceof QueueDisposable) {            this.qd = (QueueDisposable<T>)d;        }        if (beforeDownstream()) {             downstream.onSubscribe(this);             afterDownstream();        }    } }

После того, как подписались на ObservableFilter и создали объект FilterObserver, он передается в source.subscribe(). Хочу напомнить, что source это объект класса ObservableMap, переданный ранее в цепочке. У объекта ObservableMap вызывается метод subscribe().

public final void subscribe(Observer<? super T> observer) {    ObjectHelper.requireNonNull(observer, "observer is null");    try {        observer = RxJavaPlugins.onSubscribe(this, observer);         ObjectHelper.requireNonNull(observer, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");         subscribeActual(observer);    } catch (NullPointerException e) {       ......  } }

Далее происходит те же шаги, в методе subscribe() вызывается subscribeActual(), оба этих метода вызываются у ObservableMap. В subscribeActual() создается новый MapObserver с переданным в качестве параметра экземпляром FilterObserver и функцией mapper’а. 

public void subscribeActual(Observer<? super U> t) {    source.subscribe(new MapObserver<T, U>(t, function)); }

public void subscribeActual(Observer<? super T> observer) {    FromArrayDisposable<T> d = new FromArrayDisposable<T>(observer, array);     observer.onSubscribe(d);     if (d.fusionMode) {        return;    }    d.run(); }

Все рассмотренные Observer’ы расширяли абстрактный класс BasicFuseableObserver, в котором уже реализован метод onSubscribe() и, так же есть ссылка на нижележащий Observer, у которого так же вызывается метод onSubscribe().

В конце метода subscribeActual() вызывается метод run(), в котором и начинается излучение всех данных в нижележащие Observer’ы.

void run() {    T[] a = array;    int n = a.length;     for (int i = 0; i < n && !isDisposed(); i++) {        T value = a[i];        if (value == null) {            downstream.onError(new NullPointerException("The element at index " + i + " is null"));            return;        }        downstream.onNext(value);    }    if (!isDisposed()) {        downstream.onComplete();    } }

Соответственно вызываются onNext() для передачи значений в нижележащие Observer’ы, и потом onComplete() при завершении излучения данных или может произойти ошибка и вызовется onError(), который завершит всю цепочку.

Вывод

Observable’ы вкладываются друг в друга и вызывают callback’и для создания Observer’ов, которые и будут обрабатывать получаемые данные и передавать их дальше по цепочки.

Метод onSubscribe() вызывается до начала отправки данных и это надо иметь ввиду если вы пользуетесь такими оператора, как doOnSubscribe().

На каждый оператор создается как минимум 3 объекта:

• Анонимный класс передаваемый в оператор

• Observable создаваемый внутри оператора

• Observer обрабатывающий получаемые данные

Потому при использовании операторов стоит иметь ввиду, что каждый оператор аллоцирует память для несколько объектов и не стоит добавлять операторы в цепочку, только потому что “можно”.

RxJava мощный инструмент, но нужно понимать, как он работает и для каких задач его использовать. Если вам нужно просто выполнение сетевого запроса в фоновом потоке и последующее выполнение результата на главном потоке, то это как “стрелять из пушки по воробьям”, попасть можно, а вот последствия могут быть серьезными.