我还没有找到合适的现有函数，我也不太喜欢链接的解决方案，所以我玩了一下，最终得到了以下扩展函数：

fun <T> Sequence<T>.takeWhileInclusive(predicate: (T) -> Boolean) = sequence {
    with(iterator()) {
        while (hasNext()) {
            val next = next()
            yield(next)
            if (!predicate(next)) break
        }
    }
}

它使用sequence，在需要时缓慢地产生值。至少我可以省略中间的var，我认为这样可能更有益...

遗憾的是，takeWhile使用的几个函数是内部的。但是如果我们将它们复制到扩展文件中，我们也可以为流构建takeWhileInclusive：

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.FlowCollector
import kotlinx.coroutines.flow.flow
import java.util.concurrent.CancellationException

fun <T> Flow<T>.takeWhileInclusive(predicate: suspend (T) -> Boolean): Flow<T> = flow {
    // This return is needed to work around a bug in JS BE: KT-39227
    return@flow collectWhile { value ->
        emit(value)
        predicate(value)
    }
}

private suspend inline fun <T> Flow<T>.collectWhile(crossinline predicate: suspend (value: T) -> Boolean) {
    val collector = object : FlowCollector<T> {
        override suspend fun emit(value: T) {
            // Note: we are checking predicate first, then throw. If the predicate does suspend (calls emit, for example)
            // the the resulting code is never tail-suspending and produces a state-machine
            if (!predicate(value)) {
                throw AbortFlowException(this)
            }
        }
    }
    try {
        collect(collector)
    } catch (e: AbortFlowException) {
        e.checkOwnership(collector)
    }
}

private class AbortFlowException constructor(
    @JvmField @Transient val owner: FlowCollector<*>
) : CancellationException("Flow was aborted, no more elements needed") {

    override fun fillInStackTrace(): Throwable {
        // Prevent Android <= 6.0 bug, #1866
        stackTrace = emptyArray()
        return this
    }
}

private fun AbortFlowException.checkOwnership(owner: FlowCollector<*>) {
    if (this.owner !== owner) throw this
}

我不喜欢Kotlin内部代码的复制粘贴，但还没有找到更好的方法来实现它。