Comment lancer deux requêtes réseau simultanément en utilisant Coroutine et RxJava?

Il n'est pas possible de synchroniser directement l'exécution de deux Coroutines avec RxJava, car ce sont deux paradigmes différents de programmation asynchrone. Les Coroutines utilisent des fonctions suspendues et une concurrence structurée, tandis que RxJava utilise des Observable et des opérateurs.

Cependant, si vous devez attendre la fin de deux opérations asynchrones indépendantes et continuer uniquement après leur achèvement (modèle "attendre tous"), vous pouvez utiliser des opérateurs RxJava pour combiner les résultats ou attendre la fin d'un Observable.

Un exemple utilisant Coroutines pour effectuer des requêtes réseau et RxJava pour les combiner (en utilisant Completable.mergeArray ou similaire pour attendre la fin, si les requêtes renvoient Completable, ou en combinant les résultats si elles renvoient Single ou Maybe) :

Supposons que vous avez deux méthodes suspendues effectuant des requêtes réseau :

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

Et vous souhaitez les lancer en parallèle et combiner les résultats. Convertissez d'abord les fonctions suspendues en Observable ou 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())
}

Puis, utilisez les opérateurs RxJava pour les combiner :

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

// ... dans un Scope où tu as accès à CoroutineScope
val scope: CoroutineScope = ... // par exemple, viewModelScope

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

Singles.zip(
    single1,
    single2,
    Func2 { data1, data2 ->
        // Traiter les données reçues data1 et data2
        Pair(data1, data2) // retourner le résultat combiné
    }
)
.observeOn(AndroidSchedulers.mainThread()) // passer au thread principal pour UI
.subscribe(
    { combinedResult ->
        // Traiter le résultat combiné sur le thread principal
        println("Données reçues : data1: ${combinedResult.first}, data2: ${combinedResult.second}")
    },
    { error ->
        // Gérer les erreurs
        error.printStackTrace()
    }
)

Dans cette approche, les Coroutines effectuent les opérations asynchrones réelles (requêtes réseau), et RxJava est utilisé pour gérer les threads (observeOn, subscribeOn) et combiner les résultats de deux Observable indépendants (zip).

Alternativement, si vous migrez complètement vers Coroutines, l'exécution parallèle se fait avec async et await :

import kotlinx.coroutines.*

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

    // Attendre la fin des deux coroutines
    val data1 = deferred1.await()
    val data2 = deferred2.await()

    Pair(data1, data2)
}

// Depuis un CoroutineScope
scope.launch {
    try {
        val (data1, data2) = fetchDataParallel(apiService)
        // Traiter les données
    } catch (e: Exception) {
        // Gérer les erreurs
    }
}

La deuxième approche (Coroutines pures) est plus recommandée pour les nouveaux développements ou migrations, car elle élimine la nécessité de mélanger deux bibliothèques différentes pour gérer l'asynchronie.