Потоки Flow

Потоки являются частью языка программирования Kotlin и предназначены для последовательного возврата нескольких значений из асинхронных задач на основе корутин. Подробнее про потоки можно прочитать на этом в соответствующей главы из руководства по Kotlin: Асинхронные потоки. В данном же случае мы рассмотрим применение потоков в контексте мобильного приложения Android.

Создание потока Flow

Самую базовую форму потока в Kotlin представляет тип Flow Каждый поток может отправлять данные только одного типа, который должен быть указан при объявлении потока. Например, для потоковой передачи строк можно использовать тип Flow<String>.

При объявлении потока нам нужно определить код для генерации данных. Для этого можно использовать специальную функцию-строитель flow(), которая принимает в качестве параметра блок корутин с кодом генерации данным. Например:

val myFlow: Flow<String> = flow {

    // генерация значений типа String
}

В данном случае определяется поток myFlow, предназначенный для генерации значений String.

В качестве альтернативы можно использовать функцию-строитель потока flowOf() для преобразования фиксированного набора значений в поток:

val myFlow2 = flowOf("Tom", "Bob", "Sam")

Кроме того, многие типы коллекций Kotlin имеют функцию расширения asFlow(), которую можно вызывать для преобразования данных коллекции в поток. Например, следующий код преобразует массив строковых значений в поток:

val myFlow3 = arrayOf<String>("Tom", "Bob", "Sam").asFlow()

Генерация данных

После создания потока следует предоставить механизм для передачи данных во вне. Из трех построителей потоков, которые мы рассмотрели в предыдущем разделе, только построители flowOf() и asFlow() создают потоки, которые автоматически отправляют данные, как только потребитель-внешний код начинает извлекать из потока данные. Однако в случае с функцией flow() нам нужно самим написать код, который будет вручную выдавать каждое значение по мере его появления. Для этого применяется функция emmit(), в которую в качестве аргумента передается текущее значение для передачи из потока. Например:

val myFlow: Flow<Int> = flow {

    for (i in 0..9) {
        emit(i)
        delay(2000)
    }
}

Здесь myFlow в цикле перебирает числа от 0 до 9 и через вызов emit() возвращает текущее перебираемое число. Для имитации долгой работы на каждой итерации цикла выполняется 2-секундная задержка, что позволяет продемонстрировать асинхронный характер потока.

Поток как состояние и collectAsState()

Jetpack Compose предоставляет функцию collectAsState(), которая позволяет получить текущее значение из потока в виде состояния. ПРи получении нового значения произойдет рекомпозиция интерфейса. Например:

package com.example.helloapp

import android.os.Bundle
import androidx.activity.ComponentActivity
import androidx.activity.compose.setContent
import androidx.compose.foundation.layout.padding
import androidx.compose.runtime.Composable
import androidx.compose.material3.Text
import androidx.compose.runtime.collectAsState
import androidx.compose.runtime.getValue
import androidx.compose.ui.Modifier
import androidx.compose.ui.unit.dp
import androidx.compose.ui.unit.sp
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow

class MainActivity : ComponentActivity() {
    
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContent {
            val myFlow = flow {
                for (i in 0..9) {
                    emit(i)
                    delay(2000)
                }
            }
            Main(flow = myFlow)
        }
    }
}
@Composable
fun Main(flow: Flow<Int>) {
    val count by flow.collectAsState(initial = 0)
    Text("Count: $count", Modifier.padding(10.dp), fontSize = 40.sp)
}

Здесь в лямбда-выражении в методе setContent() создается простейший поток Flow<Int>, который генерирует числа от 0 до 9 с интервалом в 2 секунды. Этот поток передается в кастомный компонент Main, где текущее значение этого потока получаем в переменную состояния count. А компонент Text выводит ее значение.

Таким образом при запуске приложения каждые 2 секунды будет генерироваться новое число и помещаться в состояние count, а компонент Text будет его отображать:

Преобразование потока

И как в общем случае, к потокам Flow в приложении на Jetpack Compose можно применять различные преобразования. Основные из них были рассмотрены на этом сайте в руководстве по Kotlin в статье Операции с потоками. Например, используем ряд операций:

package com.example.helloapp

import android.os.Bundle
import androidx.activity.ComponentActivity
import androidx.activity.compose.setContent
import androidx.compose.foundation.layout.padding

import androidx.compose.runtime.Composable
import androidx.compose.ui.tooling.preview.Preview
import androidx.compose.material3.Text
import androidx.compose.runtime.collectAsState
import androidx.compose.runtime.getValue
import androidx.compose.ui.Modifier
import androidx.compose.ui.unit.dp
import androidx.compose.ui.unit.sp
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map

class MainActivity : ComponentActivity() {

    val myFlow = flow {
        for (i in 0..9) {
            emit(i)
            delay(2000)
        }
    }

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContent {
            val newFlow = myFlow.filter {it % 2 == 0}.map{ "Current value = $it"}
            Main(flow = newFlow)
        }
    }
}
@Composable
fun Main(flow: Flow<String>) {
    val count by flow.collectAsState(initial = 0)
    Text("$count", Modifier.padding(10.dp), fontSize = 40.sp)
}

На уровне класса MainActivity определяется поток myFlow, который содержит числоа от 0 до 9.

В лямбда-выражении из метода setContent этот поток изменяется - из него выбираются только четные числа, которые трансформируются в строки

val newFlow = myFlow.filter {it % 2 == 0}.map{ "Current value = $it"}

И этот измененный поток передается в компонент Main.

Или другой пример - используем функцию reduce(), которая сводит все значения потока к одному значению:

package com.example.helloapp

import android.os.Bundle
import androidx.activity.ComponentActivity
import androidx.activity.compose.setContent
import androidx.compose.foundation.layout.padding

import androidx.compose.runtime.Composable
import androidx.compose.ui.tooling.preview.Preview
import androidx.compose.material3.Text
import androidx.compose.runtime.LaunchedEffect
import androidx.compose.runtime.collectAsState
import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.remember
import androidx.compose.runtime.setValue
import androidx.compose.ui.Modifier
import androidx.compose.ui.unit.dp
import androidx.compose.ui.unit.sp
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.reduce

class MainActivity : ComponentActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContent {
            val myFlow = flow {
                for (i in 0..9) {
                    emit(i)
                    delay(2000)
                }
            }

            Main(flow = myFlow)
        }
    }
}
@Composable
fun Main(flow: Flow<Int>) {
    var count by remember { mutableStateOf<Int>(0) }
    LaunchedEffect(Unit) {
        flow.reduce { accumulator, value ->
            count = accumulator
            accumulator + value
        }
    }
    Text("$count", Modifier.padding(10.dp), fontSize = 40.sp)
}

Функция reduce() принимает два параметра в виде аккумулятора и текущего значения из потока. Первое значение из потока помещается в аккумулятор, и между аккумулятором и текущим значением выполняется указанная операция (с сохранением результата в аккумуляторе). В данном случае мы просто складываем все числа из потока. Поскольку функция reduce() является suspend-функцией, она выполняется в контексте корутины. И для ее выполнения определяем компонент LaunchedEffect.