launch(Dispatchers.Main) {
    // 第一部分
    flow {
        emit(1)
        throw NullPointerException("e")
    }.catch {
        Log.d("liduo", "onCreate1: $it")
    }.collect {
        Log.d("liudo", "onCreate2: $it")
    }
    // 第二部分
    flow {
        emit(1)
    }.onCompletion {
        Log.d("liduo", "onCreate3: $it")
    }.collect {
        Log.d("liudo", "onCreate4: $it")
    }
    // 第三部分
    flow {
        emit(1)
        throw NullPointerException("e")
    }.retryWhen { cause, attempt ->
        cause !is NullPointerException && attempt <= 2
    }.collect {
        Log.d("liudo", "onCreate5: $it")
    }
}

public fun <T> Flow<T>.catch(action: suspend FlowCollector<T>.(cause: Throwable) -> Unit): Flow<T> =
    flow { // 創(chuàng)建Flow對(duì)象
        // 觸發(fā)上游流的執(zhí)行，并捕獲異常
        val exception = catchImpl(this)
        // 捕獲到異常，則回調(diào)action處理
        if (exception != null) action(exception)
    }

internal suspend fun <T> Flow<T>.catchImpl(
    collector: FlowCollector<T>
): Throwable? {
    // 保存下游流執(zhí)行拋出的異常
    var fromDownstream: Throwable? = null
    try {
        // 觸發(fā)上游流的執(zhí)行
        collect {
            try {
                // 將上游流發(fā)送的值作為參數(shù)，觸發(fā)下游流執(zhí)行
                collector.emit(it)
            } catch (e: Throwable) { // 如果下游流在執(zhí)行中發(fā)生異常，保存并拋出
                fromDownstream = e
                throw e
            }
        }
    } catch (e: Throwable) { // 這里捕獲的異常，可能為上游流的異?！猚ollect方法，
                             // 也可能為下游流的異?！猠mit方法
        // 如果異常是下游流產(chǎn)生的異常，或者是協(xié)程取消時(shí)拋出的異常
        if (e.isSameExceptionAs(fromDownstream) || e.isCancellationCause(coroutineContext)) {
            throw e // 再次拋出，交給下游處理
        } else { // 如果是上游流的異常且不為協(xié)程取消異常
            return e // 成功捕獲
        }
    }
    // 未捕獲到異常，返回
    return null
}

public fun <T> Flow<T>.onCompletion(
    action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit
): Flow<T> = unsafeFlow { // 創(chuàng)建一個(gè)Flow對(duì)象
    try {
        // 觸發(fā)上游流的執(zhí)行
        // this表示下游的FlowCollector
        collect(this)
    } catch (e: Throwable) {// 如果下游發(fā)生異常
        // 將異常封裝成ThrowingCollector類(lèi)型的FlowCollector，并回調(diào)參數(shù)action，
        ThrowingCollector(e).invokeSafely(action, e)
        // 拋出異常
        throw e
    }
    // 如果正常執(zhí)行結(jié)束，會(huì)走到這里
    val sc = SafeCollector(this, currentCoroutineContext())
    try {
        // 回調(diào)執(zhí)行參數(shù)action
        sc.action(null)
    } finally {
        sc.releaseIntercepted()
    }
}

@PublishedApi
internal inline fun <T> unsafeFlow(@BuilderInference crossinline block: suspend FlowCollector<T>.() -> Unit): Flow<T> {
    // 返回一個(gè)匿名內(nèi)部類(lèi)
    return object : Flow<T> {
        // 回調(diào)collect方法是直接執(zhí)行block
        override suspend fun collect(collector: FlowCollector<T>) {
            collector.block()
        }
    }
}

private class ThrowingCollector(private val e: Throwable) : FlowCollector<Any?> {
    override suspend fun emit(value: Any?) {
        // 拋出異常
        throw e
    }
}

// 參數(shù)cause表示捕獲到的異常
// 參數(shù)attempt表示重試的次數(shù)
// 參數(shù)predicate返回true表示重新觸發(fā)上游流的執(zhí)行
public fun <T> Flow<T>.retryWhen(predicate: suspend FlowCollector<T>.(cause: Throwable, attempt: Long) -> Boolean): Flow<T> =
    // 創(chuàng)建一個(gè)Flow對(duì)象
    flow {
        // 記錄重試次數(shù)
        var attempt = 0L
        // 表示是否重新觸發(fā)
        var shallRetry: Boolean
        do {
            // 復(fù)位成false
            shallRetry = false
            // 觸發(fā)上游流的執(zhí)行，并捕獲異常
            val cause = catchImpl(this)
            // 如果捕獲到異常
            if (cause != null) {
                // 用戶判斷，是否要重新觸發(fā)
                if (predicate(cause, attempt)) {
                    // 表示要重新觸發(fā)
                    shallRetry = true
                    // 重試次數(shù)加1
                    attempt++
                } else { // 如果用戶不需要重新觸發(fā)
                    // 則拋出異常
                    throw cause
                }
            }
        // 判斷是否重新觸發(fā)
        } while (shallRetry)
    }