【Kotlin 协程】Flow 操作符 ② ( 末端操作符 | collect 操作符 | reduce 操作符 | single 操作符 | first 操作符 | toList 操作符 )

/**
 * 终端流操作符，使用提供的[动作]收集给定的流。
 * 如果在收集过程中或在所提供的流中发生任何异常，则此方法将重新抛出此异常。
 *
 * 使用示例:
 *
 * ```
 * val flow = getMyEvents()
 * try {
 *     flow.collect { value ->
 *         println("Received $value")
 *     }
 *     println("My events are consumed successfully")
 * } catch (e: Throwable) {
 *     println("Exception from the flow: $e")
 * }
 * ```
 */
public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value: T) -> Unit): Unit =
    collect(object : FlowCollector<T> {
        override suspend fun emit(value: T) = action(value)
    })

/**
 * 从第一个元素开始累加值，并对当前累加器值和每个元素应用[操作]。
 * 如果流为空，抛出[NoSuchElementException]。
 */
public suspend fun <S, T : S> Flow<T>.reduce(operation: suspend (accumulator: S, value: T) -> S): S {
    var accumulator: Any? = NULL

    collect { value ->
        accumulator = if (accumulator !== NULL) {
            @Suppress("UNCHECKED_CAST")
            operation(accumulator as S, value)
        } else {
            value
        }
    }

    if (accumulator === NULL) throw NoSuchElementException("Empty flow can't be reduced")
    @Suppress("UNCHECKED_CAST")
    return accumulator as S
}

package kim.hsl.coroutine

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        runBlocking {
            val result = (0..3)
                        .asFlow()
                        .reduce{a, b ->
                            a + b
                        }
            println("result : $result")
        }
    }
}

2022-12-26 15:51:20.411 24364-24364/kim.hsl.coroutine I/System.out: result : 6

/**
 * 终端操作符，等待一个且仅等待一个值发出。
 * 为空流抛出[NoSuchElementException]，为流抛出[IllegalStateException]
 * 包含多个元素的。
 */
public suspend fun <T> Flow<T>.single(): T {
    var result: Any? = NULL
    collect { value ->
        require(result === NULL) { "Flow has more than one element" }
        result = value
    }

    if (result === NULL) throw NoSuchElementException("Flow is empty")
    return result as T
}

/**
 * 终端操作符，返回流发出的第一个元素，然后取消流的收集。
 * 如果流为空，则抛出[NoSuchElementException]。
 */
public suspend fun <T> Flow<T>.first(): T {
    var result: Any? = NULL
    collectWhile {
        result = it
        false
    }
    if (result === NULL) throw NoSuchElementException("Expected at least one element")
    return result as T
}

/**
 * 从[initial]值开始累加值，并应用[operation]当前累加器值和每个元素
 */
public suspend inline fun <T, R> Flow<T>.fold(
    initial: R,
    crossinline operation: suspend (acc: R, value: T) -> R
): R {
    var accumulator = initial
    collect { value ->
        accumulator = operation(accumulator, value)
    }
    return accumulator
}

/**
 * 将给定的流收集到[destination]
 */
public suspend fun <T> Flow<T>.toList(destination: MutableList<T> = ArrayList()): List<T> = toCollection(destination)

/**
 * 将给定的流收集到[destination]
 */
public suspend fun <T> Flow<T>.toSet(destination: MutableSet<T> = LinkedHashSet()): Set<T> = toCollection(destination)