Flow invariant is violated when no obvious context switching

[digitalbuddha]

Hello, I am knew to using flow and having trouble with a seemingly easy operator I am writing. Consider the following operator castNotNull

@FlowPreview
@Suppress("UNCHECKED_CAST")
private fun <T1, T2> Flow<T1>.castNonNull(): Flow<T2> {
    val self = this
    return flow {
        self.collect {
            if (it != null) {
                try {
                    emit(it as T2)
                } catch (ils: IllegalStateException) {
                    println("wtf")
                    throw ils
                }
            }
        }
    }
}

I am unclear why the 3rd example below throws an

Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated: flow was collected in EmptyCoroutineContext, but emission happened in ScopeCoroutine{Active}@49f3e185. Please refer to 'flow' documentation or use 'flowOn' instead

suspend fun main() {
    val f = channelFlow {
        coroutineScope {
            val channel = produce {
                send(1)
            }
            channel.consumeEach {
                send(it)
            }
        }
    }
    //Example 1 WORKS
    f.collect {
        println(it)
    }
    //Example 2 WORKS
    flowOf(1,2,3).castNonNull<Int, Int>().collect {
        println(it)
    }
    //Example 3 FAILS with exception above
    f.castNonNull<Int, Int>().collect {
        println(it)
    }
}

Is there a bug or is there something else I need to be doing? Thank you kindly for your help

[yigit]

Here is simpler example that reproduces the exception:

suspend fun main() {
    channelFlow {
        send(1)
    }.transform {
        emit(it * it)
    } .collect {
        println(it)
    }
}

fails with:

Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated: flow was collected in EmptyCoroutineContext, but emission happened in ScopeCoroutine{Active}@75a1cd57. Please refer to 'flow' documentation or use 'flowOn' instead

If you replace transform with Map, works fine. (I assume thats because the internal methods use unsafeTransform)

works fine:

suspend fun main() {
    channelFlow {
        send(1)
    }.map {
        it * it
    }.collect {
        println(it)
    }
}

[qwwdfsad]

Thanks for the report.
This problem is only reproducible when running collect from suspend fun main which context is beyond our control.

Running it from any coroutine builder (even withContext(EmptyCoroutineContext)!) fixes the issue

[LouisCAD]

Does it mean using `coroutineScope { }` would fix the issue too?

…

On Thu, Aug 8, 2019, 10:57 Vsevolod Tolstopyatov ***@***.***> wrote: Thanks for the report. This problem is only reproducible when running collect from suspend fun main which context is beyond our control. Running it from any coroutine builder (even withContext(EmptyCoroutineContext)!) fixes the issue — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#1421?email_source=notifications&email_token=ABVG6BKXDYI7BPNZT52ZY33QDPNZBA5CNFSM4IKEUGO2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD326RPY#issuecomment-519432383>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ABVG6BMEPK6JMMXP6IX5ZG3QDPNZBANCNFSM4IKEUGOQ> .

[qwwdfsad]

Yes