Концептуально тип Flow<T> Котлина представляет собой асинхронный холодный поток ¹ элементов типа T, который может завершиться успешно или с исключением. Давайте посмотрим, как можно обрабатывать эти исключения и что мы можем узнать о потоках и исключениях из основных принципов.

Предположим, что мы пишем приложение пользовательского интерфейса, которое отображает поток обновлений значений в пользовательском интерфейсе и, таким образом, собирает их из потока. Это приложение имеет uiScope, то есть CoroutineScope, время жизни которого привязано к соответствующему элементу пользовательского интерфейса, отображающему данные. Существует функция dataFlow(), которая возвращает поток с отображаемыми данными, поэтому отображение данных можно активировать следующим образом:

uiScope.launch { // launch a UI display coroutine
    dataFlow().collect { value -> updateUI(value) }
}

Flow гарантирует, что updateUI всегда вызывается в контексте выполнения сборщика, который определяется здесь uiScope. Даже если dataFlow() использует другой контекст внутри, этот факт никак не вытекает из него ².

Но что будет, если в dataFlow() есть ошибка? В этом случае вызов collect генерирует исключение, которое приводит к исключительному завершению сопрограммы, которая распространяется на uiScope и, как правило, в конечном итоге вызывает обработчик неперехваченных исключений (CoroutineExceptionHandler) в своем контексте. . Это нормально, если исключение было действительно неожиданным и никогда не должно происходить в правильном коде, но что, если dataFlow(), например, читает данные из сети, и вполне ожидаемо произойдет сбой, когда что-то не так с сетью? С этим нужно справиться. Об ошибке сообщается через исключение и может обрабатываться так же, как исключения обычно обрабатываются в Kotlin - с использованием _14 _ / _ 15_ блока ³:

uiScope.launch { 
    try {
        dataFlow().collect { value -> updateUI(value) }
    } catch (e: Throwable) {
        showErrorMessage(e)
    }
}

Если мы инкапсулируем эту логику обработки исключений в оператор в потоке ⁴, то мы можем упростить этот код, уменьшить вложенность и сделать его более читаемым:

uiScope.launch {
    dataFlow()
        .handleErrors() // handle dataFlow errors
        .collect { value -> updateUI(value) }
}

Но как мы можем реализовать эту handleErrors функцию? Наивная попытка написать это показана ниже:

fun <T> Flow<T>.handleErrors(): Flow<T> = flow {
    try {
        collect { value -> emit(value) }
    } catch (e: Throwable) {
        showErrorMessage(e)
    }
}

Эта реализация собирает значения из восходящего потока, который вызывается, и передает их вниз по потоку, заключая вызов collect в блок _18 _ / _ 19_, как мы это делали раньше. Он просто абстрагирует код, который мы изначально написали. Это сработает? Да, в данном конкретном случае. Так почему именно эта реализация наивна?

Исключительная прозрачность

Подумайте о свойствах потока, возвращаемого handleErrors:

val flow = dataFlow().handleErrors()

Он выдает некоторые значения, как и любой другой поток, но у него также есть дополнительное свойство, которого нет у других потоков - любая ошибка в нисходящем потоке улавливается им. Рассмотрим приведенный ниже код с функцией Kotlin error в качестве лакмусовой бумажки:

flow.collect { error("Failed") }

Если вы запустите его с простым потоком, этот код выдает anIllegalStateException на первое выданное значение. Но с потоком, возвращаемым handleError, это исключение перехватывается и не появляется, поэтому вызов collect завершается нормально. Это совершенно удивительно для читателя этого кода, от которого не требуется знать детали реализации потока, из которого они пытаются получить данные.

Потоки Kotlin предназначены для модульного анализа потоков данных. Единственными предполагаемыми эффектами потоков являются их порожденные значения и завершение, поэтому операторы потока, такие как handleError, не разрешены спецификацией потока. Каждая реализация потока должна гарантировать прозрачность исключения - исключение нижестоящего потока всегда должно передаваться сборщику ⁵.

Обработка исключений

Потоки Kotlin предоставляют несколько операторов обработки исключений, которые обеспечивают прозрачность исключений. В нашем случае мы можем использовать оператор catch:

fun <T> Flow<T>.handleErrors(): Flow<T> = 
    catch { e -> showErrorMessage(e) }

Однако результирующий код ведет себя иначе, чем исходный код, который мы написали с использованием _27 _ / _ 28_, потому что он не улавливает ошибки, которые могут произойти внутри collect { value -> updateUI(value) } вызова из-за прозрачности исключения. Мы можем продолжить обработку ошибок в updateUI, переписав код следующим образом:

uiScope.launch { 
    dataFlow()
        .onEach { value -> updateUI(value) }
        .handleErrors() 
        .collect()
}

Перемещая updateUI из collect в оператор onEach, мы поместили его перед обработкой ошибок в handleErrors, поэтому updateUI ошибки теперь обрабатываются. В качестве последнего штриха теперь мы можем объединить вызовы launch и collect, используя оператор терминала launchIn, что еще больше уменьшит вложенность в этом коде и превратит его в простую последовательность операторов слева направо:

dataFlow()
    .onEach { value -> updateUI(value) }
    .handleErrors() 
    .launchIn(uiScope)

Статус API

Kotlin Flow - экспериментальный тип в библиотеке kotlinx.coroutines, начиная со второй промежуточной (предварительной) версии 1.3.0-M2 грядущего 1.3.0 выпуска. Некоторые дальнейшие изменения могут быть возможны до того, как он будет выпущен в качестве стабильного API, но теперь общая форма API выглядит довольно солидно.

Сноски и дополнительная литература

  1. ^ Холодные потоки, горячие каналы дает определение холодного потока.
  2. ^ Контекст выполнения Kotlin Flows содержит более подробную информацию о контекстах.
  3. ^ Справочная документация по исключениям Kotlin объясняет исключения Kotlin.
  4. ^ Простой дизайн Kotlin Flow вводит понятие оператора.
  5. ^ Потоковая документация имеет более подробное описание прозрачности исключения.