I need advice on something:

sealed class Updates{
    data class LiveUpdate(val progress: Int): Updates()
    data class History(val list: List<Int>): Updates()
}

private fun fetchUpdates(): Flow<Updates> {
    TODO()
}

private fun save(list: List<Updates.History>){
    TODO()
}

fun dispatchUpdates(): Flow<Updates.LiveUpdate>{
    val history = mutableListOf<Updates.History>()
    return fetchUpdates()
        .onEach { 
            if(it is Updates.History){
                history.add(it)
            }
        }
        .onCompletion { if(it != null) save(history) }
        .filterIsInstance<Updates.History>()
}

Is there any way to improve this code so that the save part doesn't end up a side-effect? Something like an inner flow. I'm not sure how to approach this

We have some HTTP requests which we are currently running consecutively but it's taking too long, so we would like to parallelise them. We would need to halt execution until they are all complete, and propagate an exception if any failed (wouldn't matter which exception was thrown if multiple requests fail). We don't need to capture any response data. What would be the neatest/most idiomatic way to achieve this?

Hey guys. At the moment I'm processing events in the following way:

val channel = BroadcastChannel<Event>(10)
val receiveChannel = channel.openSubscription()
val flow = receiveChannel.consumeAsFlow()

flow.onEach(someProcessor::doOnEach)
    .catch { log.error(it) { "Error while processing event" } }
    .launchIn(coroutineScope)

In my case I have to just skip an event if an exception is thrown in

someProcessor.doOnEach(event: Event)

and continue processing of next events. I can achieve what I want with

flow.onEach {
  try {
    someProcessor.doOnEach(it)
  } catch (e: Throwable) {
     //log
  }
}

but I feel like I'm missing something. Is there a better way to do this?

Can someone explain me why the compiler has executed the line 24 before starting the coroutine at the line 18 ?

hi! Is there an existing suspend memoize function? If not, is the following code idiomatic?

fun <T> (suspend () -> T).memoize() : suspend () -> T {
    var value: T? = null
    return suspend { value ?: this().also { value = it }}
}

suspend fun work() : Int = coroutineScope {
    delay(1000)
    22
}

val memoizedWork = ::work.memoize()

Another question 🙂 Why is this warning? As I understand, it's the recommended way to wrap IO-bound blocking calls with

<http://Dispatchers.IO|Dispatchers.IO>

Can StateFlow represent one off events?

Have an Android app that launches the following coroutine when the app goes in the background (all activities are stopped):

sessionTimeOutJob = GlobalScope.launch {
  delay(5.minutes)
  activity.finishAffinity()
}

So after 5 minutes the coroutine should resume and call finishAffinity(). But it doesn't happen on release builds. Only on debug builds. On release builds it first resumes when the app is selected in the Recents menu. Long past the 5 minute mark. Is it somehow the app lifecycle that prevents the coroutine from resuming?

if

p

is

Deferred

, then I expected it to be cancelled on timeout:

select<Unit> {
    p.onAwait { println("ok") }
    onTimeout(1000) { println("timed out!") }
}

but it's not, it prints "timeout out!",

p

is not cancelled. What is the best way to cancel a

Deferred

by timeout?

Has anyone ran into use cases for overloads of

Flow

operators for

StateFlow

that preserve the

value

property? I filed an issue for

map

here, curious if people have other use cases or use cases for other operators (or opinions about this proposal). https://github.com/Kotlin/kotlinx.coroutines/issues/2081

Hi everyone, curious about everyone’s preferred nomenclature for methods returning

Flow<T>

For example:

// For Flow<List<*>>
fun postsFlow(): Flow<List<Post>> 
// OR
fun postLists(): Flow<List<Post>>


// For Flow<*>
fun eventFlow(): Flow<Event>
// OR
fun events(): Flow<Event>

Would be interesting to know if there are other preferred styles.

Hi need some help, i got error for the suspend function

transactionAwait

: "suspend function can be called only within coroutine body".

callRealmSuspend { realm ->
            realm.transactionAwait {
                realm.delete(RegionRealmObject::class.java)
                realm.delete(ServiceRealmObject::class.java)
            }
        }

and my

callRealmSuspend

function looks like this:

suspend fun callRealmSuspend(call: (realm: Realm) -> Unit): Realm? {
    var realm: Realm? = null

    try {
        realm = Realm.getDefaultInstance()
        call(realm)
    }  catch (error: java.lang.Exception) {
        LogHelper.e(messages = *arrayOf("Unhandled error occurred: ${error.localizedMessage}."))
    }

    return realm
}

so how do I change

callRealmSuspend

function to fix the error/ provide a coroutine body? thanks!

Hi all! I am learning Flow, and was trying to understand how a flow can consume other flows in parallel and what happens in error scenarios. Here’s a test that I can’t get to work (requirements are documented in the

handler

function): https://pl.kotl.in/zyHCNXVLg

Why cancelling

channelFlow

does not cancel the child?

suspend fun run() = channelFlow {
    suspendCancellableCoroutine<Unit> { cont ->
        cont.invokeOnCancellation { println("Cancelled.") }
        cont.resume(Unit) { }
    }

    for (x in 1..10) {
        send(x)
    }
}

suspend fun main(): Unit = coroutineScope {
    withTimeout(1000) {
        run().collect {
            println(it)
            delay(2000)
        }
    }
}

It produces:

1
Exception in thread "main" kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1000 ms
	at kotlinx.coroutines.TimeoutKt.TimeoutCancellationException(Timeout.kt:158)
	at kotlinx.coroutines.TimeoutCoroutine.run(Timeout.kt:128)
	at kotlinx.coroutines.EventLoopImplBase$DelayedRunnableTask.run(EventLoop.common.kt:497)
	at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:274)
	at kotlinx.coroutines.DefaultExecutor.run(DefaultExecutor.kt:68)
	at java.base/java.lang.Thread.run(Thread.java:834)

Process finished with exit code 1

Hi, I get Inappropriate blocking method call (

BlockingMethodInNonBlockingContext

) on the

exec()

call in this code, but this is a false positive, right? Since I run with

<http://Dispatchers.IO|Dispatchers.IO>

I don’t see how the code would be problematic:

suspend fun String.test() = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    Runtime.getRuntime().exec(this@test)
}

I’m looking for an alternative to flow.emit(). I would need of a data structure (a sized circular buffered list) that would be “coroutine-safe” and the “put” operation is not blocking. The producer should not depend on the consumer and, it should be able to emit data even if there are no consumers listening. I was thinking to Channels, but they offers non-suspending send() only when declared Channel.UNLIMITED. Do you have any suggestion?

Hi. How do I figure out that the downstream is exhausted (completed/whatever the right terminology is) in a flow builder? This is the code I currently have: https://pl.kotl.in/tCjm9ICUp - it will time out because my loop condition is obviously not working. In general, what I'm trying to do is have this

while (...) { delay(1000); emit(buffer); buffer.clear() }

to have timed buffering. I'm aware there are operators for this, just trying to figure things out for myself. Also, any other notes on pitfalls of my implementation would be appreciated.

Hi, What is the idiomatic way to make suspend functions cross Java callback boundaries? This code

suspend fun g(): String = "bar"
suspend fun f(mutableMap: MutableMap<String, String>) {
    mutableMap.computeIfAbsent("foo") {
        g()
    }
}

gives me

Suspension functions can be called only within coroutine body

compilation errror on the

g()

call. The only solution I can come up with is to re-implement

computeIfAbsent

in Kotlin and make the

fun

inline

, but that does not seem like an elegant, idiomatic solution. What is the proper solution to this problem?

📣 Shootout to all Android devs using MVI, MVVM, etc with coroutines. How do you handle events that need to be processed exactly once by your views (like navigation events, etc)? That is, what approach do you use with coroutines to work around the problem explained here for LiveData: https://medium.com/androiddevelopers/livedata-with-snackbar-navigation-and-other-events-the-singleliveevent-case-ac2622673150 0️⃣ I’m avoiding this problem altogether, not using anything like

Rx/Channel/Flow/LiveData

to deliver those events (comment in :thread-please: what you do instead). 1️⃣ Using an

Event

wrapper class, adapting solution in the above story to coroutines. 2️⃣ Using a [buffered or rendezvous]

Channel

for events, exposing it to the views either as

Channel

or as a

Flow

via

receiveAsFlow()

extension. 3️⃣ Some other approach (comment in :thread-please:).

📣 For all of you who picked 2️⃣ (using

Channel

) how do you work around the fact that

Channel.receive

can get a message in an activity that is already destroyed and thus attempt to work with it will cause an

IllegalStateException

? (see this demo https://pl.kotl.in/Rs18kihuJ) The same problem happens when you use

Channel.receiveAsFlow

and then collect from the flow (see this demo https://pl.kotl.in/-ehZrPMl4). 😱 I never knew it could happen! All my code is broken! 👍 It never happens to me because I use

immediate

dispatcher (Android provides that by default) and I always post from the main thread, so between the event is posted and is being processed my activities cannot get destroyed. 👉 I do something else (comment in :thread-please:)

@elizarov What is the motivation behind these two polls? What are the answers going to be used for?

Hi, I got warning

Inappropriate blocking method call

when call blocking method in suspend function. I don't know that why this code was warned... How to fix this?

import java.net.URL

suspend fun foo(bar: String): String {
    if (bar.isEmpty()) {
        return ""
    }
    return withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
        // warning: Inappropriate blocking method call
        URL("<https://www.example.com>")
            .readText()
    }
}

Hey Flow.zip works with hot flows correct?

I'm planning to call method in suspend function that will throw IOException. In this case, IDE warn it as

Inappropriate blocking method call

. So, Should not I call method(will throw IOE) in suspend function?

suspend fun foo() {
    // warning: Inappropriate blocking method call
    throwIoException()

    withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
        // warning: Inappropriate blocking method call
        throwIoException()
    }
}

@Throws(IOException::class)
fun throwIoException() {
    // something blocking process
}

Hey, I try to use Kotlin

1.4-M2

and related

1.3.7-1.4-M2

Coroutines dependencies with Gradle 6.5 and get this error:

> Could not resolve all files for configuration ':kotlin-coroutines:kotlinCompilerPluginClasspath'.
   > Could not resolve org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.1.
     Required by:
         project :kotlin-coroutines > org.jetbrains.kotlin:kotlin-scripting-compiler-embeddable:1.4-M2 > org.jetbrains.kotlin:kotlin-scripting-compiler-impl-embeddable:1.4-M2
         project :kotlin-coroutines > org.jetbrains.kotlin:kotlin-scripting-compiler-embeddable:1.4-M2 > org.jetbrains.kotlin:kotlin-scripting-compiler-impl-embeddable:1.4-M2 > org.jetbrains.kotlin:kotlin-scripting-common:1.4-M2
      > Cannot choose between the following variants of org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.7-1.4-M2:
          - iosArm32-api
          - iosArm32-iosArm32MetadataElements
          - iosArm64-api
          - ...
  All of them match the consumer attributes:
          - Variant 'iosArm32-api' capability org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.7-1.4-M2:
              - Unmatched attributes:
                  - Provides org.gradle.status 'release' but the consumer didn't ask for it
                  - Provides org.gradle.usage 'kotlin-api' but the consumer didn't ask for it
                  - Provides org.jetbrains.kotlin.native.target 'ios_arm32' but the consumer didn't ask for it
                  - Provides org.jetbrains.kotlin.platform.type 'native' but the consumer didn't ask for it
          - ...

Any hint about how to fix this (worked fine with Kotlin

1.3.72

and Coroutines

1.3.5

)?

I am using

runBlocking(someContext)

, and within the block a call is made to a method that is calling

launch

using a scope that was created with the same context. however, it is not blocking. am I doing something wrong?

If I cancel an entire coroutineScope, will launches still be accepted with the same scope? I'm trying to resolve a debate.

In this code will the fist collect finish before the second collect is started or is possible that the execution will interleave?

val singleThread =   newSingleThreadContext("SingleThread")
val flow = flow { 
 emit("a")
 blocking call
 delay(1000)
 emit("b")
}.flowOn(singleThread)

launch(<http://Dispachers.IO|Dispachers.IO>) {
    flow.collect {
        println(it)
    }
}
launch(<http://Dispachers.IO|Dispachers.IO>)  { 
    flow.collect {
            println(it)
    }
}

will the return value from

scope.actor<>{}

always implement

Job

? I want to implement some code that calls

actor.close()

, and then just to keep things synchronous I want to call

(actor as Job).join()

, will that work in future kotlinx-coroutines?