Android'de Kotlin akışı

Coroutine'larda flow, yalnızca tek bir değer döndüren askıya alma işlevlerinin aksine, birden fazla değeri sırayla yayınlayabilen bir türdür. Örneğin, bir akışı kullanarak veritabanından canlı güncellemeler alabilirsiniz.

Akışlar, eş yordamların üzerine kurulur ve birden fazla değer sağlayabilir. Akış, kavramsal olarak eşzamansız şekilde hesaplanabilen bir veri akışıdır. Yayılan değerler aynı türde olmalıdır. Örneğin, Flow<Int>, tam sayı değerleri veren bir akıştır.

Akış, bir değer dizisi üreten Iterator'ye çok benzer ancak değerleri eşzamansız olarak üretmek ve kullanmak için askıya alma işlevlerini kullanır. Bu, örneğin akışın ana iş parçacığını engellemeden bir sonraki değeri üretmek için güvenli bir şekilde ağ isteğinde bulunabileceği anlamına gelir.

Veri akışlarında üç öğe bulunur:

  • Üretici, akışa eklenen veriler üretir. Coroutines sayesinde akışlar, verileri eşzamansız olarak da üretebilir.
  • (İsteğe bağlı) Aracı kuruluşlar, akışa verilen her değeri veya akışın kendisini değiştirebilir.
  • Bir tüketici, akıştaki değerleri tüketir.

Veri akışlarına dahil olan öğeler; tüketici, isteğe bağlı aracılar ve üretici
Şekil 1. Veri akışlarına dahil olan öğeler: tüketici, isteğe bağlı aracı öğeler ve üretici.

Android'de depo, genellikle kullanıcı arayüzünü (UI) tüketici olarak kullanan ve verileri nihai olarak görüntüleyen bir kullanıcı arayüzü veri üreticisidir. Bazı durumlarda ise kullanıcı arayüzü katmanı, kullanıcı girişi etkinliklerinin üreticisi olur ve hiyerarşinin diğer katmanları bunları tüketir. Üretici ile tüketici arasındaki katmanlar genellikle, veri akışını bir sonraki katmanın gereksinimlerine uyacak şekilde değiştiren aracı olarak hareket eder.

Akış oluşturma

Akış oluşturmak için akış oluşturma aracı API'lerini kullanın. flow oluşturucu işlevi, emit işlevini kullanarak veri akışına manuel olarak yeni değerler gönderebileceğiniz yeni bir akış oluşturur.

Aşağıdaki örnekte, bir veri kaynağı en son haberleri sabit aralıklarla otomatik olarak getiriyor. Askıya alma işlevi birden fazla ardışık değer döndüremediğinden, veri kaynağı bu koşulu karşılamak için bir akış oluşturup döndürür. Bu durumda veri kaynağı üretici olarak hareket eder.

class NewsRemoteDataSource(
    private val newsApi: NewsApi,
    private val refreshIntervalMs: Long = 5000
) {
    val latestNews: Flow<List<ArticleHeadline>> = flow {
        while (true) {
            val latestNews = newsApi.fetchLatestNews()
            emit(latestNews) // Emits the result of the request to the flow
            delay(refreshIntervalMs) // Suspends the coroutine for some time
        }
    }
}

// Interface that provides a way to make network requests with suspend functions
interface NewsApi {
    suspend fun fetchLatestNews(): List<ArticleHeadline>
}

flow oluşturucu, bir eş yordam içinde yürütülür. Bu nedenle, aynı eşzamansız API'lerden yararlanır ancak bazı kısıtlamalar geçerlidir:

  • Akışlar sıralıdır. Üretici bir coroutine içinde olduğundan, askıya alma işlevi çağrıldığında askıya alma işlevi döndürülene kadar üretici askıya alınır. Örnekte, üretici fetchLatestNews ağ isteği tamamlanana kadar askıya alınır. Sonuç ancak bu durumda akışa gönderilir.
  • flow oluşturucu ile yapımcı, farklı bir CoroutineContext değerini emit yapamaz. Bu nedenle, yeni coroutine'ler oluşturarak veya withContext kod bloklarını kullanarak emit işlevini farklı bir CoroutineContext içinde çağırmayın. Bu durumlarda callbackFlow gibi diğer akış oluşturucuları kullanabilirsiniz.

Akışı değiştirme

Aracılar, değerleri kullanmadan veri akışını değiştirmek için ara operatörleri kullanabilir. Bu operatörler, bir veri akışına uygulandığında değerler gelecekte kullanılana kadar yürütülmeyen bir işlem zinciri oluşturan işlevlerdir. Ara operatörler hakkında daha fazla bilgiyi Akış referans belgelerinde bulabilirsiniz.

Aşağıdaki örnekte, depo katmanı View üzerinde gösterilecek verileri dönüştürmek için ara operatör map kullanır:

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData
) {
    /**
     * Returns the favorite latest news applying transformations on the flow.
     * These operations are lazy and don't trigger the flow. They just transform
     * the current value emitted by the flow at that point in time.
     */
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            // Intermediate operation to filter the list of favorite topics
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            // Intermediate operation to save the latest news in the cache
            .onEach { news -> saveInCache(news) }
}

Ara operatörler, akışa bir öğe gönderildiğinde tembel bir şekilde yürütülen bir işlem zinciri oluşturarak art arda uygulanabilir. Bir akışa yalnızca ara operatör uygulamanın akış toplama işlemini başlatmadığını unutmayın.

Akıştan toplama

Değerleri dinlemeye başlamak için akışı tetiklemek üzere terminal operatörü kullanın. Akışta yayınlanan tüm değerleri almak için collect kullanın. Terminal operatörleri hakkında daha fazla bilgiyi resmi akış belgelerinde bulabilirsiniz.

collect bir askıya alma işlevi olduğundan bir coroutine içinde yürütülmesi gerekir. Her yeni değerde çağrılan bir lambda'yı parametre olarak alır. Askıya alma işlevi olduğundan, collect işlevini çağıran ortak yordam, akış kapatılana kadar askıya alınabilir.

Önceki örneğe devam ederek, depo katmanındaki verileri tüketen bir ViewModel öğesinin basit bir uygulamasını aşağıda bulabilirsiniz:

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            // Trigger the flow and consume its elements using collect
            newsRepository.favoriteLatestNews.collect { favoriteNews ->
                // Update UI with the latest favorite news
            }
        }
    }
}

Toplama işlemi, en son haberleri yenileyen ve ağ isteğinin sonucunu sabit aralıklarla yayınlayan üreticiyi tetikler. Üretici, while(true) döngüsüyle her zaman etkin kaldığından ViewModel temizlendiğinde ve viewModelScope iptal edildiğinde veri akışı kapatılır.

Akış toplama işlemi aşağıdaki nedenlerle durabilir:

  • Önceki örnekte gösterildiği gibi, toplayan coroutine iptal edilir. Bu işlem, temel üreticiyi de durdurur.
  • Üretici, öğe yayınlamayı tamamlar. Bu durumda, veri akışı kapatılır ve collect işlevini çağıran eş yordam yürütmeye devam eder.

Diğer ara operatörlerle belirtilmediği sürece akışlar cold (soğuk) ve lazy (tembel) olur. Bu, akışta bir terminal operatörü her çağrıldığında üretici kodunun yürütüldüğü anlamına gelir. Önceki örnekte, birden fazla akış toplayıcının olması, veri kaynağının en son haberleri farklı sabit aralıklarla birden çok kez getirmesine neden olur. Birden fazla tüketici aynı anda veri topladığında bir akışı optimize etmek ve paylaşmak için shareIn operatörünü kullanın.

Beklenmeyen istisnaları yakalama

Üreticinin uygulanması, üçüncü taraf kitaplığından gelebilir. Bu, beklenmedik istisnalar oluşturabileceği anlamına gelir. Bu istisnaları işlemek için catch ara operatörünü kullanın.

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            newsRepository.favoriteLatestNews
                // Intermediate catch operator. If an exception is thrown,
                // catch and update the UI
                .catch { exception -> notifyError(exception) }
                .collect { favoriteNews ->
                    // Update UI with the latest favorite news
                }
        }
    }
}

Önceki örnekte, bir istisna oluştuğunda yeni bir öğe alınmadığı için collect lambda çağrılmaz.

catch öğeleri akışa da emit. Örnek depo katmanı, bunun yerine önbelleğe alınmış değerleri emit olabilir:

class NewsRepository(
    // ...
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            .onEach { news -> saveInCache(news) }
            // If an error happens, emit the last cached values
            .catch { exception -> emit(lastCachedNews()) }
}

Bu örnekte, bir istisna oluştuğunda collect lambda'sı çağrılır. Bunun nedeni, istisna nedeniyle akışa yeni bir öğe yayınlanmış olmasıdır.

Farklı bir CoroutineContext'te yürütme

Varsayılan olarak, bir flow oluşturucunun üreticisi, kendisinden toplayan coroutine'in CoroutineContext içinde yürütülür ve daha önce belirtildiği gibi, farklı bir CoroutineContext değerlerini emit edemez. Bu davranış bazı durumlarda istenmeyebilir. Örneğin, bu konu boyunca kullanılan örneklerde, depo katmanı viewModelScope tarafından kullanılan Dispatchers.Main üzerinde işlemler yapmamalıdır.

Bir akışın CoroutineContext değerini değiştirmek için ara operatörü flowOn kullanın. flowOn, yukarı akışın flowOn değerini değiştirir. Bu, üreticinin ve flowOn'dan önce (veya yukarıda) uygulanan tüm ara operatörlerin flowOn değerini değiştirdiği anlamına gelir.CoroutineContext Akışın sonraki kısmı (tüketiciyle birlikte flowOn sonrasındaki ara operatörler) etkilenmez ve akıştan collect için kullanılan CoroutineContext üzerinde yürütülür. Birden fazla flowOn operatörü varsa her biri akışı mevcut konumundan değiştirir.

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData,
    private val defaultDispatcher: CoroutineDispatcher
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> // Executes on the default dispatcher
                news.filter { userData.isFavoriteTopic(it) }
            }
            .onEach { news -> // Executes on the default dispatcher
                saveInCache(news)
            }
            // flowOn affects the upstream flow ↑
            .flowOn(defaultDispatcher)
            // the downstream flow ↓ is not affected
            .catch { exception -> // Executes in the consumer's context
                emit(lastCachedNews())
            }
}

Bu kodla, onEach ve map operatörleri defaultDispatcher kullanırken catch operatörü ve tüketici, viewModelScope tarafından kullanılan Dispatchers.Main üzerinde yürütülür.

Veri kaynağı katmanı G/Ç çalışması yaptığından G/Ç işlemleri için optimize edilmiş bir gönderici kullanmanız gerekir:

class NewsRemoteDataSource(
    // ...
    private val ioDispatcher: CoroutineDispatcher
) {

    val latestNews: Flow<List<ArticleHeadline>> = flow {
        // Executes on the IO dispatcher
        // ...
    }
        .flowOn(ioDispatcher)
}

Jetpack kitaplıklarındaki akışlar

Flow, birçok Jetpack kitaplığına entegre edilmiştir ve Android üçüncü taraf kitaplıkları arasında popülerdir. Flow, canlı veri güncellemeleri ve sonsuz veri akışları için idealdir.

Veritabanındaki değişikliklerden haberdar olmak için Oda ile Akış'ı kullanabilirsiniz. Veri erişimi nesnelerini (DAO) kullanırken anlık güncellemeleri almak için Flow türünü döndürün.

@Dao
abstract class ExampleDao {
    @Query("SELECT * FROM Example")
    abstract fun getExamples(): Flow<List<Example>>
}

Example tablosunda her değişiklik yapıldığında, veritabanındaki yeni öğeleri içeren yeni bir liste yayınlanır.

Geri çağırmaya dayalı API'leri akışlara dönüştürme

callbackFlow geri çağırmaya dayalı API'leri akışlara dönüştürmenize olanak tanıyan bir akış oluşturucudur. Örneğin, Firebase Firestore Android API'leri geri çağırmaları kullanır.

Bu API'leri akışa dönüştürmek ve Firestore veritabanı güncellemelerini dinlemek için aşağıdaki kodu kullanabilirsiniz:

class FirestoreUserEventsDataSource(
    private val firestore: FirebaseFirestore
) {
    // Method to get user events from the Firestore database
    fun getUserEvents(): Flow<UserEvents> = callbackFlow {

        // Reference to use in Firestore
        var eventsCollection: CollectionReference? = null
        try {
            eventsCollection = FirebaseFirestore.getInstance()
                .collection("collection")
                .document("app")
        } catch (e: Throwable) {
            // If Firebase cannot be initialized, close the stream of data
            // flow consumers will stop collecting and the coroutine will resume
            close(e)
        }

        // Registers callback to firestore, which will be called on new events
        val subscription = eventsCollection?.addSnapshotListener { snapshot, _ ->
            if (snapshot == null) { return@addSnapshotListener }
            // Sends events to the flow! Consumers will get the new events
            try {
                trySend(snapshot.getEvents())
            } catch (e: Throwable) {
                // Event couldn't be sent to the flow
            }
        }

        // The callback inside awaitClose will be executed when the flow is
        // either closed or cancelled.
        // In this case, remove the callback from Firestore
        awaitClose { subscription?.remove() }
    }
}

flow oluşturucunun aksine, callbackFlow send işleviyle farklı bir CoroutineContext'dan veya trySend işleviyle bir eş yordamın dışından değerlerin yayınlanmasına olanak tanır.

callbackFlow, dahili olarak bir kanal kullanır. Bu kanal, kavramsal olarak engelleme kuyruğuna çok benzer. Bir kanal, arabelleğe alınabilecek maksimum öğe sayısı olan kapasite ile yapılandırılır. callbackFlow içinde oluşturulan kanalın varsayılan kapasitesi 64 öğedir. Tam kanala yeni bir öğe eklemeye çalıştığınızda send, yeni öğe için yer açılana kadar üreticiyi askıya alır. trySend ise öğeyi kanala eklemez ve hemen false döndürür.

trySend, belirtilen öğeyi kapasite kısıtlamalarını ihlal etmemesi koşuluyla kanala hemen ekler ve başarılı sonucu döndürür.

Ek akış kaynakları