如何在不使用异常的情况下为流程提供 'close reason'？

Question

假设我正在生成有限的消息流作为 Flow<Message>。我循环一些消息源，当没有更多消息时，跳出循环以正常完成流程。

val messages = flow {
    while (source.hasNext()) {
        emit(source.next())
    }
    println("Finished sending all the messages")
}

消费者只需调用 collect 依次接收每条消息，直到流程结束。

messages.collect { doSomethingWith(it) }
println("All the messages have been processed")

但是当我需要告诉调用者为什么 流程停止时会发生什么？例如，假设消息生产者需要身份验证，而我必须处理令牌过期的问题。现在有两个原因可能会终止流程，所以我的生产者看起来像这样：

val messages = flow {
    while(source.hasNext()) {
        if (authentication.isExpired) {
            break
        }
        emit(source.next())
    }
    println("Finished sending all the messages")
}

问题是消费者不知道流程结束的原因。消息只是运行出来了，还是需要重新验证并继续获取消息？现在，我可以想到两种可能的处理方式，我都不喜欢。

例外情况

一个简单的解决方案是向 'abnormal' 流终止发出异常信号，如下所示：

val messages = flow {
    while (source.hasNext()) {
        if (authentication.isExpired) {
            throw AuthenticationExpiredException()
        }
        emit(source.next())
    }
    println("Finished sending all the messages")
}

太好了，因为现在消费者可以使用 try/catch 来确定流终止的原因：

try {
    messages.collect { doSomethingWith(it) }
    println("There are no more messages")
} catch (e: AuthenticationExpiredException) {
    println("The auth token expired")
}

但是像这样使用异常绝对不是我想做的事情。在他关于 Kotlin and Exceptions 的文章中，Kotlin 库负责人 Roman Elizarov 明确告诉我们：

don’t use exceptions to return a result value, [and] avoid try/catch in general application code.

这是一篇很棒的文章，我同意它的原因有很多，但不想使用异常的一个关键原因是它引入了未经类型检查的行为。如果我使用我的 AuthenticationExpiredException，那么在不阅读代码的情况下，流程的调用者不知道他们必须捕获并处理它。如果没有 try/catch，代码将编译得很好，并且会在运行时失败。这是 Kotlin 中有意的语言设计选择：编译时不检查异常，因为它们根本不打算以这种方式使用。

密封类

在 Kotlin 的其他地方 return 不同类型的结果的一个好方法是使用密封类。所以我可能会这样写：

sealed class FlowResult<out T> {
    data class Value<T>(val value: T): FlowResult<T>
    object: NoMoreMessages: FlowResult<Nothing>()
    object: AuthenticationExpired: FlowResult<Nothing>()
}

val messages = flow {
    while(source.hasNext()) {
        if (authentication.isExpired) {
            emit(AuthenticationExpired)
            break
        }
        emit(Value(source.next()))
    }
    emit(NoMoreMessages)
}

这要好得多，因为现在两个可能的关闭原因得到了同等对待。消费者可以使用 when 块来处理不同的结果，如果它们不包含处理 AuthenticationExpired.

的分支，则会出现编译错误

我使用这种方法的问题是我无法强制执行“失败”结果必须始终是流程中的最后一项的规则。例如，假设我犯了一个错误，不小心错过了循环中的 break。

val messages = flow {
    while(source.hasNext()) {
        if (authentication.isExpired) {
            emit(AuthenticationExpired)
        }
        emit(Value(source.next()))
    }
    emit(NoMoreMessages)
}

现在我将在不实际终止流程的情况下发出 AuthenticationExpired 值。它只会继续发出消息。这可能会导致各种错误和安全问题，但它会编译得很好。

感觉我真正在寻找的是一种机制，借此 collect 可以 return 一个值，像这样：

val closeReason = messages.collect { doSomethingWith(it) }
when (closeReason) {
    ...
}

有什么方法可以实现这样的目标吗？或者是否有另一种方法可以解决我所描述的问题？

Answer 1

没有 built-in 方法可以实现您想要的，但可以在现有 API 之上构建并将其与密封类:

// Essentially a functional Either type
sealed class Signal<out V, out T> {
    data class Value<V>(val value: V): Signal<V, Nothing>
    data class Terminal<T>(val value: T): Signal<Nothing, T>
}

然后构建一组扩展方法来执行您想要的处理

inline suspend fun <reified V, T> Flow<Signal<V,T>>.collect(
    valueCollector: suspend (V) -> Unit
): T = onEach {
    if (it is Signal.Value) {
        valueCollector(it.value)
    }
}.filterIsInstance<Signal.Terminal<T>>().first().value

这使用提供的收集器处理了 value 个项目，并且 return 将第一个 Terminal 实例的值视为 return 值。

如何在不使用异常的情况下为流程提供 'close reason'？

How can I provide a 'close reason' for a Flow without using exceptions?

kotlin

kotlin-coroutines

kotlin-flow

例外情况

密封类