Flow — nedir, Kotlin koroutine'lerinde cold ve hot akışlar

Kotlin'de Flow Nedir?

Flow, kotlinx.coroutines.flow paketinden, cold asenkron veri akışını temsil eden bir türdür. Özünde Flow, emit() işlevi aracılığıyla değerler yayınlayan ve başarıyla veya bir istisnayla sonlanan bir koroutine dizisidir. Akış toplama, bir suspend işlevi olan terminal operatörü collect() aracılığıyla gerçekleştirilir.

Cold Semantiği

Cold akış, flow oluşturucu içindeki kodun her abone için yeniden çalıştırılması anlamına gelir. RxJava'daki Observable.fromIterable benzer şekilde davranır: yeni bir abone tüm değerleri baştan alır. Flow'da bu, veri toplama süresi boyunca koroutine'i bloke eden suspend işlevi collect aracılığıyla uygulanır.

Flow Oluşturucular

Kotlin, Flow oluşturmanın birkaç yolunu sağlar: flow { } — emit() ile temel yapı, flowOf(vararg values) — sabit bir değer kümesi için, .asFlow() — koleksiyonlar ve Sequence için bir uzantı. Tüm oluşturucular cold'dur — veriler yalnızca terminal operatörü çağrıldığında üretilir.

Cold ve Hot Akışlar

Cold ve hot akışlar arasındaki ayrım, reaktif programlamanın temel bir kavramıdır. Cold akış (Flow, Observable) abonelikte veri üretimini başlatır. Hot akış (Channel, SharedFlow) bağımsız olarak veri yayınlar — abone, dizinin başlangıcı olmadan yalnızca abonelikten sonra olanları alır.

SharedFlow, birden çok aboneye sahip olabilen ve replay yapılandırıldığında son değerleri yeniden oynatabilen bir hot Flow'dur. SharedFlow, olaylar (tek seferlik bildirimler) için uygundur. StateFlow, yeni aboneler için son değeri önbelleğe alan sabit bir durum değerine sahip varyantıdır.

ChannelFlow, dahili olarak Channel kullanır, Flow ve Channel'ın özelliklerini birleştirir. capacity aracılığıyla tamponlama ve backpressure'ı destekler. ChannelFlow, değerlerin farklı koroutine'lerden yayınlandığı callback API'lerini reaktif akışa dönüştürürken kullanışlıdır.

Cold ve Hot Arasında Dönüşüm

Cold Flow'u hot SharedFlow'a dönüştürmek için shareIn(scope, started, replay) operatörü kullanılır. started parametresi başlangıç anını kontrol eder: SharingStarted.WhileSubscribed() — aboneler olduğu sürece aktif, Lazily — ilk abonede başlatma, Eagerly — hemen başlatma. Ters dönüşüm — hot'tan cold'a: StateFlow.asFlow(), collect'te StateFlow'un geçerli değerini yayınlayan bir cold Flow döndürür. Bu test için kullanışlıdır.

Flow Operatörleri

Flow, bir koroutine içinde suspend işlevleri olarak çalışan zengin bir operatör kümesi sağlar. Operatörler durumsuzdur ve yeni bir Flow döndürür — orijinal akış değişmeden kalır. Bu, yan etkiler olmadan güvenli dönüşüm zincirleri oluşturmayı sağlar.

map operatörü, her akış değerini asenkron veya senkron bir dönüşüm aracılığıyla dönüştürür. filter yalnızca koşulu karşılayan değerleri geçirir. catch, terminal operatöründen önce istisnaları yakalar ve akışın kurtarılmasına izin verir. flatMapLatest, yeni bir değer geldiğinde önceki yayını iptal eder — Rx'teki switchMap'e benzer.

Flow'taki debounce operatörü, değer yayınını belirtilen bir zaman aşımına kadar geciktirir. Bu süre içinde yeni bir değer gelirse, zamanlayıcı sıfırlanır. Android'de debounce, arama için kullanılır: istek yalnızca 300-400 ms'lik bir duraklamadan sonra gönderilir ve API çağrılarını 3-5 kat azaltır.

Terminal Operatörler

collect()'e ek olarak, Flow diğer terminal operatörlerini destekler: toList() tüm değerleri bir listede toplar — testler için kullanışlıdır, first() ilk öğeyi döndürür ve akışı iptal eder, single() tam olarak bir öğe bekler. fold(initial) iletilen bir işlev aracılığıyla değerleri biriktirir. Tüm terminal operatörler suspend işlevleridir ve bir koroutine veya başka bir suspend işlevi içinde çağrılmalıdır.

Flow Kod Örnekleri

İlk örnek — map operatörü aracılığıyla dönüşümle sayılar üreten temel bir Flow:

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

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

İkinci örnek — catch aracılığıyla filtreleme ve hata yönetimi içeren akış dönüşümü:

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

Üçüncü örnek — Jetpack Compose'da reaktif UI için ViewModel'de StateFlow kullanımı:

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

StateFlow, tek bir geçerli değere sahip bir hot Flow'dur. Son değeri önbelleğe alır ve yeni bir aboneye hemen iletir. StateFlow, durum için gözlemlenebilir bir kapsayıcıdır, equals karşılaştırmasını destekler — yeni değer mevcut değerle eşleşirse, hiçbir yayın gerçekleşmez. Jetpack Compose, collectAsState() aracılığıyla StateFlow kullanır.

SharedFlow, zorunlu bir başlangıç değeri olmadan daha esnek bir hot Flow'dur. SharedFlow, replay (yeni aboneler için değer sayısı), extraBufferCapacity (replay ötesinde tampon) ve onBufferOverflow (taşma durumunda strateji) aracılığıyla yapılandırılır. SharedFlow, tek seferlik olaylar için idealdir: navigasyon, Snackbar, analitik.

Android mimarisinde Flow, Google tarafından birincil veri kaynağı (Katman: Repository → UseCase → ViewModel) olarak önerilir. LiveData, esneklik açısından Flow'dan daha düşüktür: Flow, koroutine'leri, operatörleri, backpressure'ı destekler ve UI katmanının dışında çalışır. LiveData'dan Flow'a geçiş, modern Android projelerinde standart bir uygulamadır.

ViewModel'de Flow kullanırken doğru türü seçmek önemlidir. StateFlow, ekran döndürmeye dayanması gereken UI durumu için idealdir. SharedFlow, yeniden işlemenin kabul edilemez olduğu olaylar için uygundur — örneğin, navigasyon. lifecycleScope'da collect() ile Flow, yürütme bağlamı üzerinde maksimum kontrol sağlar ancak ekrandan çıkarken manuel iptal gerektirir.

Flow testi kotlinx-coroutines-test aracılığıyla yapılır. Kütüphane, gecikmeleri (delay) hızlandırmaya ve koroutine yürütme sırasını kontrol etmeye olanak tanıyan sanal zaman olan TestDispatcher'ı sağlar. TestScope.runTest { } Flow'u test etmek için izole bir ortam oluşturur. toList() operatörü, akışın doğru veri dizisini yayınlayıp yayınlamadığını doğrulamak için genellikle testlerde bir zaman aşımı ile tüm flow değerlerini toplamak için kullanılır.

Flow, Room (Android DB kütüphanesi) ile iyi entegre olur: DAO yöntemleri Flow<List<Entity>> döndürebilir. Room, her tablo değişikliğinde otomatik olarak yeni bir değer yayınlar — UI manuel tetikleyici olmadan güncellenir. Bu, perde arkasında callbackFlow ile Flow kullanan InvalidationTracker aracılığıyla uygulanır. Bu yaklaşım, LiveData ihtiyacını ortadan kaldırır ve veri katmanını tamamen koroutine odaklı hale getirir. Jetpack Compose, collectAsState() aracılığıyla StateFlow'a abone olur ve yalnızca verileri değişen bileşenleri yeniden çizer — bu, LiveData odaklı mimarilerle ulaşılamaz bir performans sağlar. DataStore (SharedPreferences'ın yerine geçen) da Flow<Preferences> döndürerek, manuel güncelleme tetikleyicileri olmadan uygulama ayarlarının reaktif okunmasını sağlar.

Flow, JVM'de kotlinx-coroutines-core aracılığıyla ek kütüphane olmadan süreçler arası iletişimi destekler. Örneğin, Ktor'daki sunucu uygulamalarında Flow, gelen WebSocket mesaj akışını temsil edebilir. Her mesaj akışa yayınlanır, operatörler aracılığıyla filtreleme ve toplamadan geçer ve sonuç istemciye gönderilir. Bu yaklaşım, Kotlin projelerinde Reactor veya RxJava gibi reaktif kütüphanelerin yerini alır.

Mevcut RxJava koduyla Flow uyumluluğu kotlinx-coroutines-rx3 modülü tarafından sağlanır. Uzantı işlevi Flow.asObservable(), Flow'u RxJava 3'ün Observable'ına dönüştürür. Ters dönüşüm — CompletableSource.asFlow(), Observable.asFlow(). Bu, RxJava'dan koroutine'lere geçişi basitleştirir: proje aşamalı olarak yeniden yazılabilir, bazı katmanlar RxJava'da bırakılabilir. Dönüşüm sırasında, cold/hot semantiğindeki fark dikkate alınmalıdır: Observable hem cold hem de hot olabilir, Flow normal Flow için her zaman cold, SharedFlow için hot'tur.

Flow Hata Yönetimi ve Test

Flow'da hata yönetiminin bir özelliği vardır: terminal operatöründen önce flow oluşturucu içinde bir istisna oluşursa, catch'e iletilir. Oluşturucudan sonra bir operatörde istisna oluşursa, bu operatörden sonraki catch tarafından yakalanır. retryWhen, bir koşul ile aboneliği yeniden denemeye izin verir: ağ hatasında 3 defaya kadar yeniden dene, ancak CancellationException'da yeniden deneme. Flow, durum depolamadığı için duruma bağlı hataları ortadan kaldırır — bu, Subject'in dahili durum depoladığı Observable'a kıyasla hata ayıklamayı basitleştirir.

kotlinx-coroutines-test ile Flow testi, gecikmeleri simüle etmek için TestDispatcher kullanır. Turbine, Flow testi için popüler bir topluluk kütüphanesidir: test { } Flow'u başlatır, awaitItem() sonraki değeri bekler, awaitComplete() tamamlanmayı bekler. Turbine, testlerin takılmasını önleyen varsayılan bir zaman aşımı ekler. StateFlow'u test etmek için, kronolojik sırada değer doğrulamasıyla .testIn(scope) kullanın.