Jak jednocześnie uruchomić dwa zapytania sieciowe za pomocą Coroutine i RxJava?

Synchronizacja wykonania dwóch Coroutine za pomocą RxJava bezpośrednio jest niemożliwa, ponieważ są to dwa różne paradygmaty programowania asynchronicznego. Coroutine używają funkcji suspend i strukturalnej konkurencji, podczas gdy RxJava korzysta z Observable i operatorów.

Jednak, jeśli trzeba poczekać na zakończenie dwóch niezależnych operacji asynchronicznych i kontynuować dopiero po ich zakończeniu (wzorzec "czekaj na wszystkich"), można użyć operatorów RxJava do łączenia wyników lub oczekiwania na zakończenie Observable.

Przykład z użyciem Coroutine do wykonywania zapytań sieciowych i RxJava do ich łączenia (z użyciem Completable.mergeArray lub podobnego do oczekiwania na zakończenie, jeśli same zapytania zwracają Completable, lub łączenia wyników, jeśli zwracają Single lub Maybe):

Załóżmy, że mamy dwie metody suspend wykonujące zapytania sieciowe:

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

I chcemy uruchomić je równolegle i połączyć wyniki. Najpierw konwertujemy funkcje suspend na Observable lub 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())
}

Następnie, używamy operatorów RxJava do ich połączenia:

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

// ... w jakimś Scope, gdzie masz dostęp do CoroutineScope
val scope: CoroutineScope = ... // np. viewModelScope

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

Singles.zip(
    single1,
    single2,
    Func2 { data1, data2 ->
        // Przetwarzanie otrzymanych danych data1 i data2
        Pair(data1, data2) // zwracamy wynik połączony
    }
)
.observeOn(AndroidSchedulers.mainThread()) // przełącz na główny wątek dla UI
.subscribe(
    { combinedResult ->
        // obsługa połączonego wyniku na głównym wątku
        println("Otrzymane dane: data1: ${combinedResult.first}, data2: ${combinedResult.second}")
    },
    { error ->
        // obsługa błędów
        error.printStackTrace()
    }
)

W tym podejściu Coroutine wykonują rzeczywiste operacje asynchroniczne (zapytania sieciowe), a RxJava służy do zarządzania wątkami (observeOn, subscribeOn) i łączenia wyników dwóch niezależnych Observable (zip).

Alternatywnie, przy pełnym przejściu na Coroutine, równoległe wykonanie osiąga się za pomocą async i await:

import kotlinx.coroutines.*

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

    // Oczekiwanie na zakończenie obu korutyn
    val data1 = deferred1.await()
    val data2 = deferred2.await()

    Pair(data1, data2)
}

// Wywołanie z CoroutineScope
scope.launch {
    try {
        val (data1, data2) = fetchDataParallel(apiService)
        // Przetwarzanie danych
    } catch (e: Exception) {
        // Obsługa błędów
    }
}

Drugi sposób (czyste Coroutine) jest bardziej zalecany przy nowym rozwoju lub migracji, ponieważ eliminuje konieczność mieszania dwóch różnych bibliotek do zarządzania asynchronicznością.