Hoe start je twee netwerkverzoeken tegelijk met…

Het synchroniseren van de uitvoering van twee Coroutine's is direct onmogelijk met RxJava, omdat dit twee verschillende asynchrone programmeerparadigma's zijn. Coroutine gebruikt suspend-functies en gestructureerde concurrency, terwijl RxJava Observable en operators gebruikt.

Echter, als je wilt wachten op de voltooiing van twee onafhankelijke asynchrone operaties en alleen verder wilt gaan nadat beide voltooid zijn (het patroon "wacht op allemaal"), kun je RxJava operators gebruiken om de resultaten te combineren of te wachten op de voltooiing van Observable.

Een voorbeeld met Coroutine voor het uitvoeren van netwerkverzoeken en RxJava voor het combineren ervan (met gebruik van Completable.mergeArray of vergelijkbaar voor wachten op voltooiing, als de verzoeken zelf Completable teruggeven, of het combineren van resultaten als ze Single of Maybe teruggeven):

Stel dat we twee suspend-methoden hebben die netwerkverzoeken uitvoeren:

interface ApiService {
    suspend fun fetchData1(): Data1
    suspend fun fetchData2(): Data2
}

En we willen ze parallel uitvoeren en de resultaten combineren. Eerst converteren we de suspending-functies naar Observable of Single:

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.launch
import rx.Single
import rx.schedulers.Schedulers

fun CoroutineScope.fetchData1AsSingle(apiService: ApiService): Single<Data1> {
    return Single.create { subscriber ->
        launch {
            try {
                val data = apiService.fetchData1()
                subscriber.onSuccess(data)
            } catch (e: Throwable) {
                subscriber.onError(e)
            }
        }
    }.subscribeOn(Schedulers.io())
}

fun CoroutineScope.fetchData2AsSingle(apiService: ApiService): Single<Data2> {
    return Single.create { subscriber ->
        launch {
            try {
                val data = apiService.fetchData2()
                subscriber.onSuccess(data)
            } catch (e: Throwable) {
                subscriber.onError(e)
            }
        }
    }.subscribeOn(Schedulers.io())
}

Nu gebruiken we RxJava operators om de resultaten te combineren:

import rx.Single
import rx.android.schedulers.AndroidSchedulers
import rx.functions.Func2
import rx.singles.Singles

// ... binnen een Scope waar CoroutineScope beschikbaar is
val scope: CoroutineScope = ... // Bijvoorbeeld viewModelScope

val single1: Single<Data1> = scope.fetchData1AsSingle(apiService)
val single2: Single<Data2> = scope.fetchData2AsSingle(apiService)

Singles.zip(
    single1,
    single2,
    Func2 { data1, data2 ->
        // Verwerking van de verkregen data data1 en data2
        Pair(data1, data2) // Teruggeven van de gecombineerde resultaat
    }
)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
    { combinedResult ->
        // Verwerking van het gecombineerde resultaat op de hoofdthread
        println("Got data1: ${combinedResult.first}, data2: ${combinedResult.second}")
    },
    { error ->
        // Foutafhandeling
        error.printStackTrace()
    }
)

In deze aanpak worden Coroutine gebruikt voor het uitvoeren van de feitelijke asynchrone operaties (netwerkverzoeken), en RxJava voor het beheren van threads (observeOn, subscribeOn) en het combineren van de resultaten van twee onafhankelijke Observable (zip).

Alternatief, bij volledige overgang naar Coroutine, wordt parallelle uitvoering bereikt met async en await:

import kotlinx.coroutines.*

suspend fun fetchDataParallel(apiService: ApiService): Pair<Data1, Data2> = coroutineScope {
    val deferred1 = async { apiService.fetchData1() }
    val deferred2 = async { apiService.fetchData2() }

    // Wachten op de voltooiing van beide coroutines
    val data1 = deferred1.await()
    val data2 = deferred2.await()

    Pair(data1, data2)
}

// Aanroepen vanuit CoroutineScope
scope.launch {
    try {
        val (data1, data2) = fetchDataParallel(apiService)
        // Verwerking van data
    } catch (e: Exception) {
        // Foutafhandeling
    }
}

De tweede aanpak (pure Coroutine) is meer aanbevolen bij nieuwe ontwikkeling of migratie, omdat het de noodzaak wegneemt om twee verschillende bibliotheken voor asynchrone controle te combineren.

Hoe start je twee netwerkverzoeken tegelijk met behulp van Coroutine en RxJava?