Flow — bu nədir, Kotlin korutinlərində cold və hot strimlər

Kotlin-də Flow nədir?

Flow — kotlinx.coroutines.flow paketindən bir növdür, soyuq asinxron məlumat axınını təmsil edir. Mahiyyətcə Flow, emit() funksiyası vasitəsilə dəyərləri emit edən və ya uğurla, ya da istisna ilə bitən korutin ardıcıllığıdır. Axının toplanması terminal operator olan collect() vasitəsilə həyata keçirilir, bu da suspend-funksiyadır.

Soyuq semantika

Cold stream o deməkdir ki, flow-builder daxilindəki kod hər bir abunəçi üçün yenidən icra olunur. RxJava-da Observable.fromIterable eyni şəkildə davranır: yeni abunəçi bütün dəyərləri əvvəldən alır. Flow-da bu, məlumatların toplanması müddətində korutini bloklayan suspend-funksiya collect vasitəsilə tətbiq olunur.

Flow builder-lar

Kotlin Flow yaratmaq üçün bir neçə üsul təqdim edir: flow { } — emit() ilə əsas konstruksiya, flowOf(vararg values) — sabit dəyərlər dəsti üçün, .asFlow() — kolleksiyalar və Sequence üçün genişləndirmə. Bütün builder-lar soyuqdur — məlumatlar yalnız terminal operator çağırıldıqda yaradılır.

Cold və Hot strimlər

Cold və hot strimlərə bölünmə reaktiv proqramlaşdırmanın əsas konsepsiyasıdır. Cold stream (Flow, Observable) abunə zamanı məlumat yaradılmasını başladır. Hot stream (Channel, SharedFlow) məlumatı müstəqil emit edir — abunəçi yalnız abunədən sonra baş verənləri alır, ardıcıllığın əvvəlini almır.

SharedFlow — çox sayda abunəçisi ola bilən və replay qurulduqda son dəyərləri təkrarlaya bilən isti Flow-dur. SharedFlow hadisələr üçün uyğundur (birdəfəlik bildirişlər). StateFlow — yeni abunəçilər üçün son dəyəri keşləyən sabit vəziyyət dəyəri olan onun növüdür.

ChannelFlow altlıqda Channel istifadə edir, Flow və Channel xüsusiyyətlərini birləşdirir. O, tutum (capacity) vasitəsilə buferləşdirmə və backpressure dəstəkləyir. ChannelFlow, dəyərlər müxtəlif korutinlərdən emit edildikdə, callback-API-ni reaktiv axına çevirmək üçün faydalıdır.

Cold və hot arasında çevirmə

Cold Flow-u hot SharedFlow-a çevirmək üçün shareIn(scope, started, replay) operatoru istifadə olunur. started parametri başlanma anını idarə edir: SharingStarted.WhileSubscribed() — abunəçilər olduqca aktiv, Lazily — ilk abunəçidə başlanma, Eagerly — dərhal başlanma. Əks çevirmə — hot-dan cold-a: StateFlow.asFlow() collect zamanı StateFlow-un cari dəyərini emit edən soyuq Flow qaytarır. Bu test etmək üçün əlverişlidir.

Flow operatorları

Flow korutin daxilində suspend-funksiyalar kimi işləyən zəngin operatorlar dəsti təqdim edir. Operatorların vəziyyəti yoxdur və yeni Flow qaytarır — orijinal axın dəyişməz qalır. Bu, yan təsirlər olmadan təhlükəsiz çevrilmə zəncirləri qurmağa imkan verir.

map operatoru axının hər bir dəyərini asinxron və ya sinxron çevrilmə vasitəsilə dəyişdirir. filter yalnız şərti ödəyən dəyərləri buraxır. catch terminal operatora qədər istisnaları tutur və axını bərpa etməyə imkan verir. flatMapLatest yeni dəyər gəldikdə əvvəlki emisiyanı ləğv edir — Rx-də switchMap-ə bənzəyir.

debounce operatoru Flow-da dəyərin nəşrini göstərilən vaxta qədər gecikdirir. Əgər bu müddət ərzində yeni dəyər gələrsə — taymer sıfırlanır. Android-də debounce axtarış üçün istifadə olunur: sorğu yalnız 300-400 ms pauzadan sonra göndərilir, bu da API çağırışlarının sayını 3-5 dəfə azaldır.

Terminal operatorlar

collect()-dən əlavə, Flow digər terminal operatorları dəstəkləyir: toList() bütün dəyərləri siyahıya toplayır — testlər üçün faydalıdır, first() ilk elementi qaytarır və axını ləğv edir, single() tam bir element gözləyir. fold(initial) dəyərləri verilən funksiya vasitəsilə yığır. Bütün terminal operatorlar suspend-funksiyalardır və korutin və ya digər suspend-funksiya daxilində çağırılmalıdır.

Flow kod nümunələri

Birinci nümunə — map operatoru ilə ədədlərin yaradılması və çevrilməsi ilə əsas Flow:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "Sayı: $it" }
        .collect { value ->
            println(value)
        }
}

İkinci nümunə — filtrasiya və catch ilə xəta idarəetməsi ilə axın çevrilməsi:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

Üçüncü nümunə — Jetpack Compose-da reaktiv UI üçün ViewModel-də StateFlow istifadəsi:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow və SharedFlow

StateFlow — tək cari dəyəri olan isti Flow-dur. Son dəyəri keşləyir və onu dərhal yeni abunəçiyə ötürür. StateFlow vəziyyət üçün Observable konteyneridir, equals müqayisəsini dəstəkləyir — əgər yeni dəyər cari ilə üst-üstə düşərsə, emisiya baş vermir. Jetpack Compose StateFlow-dan collectAsState() vasitəsilə istifadə edir.

SharedFlow — məcburi ilkin dəyər olmadan daha çevik isti Flow-dur. SharedFlow replay (yeni abunəçilər üçün dəyərlərin sayı), extraBufferCapacity (replay-dən kənar bufer) və onBufferOverflow (daşma strategiyası) vasitəsilə konfiqurasiya olunur. SharedFlow birdəfəlik hadisələr üçün idealdır: naviqasiya, Snackbar, analitika.

Flow Android arxitekturasında Google tərəfindən əsas məlumat mənbəyi kimi tövsiyə olunur (Təbəqə: Repository → UseCase → ViewModel). LiveData çeviklik baxımından Flow-dan geri qalır: Flow korutinləri, operatorları, backpressure dəstəkləyir və UI təbəqəsindən kənarda işləyir. LiveData-dan Flow-a miqrasiya müasir Android layihələrində standart təcrübədir.

ViewModel-də Flow istifadə edərkən düzgün tipi seçmək vacibdir. StateFlow ekranın dönməsinə dözməli olan UI vəziyyəti üçün idealdır. SharedFlow təkrar emalı yolverilməz olan hadisələr üçün uyğundur — məsələn, naviqasiya. lifecycleScope-də collect() ilə Flow icra konteksti üzərində maksimum nəzarət verir, lakin ekrandan çıxarkən əl ilə ləğv tələb edir.

Flow-un test edilməsi kotlinx-coroutines-test vasitəsilə həyata keçirilir. Kitabxana TestDispatcher — gecikmələri (delay) sürətləndirməyə və korutinlərin icra sırasını idarə etməyə imkan verən virtual vaxt təqdim edir. TestScope.runTest { } Flow-u test etmək üçün izolyasiya edilmiş mühit yaradır. toList() operatoru tez-tez testlərdə axının düzgün məlumat ardıcıllığını emit etdiyini yoxlamaq üçün timeout ilə bütün flow dəyərlərini toplamaq üçün istifadə olunur.

Flow Room (Android üçün verilənlər bazası kitabxanası) ilə yaxşı inteqrasiya olunur: DAO metodları Flow<List<Entity>> qaytara bilər. Room cədvəldə hər hansı dəyişiklik olduqda avtomatik olaraq yeni dəyər emit edir — UI əl ilə tətik olmadan yenilənir. Bu, altlıqda callbackFlow ilə Flow istifadə edən InvalidationTracker vasitəsilə tətbiq olunur. Belə yanaşma LiveData ehtiyacını aradan qaldırır və məlumat təbəqəsini tam korutin yönümlü edir. Jetpack Compose collectAsState() vasitəsilə StateFlow-a abunə olur və yalnız məlumatları dəyişən komponentləri yenidən çəkir — bu, LiveData yönümlü arxitekturalarla əldə olunmayan performans verir. DataStore (SharedPreferences-in əvəzi) də Flow<Preferences> qaytarır, əl ilə yeniləmə tətikləri olmadan tətbiq parametrlərinin reaktiv oxunmasını təmin edir.

Flow əlavə kitabxanalar olmadan JVM-də kotlinx-coroutines-core vasitəsilə proseslərarası əlaqəni dəstəkləyir. Məsələn, Ktor-da server tətbiqlərində Flow daxil olan WebSocket mesajlarının axınını təmsil edə bilər. Hər bir mesaj axına emit edilir, operatorlar vasitəsilə filtrasiya və aqreqasiyadan keçir və nəticə müştəriyə göndərilir. Belə yanaşma Kotlin layihələrində Reactor və ya RxJava kimi reaktiv kitabxanaları əvəz edir.

Flow-un mövcud RxJava kodu ilə uyğunluğu kotlinx-coroutines-rx3 modulu ilə təmin edilir. Flow.asObservable() genişləndirmə funksiyası Flow-u RxJava 3-dən Observable-a çevirir. Əks çevirmə — CompletableSource.asFlow(), Observable.asFlow(). Bu, RxJava-dan korutinlərə miqrasiyanı asanlaşdırır: layihəni mərhələli şəkildə yenidən yazmaq, təbəqələrin bir hissəsini RxJava-da saxlamaq olar. Çevirmə zamanı cold/hot semantikasındakı fərqi nəzərə almaq lazımdır: Observable həm cold, həm də hot ola bilər, Flow adi Flow üçün həmişə cold, SharedFlow üçün hot-dur.

Flow-da xəta idarəetməsi və test etmə

Flow-da xətaların idarə edilməsinin özəlliyi var: əgər istisna flow-builder daxilində terminal operatora qədər yaranarsa, catch-ə ötürülür. Əgər istisna builder-dən sonrakı operatorda yaranarsa, onu bu operatordan sonrakı catch tutur. retryWhen şərtlə abunəni təkrarlamağa imkan verir: şəbəkə xətasında 3 dəfəyə qədər təkrarla, lakin CancellationException-da təkrarlama. Flow vəziyyətdən asılı xətaları aradan qaldırır, çünki vəziyyət saxlamır — bu, daxili vəziyyəti saxlayan Subject-in olduğu Observable ilə müqayisədə debug-u asanlaşdırır.

Flow-un kotlinx-coroutines-test ilə test edilməsi gecikmələri simulyasiya etmək üçün TestDispatcher istifadə edir. Turbine — Flow-u test etmək üçün icma tərəfindən məşhur kitabxanadır: test { } Flow-u işə salır, awaitItem() növbəti dəyəri gözləyir, awaitComplete() tamamlanmanı gözləyir. Turbine standart timeout əlavə edir, bu da testlərin donmasının qarşısını alır. StateFlow-u test etmək üçün dəyərləri xronoloji ardıcıllıqla yoxlamaqla .testIn(scope) istifadə edin.