Akışlar

Önceki ders tek bir değer üreten işleri ele aldı. Çoğu iş ise tek değer değil, zamanla gelen bir değer dizisi üretir: bir sensörden okumalar, bir listenin sayfaları, bir sayacın her adımı. Tembel dizi bu dizileri bekleyemez; elemanı üretirken çağıranı bloke eder. Akış (Flow) aynı fikri eş yordamlarla kurar: değerleri bir bir üretir ve üretici ile toplayıcı arada askıya alınabilir.

Akışın iki yüzü vardır. Soğuk akış (cold flow) flow { } ya da flowOf ile kurulur ve kendi başına hiçbir şey yapmaz; biri toplayana kadar bekler. Her toplayıcı için üretici blok baştan çalışır. Sıcak akış (hot flow) ise toplayıcıdan bağımsız bir durumu taşır; bu derste yalnız MutableStateFlow görülür, elde tutulan son değeri olan bir sıcak akıştır. Toplama işlemi için collect kullanılır; akışın bütün değerlerini bir listeye indirmek için toList() yeterlidir. Her ikisi de askıya alınan fonksiyondur ve bir eş yordam içinde çağrılır; bu derste yine runBlocking kapsamı kurulur ve import kotlinx.coroutines.* ile birlikte akışlar için import kotlinx.coroutines.flow.* gerekir. Zamanlama ölçülmez, yalnız değerler ve sayılar okunur. Ölçü sorusu önceki derslerdeki gibidir: bu iki değeri eşit yapan ne?

// Criterion.kt — iki degeri esit yapan olcutu siniflandirir
class Criterion(val label: String, first: Any?, twin: Any?, other: Any?) {
    val sameObject = first === twin
    val equalTwin = first == twin
    val equalOther = first == other

    val name: String = when {
        sameObject -> "tekil"
        !equalTwin -> "kimlik"
        !equalOther -> "icerik"
        else -> "kismi-icerik"
    }

    fun report() {
        println("$label: ayni-nesne=$sameObject, ikiz=$equalTwin, farkli=$equalOther -> $name")
    }
}

fun summarize(checks: List<Criterion>) {
    for (name in listOf("tekil", "kimlik", "icerik", "kismi-icerik")) {
        println("$name: ${checks.count { it.name == name }}")
    }
}

Soğuk Akış, Dönüşüm ve Sıcak Akış

Aşağıdaki blok önce flow { } ile bir akış kurar; üretici blok her çalıştığında bir sayacı artırır. Akış iki kez toplanır ve sayaç okunur. Sonra map ve filter ile dönüşüm ve toList() ile toplama gösterilir. Son olarak bir MutableStateFlow kurulur, değeri değiştirilir ve value ile okunur.

// Main.kt — soguk akis ve sicak akis
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    var runs = 0
    val numbers = flow {
        runs++
        emit(1)
        emit(2)
        emit(3)
    }
    println(runs)
    println(numbers.toList())
    println(numbers.toList())
    println(runs)
    println(flowOf(1, 2, 3, 4).map { it * 10 }.filter { it > 10 }.toList())
    val state = MutableStateFlow(5)
    state.value = 6
    println(state.value)
}

0
[1, 2, 3]
[1, 2, 3]
2
[20, 30, 40]
6

İlk satır, akış kurulduğunda üretici bloğun hiç çalışmadığını gösterir: runs sıfırdır. İki toList() çağrısı her seferinde [1, 2, 3] verir ve dördüncü satır üretici bloğun iki kez çalıştığını söyler. Bu soğuk akışın tanımıdır: her toplayıcı akışı baştan çalıştırır, akış önceki toplamadan bir şey hatırlamaz. Beşinci satırda map her değeri on ile çarpar, filter on’u eler ve [20, 30, 40] kalır; tembel dizideki gibi, ara işlemler akışı yalnız tarif eder ve toplama anında çalışır. Son satırda MutableStateFlow bir değer tutar ve value ile atanan son değer okunur: 6.

KE27. Bu ölçümler tek toplayıcılı akışlar ve tek iş parçacıklı runBlocking kapsamı için. Birden çok toplayıcının aynı anda toplaması, flowOn ile dağıtıcı değiştirme ve akışın dışarıdan cancel ile iptali ölçülmedi; take ile erken kesme ilerideki bir bölümde ölçülüyor.

Ölçüt: İçerik

MutableStateFlow ve distinctUntilChanged aynı soruyu sorar: yeni gelen değer öncekine eşit mi? Eşitse atlarlar. Bu soru bu kursun ölçü sorusudur, yani akışta eşitliği bir ölçüt belirler. Aşağıda önce iki yardımcı sınıf tanımlanır: Reading bir veri sınıfıdır, Raw ise aynı alana sahip sıradan sınıftır.

// Reading.kt — veri sinifi
data class Reading(val id: Int)

// Raw.kt — siradan sinif
class Raw(val id: Int)

Aşağıdaki blok Reading öğeli flowOf(A, A, B, B, A) akışına distinctUntilChanged() uygular ve kaç değerin kaldığını sayar. Hemen yanında Reading için Criterion ölçümü yapılır.

// Main.kt — art arda eslik
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val a = Reading(1)
    val b = Reading(2)
    val kept = flowOf(a, Reading(1), b, Reading(2), Reading(1)).distinctUntilChanged().toList()
    println(kept.size)
    Criterion("veri-sinifi", Reading(1), Reading(1), Reading(2)).report()
}

3
veri-sinifi: ayni-nesne=false, ikiz=true, farkli=false -> icerik

Beş değerden üçü kaldı: A, B, A. Art arda gelen ikinci Reading(1) ve ikinci Reading(2) elendi, çünkü distinctUntilChanged öncekiyle == karşılaştırır ve veri sınıfı equals’ı alanlara göre tanımlar. Ayrı bir nesne olan Reading(1), öncekine eşit sayıldı. İkinci satır aynı sonucu Criterion ile söyler: ikiz eşittir, farklı olan eşit değildir, ölçüt içerik. Bu dersin ölçütü budur: içerik.

KE28. Bu ölçümler Int alanlı bir veri sınıfı ve beş öğeli bir flowOf için. Anahtar seçen distinctUntilChangedBy, karşılaştırıcı alan distinctUntilChanged { a, b -> ... } biçimi ve uzun akışlarda kaç değerin elenebileceği ölçülmedi.

Sınırlayıcı Ölçüm

Aynı akışı sıradan sınıf öğeleriyle, MutableStateFlow’a eşit değer atamayı ve iki ayrı akışın kendisini ölçen blok:

// Main.kt — siradan sinif, durum ve akis
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val raws = flowOf(Raw(1), Raw(1), Raw(2), Raw(2), Raw(1)).distinctUntilChanged().toList()
    println(raws.size)
    val original = Reading(1)
    val state = MutableStateFlow(original)
    state.value = Reading(1)
    println(state.value === original)
    val checks = listOf(
        Criterion("akis", flowOf(1, 2), flowOf(1, 2), flowOf(1, 3)),
        Criterion("veri-sinifi", Reading(1), Reading(1), Reading(2)),
        Criterion("siradan-sinif", Raw(1), Raw(1), Raw(2)),
    )
    for (check in checks) check.report()
    summarize(checks)
}

5
true
akis: ayni-nesne=false, ikiz=false, farkli=false -> kimlik
veri-sinifi: ayni-nesne=false, ikiz=true, farkli=false -> icerik
siradan-sinif: ayni-nesne=false, ikiz=false, farkli=false -> kimlik
tekil: 0
kimlik: 2
icerik: 1
kismi-icerik: 0

Birinci satır, aynı distinctUntilChanged çağrısının sıradan sınıf öğelerde hiçbir değeri elemediğini gösterir: beş öğenin beşi de kaldı. Raw(1) ile Raw(1) içerik olarak aynı alanı taşısa da equals varsayılan kimliğe düşer ve ardışık değerler hiç eşit sayılmaz. İkinci satırda MutableStateFlow’a Reading(1) ile eşit ama ayrı bir nesne atandı ve value hâlâ ilk nesneye işaret ediyor (true): akış eşit değeri yeni değer saymadı ve eskisini yerinde bıraktı. Sıcak akış da soğuk akış gibi eşitliği == ile ölçer.

Kalan satırlar üç ölçümü yan yana koyar. İki ayrı flowOf(1, 2) akışı kimlik ölçütünü verir: akışın kendisi eşitliği içerikle tanımlamaz, tıpkı tembel dizi ve Deferred gibi. Veri sınıfı icerik, sıradan sınıf kimlik çıkar. Sınırlayıcı ölçüm şudur: akışın içindeki değerler içerik ölçütünü yalnız öğenin tipi taşıyorsa uygular. Akışı kuran tarif kimlikle, akışın içinden geçen değer ise tipinin eşitliğiyle karşılaştırılır. Özet tablosu da bunu söyler: iki kimlik (akış ve sıradan sınıf), bir icerik (veri sınıfı).

KE29. Bu ölçümler MutableStateFlow için ve eşit değerin value ile atanmasını kapsar. StateFlow’un toplayıcıya ilettiği değerler ve SharedFlow ölçülmedi; update ve compareAndSet ile atamalar ilerideki bir bölümde ölçülüyor.

KE30. Bu derste hiçbir zamanlama, iş parçacığı adı ya da dağıtıcı sıralaması basılmadı. Akışın değerleri toList() ile toplandı, çıktıdaki sıra üreticinin sırasıdır.

Akışı Erken Kesmek

Bir akışın üreticisi çok değer üretebilir, ama toplayıcı hepsini istemeyebilir. take(n) akıştan ilk n değeri alır ve sonra akışı iptal eder; üretici blok o noktadan sonra devam etmez. Tembel dizideki take ile aynı fikirdir, ama burada kesilen şey bir eş yordamdır. Aşağıdaki blok beş değer üretmeye hazır bir akış kurar. Üretici her değerden önce bir kayıt bırakır, böylece take(2) sonrasında üreticinin ne kadar ilerlediği listeden okunur. Ayrıca onStart ve onCompletion ile akışın başlangıcı ve bitişi kaydedilir.

// Main.kt — take akisi erken keser
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val log = mutableListOf<String>()
    val source = flow {
        for (i in 1..5) {
            log.add("uret$i")
            emit(i)
        }
    }
    val firstTwo = source
        .onStart { log.add("basla") }
        .onCompletion { log.add("bitti") }
        .take(2)
        .toList()
    println(firstTwo)
    println(log)
}

[1, 2]
[basla, uret1, uret2, bitti]

Toplanan liste [1, 2] oldu, yani yalnız iki değer içerir. Kayıt listesi basla ile açılır, uret1 ve uret2 ile sürer, bitti ile kapanır; uret3 kaydı hiç yoktur. Kayıtlar üreticinin üçüncü değere hiç uzanmadığını gösterir: take(2) ikinci değeri aldıktan sonra akışı kesti. onStart ve onCompletion kayıtları da listede yer alır ve bitiş kaydı kesilmiş akışta da düşer. Bu sıra güvencelidir, çünkü toList() döndüğünde akış tamamlanmıştır; kayıtlar bu noktadan sonra okunur. Pratik sonuç şudur: bir akışın ilk birkaç değeri yetiyorsa take ya da first() kullanmak, üretici bloğun geri kalanını çalıştırmaz. Üreticinin yan etkisi olan bir iş yaptığı yerde bu fark, çalışan ya da çalışmayan işin farkıdır.

KE31. Bu ölçümler take, onStart ve onCompletion için, tek toplayıcıyla. Akışın kesilmesi sırasında üretici bloktaki try/finally temizliği ve takeWhile ile kesme ölçülmedi.

Uç İşlemler ve Boş Akış

Akışı bir listeye çevirmek tek yol değildir. fold ve reduce bütün değerleri tek bir değere indirger, count değer sayısını verir, first() ilk değeri döndürür. Boş akış sınır durumudur: first() istisna fırlatır, firstOrNull() ise null verir. Aşağıdaki blok dört değerli bir akışta bu işlemleri çağırır; sonra boş bir akışta ikisini dener. fold bir başlangıç değeri ister; reduce ister istemez ilk değeri başlangıç sayar ve boş akışta çalışamaz.

// Main.kt — uc islemler ve bos akis
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val numbers = flowOf(1, 2, 3, 4)
    println(numbers.fold(100) { sum, n -> sum + n })
    println(numbers.reduce { sum, n -> sum + n })
    println(numbers.count { it % 2 == 0 })
    println(numbers.first())
    val empty = emptyFlow<Int>()
    try {
        empty.first()
    } catch (e: NoSuchElementException) {
        println(e::class.simpleName)
    }
    println(empty.firstOrNull())
}

110
10
2
1
NoSuchElementException
null

fold başlangıç değeri olan 100’e dört değeri ekledi. reduce başlangıç değeri olmadan ilk değerden başladı ve yalnız dört değerin toplamını verdi. count yalnız çift olanları saydı. first() akışın ilk değeridir. Boş akışta first() NoSuchElementException fırlattı, firstOrNull() ise null döndürdü. Koleksiyonlardaki first ve firstOrNull ikilisinin akıştaki karşılığı budur: hata görünür olsun isteniyorsa ilki, boşluk kabul edilebilirse ikincisi seçilir.

KE32. Bu ölçümler sonlu, Int değerli akışlar için. Hiç bitmeyen bir akışta fold ile count çağrısı ve reduce’un boş akıştaki istisnasının sınıf adı ölçülmedi.

Dönüştürmek ve Birleştirmek

map her değerden bir değer üretir. Bir değerden birden çok değer ya da hiç değer üretmek için transform kullanılır: blok gelen değeri alır ve istediği kadar emit çağırır. İki akışı yan yana okumak için zip iki akıştan sırayla birer değer alıp birleştirir; kısa olan akış bitince sonuç da biter. Aşağıdaki blok her ikisini de ölçer. İkinci akış üç yerine iki değer taşır.

// Main.kt — transform ve zip
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val doubled = flowOf(1, 2, 3).transform { n ->
        emit(n)
        if (n % 2 == 1) emit(n * 10)
    }.toList()
    println(doubled)
    val pairs = flowOf(1, 2, 3).zip(flowOf("a", "b")) { n, s -> "$n$s" }.toList()
    println(pairs)
}

[1, 10, 2, 3, 30]
[1a, 2b]

transform tek sayılar için iki değer üretti, çift sayı için tek değer üretti. Sonuç listesi giriş değerlerini ve türeyen değerleri art arda taşır. zip iki akıştan birer değer çiftleyip "1a" ve "2b" üretti; birinci akışın üçüncü değeri eşsiz kaldığı için hiçbir sonuç üretmedi. Bu iki işlemin sonucu zamanlamaya değil, yalnız giriş değerlerine ve sırasına bağlıdır, bu yüzden güvenle ölçülebilir. Kural açık: map bire bir, transform bire çok ya da bire sıfır dönüşüm yapar; zip ise kısa olanın uzunluğunda durur.

KE33. Bu ölçümler transform ve zip için, iki sonlu akışla. İki akışın zaman içinde değişen son değerlerini birleştiren combine ve flatMapConcat ölçülmedi; bunların sonucu zamanlamaya dayanabilir.

Durumu Güncellemek

MutableStateFlow’a değer atamak için value yeterlidir, ama okuma ile atama arasındaki boşluk güvenli değildir. update { } bloğu yeni değeri eskisinden hesaplar ve atamayı tek adımda yapar. compareAndSet(beklenen, yeni) ise yalnız mevcut değer beklenene eşitse atar ve atayıp atamadığını döndürür. Karşılaştırma == iledir, bu yüzden bu iki işlem de önceki bölümlerdeki eşitlik kuralına bağlıdır. Aşağıdaki blok bir sayıyı güncelleyip denetler.

// Main.kt — update ve compareAndSet
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val state = MutableStateFlow(0)
    state.update { it + 1 }
    println(state.value)
    println(state.compareAndSet(1, 5))
    println(state.value)
    println(state.compareAndSet(1, 9))
    println(state.value)
    println(state.first())
}

1
true
5
false
5
5

update değeri 0’dan 1’e çıkardı. İlk compareAndSet(1, 5) mevcut değer 1 olduğu için atadı ve true döndürdü. İkinci compareAndSet(1, 9) değer artık 5 olduğundan atamadı ve false döndürdü, değer 5 olarak kaldı. first() sıcak akışın o anki değerini verdi: durum akışı toplanınca hep mevcut değerle başlar. Bu, update ve compareAndSet işlemlerini sıradan bir value atamasından ayırır: ikisi de mevcut değeri okuyup karar verir ve bunu tek işlemde yapar. Soğuk akıştan farkı budur: sıcak akışın elinde bir değer vardır ve yeni toplayıcıya önce onu verir.

KE34. Bu ölçümler Int değerli bir MutableStateFlow ve tek eş yordam için. Birden çok eş yordamın aynı anda update çağırması ve StateFlow toplayıcısının ara değerleri görüp görmemesi ölçülmedi.

Özet

• Soğuk akış her toplamada üretici bloğu baştan çalıştırır; kurulduğunda hiçbir şey çalışmaz. map ve filter akışı dönüştürür, toList() toplar.
• MutableStateFlow son değeri tutan bir sıcak akıştır; eşit değer atanınca eski nesne yerinde kalır.
• distinctUntilChanged art arda gelen eşit değerleri == ile eler. Bu dersin ölçütü: içerik. Veri sınıfında 5 değerden 3’ü, sıradan sınıfta 5’i kalır.
• Sınırlayıcı ölçüm: akışın kendisi kimlikle, içindeki değerler ise tipinin eşitliğiyle karşılaştırılır.

Sonraki Adım

Şimdiye kadar her ders değerin eşit olup olmadığını sordu. Son ders, değer yerine bir hatanın eşit olup olmadığını sorar: aynı mesajı taşıyan iki istisna eşit midir? İstisna Yönetimi dersi bunu try, catch ve finally ile ölçer.