如何在Kotlin/Arrow.kt中实现生产者频道的“铁路模式”

import arrow.core.Either
import arrow.core.Left
import arrow.core.Right
import arrow.fx.coroutines.ExitCase
import arrow.fx.coroutines.Fiber
import arrow.fx.coroutines.bracket
import arrow.fx.coroutines.fork
import arrow.fx.coroutines.join
import arrow.fx.coroutines.produce
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.runBlocking

// 定义生产者频道的“铁路模式”函数
fun <A, B, C> Flow<A>.railwayMap(
    mapper: suspend (A) -> Either<B, C>
): Flow<Either<B, C>> = flow {
    collect { value ->
        val result = mapper(value)
        emit(result)
    }
}

// 示例使用
fun main() = runBlocking {
    val input = listOf(1, 2, 3, 4, 5)

    val processed: Flow<Either<String, Int>> = input.asFlow()
        .railwayMap { value ->
            if (value % 2 == 0) {
                Left("Even number")
            } else {
                Right(value * 2)
            }
        }
        .flowOn(Dispatchers.Default)

    processed
        .onEach { result ->
            when (result) {
                is Left -> println("Error: ${result.value}")
                is Right -> println("Result: ${result.value}")
            }
        }
        .onCompletion { cause ->
            if (cause != null) {
                println("Flow completed with an error: $cause")
            } else {
                println("Flow completed successfully")
            }
        }
        .collect()

    println("Done")
}