Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/jvm/src/flow/internal/SafeCollector.kt (Kotlin)

Code before:
     * This is a crafty implementation of state-machine reusing.
     *
     * First it checks that it is not used concurrently (which we explicitly prohibit), and
     * then just caches an instance of the completion in order to avoid extra allocation on each emit,
     * making it effectively garbage-free on its hot-path.
     *
     * See `emit` overload.
     */
    actual override suspend fun emit(value: T) {
        // NB: it is a tail-call, so we are sure `uCont` is the completion of the emit's **caller**.
        return suspendCoroutineUninterceptedOrReturn sc@{ uCont ->
            try {
                emit(uCont, value)
            } catch (e: Throwable) {
                // Save the fact that exception from emit (or even check context) has been thrown
                // Note, that this can the first emit and lastEmissionContext may not be saved yet,
                // hence we use `uCont.context` here.
                lastEmissionContext = DownstreamExceptionContext(e, uCont.context)
                throw e
            }
        }
    }

    /**
     * Here we use the following trick:
     * - Proceed with all checks required
     * - Having a non-intercepted, non-cancellable caller's `uCont`, we leverage our implementation knowledge
     *   and invoke `collector.emit(T)` as `collector.emit(value: T, completion: Continuation), passing `this`
     *   as the completion. We also setup `this` state, so if the `completion.resume` is invoked, we are
     *   invoking `uCont.resume` properly in accordance with `ContinuationImpl`/`BaseContinuationImpl` internal invariants.
     *
     * Note that in such scenarios, `collector.emit` completion is the current instance of SafeCollector and thus is reused.
     */
    private fun emit(uCont: Continuation<Unit>, value: T): Any? {
        val currentContext = uCont.context
        currentContext.ensureActive()
        // This check is triggered once per flow on a happy path.
        val previousContext = lastEmissionContext
        if (previousContext !== currentContext) {
            checkContext(currentContext, previousContext, value)
            lastEmissionContext = currentContext
        }
        completion = uCont
        val result = emitFun(collector as FlowCollector<Any?>, value, this as Continuation<Unit>)
        /*
         * If the callee hasn't suspended, that means that it won't (it's forbidden) call 'resumeWith` (-> `invokeSuspend`)
         * and we don't have to retain a strong reference to it to avoid memory leaks.
         */
        if (result != COROUTINE_SUSPENDED) {
            completion = null
        }

Review comment (suggestion):
```suggestion
     * - Perform all the required checks
```
? Otherwise, it reads to me as "proceed, also requiring all checks," which doesn't seem aligned with the context.

Code after:
     * This is a crafty implementation of state-machine reusing.
     *
     * First it checks that it is not used concurrently (which we explicitly prohibit), and
     * then just caches an instance of the completion_ in order to avoid extra allocation on each emit,
     * making it effectively garbage-free on its hot-path.
     *
     * See `emit` overload.
     */
    actual override suspend fun emit(value: T) {
        // NB: it is a tail-call, so we are sure `uCont` is the completion of the emit's **caller**.
        return suspendCoroutineUninterceptedOrReturn sc@{ uCont ->
            try {
                emit(uCont, value)
            } catch (e: Throwable) {
                // Save the fact that exception from emit (or even check context) has been thrown
                // Note, that this can the first emit and lastEmissionContext may not be saved yet,
                // hence we use `uCont.context` here.
                lastEmissionContext = DownstreamExceptionContext(e, uCont.context)
                throw e
            }
        }
    }

    /**
     * Here we use the following trick:
     * - Perform all the required checks
     * - Having a non-intercepted, non-cancellable caller's `uCont`, we leverage our implementation knowledge
     *   and invoke `collector.emit(T)` as `collector.emit(value: T, completion: Continuation), passing `this`
     *   as the completion. We also setup `this` state, so if the `completion.resume` is invoked, we are
     *   invoking `uCont.resume` properly in accordance with `ContinuationImpl`/`BaseContinuationImpl` internal invariants.
     *
     * Note that in such scenarios, `collector.emit` completion is the current instance of SafeCollector and thus is reused.
     */
    private fun emit(uCont: Continuation<Unit>, value: T): Any? {
        val currentContext = uCont.context
        currentContext.ensureActive()
        // This check is triggered once per flow on a happy path.
        val previousContext = lastEmissionContext
        if (previousContext !== currentContext) {
            checkContext(currentContext, previousContext, value)
            lastEmissionContext = currentContext
        }
        completion_ = uCont
        val result = emitFun(collector as FlowCollector<Any?>, value, this as Continuation<Unit>)
        /*
         * If the callee hasn't suspended, that means that it won't (it's forbidden) call 'resumeWith` (-> `invokeSuspend`)
         * and we don't have to retain a strong reference to it to avoid memory leaks.
         */
        if (result != COROUTINE_SUSPENDED) {
            completion_ = null
        }