Этот урок начнем с паттерна Наблюдатель и разберемся, как он используется в RxJava. Рассмотрим основные понятия: Observable и Observer, и какие типы событий они используют. Далее разберем один теоретический и один практический примеры.

Версии RxJava

На момент создания этого курса, вторая версия RxJava была в статусе Release Candidate. Поэтому начало курса описывает первую версию. Но, во-первых, вторая версия очень похожа на первую, и уроки актуальны для обоих версий. А, во-вторых, если вы новичок в теме Rx, то вам будет сложно сразу понять главное новшество второй версии.

Поэтому предлагаю вам спокойно читать уроки, написанные по первой версии, т.к. все эти знания будут актуальны и для второй. А начиная с 11 урока мы просто перейдем на вторую версию. К тому моменту вы уже без проблем сможете понять главные отличия между ними.

Теория

Прежде чем начать обсуждать механизмы RxJava, давайте вспомним паттерн Наблюдатель. В нем есть объект, который генерирует какие-то события, и есть объект или объекты, которые подписываются и получают эти события. Я думаю, что в работе вы постоянно используете этот паттерн. Самый простой пример - обработчик нажатия кнопки.

В Java даже есть инструменты для этого паттерна - класс Observable и интерфейс Observer. Реализация интерфейса Observer - это объект, который ожидает событие. Observable - это класс, который всем переданным ему Observer-объектам сообщит о том, что событие наступило.

Эти же названия используются и в RxJava. И смысл их остался тем же: Observable генерирует событие, а Observer получает его. Но было значительно расширено само понятие "событие". В RxJava события, которые Observable передает в Observer, можно рассматривать как поток данных. И события в этом потоке имеют три типа:
1) Next - очередная порция данных
2) Error - произошла ошибка
3) Completed - поток завершен и данных больше не будет.

В качестве примера давайте рассмотрим поиск авиарейса. Пусть у нас есть метод, который умеет бегать по сайтам авиакомпаний, искать там по заданным критериям необходимый рейс и все полученные рейсы возвращать нам. Мы задаем ему пункт назначения, пункт отправления, дату - и он начинает работу.

По мере того, как он один за другим обрабатывает сайты, он генерирует события Next, в которые передает результат. Т.е. каждый найденный рейс - новое событие Next.

Если в работе метода произошла какая то серьезная ошибка и продолжение работы невозможно, метод отправит событие Error.

Если метод успешно обработал все известные ему сайты и закончил работу, он отправит нам событие Completed.

Какие места занимают во всей этой схеме Observable и Observer? Метод поиска при вызове возвращает нам объект Observable. А мы создаем Observer, в котором пишем код для обработки полученных событий, и подписываемся на этот Observable. По мере работы метода, Observable будет генерировать события, которые мы будем получать в нашем созданном Observer:
- если пришло событие Next с очередным рейсом, то берем рейс и, например, добавляем его в адаптер списка. Таким образом рейсы один за другим будут появляться в списке по мере их нахождения.
- если пришло событие Completed, значит мы можем выключить ProgressBar и уведомить пользователя, что поиск завершен
- если пришло событие Error, то уведомляем пользователя, что поиск был прерван с ошибкой

В этом примере я описал все три типа событий, чтобы было понятно, зачем они нужны. Но, вовсе необязательно в каждом потоке данных вы встретите все эти типы. Например, Observable, который сообщает нам о нажатиях на кнопку. В этом случае мы будем получать только событие Next при каждом нажатии. Событие Completed нам не придет, потому что пользователь может сколько угодно раз нажимать эту кнопку и никакого последнего нажатия там не будет. Ну и ошибку тут мы вряд ли получим.

Надеюсь, после этого введения и примеров у вас появилось понимание роли объектов Observable и Observer. Самое сложное поначалу - это просто отличать их друг от друга )

Практика

Давайте рассмотрим простейший пример Observable.

Создание Observable выглядит так:

Observable<String> observable = Observable.from(new String[]{"one", "two", "three"});

Observable<String> - это описание означает, что Observable будет предоставлять данные типа String, т.е. каждое событие Next, которое он будет генерировать, будет приходить с объектом типа String. Метод Observable.from создает для нас Observable, который возьмет данные из указанного String массива и передаст их получателям

Создаем получателя, т.е. Observer: 

    Observer<String> observer = new Observer<String>() {
        @Override
        public void onNext(String s) {
            log("onNext: " + s);
        }

        @Override
        public void onError(Throwable e) {
            log("onError: " + e);
        }

        @Override
        public void onCompleted() {
            log("onCompleted");
        }
    };

Observer<String> - получатель данных типа String. Напомню, что он от Observable ожидает получения событий трех типов Next, Error и Completed. И под каждый тип у Observer есть свой одноименный метод:
onNext(String s) - в этот метод будут приходить данные
onError(Throwable e) - будет вызван в случае какой-либо ошибки и на вход получит данные об ошибке
onCompleted() - уведомление о том, что все данные переданы

Оба объекта созданы, осталось подписать Observer на Observable методом subscribe:

observable.subscribe(observer);

Сразу после подписки Observable передаст в Observer все данные  (в метод onNext) и сигнал о том, что передача завершена (метод onCompleted).

Результат:
onNext: one
onNext: two
onNext: three
onCompleted

Этот простой пример призван показать взаимодействие между Observable и Observer. Мы использовали в нем Observable, который умеет передавать данные из предоставленного ему массива. Но это только один из видов Observable. Дальше мы научимся создавать различные Observable.

Что дальше

Когда я только начинал изучать RxJava, у меня после рассмотрения таких примеров возникали вопросы типа:
- В какой момент Observable начал генерировать события: после создания или после подписки на него Observer-а?
- Что будет если подписать несколько Observer-ов: каждый получит свои данные или те, кто подписался позже, не получит ничего?
- Как создать свой Observable, который будет отправлять результаты работы моего кода?
- Как сделать, чтобы работа в Observable выполнялась в одном потоке (в смысле Thread), а результаты приходили в другом?

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

Кроме того, мы рассмотрим, какие возможности работы с потоками данных предоставляет RxJava. Например, вы можете взять поток данных и выполнять над ним различные преобразования: фильтровать данные или конвертировать данные из одного типа в другой. Можете объединять данные из разных потоков данных последовательно, параллельно или попарно.

Также в RxJava присутствует отличный инструментарий для работы с потоками (речь уже не о потоках данных, а потоках, которые Thread). Вы можете указать один поток для генерации данных в Observable, другой поток для выполнения каких либо операций преобразования над этими данными и третий поток, в котором данные будут приходить в Observer.

В общем, тема очень интересная и полезная, и этот курс поможет вам в ней разобраться.

Курс RxJava