Начало работы с ReactiveX на Android

() translation by (you can also view the original English article)

Введение

Разработка сложного Android-приложения с множеством сетевых подключений, взаимодействий пользователей и анимаций часто означает написание кода с кучей вложенных обратных вызовов. Такой код, иногда называемый адом обратных вызовов, не только длинный и трудно понимаемый, но и подвержен ошибкам. ReactiveX предлагает альтернативный подход, который является ясным и кратким, для управления асинхронными задачами и событиями.

RxJava - это JVM-реализация ReactiveX, разработанная NetFlix, и очень популярна среди разработчиков Java. В этом уроке вы узнаете, как использовать привязки RxJava для Android или RxAndroid для краткости в ваших проектах на Android.

1. Настройка RxAndroid

Чтобы использовать RxAndroid в проекте Android Studio, добавьте его как compile зависимость в build.gradle модуля приложения.

1	compile 'io.reactivex:rxandroid:0.25.0'

2. Основы наблюдателей и наблюдателей

При работе с ReactiveX вы будете широко использовать наблюдаемых и наблюдателей. Вы можете думать о наблюдаемом как объект, как об объекте который испускает данные и о наблюдателе как об объекте, который потребляет эти данные. В RxJava и RxAndroid наблюдатели являются экземплярами интерфейса Observer, а наблюдаемые - экземплярами класса Observable.

Класс Observable имеет множество статических методов, называемых operators, для создания объектов Observable. Следующий код показывает вам, как использовать оператор just для создания очень простого Observable, который испускает одну String.

1	Observable<String> myObservable
2	= Observable.just("Hello"); // Emits "Hello"

Наблюдаемый, которого мы только что создали, будет излучать его данные только тогда, когда у него есть хотя бы один наблюдатель. Чтобы создать наблюдателя, вы создаете класс, который реализует интерфейс Observer. Интерфейс Observer имеет интуитивно названные методы обработки различных типов уведомлений, которые он может получать от наблюдаемого. Вот наблюдатель, который может печатать String, испущенную наблюдаемым нами ранее:

Observer<String> myObserver = new Observer<String>() {
    @Override
    public void onCompleted() {
        // Called when the observable has no more data to emit
    }

    @Override
    public void onError(Throwable e) {
        // Called when the observable encounters an error
    }

    @Override
    public void onNext(String s) {
        // Called each time the observable emits data
        Log.d("MY OBSERVER", s);
    }
};

Чтобы назначить наблюдателя наблюдаемому, вы должны использовать метод subscribe, который возвращает объект Subscription. Следующий код заставляет myObserver наблюдать myObservable:

1	Subscription mySubscription = myObservable.subscribe(myObserver);

Как только наблюдатель добавляется к наблюдаемому, он испускает свои данные. Поэтому, если вы сейчас выполняете код, вы увидите Hello в окне Android logcat.

Возможно, вы заметили, что мы не использовали методы onCompleted и onError в myObserver. Поскольку эти методы часто остаются неиспользованными, у вас также есть возможность использовать интерфейс Action1, который содержит один метод с именем call.

Action1<String> myAction = new Action1<String>() {
    @Override
    public void call(String s) {
        Log.d("My Action", s);
    }
};

Когда вы передаете экземпляр Action1 методу subscribe, метод call вызывается всякий раз, когда наблюдаемый испускает данные.

1	Subscription mySubscription = myObservable.subscribe(myAction1);

Чтобы отсоединить наблюдателя от его наблюдаемого, в то время как наблюдаемый все еще излучает данные, вы можете вызвать метод unsubscribe на объекте Subscription.

1	mySubscription.unsubscribe();

3. Использование операторов

Теперь, когда вы знаете, как создавать наблюдателей и наблюдаемых, позвольте мне показать вам, как использовать операторы ReactiveX, которые могут создавать, преобразовывать и выполнять другие операции над наблюдаемыми. Начнем с создания немного более продвинутого Observable, который испускает элементы из массива объектов Integer. Для этого вам необходимо использовать оператор from, который может генерировать Observable из массивов и списков.

Observable<Integer> myArrayObservable 
    = Observable.from(new Integer[]{1, 2, 3, 4, 5, 6}); // Emits each item of the array, one at a time

myArrayObservable.subscribe(new Action1<Integer>() {
    @Override
    public void call(Integer i) {
        Log.d("My Action", String.valueOf(i)); // Prints the number received
    }
});

Когда вы запустите этот код, вы увидите каждый из номеров массива, напечатанных один за другим.

Если вы знакомы с JavaScript, Ruby или Kotlin, вы можете быть знакомы с функциями более высокого порядка, такими как map и filter, которые можно использовать при работе с массивами. ReactiveX имеет операторы, которые могут выполнять аналогичные операции с наблюдаемыми. Однако, поскольку Java 7 не имеет функций lambdas и более высокого порядка, нам придется делать это с помощью классов, которые имитируют lambdas. Чтобы имитировать лямбду, которая принимает один аргумент, вам нужно создать класс, который реализует интерфейс Func1.

Вот как вы можете использовать оператор map для каждого элемента myArrayObservable:

myArrayObservable = myArrayObservable.map(new Func1<Integer, Integer>() { // Input and Output are both Integer
    @Override
    public Integer call(Integer integer) {
        return integer * integer; // Square the number
    }
});

Обратите внимание, что вызов оператора map возвращает новый Observable, он не меняет исходный Observable. Если теперь вы подписаны на myArrayObservable, вы получите квадраты чисел.

Операторы могут быть выстроены в цепочку. Например, следующий блок кода использует оператор skip для пропуска первых двух чисел, а затем оператор filter игнорирует нечетные числа:

myArrayObservable
    .skip(2) // Skip the first two items
    .filter(new Func1<Integer, Boolean>() {
        @Override
        public Boolean call(Integer integer) { // Ignores any item that returns false
            return integer % 2 == 0; 
        }
    });

// Emits 4 and 6

4. Обработка асинхронных заданий

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

Предположим, что у вас есть метод с именем fetchData, который можно использовать для извлечения данных из API. Предположим, что он принимает URL как свой параметр и возвращает содержимое ответа как String. Следующий фрагмент кода показывает, как его можно использовать.

1	String content = fetchData("http://www.google.com");
2	// fetches the contents of google.com as a String

Этот метод должен запускаться в своем потоке, потому что Android не разрешает сетевые операции в потоке пользовательского интерфейса. Это означает, что вы либо создадите AsyncTask, либо создаете Thread, который использует Handler.

Однако с ReactiveX у вас есть третий вариант, который немного более короче. Используя операторы subscribeOn и observOn, вы можете явно указать, какой поток должен запускать фоновое задание, и какой поток должен обрабатывать обновления пользовательского интерфейса.

Следующий код создает пользовательский Observable с помощью оператора create. Когда вы создаете Observable таким образом, вы должны реализовать интерфейс Observable.OnSubscribe и управлять тем, что он испускает, вызывая сами методы onNext, onError и onCompleted.

Observable<String> fetchFromGoogle = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        try {
            String data = fetchData("http://www.google.com");
            subscriber.onNext(data); // Emit the contents of the URL
            subscriber.onCompleted(); // Nothing more to emit
        }catch(Exception e){
            subscriber.onError(e); // In case there are network errors
        }
    }
});

Когда Observable готов, вы можете использовать subscribeOn и observeOn, чтобы указать потоки, которые он должен использовать, и подписаться на него.

fetchFromGoogle
    .subscribeOn(Schedulers.newThread()) // Create a new Thread
    .observeOn(AndroidSchedulers.mainThread()) // Use the UI thread
    .subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
            view.setText(view.getText() + "\n" + s); // Change a View
        }
    });

Вы все еще можете подумать, что реактивный подход не будет значительно лучше, чем использование классов AsyncTask или Handler. Вы правы, вам действительно не нужен ReactiveX, если вам нужно управлять только одним заданием в фоне.

Теперь рассмотрим сценарий, который приведет к сложной базе кода, если вы использовали обычный подход. Предположим, вы должны получать данные с двух (или более) веб-сайтов параллельно и обновлять View только тогда, когда все запросы завершены. Если вы придерживаетесь обычного подхода, вам придется написать много ненужного кода, чтобы убедиться, что запросы выполнены без ошибок.

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

С операторами ReactiveX оба сценария могут обрабатываться очень просто. Например, если вам нужно использовать fetchData для извлечения содержимого двух веб-сайтов, например Google и Yahoo, вы должны создать два объекта Observable и использовать метод subscribeOn, чтобы заставить их работать на разных потоках.

1	fetchFromGoogle = fetchFromGoogle.subscribeOn(Schedulers.newThread());
2	fetchFromYahoo = fetchFromYahoo.subscribeOn(Schedulers.newThread());

Чтобы справиться с первым сценарием, в котором оба запроса должны выполняться параллельно, вы можете использовать zip оператор и подписаться на Observable, который он возвращает.

// Fetch from both simultaneously
Observable<String> zipped 
	= Observable.zip(fetchFromGoogle, fetchFromYahoo, new Func2<String, String, String>() {
		@Override
		public String call(String google, String yahoo) { 
			// Do something with the results of both threads
		    return google + "\n" + yahoo;
		}
	});

Аналогично, для обработки второго сценария вы можете использовать оператор concat для запуска потоков один за другим.

1	Observable<String> concatenated = Observable.concat(fetchFromGoogle, fetchFromYahoo);
2	// Emit the results one after another

5. Обработка событий

RxAndroid имеет класс с именем ViewObservable, который упрощает обработку событий, связанных с объектами View. В следующем фрагменте кода показано, как создать ViewObservable, который можно использовать для обработки событий нажатия Button.

Button myButton 
    = (Button)findViewById(R.id.my_button); // Create a Button from a layout

Observable<OnClickEvent> clicksObservable
    = ViewObservable.clicks(myButton); // Create a ViewObservable for the Button

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

clicksObservable
    .skip(4)    // Skip the first 4 clicks
    .subscribe(new Action1<OnClickEvent>() {
        @Override
        public void call(OnClickEvent onClickEvent) {
            Log.d("Click Action", "Clicked!"); 
                // Printed from the fifth click onwards
        }
    });

Заключение

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

Чтобы узнать больше о реактивных расширениях, я рекомендую вам просматривать ресурсы, доступные в ReactiveX.