View source for RxJava

=Introduction=
These are notes on RxJava which I have used with Android/Kotlin. To use with Android at the time you specify the following
<syntaxhighlight lang="groovy">
dependencies {
...
    implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
    implementation 'io.reactivex.rxjava3:rxjava:3.0.13'
...
}
</syntaxhighlight>

=Observable Types=
==Singles==
Single is an Observable which only emits one item or throws an error. Single emits only one value and applying some of the operator makes no sense. Like we don’t want to take value and collect it to a list.
<syntaxhighlight lang="java">
interface SingleObserver<T> {
    void onSubscribe(Disposable d);
    void onSuccess(T value);
    void onError(Throwable error);
}
</syntaxhighlight>
==Maybes==
Maybe is similar to Single only difference being that it allows for no emissions as well.
<syntaxhighlight lang="java">
interface MaybeObserver<T> {
    void onSubscribe(Disposable d);
    void onSuccess(T value);
    void onError(Throwable error);
    void onComplete();
}
</syntaxhighlight>
==Completables==
Completable is only concerned with execution completion whether the task has reach to completion or some error has occurred.
<syntaxhighlight lang="java">
interface CompletableObserver<T> {
    void onSubscribe(Disposable d);
    void onComplete();    
    void onError(Throwable error);
}
</syntaxhighlight>
=Using RxJava with Retrofit=
This example is just to show a practical use of RxJava. 
==Retrofit==
Here is a very simple example of Retrofit which manages the API calls for use. It defines a service which returns a RxJava Single from an endpoint
<syntaxhighlight lang="kotlin">
class CountriesService {
    private val BASE_URL = "https://raw.githubusercontent.com"
    private val api: CountriesApi

    init {
        api = Retrofit.Builder()
            .baseUrl(BASE_URL)
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJava3CallAdapterFactory.create())
            .build()
            .create(CountriesApi::class.java)
    }

    fun getCountries(): Single<List<Country>> = api.getCountries()
}
</syntaxhighlight>
==Using in ViewModel==
We can now call the service by adding it to a CompositeDisposable. We need to create a new thread, specify the thread to receive the response and what to do with it for both success and error because it is a single. We also need to ensure we clean up should thing go south.
<syntaxhighlight lang="kotlin">
class ListViewModel : ViewModel() {

    private val countriesService = CountriesService()
    private val disposable = CompositeDisposable()

    val countries = MutableLiveData<List<Country>>()
    val countryLoadError = MutableLiveData<Boolean>()
    val loading = MutableLiveData<Boolean>()

    fun refresh() {
        fetchCountries()
    }

    private fun fetchCountries() {
        loading.value = true
        disposable.add(
            countriesService.getCountries()
                // Create new Thread
                .subscribeOn(Schedulers.newThread())
                // Provide Thread to received Response
                .observeOn(AndroidSchedulers.mainThread())
                // What to do with it
                .subscribeWith(object: DisposableSingleObserver<List<Country>>() {

                    // RxJava Success
                    override fun onSuccess(value: List<Country>?) {
                        countries.value = value
                        countryLoadError.value = false
                        loading.value = false
                    }
                    // RxJava error
                    override fun onError(e: Throwable) {
                        countryLoadError.value = true
                        loading.value = false
                    }
                })
        )
    }

    override fun onCleared() {
        super.onCleared()
        disposable.clear()
    }
}
</syntaxhighlight>