Hey everyone, I wrote a project that uses flow with mvi architecture You can check it out here https://github.com/Ezike/Baking-App-Kotlin

Does calling

cancel

on a

CoroutineScope

preven't it from launching coroutines in the near future?

I’ve got a function like so:

suspend fun getAudio(songId: Long): AudioStream? {
        return withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
            songRepository.getSongs(SongQuery.SongIds(listOf(songId))).firstOrNull()?.firstOrNull()?.let { song ->
                val uri = Uri.parse(song.path)
                if (song.path.startsWith("content://")) {
                    if (DocumentsContract.isDocumentUri(context, uri)) {
                        val documentFile = DocumentFile.fromSingleUri(context, uri)
                        documentFile?.let {
                            context.contentResolver.openInputStream(documentFile.uri)?.let { inputStream ->
                                AudioStream(inputStream, documentFile.length(), documentFile.type ?: "audio/*")
                            }
                        }
                    } else {
                        context.contentResolver.openInputStream(uri)?.let { inputStream ->
                            AudioStream(inputStream, song.size, song.mimeType)
                        }
                    }
                } else {
                    AudioStream(File(uri.toString()).inputStream(), song.size, song.mimeType)
                }
            }
        }
    }

AS complains that

openInputStream()

is an ‘inappropriate blocking call’ Is this a false positive by AS, or am I doing something wrong here?

Hello, did you ever forget to activate/collect a

Flow

? If so, please react with ➕. I personally did (and that made a minor bug reach production), so I submitted a feature request: https://youtrack.jetbrains.com/issue/KT-40228

Hi! How do I make this compile?

private suspend fun myStruct(): MyStruct = MyStruct(1) {
    foo()
}

suspend fun foo() {}

private data class MyStruct(
    val a: Int,
    val callback: (() -> Unit)?
)

Why is it necessary to call

BroadcastChannel.send

in a suspended function? With e.g. RxJava pushing events to a subject doesn't require anything special.

Hello ✋ , I have an use-case where I need to implement a

backoff

strategy using the flow.

fun main() {
    runBlocking {
        (0..8).asFlow()
            .flatMapConcat { input ->
                // To simulate a flow which fetches data based on a time-window which is exponentially increased if there's no data
                // available with in that time frame.
                flow {
                    println("Input: $input")
                    if (input < 5) {
                        emit(emptyList<String>())
                    } else { // After emitting this once the flow should complete
                        emit(listOf("Available"))
                    }
                }.retryWhenThrow(DummyException(), predicate = {
                    it.isNotEmpty()
                })
            }.collect {
                //println(it)
            }
    }
}

class DummyException : Exception("Collected size is empty")

private inline fun <T> Flow<T>.retryWhenThrow(
    throwable: Throwable,
    crossinline predicate: suspend (T) -> Boolean
): Flow<T> {
    return flow {
        collect { value ->
            if (!predicate(value)) {
                throw throwable // informing the upstream to keep emitting since the condition is met
            }
            println("Value: $value")
            emit(value)
        }
    }.catch { e ->
        if (e::class != throwable::class) throw e
    }
}

If you run this example in kotlin play ground you will notice that the flow keep emitting the value even after my condition is met(Where the list is not empty anymore). After much digging I found that I need something like takeUntil from rxJava. I need some help how I should proceed this.

What's the best way to flatten a

Flow<Flow<T>>

to

Flow<T>

?

For some reason I can't do this in my project

val f: Flow<Int> = flow<Int> { emit(1) }.transform<Int, Int> { emit(it) }

When I compile, I get

Error:(6, 32) Kotlin: Unresolved reference: transform
Error:(10, 47) Kotlin: Unresolved reference: transform
Error:(10, 69) Kotlin: Unresolved reference: emit
Error:(10, 74) Kotlin: Unresolved reference: it

So I set up a new project with the same dependency on

kotlinx-coroutines-core:1.3.7

. That works fine. Also, oddly, all my other flow-related code that doesn't use

transform

is fine. Has anyone else encountered this before?

When testing suspend functions, does anyone have advice on whether to use

runBlockingTest

vs

runBlocking

?

runBlockingTest

seems to have some nice features, but this line in the documentation makes it sound like there is some risk of deprecation

This API is experimental and it is may change before migrating out of experimental (while it is marked as @ExperimentalCoroutinesApi). Changes during experimental may have deprecation applied when possible, but it is not advised to use the API in stable code before it leaves experimental due to possible breaking changes.

Hey guys how do run flows concurrently then collect the output

@Test
    fun `experimental`() {
        runBlocking {
            val scope = CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>) 
            val songs = listOf("a", "b", "c", "d", "e")
            songs.asFlow().flatMapMerge {
                requestFlow("${Thread.currentThread().name} $it check")
            }
            .collect {
                println("$it")
            }
        }
    }

    fun requestFlow(i: String): Flow<String> = flow {
        emit("$i: First")
        Thread.sleep(1000)
        emit("$i: Second")
    }

Wanted to share the slides for my presentation

Kotlin Actors - No Drama Concurrency

https://speakerdeck.com/handstandsam/kotlin-actors-no-drama-concurrency

How do you document that the

async

-coroutine-builder of the kotlinx.coroutines library can throw exceptions since the coroutine builder itself does not throw an exception but the

.await()

does?

/**
 * @throws[IllegalStateException] ... <- would this be correct?
 */
fun CoroutineScope.doStuffAsync() = async { ... }

hello guys, I am fighting with intellij and coroutines. I try to use

produce

I have imported it from

import kotlinx.coroutines.channels.produce

but the ide still displays it red, and says

create function

on alt-enter. my dependencies are

implementation 'org.jetbrains.kotlin:kotlin-stdlib-jdk8'
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-jdk9:1.3.7'
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-javafx:1.3.7'

I’ve got the following code (Android):

coroutineScope.launch {
    val songs = mutableListOf<Song>()
    mediaProvider.findSongs().collect { (song, progress) ->
        songs.add(song)
        listeners.forEach { listener -> listener.onProgress(progress, song) }
    }

    listeners.forEach { listener -> listener.onComplete() }
}

The scope is instantiated via

CoroutineScope(Dispatchers.Main)

Despite

listener.onProgress

being called on the main thread - and successful log messages when it is called, the UI updates I’m making in response don’t seem to have any effect. Are the

listener.onProgress(progress, song)

calls somehow being gathered up and then all executed at the end? Ooh I think I might be blocking the main thread somewhere

Will contracts be non-experimental in 1.4? What will no longer be experimental in 1.4?

I apologise for all these dumb questions, but anyway.. Why does this work:

override fun loadData() {
        launch {
            songRepository.getSongs(SongQuery.Album(name = album.name, albumArtistName = album.albumArtist))
                .collect { songs ->
                    this@AlbumDetailPresenter.songs = songs
                    view?.setData(songs)
                }
        }
    }

But this doesn’t:

override fun loadData() {
        launch {
            songRepository.getSongs(SongQuery.Album(name = album.name, albumArtistName = album.albumArtist))
                .collect { songs ->
                    this@AlbumDetailPresenter.songs = songs
                }
            view?.setData(songs)
        }
    }

How can I tell if any continuations are in the queue to be run, or if there is a job waiting in delay() etc?

📣

kotlinx.coroutines

version

1.3.8

is released. It is an incremental update with several bug-fixes, a few improvements, and a couple minor features (experimental). Change log is here: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.3.8

QQ: I have a Java API with a callback and I want to wrap that in a kotlin suspend function, so I can avoid callback hell What is the most idiomatic way to do that? One obvious solution is a Countdown latch, but I imagine there being something more elegant.

Would this onEach work to keep the current state of the flow?

private val state = MutableStateFlow(initialState)

   fun getState(events: Flow<Event>): Flow<State> =
        merge(
            state, 
            events.transformLatest {
                emit(state.value.copy(...))
                ...
                emit(state.value.copy(...))
            }
        ).onEach { state.value = it }

Think I'm missing something here, the second emit calling

state.value

gets the same value as the first one

StateFlow does not emit an updated object value if a property of the object is changed, is this the intentional design? Example:

data class Count(var value: Int = 0)

private val count = MutableStateFlow(Count())

val newCount = count.value
newCount.value++ // no emission though value is updated
count.value = newCount // this does not emit either

Hello people, is it okay to make use of

yield

in production code? I rarely see it being used and only seen it used in tests. I wanted to know people's opinion if they ever used this function.

This message was deleted.

I need some help. I have 2 unrelated useCases sharing a common dependency, I want them to share the execution of their dependency if that’s possible, because it’s critical that the dependency is only called once. Is there a way to do this nicely with coroutines?

I've got two deferred values. I want to wait for both, execute something after one of them completes and cancel the other one, not yet completed. What'd be the mechanism to achieve it?

In this post https://medium.com/@elizarov/structured-concurrency-anniversary-f2cc748b2401 by Elizarov, Roman mentions a use case which is very similar to a problem I have.

In fact, there are valid use-cases where you have an entity with its own life-time, like a HTTP connection, and a client code in another module of your application making a request in its own scope. You should be able to easily combine them so that request is restricted to both life-times. Currently, this is cumbersome to implement.

What's the right approach to combine two different scopes/lifecycles when launching a coroutine? In my case I have a BLE connection which creates a scope bound to the lifecycle of the connection, and I'd like a client to be able to launch a task to send a message, but also within the scope in which this client lives. This is for an Android project, so client would normally be a Fragment or ViewModel

Hey stupid question but how do I use the new StateFlow? Im using kotlin version 1.3.72 but it doesnt seem to have StateFlow

Beginner here… Is it a typical pattern to pass around the coroutine scope as a parameter to my suspen functions?