Is Kotlin Flow's Collect is only internal kotlinx.coroutines API?

Question

Taking the direct example from https://kotlinlang.org/docs/reference/coroutines/flow.html#flows-are-cold

fun simple(): Flow<Int> = flow { 
    println("Flow started")
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}

fun main() = runBlocking<Unit> {
    println("Calling simple function...")
    val flow = simple()
    println("Calling collect...")
    flow.collect { value -> println(value) } 
    println("Calling collect again...")
    flow.collect { value -> println(value) } 
}

I got the error on collect.

This is an internal kotlinx.coroutines API that should not be used from outside of kotlinx.coroutines. No compatibility guarantees are provided.It is recommended to report your use-case of internal API to kotlinx.coroutines issue tracker, so stable API could be provided instead

When I add @InternalCoroutinesApi

@InternalCoroutinesApi
fun main() = runBlocking<Unit> {
    println("Calling simple function...")
    val flow = simple()
    println("Calling collect...")
    flow.collect { value -> println(value) }
    println("Calling collect again...")
    flow.collect { value -> println(value) }
}

I get an error in the collect's lambda (function of value -> println(value) as below

Type mismatch.
Required:
FlowCollector<Int>
Found:
([ERROR :  ]) → Unit
Cannot infer a type for this parameter. Please specify it explicitly.

I am using Kotlin version 1.4.21.

    implementation "org.jetbrains.kotlin:kotlin-stdlib:1.4.2"
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.4.2'
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.4.1'
    testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-test:1.4.2'

Did I do anything wrong that I cannot compile the example code in Android Studio?

Answer 1

The answer is, NO, collect is not only internal kotlinx.coroutines API. The error message is misleading.

As per @ir42's comment, add import kotlinx.coroutines.flow.collect solve the problem.

Additional info, why I didn't pick collectLatest as the answer

collect and collectLatest is different.

Using this example

fun simple(): Flow<Int> = flow { // flow builder
    for (i in 1..3) {
        delay(100) // pretend we are doing something useful here
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> {
    // Launch a concurrent coroutine to check if the main thread is blocked
    launch {
        for (k in 1..3) {
            println("I'm not blocked $k")
            delay(100)
        }
    }
    // Collect the flow
    simple().collect { value -> println(value) } 
}

Collect will produce

I'm not blocked 1
1
I'm not blocked 2
2
I'm not blocked 3
3

as per https://kotlinlang.org/docs/reference/coroutines/flow.html

But collectLatest

fun simple(): Flow<Int> = flow { // flow builder
    for (i in 1..3) {
        delay(100) // pretend we are doing something useful here
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> {
    // Launch a concurrent coroutine to check if the main thread is blocked
    launch {
        for (k in 1..3) {
            println("I'm not blocked $k")
            delay(100)
        }
    }
    // Collect the flow
    simple().collectLatest { value -> println(value) } 
}

will produce

I'm not blocked 1
I'm not blocked 2
1
I'm not blocked 3
2

Answer 2

Add below import:

import kotlinx.coroutines.flow.collect

Answer 3

I finally found the solution for it after it almost drove me mad. As none of the solutions here worked for me, I had a special case where I was working on a multi-modules project and collect() kept throwing out this error everywhere and even manually importing it didn't fix it.

What solved the issue for me is explicitly adding coroutines dependency to ALL of my project's modules build.gradle files, not just the app's one, even those modules that don't use coroutines.

implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.2")

Answer 4

For version 1.6.0-RC3 and newer:

The @InternalCoroutinesApi annotations have been removed from all collect() functions (see this PR).

That means we won't see the error anymore and also won't need the import

import kotlinx.coroutines.flow.collect

anymore because (together with the changes of another PR) we can now use the collect() member function directly.

Answer 5

The Api changed in newer versions of kotlinx coroutines and instead of a lambda in the collect function, it should go as the final call in the chain. The documentation example:

flow
  .onEach { value -> process(value) }
  .catch { e -> handleException(e) }
  .collect() // trigger collection of the flow

Answer 6

Just casting as the same object (here, StateFlow) worked for me, like below:

(flow as StateFlow).collect { value -> println(value) }

Answer 7

Replace collect with collectLatest.