如何扁平化Arrow-KT的Either与Kotlin协程的Flow？

我使用Arrow-KT已有较长时间，但刚接触Kotlin协程及其编程模型，这是我首个使用协程的项目。当前代码如下：

Either.catch { funcThatReturnsFlow() }
    .map { flow ->
        flow.map { flowItem -> mappingFuncThatReturnsEither(flowItem) }
    }
// 最终类型：Either<Throwable, Flow<Either<Throwable, Item>>>

我希望将这个嵌套结构转换为Either<Throwable, Flow<Item>>，同时保留Flow的流式特性，而不是像下面这样将Flow转为List：

resultFromBefore.flatMap { flow -> flow.toList().sequence() }
// 类型变为Either<Throwable, List<Item>>

得先搞清楚：Flow是异步流式的数据结构，元素是逐个发射的，你没法提前知道后面的元素会不会转换失败。如果要得到Either<Throwable, Flow<Item>>，意味着只有当整个Flow里的所有元素都转换成功（所有Either都是Right），才返回Right(Flow<Item>)；只要有一个元素失败，就返回Left(Throwable)。这种逻辑下，必须先收集Flow的所有元素——毕竟流式处理阶段，你没法预判后续元素的结果。

如果一定要实现你要的类型转换 #

基于Arrow-KT的现有扩展，你可以自己写个扩展函数来实现：

// 扩展函数：把Flow<Either<E, A>>转成Either<E, Flow<A>>
suspend fun <E, A> Flow<Either<E, A>>.sequenceToEither(): Either<E, Flow<A>> {
    return this.toList()
        .sequence() // Arrow自带的List<Either<E,A>>转Either<E,List<A>>
        .map { it.asFlow() } // 把List转回Flow
}

// 调用示例
val result: Either<Throwable, Flow<Item>> = Either.catch { funcThatReturnsFlow() }
    .flatMap { flow ->
        flow.map { mappingFuncThatReturnsEither(it) }
            .sequenceToEither()
    }

如果想真正发挥Flow的流式优势 #

要是你不想等所有元素处理完再出结果，更合理的做法是把嵌套结构转成Flow<Either<Throwable, Item>>，这样可以逐个处理每个元素的成功或失败：

val resultFlow: Flow<Either<Throwable, Item>> = Either.catch { funcThatReturnsFlow() }
    .fold(
        ifLeft = { flowOf(Left(it)) }, // 外层捕获的异常直接转为Flow里的单个Left元素
        ifRight = { flow -> flow.map { mappingFuncThatReturnsEither(it) } }
    )

消费这个Flow的时候，就能逐个处理元素：

resultFlow.collect { either ->
    either.fold(
        ifLeft = { error -> // 处理单个元素的转换错误 },
        ifRight = { item -> // 处理成功转换的元素 }
    )
}

用Arrow-KT的FlowK简化操作 #

Arrow-KT提供了FlowK（对Kotlin Flow的包装），支持更多函数式操作，你可以用它来简化流程：

import arrow.fx.coroutines.FlowK
import arrow.fx.coroutines.sequence

val result: Either<Throwable, FlowK<Item>> = Either.catch { funcThatReturnsFlow() }
    .map { flow -> FlowK(flow).map { mappingFuncThatReturnsEither(it) } }
    .flatMap { flowK -> flowK.sequence() }

// 要是需要转回普通Flow，调用flowK.flow即可
val regularFlow: Either<Throwable, Flow<Item>> = result.map { it.flow }

内容的提问来源于stack exchange，提问作者craigmiller160