Flow — bu nima, Kotlin korutinlarida cold va hot strimlar

Kotlin-da Flow nima?

Flow — kotlinx.coroutines.flow paketidagi tur bo'lib, sovuq asinxron ma'lumotlar oqimini ifodalaydi. Mohiyatan, Flow emit() funksiyasi orqali qiymatlarni emit qiladigan va yoki muvaffaqiyatli yoki istisno bilan tugaydigan korutin ketma-ketligidir. Oqimni yig'ish terminal operator collect() orqali amalga oshiriladi, bu suspend-funksiyadir.

Sovuq semantika

Cold stream flow-builder ichidagi kod har bir obunachi uchun qayta bajarilishini anglatadi. RxJava-dagi Observable.fromIterable xuddi shunday harakat qiladi: yangi obunachi barcha qiymatlarni boshidan oladi. Flow-da bu ma'lumotlarni yig'ish vaqti davomida korutinani bloklaydigan suspend-funksiya collect orqali amalga oshiriladi.

Flow builder-lar

Kotlin Flow yaratish uchun bir necha usulni taqdim etadi: flow { } — emit() bilan asosiy konstruksiya, flowOf(vararg values) — sobit qiymatlar to'plami uchun, .asFlow() — kolleksiyalar va Sequence uchun kengaytma. Barcha builder-lar sovuq — ma'lumotlar faqat terminal operator chaqirilganda yaratiladi.

Cold va Hot strimlar

Cold va hot strimlarga bo'linish reaktiv dasturlashning asosiy kontseptsiyasidir. Cold stream (Flow, Observable) obuna bo'lganda ma'lumotlar yaratilishini boshlaydi. Hot stream (Channel, SharedFlow) ma'lumotni mustaqil emit qiladi — obunachi faqat obunadan keyin sodir bo'layotgan narsalarni oladi, ketma-ketlikning boshini olmaydi.

SharedFlow — ko'p sonli obunachilarga ega bo'lishi mumkin bo'lgan va replay o'rnatilganda oxirgi qiymatlarni takrorlay oladigan issiq Flow. SharedFlow hodisalar uchun mos keladi (bir martalik bildirishnomalar). StateFlow — yangi obunachilar uchun oxirgi qiymatni keshlaydigan sobit holat qiymatiga ega uning turidir.

ChannelFlow ostida Channel-dan foydalanadi, Flow va Channel xususiyatlarini birlashtiradi. Bu sig'im (capacity) orqali buferlash va backpressure-ni qo'llab-quvvatlaydi. ChannelFlow qiymatlar turli korutinlardan emit qilinganda, callback-API-ni reaktiv oqimga aylantirishda foydalidir.

Cold va hot o'rtasida konvertatsiya

Cold Flow-ni hot SharedFlow-ga aylantirish uchun shareIn(scope, started, replay) operatori ishlatiladi. started parametri boshlanish momentini boshqaradi: SharingStarted.WhileSubscribed() — obunachilar bor ekan faol, Lazily — birinchi obunachida boshlanish, Eagerly — darhol boshlanish. Teskari konvertatsiya — hot dan cold-ga: StateFlow.asFlow() collect vaqtida StateFlow ning joriy qiymatini emit qiladigan sovuq Flow-ni qaytaradi. Bu testlash uchun qulay.

Flow operatorlari

Flow korutin ichida suspend-funksiyalar sifatida ishlaydigan boy operatorlar to'plamini taqdim etadi. Operatorlarning holati yo'q va ular yangi Flow qaytaradi — asl oqim o'zgarishsiz qoladi. Bu yon ta'sirlarsiz xavfsiz transformatsiya zanjirlarini qurish imkonini beradi.

map operatori oqimning har bir qiymatini asinxron yoki sinxron transformatsiya orqali o'zgartiradi. filter faqat shartni qondiradigan qiymatlarni o'tkazadi. catch terminal operatorgacha bo'lgan istisnolarni ushlaydi va oqimni tiklash imkonini beradi. flatMapLatest yangi qiymat kelganda oldingi emissiyani bekor qiladi — Rx-dagi switchMap-ga o'xshaydi.

Flow-da debounce operatori qiymatning nashr etilishini belgilangan taym-autga kechiktiradi. Agar bu vaqt ichida yangi qiymat kelsa — taymer qayta o'rnatiladi. Android-da debounce qidirish uchun ishlatiladi: so'rov faqat 300-400 ms pauzadan so'ng yuboriladi, bu API chaqiruvlari sonini 3-5 marta kamaytiradi.

Terminal operatorlar

collect() dan tashqari, Flow boshqa terminal operatorlarni qo'llab-quvvatlaydi: toList() barcha qiymatlarni ro'yxatga yig'adi — testlar uchun foydali, first() birinchi elementni qaytaradi va oqimni bekor qiladi, single() aynan bitta elementni kutadi. fold(initial) qiymatlarni berilgan funksiya orqali yig'adi. Barcha terminal operatorlar suspend-funksiyalardir va korutin yoki boshqa suspend-funksiya ichida chaqirilishi kerak.

Flow kod misollari

Birinchi misol — map operatori bilan raqamlarni yaratish va transformatsiya qilish bilan asosiy Flow:

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

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

Ikkinchi misol — filtrlash va catch orqali xatolarni boshqarish bilan oqim transformatsiyasi:

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

Uchinchi misol — Jetpack Compose-da reaktiv UI uchun ViewModel-da StateFlow dan foydalanish:

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 va SharedFlow

StateFlow — yagona joriy qiymatga ega issiq Flow. Oxirgi qiymatni keshlaydi va uni darhol yangi obunachiga uzatadi. StateFlow holat uchun Observable konteyneri bo'lib, equals taqqoslashni qo'llab-quvvatlaydi — agar yangi qiymat joriy bilan mos kelsa, emissiya sodir bo'lmaydi. Jetpack Compose StateFlow-dan collectAsState() orqali foydalanadi.

SharedFlow — majburiy boshlang'ich qiymatsiz yanada moslashuvchan issiq Flow. SharedFlow replay (yangi obunachilar uchun qiymatlar soni), extraBufferCapacity (replay-dan tashqari bufer) va onBufferOverflow (to'lib ketish strategiyasi) orqali sozlanadi. SharedFlow bir martalik hodisalar uchun ideal: navigatsiya, Snackbar, analitika.

Flow Android arxitekturasida Google tomonidan asosiy ma'lumot manbai sifatida tavsiya etiladi (Qatlam: Repository → UseCase → ViewModel). LiveData moslashuvchanlik bo'yicha Flow-dan past: Flow korutinlar, operatorlar, backpressure ni qo'llab-quvvatlaydi va UI qatlamidan tashqarida ishlaydi. LiveData-dan Flow-ga migratsiya zamonaviy Android loyihalarida standart amaliyotdir.

ViewModel-da Flow dan foydalanishda to'g'ri turni tanlash muhimdir. StateFlow ekran aylanishiga bardosh berishi kerak bo'lgan UI holati uchun idealdir. SharedFlow qayta ishlash mumkin bo'lmagan hodisalar uchun mos keladi — masalan, navigatsiya. lifecycleScope da collect() bilan Flow bajarish konteksti ustida maksimal nazorat beradi, lekin ekrandan chiqishda qo'lda bekor qilishni talab qiladi.

Flow ni testlash kotlinx-coroutines-test orqali amalga oshiriladi. Kutubxona TestDispatcher — kechikishlarni (delay) tezlashtirish va korutinlarning bajarilish tartibini boshqarish imkonini beruvchi virtual vaqtni taqdim etadi. TestScope.runTest { } Flow ni testlash uchun izolyatsiyalangan muhit yaratadi. toList() operatori tez-tez testlarda oqim to'g'ri ma'lumotlar ketma-ketligini emit qilganligini tekshirish uchun barcha flow qiymatlarini taym-aut bilan yig'ish uchun ishlatiladi.

Flow Room (Android ma'lumotlar bazasi kutubxonasi) bilan yaxshi integratsiyalashadi: DAO metodlari Flow<List<Entity>> qaytarishi mumkin. Room jadvaldagi har qanday o'zgarishda avtomatik ravishda yangi qiymat emit qiladi — UI qo'lda trigersiz yangilanadi. Bu ostida callbackFlow bilan Flow dan foydalanadigan InvalidationTracker orqali amalga oshiriladi. Bunday yondashuv LiveData ehtiyojini yo'q qiladi va ma'lumotlar qatlamini to'liq korutinga yo'naltirilgan qiladi. Jetpack Compose collectAsState() orqali StateFlow-ga obuna bo'ladi va faqat ma'lumotlari o'zgargan komponentlarni qayta chizadi — bu LiveData-ga yo'naltirilgan arxitekturalar bilan erishib bo'lmaydigan samaradorlikni beradi. DataStore (SharedPreferences o'rnini bosuvchi) ham Flow<Preferences> qaytaradi, qo'lda yangilash trigerlarisiz ilova sozlamalarining reaktiv o'qilishini ta'minlaydi.

Flow qo'shimcha kutubxonalarsiz JVM-da kotlinx-coroutines-core orqali jarayonlararo aloqani qo'llab-quvvatlaydi. Masalan, Ktor-dagi server ilovalarida Flow kiruvchi WebSocket xabarlarining oqimini ifodalashi mumkin. Har bir xabar oqimga emit qilinadi, operatorlar orqali filtrlash va agregatsiyadan o'tadi va natija mijozga yuboriladi. Bunday yondashuv Kotlin loyihalarida Reactor yoki RxJava kabi reaktiv kutubxonalarni almashtiradi.

Flow ning mavjud RxJava kodi bilan mosligi kotlinx-coroutines-rx3 moduli bilan ta'minlanadi. Flow.asObservable() kengaytma funksiyasi Flow-ni RxJava 3 dan Observable-ga aylantiradi. Teskari konvertatsiya — CompletableSource.asFlow(), Observable.asFlow(). Bu RxJava-dan korutinlarga migratsiyani soddalashtiradi: loyihani bosqichma-bosqich qayta yozish, qatlamlarning bir qismini RxJava-da qoldirish mumkin. Konvertatsiyada cold/hot semantikasidagi farqni hisobga olish kerak: Observable ham cold ham hot bo'lishi mumkin, Flow oddiy Flow uchun har doim cold va SharedFlow uchun hot.

Flow-da xatolarni boshqarish va testlash

Flow-da xatolarni boshqarishning o'ziga xosligi bor: agar istisno flow-builder ichida terminal operatorgacha yuz bersa, catch-ga uzatiladi. Agar istisno builder-dan keyingi operatorda yuz bersa, uni ushbu operatordan keyingi catch ushlaydi. retryWhen shart bilan obunani takrorlash imkonini beradi: tarmoq xatosida 3 martagacha takrorla, lekin CancellationException da takrorlama. Flow holatga bog'liq xatolarni bartaraf qiladi, chunki holatni saqlamaydi — bu ichki holatni saqlaydigan Subject mavjud bo'lgan Observable bilan solishtirganda disk raskadrovkani soddalashtiradi.

Flow ni kotlinx-coroutines-test bilan testlash kechikishlarni simulyatsiya qilish uchun TestDispatcher dan foydalanadi. Turbine — Flow ni testlash uchun jamoa tomonidan mashhur kutubxona: test { } Flow-ni ishga tushiradi, awaitItem() keyingi qiymatni kutadi, awaitComplete() tugashni kutadi. Turbine standart taym-aut qo'shadi, bu testlarning osilib qolishining oldini oladi. StateFlow ni testlash uchun qiymatlarni xronologik tartibda tekshirish bilan .testIn(scope) dan foydalaning.