Реактивное программирование с RxJava и RxKotlin

С тех пор, как стал официально поддерживаемым языком для разработки Android, Kotlin быстро завоевал популярность среди разработчиковAndroid, и Google сообщает о 6-кратном увеличении приложений, созданных с помощью Kotlin.

Если вы ранее использовали RxJava или RxAndroid и хотите переключиться на Kotlin или хотите начать реактивное программирование с Kotlin,этот учебник для вас. Мы рассмотрим основные вопросы создания RxJava 2.0 Observers, Observables и потоков данных в Kotlin, прежде чем смотреть на то, как вы можете обрезать тонны кода шаблона из своих проектов, путем объединенияRxJava с функциями Kotlin.

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

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

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

Что такое RxJava, во всех случаях?

RxJava - это реализация библиотеки ReactiveX с открытым исходным кодом, которая помогает создавать приложения в стиле реактивного программирования. Хотя RxJava предназначен для обработки синхронных и асинхронных потоков данных, он не ограничивается «традиционными» типами данных. Определение «данных» RxJava довольно широко и включает такие вещи, как кеши, переменные, свойства и даже пользовательские события ввода, такие как щелчки и нажатия. Просто потому, что ваше приложение не имеет дело с огромными числами или выполняет сложные преобразования данных, это не значит, что он не может извлечь выгоду из RxJava!

За небольшим опытом использования приложений RxJava для Android вы можете ознакомиться с некоторыми другими моими сообщениями здесь, на Envato Tuts +.

• Android SDK
  Начало работы с RxJava 2 для Android
  Jessica Thornsby
• Android SDK
  RxJava 2 для Android-приложений: RxBinding и RxLifecycle
  Jessica Thornsby

Итак, как работает RxJava?

RxJava расширяет шаблон разработки программного обеспечения Observer, основанный на концепции наблюдателей и наблюдаемых. Чтобы создать базовый конвейер данных RxJava, вам необходимо:

• Создать наблюдаемый.
• Дайте наблюдаемым некоторые данные для выделения.
• Создайте наблюдателя.
• Подписывайте Наблюдателя в Наблюдаемого.

Как только Observable имеет хотя бы одного наблюдаемого, он начнет выпускать данные. Каждый раз, когда Observable испускает часть данных, он уведомляет об этом назначенного Observer, вызывая метод onNext(), и Observerзатем обычно выполняет некоторые действия в ответ на эту эмиссию данных. Как только Observable закончит выдавать данные, он уведомит Observer, вызвав onComplete(). После этого Observable завершит работу, и поток данных завершится.

Если возникает исключение, тогда вызывается onError(), и Observable будет немедленно завершен без каких-либо дополнительных данных или вызова onComplete().

Но RxJava - это не просто передача данных из Observable в Observer! RxJava имеет огромную коллекцию операторов, которую вы можете использовать для фильтрации, слияния и преобразования этих данных. Например, представьте себе, что в вашем приложении есть кнопка «Оплатить сейчас», которая определяет события onClick, и вы опасаетесь,что нетерпеливый пользователь может нажать кнопку несколько раз, в результате чего ваше приложение обработает несколько платежей.

RxJava позволяет преобразовывать эти события onClick в поток данных, который затем можно манипулировать с помощью различных операторовRxJava. В этом конкретном примере вы можете использовать оператор debounce()для фильтрации данных, которые происходят в быстрой последовательности,поэтому, даже если пользователь нажимает на кнопку «Оплатить сейчас», ваше приложение будет регистрировать только на один платеж.

Каковы преимущества использования RxJava?

Мы видели, как RxJava может помочь вам решить определенную проблему в конкретном приложении, но что он должен предлагать проектам Android в целом?

RxJava может упростить ваш код, предоставив вам способ написать то, что вы хотите достичь, вместо того, чтобы писать список инструкций, которые должны выполнять ваше приложение. Например, если вы хотите игнорировать все данные, которые происходят в течение одного и того же 500-миллисекундного периода, тогда вы должны написать:

.debounce(500, TimeUnit.MILLISECONDS)

Кроме того, поскольку RxJava обрабатывает почти все как данные, он предоставляет шаблон, который вы можете применить к широкому спектру событий: создать Observable, создать Observer, подписать Observer к наблюдаемому, проверить и повторить. Этот формульный подход дает гораздо более простой, удобочитаемый код.

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

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

В RxJava вы создаете и управляете дополнительными потоками, используя комбинацию операторов и планировщиков. Вы можете легко изменить поток, в котором выполняется работа, используя оператор subscribeOn плюс планировщик. Например, здесь мы планируем работу над новым потоком:

.subscribeOn(Schedulers.newThread())

Вы можете указать, где результаты этой работы должны быть опубликованы, используя оператор observOn. Здесь мы публикуем результаты для всего важного основного потока пользовательского интерфейса Android, используя планировщик AndroidSchedulers.mainThread, который доступен как часть библиотекиRxAndroid:

.observeOn(AndroidSchedulers.mainThread())

По сравнению с встроенными многопоточными решениями Android, подход RxJava гораздо более краток и понятен.

Опять же, вы можете узнать больше о том, как работает RxJava, и о преимуществах добавления этой библиотеки в ваш проект в статье«Начало работы с RxJava 2 для Android».

Должен ли я использовать RxJava или RxKotlin?

Поскольку Kotlin на 100% совместим с Java, вы можете использовать большинство библиотек Java в своих проектах Kotlin без каких-либо трудностей, а библиотека RxJava не является исключением.

Существует специальная библиотека RxKotlin, которая представляет собой обертку Kotlin вокруг обычной библиотеки RxJava. Эта оболочка предоставляет расширения, которые оптимизируют RxJava для среды Kotlinи могут дополнительно уменьшить количество кода шаблона, который вам нужно написать.

Поскольку вы можете использовать RxJava в Kotlin, не требуя RxKotlin, мы будем использовать RxJava в этой статье, если не указано иное.

Создание простых наблюдателей и наблюдаемых в Kotlin

Наблюдатели и наблюдаемые являются строительными блоками RxJava, поэтому давайте начнем с создания:

• Простой Observable, который испускает короткий поток данных в ответ на событие нажатия кнопки.
• Наблюдаемый, который реагирует на эти данные, печатая различные сообщения в Logcat Android Studio.

Создайте новый проект с настройками по вашему выбору, но при появлении запроса установите флажок Включить поддержку Kotlin. Затем откройте файл build.gradle вашего проекта и добавьте библиотеку RxJava в зависимости от проекта:

dependencies {
  implementation fileTree(dir: 'libs', include: ['*.jar'])
  implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:$kotlin_version"
  implementation 'androidx.appcompat:appcompat:1.0.0-alpha1'
  implementation 'androidx.constraintlayout:constraintlayout:1.1.0'
  implementation 'io.reactivex.rxjava2:rxjava:2.1.9'

}

Затем откройте файл activity_main.xml вашего проекта и добавьте кнопку, которая начнет поток данных:

<?xml version="1.0" encoding="utf-8"?>
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
  xmlns:tools="http://schemas.android.com/tools"
  android:layout_width="match_parent"
  android:layout_height="match_parent"
  android:orientation="vertical"
  tools:context=".MainActivity" >

  <Button
      android:id="@+id/button"
      android:layout_width="wrap_content"
      android:layout_height="wrap_content"
      android:text="Start RxJava stream" />

</LinearLayout>

Существует несколько способов создания Observable, но одним из самых простых является использование оператора just() для преобразования объекта или списка объектов в Observable.

В следующем коде мы создаем Observable (myObservable) и даем ему элементы 1, 2, 3, 4 и 5 для выдиления. Мы также создаем Observer(myObserver), подписываемся на myObservable, а затем сообщаем ему печатать сообщение Logcat каждый раз, когда он получает новую эмиссию.

import androidx.appcompat.app.AppCompatActivity
import android.os.Bundle
import android.util.Log
import io.reactivex.Observable
import io.reactivex.Observer
import io.reactivex.disposables.Disposable
import kotlinx.android.synthetic.main.activity_main.*

class MainActivity : AppCompatActivity() {

  private var TAG = "MainActivity"

  override fun onCreate(savedInstanceState: Bundle?) {
      super.onCreate(savedInstanceState)
      setContentView(R.layout.activity_main)

//Start the stream when the button is clicked//

      button.setOnClickListener { startRStream() }

  }

  private fun startRStream() {

//Create an Observable//

      val myObservable = getObservable()

//Create an Observer//

      val myObserver = getObserver()

//Subscribe myObserver to myObservable//

      myObservable
              .subscribe(myObserver)
  }

  private fun getObserver(): Observer<String> {
      return object : Observer<String> {
          override fun onSubscribe(d: Disposable) {
          }

//Every time onNext is called, print the value to Android Studio’s Logcat//

          override fun onNext(s: String) {
              Log.d(TAG, "onNext: $s")
          }

//Called if an exception is thrown//

          override fun onError(e: Throwable) {
              Log.e(TAG, "onError: " + e.message)
          }

//When onComplete is called, print the following to Logcat//

          override fun onComplete() {
              Log.d(TAG, "onComplete")
          }
      }
  }

//Give myObservable some data to emit//

  private fun getObservable(): Observable<String> {
      return Observable.just("1", "2", "3", "4", "5")
  }
}

Теперь вы можете проверить это приложение:

• Установите свой проект на физическом Android-смартфоне или планшете или виртуальном устройстве Android (AVD).
• Нажмите на Start RxJava stream
• Откройте монитор Logcat для Android Studio, выбрав вкладку «Монитор Android» (где курсор расположен на следующем снимке экрана), а затем выберите вкладку «Logcat».

На этом этапе Observable начнет выдавать свои данные, и Observer будет печатать свои сообщения в Logcat. Вывод Logcat должен выглядеть примерно так:

Вы можете загрузить этот проект из GitHub, если хотите попробовать его сами.

Kotlin Extensions для RxJava

Теперь, когда мы увидели, как настроить простой RxJava-конвейер в Kotlin, давайте посмотрим, как вы можете добиться этого в меньшем коде, используя функции расширения RxKotlin.

Чтобы использовать библиотеку RxKotlin, вам нужно добавить ее как зависимость от проекта:

dependencies {
  implementation fileTree(dir: 'libs', include: ['*.jar'])
  implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:$kotlin_version"
  implementation 'androidx.appcompat:appcompat:1.0.0-alpha1'
  implementation 'androidx.constraintlayout:constraintlayout:1.1.0'
  implementation 'io.reactivex.rxjava2:rxjava:2.1.9'

//Add the following//

  implementation 'io.reactivex.rxjava2:rxkotlin:2.2.0'

}

В следующем примере мы используем функцию расширения RxKotlin toObservable() для преобразования List в наблюдаемый. Мы также используем функцию расширения subscribeBy(), поскольку она позволяет нам построить Observer с использованием именованных аргументов, что приводит к более четкому коду.

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import io.reactivex.rxkotlin.subscribeBy
import io.reactivex.rxkotlin.toObservable
import kotlinx.android.synthetic.main.activity_main.*

class MainActivity : AppCompatActivity() {

  override fun onCreate(savedInstanceState: Bundle?) {
      super.onCreate(savedInstanceState)
      setContentView(R.layout.activity_main)

//Start the stream when the button is clicked//

      button.setOnClickListener { startRStream() }

  }

  private fun startRStream() {

      val list = listOf("1", "2", "3", "4", "5")

//Apply the toObservable() extension function//

      list.toObservable()

//Construct your Observer using the subscribeBy() extension function//

              .subscribeBy(

                      onNext = { println(it) },
                      onError = { it.printStackTrace() },
                      onComplete = { println("onComplete!") }

              )
  }
}

Вот результат, который вы должны увидеть:

Решение проблемы неоднозначности SAM RxJava

RxKotlin также обеспечивает важное обходное решение для проблемы с преобразованием SAM, которое может возникать при наличии нескольких перегрузок параметров SAM для данного метода Java. Эта двусмысленность SAM смущает компилятор Kotlin, поскольку он не может решить, какой интерфейс он должен преобразовать, и ваш проект не сможет получить результат.

Эта двусмысленность SAM является особой проблемой при использовании RxJava 2.0 с Kotlin, так как многие операторы RxJava принимают несколько SAM-совместимых типов.

Давайте рассмотрим проблему преобразования SAM в действии. В следующем коде мы используем оператор zip() для объединения вывода двух наблюдаемых:

import androidx.appcompat.app.AppCompatActivity
import android.os.Bundle
import io.reactivex.Observable
import kotlinx.android.synthetic.main.activity_main.*

class MainActivity : AppCompatActivity() {

  override fun onCreate(savedInstanceState: Bundle?) {
      super.onCreate(savedInstanceState)
      setContentView(R.layout.activity_main)

//Start the stream when the button is clicked//

      button.setOnClickListener { startRStream() }
      
  }

  private fun startRStream() {

      val numbers = Observable.range(1, 6)

      val strings = Observable.just("One", "Two", "Three",

              "Four", "Five", "Six" )

      val zipped = Observable.zip(strings, numbers) { s, n -> "$s $n" }
      zipped.subscribe(::println)
  }
}

Это заставит компилятор Kotlin сбросить ошибку вывода типа. Однако RxKotlin предоставляет вспомогательные методы и функции расширения для затронутых операторов, включая Observables.zip(), которые мы используем в следующем коде:

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import io.reactivex.Observable
import io.reactivex.rxkotlin.Observables
import kotlinx.android.synthetic.main.activity_main.*

class MainActivity : AppCompatActivity() {

  override fun onCreate(savedInstanceState: Bundle?) {
      super.onCreate(savedInstanceState)
      setContentView(R.layout.activity_main)

//Start the stream when the button is clicked//

      button.setOnClickListener { startRStream() }

  }

  private fun startRStream() {

      val numbers = Observable.range(1, 6)

      val strings = Observable.just("One", "Two", "Three",

              "Four", "Five", "Six" )

      val zipped = Observables.zip(strings, numbers) { s, n -> "$s $n" }
      zipped.subscribe(::println)
  }


}

Вот результат этого кода:

Заключение

В этом уроке я показал вам, как начать использовать библиотеку RxJava в проектах Kotlin, в том числе с использованием ряда дополнительных поддерживающих библиотек, таких как RxKotlin и RxBinding. Мы рассмотрели, как вы можете создавать простые наблюдатели и наблюдаемых в RxKotlin, чтобы оптимизировать RxJava для платформы Kotlin, используя функции расширения.

До сих пор мы использовали RxJava для создания простых Observables, которые излучают данные, и наблюдателей, которые печатают эти данные в Logcat Android Studio, но это не то, как вы будете использовать RxJava в реальном мире!

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