Введение в Kotlin Flow

Into the Flow: Kotlin cold streams primer — одна из лучших статей о новой возможности Kotlin под названием Flow.

Введение в Kotlin Flow

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

1	suspend fun doSomething(): List<Something>

Допустим, она будет выполнять сложные расчеты в фоне, а затем вернет результат в виде списка. Это отлично работает, за исключением того, что функция должна заранее выполнить расчет всех элементов списка. Если элементов слишком много или мы имеем дело с неопределенным количеством элементов, начинаются проблемы.

РЕКОМЕНДУЕМ:
Полезные советы разработчику на Kotlin

Для их решения в Kotlin есть другой инструмент под названием Flow (поток):

1	fun doSomething(): Flow<Something>

Обрати внимание, что при объявлении функции мы не использовали ключевое слово suspend. Это потому, что на самом деле, какой бы сложной ни была функция, при запуске она ничего не делает и сразу возвращает управление. Функция начинает работать только после того, как мы вызовем метод collect() полученного от функции объекта. Например:

1	flow.collect { something -> println(something) }

Именно поэтому разработчики Kotlin называют потоки холодными, в противовес горячим каналам (Channel), которые также присутствуют в языке.

Для создания самого потока можно использовать функцию flow:

fun doSomething(): Flow<Int> = flow {

for (i in 1..3) {

delay(100)

emit(i)

}

В данном случае функция «выпускает» в поток три объекта типа Int с перерывом в 100 миллисекунд. Обрати внимание, что flow — это suspend-функция, которая может запускать другие suspend-функции (в данном случае delay()).

Как уже было сказано выше, функция, возвращающая поток, не должна быть suspend-функцией. Но метод collect() объекта типа Flow — suspend-функция, которая должна работать внутри CoroutineScope:

val something = doSomething()

viewModelScope.launch {

something.collect { value -> println(value) }

}

Создать поток можно и другими способами, например с помощью метода asFlow():

1 2	listOf(1,2,3).asFlow() (1..3).asFlow()

Завершить поток возможно несколькими способами — от вызова collect() до методов типа first(), fold(), toList(), знакомых тебе по работе с коллекциями.

Сам поток можно трансформировать, чтобы получить новый поток с помощью методов map(), filter(), take():

(1..3).asFlow()

.transform { number ->

emit(number*2)

delay(100)

emit(number*4)

}

.collect { number -> println(number) }

По умолчанию функции flow и collect запускаются внутри текущего CoroutineScope, но его можно изменить, используя метод flowOn():

fun doSomething(): Flow<Int> = flow {

// Этот код будет выполнен внутри Dispatchers.IO (фоновый поток)

for (i in 1..3) {

delay(100)

emit(i)

}

}.flowOn(Dispatchers.IO)

[...]

viewModelScope.launch {

doSomething().collect { value ->

// Этот код будет выполнен внутри основного потока приложения

print (value)

}

Несколько потоков можно объединить в один с помощью метода zip():

val flowA = (1..3).asFlow()

val flowB = flowOf("one", "two", "three")

flowA.zip(flowB) { a, b -> "$a and $b" }

.collect { println(it) }

Этот код объединит каждый элемент первого потока с соответствующим элементом второго потока:

1 and one

2 and two

3 and three

Другой вариант объединения — функция combine():

al flowA = (1..3).asFlow()

val flowB = flowOf("single item")

flowA.combine(flowB) { a, b -> "$a and $b" }

.collect { println(it) }

В данном случае каждый элемент первого потока будет объединен с последним элементом второго потока:

1 and single item

2 and single item

3 and single item

После трансформации потоков мы можем получить структуры данных, включающие в себя потоки потоков (Flow<Flow<X>>). Чтобы «выровнять» такие данные, можно использовать один из следующих методов:

flatMapConcat() — возвращает поток, который возвращает все элементы первого вложенного потока, затем все элементы второго потока и так далее;
flatMapMerge() — возвращает поток, в который попадают элементы из всех вложенных потоков в порядке очередности;
flatMapLatest() — возвращает последний вложенный поток.