cancelWhen operator for Kotlin Flow

CancelFlow.kt

/**
 * Cancels current flow when the predicate is matched, throws [CancellationException] up in 
 * current coroutine. 
 * @param cause class that will be used as cause for [CancellationException]
 * @param predicate if result of this function is [true] then the flow is stopped
 */
inline fun <T> Flow<T>.cancelWhen(
    cause: KClass<out Throwable>? = null,
    crossinline predicate: (T) -> Boolean
): Flow<T> = transform {
    if (!predicate(it)) {
        emit(it)
        return@transform
    }
    currentCoroutineContext().cancel(
        CancellationException(
            message = "Cancelling flow on predicate match for value: $it",
            cause = cause?.java?.newInstance()
        )
    )
}