Coroutine və RxJava istifadə edərək eyni zamanda iki…

İki Coroutine-in icra edilməsini birbaşa RxJava ilə sinxronizasiya etmək mümkün deyil, çünki bunlar iki fərqli asinxron proqramlaşdırma paradigmasıdır. Coroutine suspend-funksiyalar və strukturlaşdırılmış konkurrentlik istifadə edir, RxJava isə Observable və operatorlar vasitəsilə işləyir.

Ancaq, əgər iki müstəqil asinxron əməliyyatın tamamlanmasını gözləmək və yalnız hər ikisi tamamlandıqdan sonra davam etmək istəyirsinizsə ("hamısı üçün gözləmə" nümunəsi), RxJava operatorlarından istifadə edərək nəticələri birləşdirə və ya Observable-in tamamlanmasını gözləyə bilərsiniz.

Coroutine istifadə edərək şəbəkə sorğularını yerinə yetirmək və RxJava ilə onları birləşdirmək üçün nümunə (Completable.mergeArray və ya oxşar istifadə edərək tamamlanma gözləmək üçün, əgər sorğular Completable qaytarırsa, və ya nəticələri birləşdirmək üçün, əgər Single və ya Maybe qaytarırsa):

Tutaq ki, iki suspend-metodumuz var və şəbəkə sorğuları yerinə yetirir:

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

Və onları paralel şəkildə işə salıb nəticələri birləşdirmək istəyirik. Əvvəlcə suspend-funksiyaları Observable və ya Single-ə çeviririk:

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())
}

İndi isə nəticələri birləşdirmək üçün RxJava operatorlarından istifadə edirik:

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

// ... CoroutineScope daxilində, mövcuddur
val scope: CoroutineScope = ... // Məsələn, viewModelScope

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

Singles.zip(
    single1,
    single2,
    Func2 { data1, data2 ->
        // alınan data1 və data2 ilə işləmə
        Pair(data1, data2) // Birləşdirilmiş nəticəni qaytarır
    }
)
.observeOn(AndroidSchedulers.mainThread()) // UI üçün əsas thread-ə keçid
.subscribe(
    { combinedResult ->
        // əsas thread-də birləşdirilmiş nəticənin işlənməsi
        println("Got data1: ${combinedResult.first}, data2: ${combinedResult.second}")
    },
    { error ->
        // xətaların işlənməsi
        error.printStackTrace()
    }
)

Bu yanaşmada Coroutine-lər faktiki asinxron əməliyyatları (şəbəkə sorğuları) yerinə yetirmək üçün istifadə olunur, RxJava isə thread-ləri idarə etmək (observeOn, subscribeOn) və iki müstəqil Observable-in nəticələrini birləşdirmək üçün (zip) istifadə olunur.

Alternativ olaraq, tamamilə Coroutine-ə keçid edildikdə, paralel yerinə yetirmə async və await vasitəsilə əldə edilir:

import kotlinx.coroutines.*

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

    // Hər iki coroutine-in tamamlanmasını gözləmək
    val data1 = deferred1.await()
    val data2 = deferred2.await()

    Pair(data1, data2)
}

// CoroutineScope daxilində çağırış
scope.launch {
    try {
        val (data1, data2) = fetchDataParallel(apiService)
        // məlumatların işlənməsi
    } catch (e: Exception) {
        // xətaların işlənməsi
    }
}

İkinci yanaşma (tam Coroutine) yeni inkişaf və ya miqrasiya üçün daha tövsiyə olunur, çünki iki fərqli kitabxananın idarə olunması ehtiyacını aradan qaldırır.

Coroutine və RxJava istifadə edərək eyni zamanda iki şəbəkə sorğusunu necə başlatmaq olar?