Kotlin et RxJava

Pour commencer avec RXJava sous Kotlin, vous trouverez un article détaillé sur le lien suivant :

https://code.tutsplus.com/fr/tutorials/kotlin-reactive-programming-with-rxjava-and-rxkotlin–cms-31577

 

Je me suis appuyé sur cet article pour créer le projet disponible sur github

Dans ce projet, vous trouverez les méthodes suivantes dans l’activité principale :

 

Méthode 1

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

       button.setOnClickListener {

           // Méthode 1
           startRxStream()
       }
   }

   private fun startRxStream() {

       // Creation d'un observable
       val myObservable = getObservable()

       // Création d'un observer
       var myObserver = getObserver()

       myObservable.subscribe(myObserver)
   }

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

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

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

           override fun onError(e: Throwable) {
               Log.d(TAG, "On Error : ${e.message}")
           }

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

Cette méthode correspond à celle décrite dans l’article ci-dessus.

 

Méthode 2

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

    button.setOnClickListener {

        // Méthode 2
        startRxStream2()
    }
}

private fun startRxStream2() {
    Observable.just("14", "15", "16", "17")
        .doOnNext {
            Log.d(TAG, "onNext : $it")
        }
        .doOnComplete {
            Log.d(TAG, "OnCompletable startRxStream2")
        }
        .doOnError {
            Log.d(TAG, "On Error : ${it.message}")
        }
        .subscribe()
}

 

Méthode 4

Class MainActivity

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

       button.setOnClickListener {

           // Méthode 4
           startRxStream4()
       }
   }

private fun startRxStream4() {
       Observable.just("21", "22", "23", "24")
           .ioToMainThread()   // Utilisation d'une extension pour factorisation
           .subscribe()

   }

Class ThreadExtension.kt – Permettant de créer une extension ioToMainThread() à Observable

import android.util.Log
import io.reactivex.Observable
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.schedulers.Schedulers

fun <T> Observable<T>.ioToMainThread(): Observable<T> =
    subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .doOnNext {
        Log.d("ThreadExtension", "onNext : $it")
    }

 

Le projet est disponible sur github.