flow {
    while (true) {
        delay(1000)
        emit(Unit)
    }
}

?

Is there something like

Flow.distinctUntilChangedOrCollected()

or a simple way to implement it? Basically a value in the

Flow

should be ignored if a) it’s equal to the previous value and b) the previous value hasn’t been collected downstream yet (due to

mapLatest

being slow). If I only use

distinctUntilChanged()

then the second value would obviously be ignored independent of b).

Hi guys. What is the equivalent of “onEachLatest” operator in Flow? I need to cancel previously running suspend fun in onEach block

TIL that my coworkers are just annoyed as I am that the first auto-suggestion that comes for

flow.colle

is not the extension function that takes a lambda but the collect that takes a flowCollector. Not sure if there is something the library could do to improve that, or if it's more something for intellij (if I'm always selecting the 2nd suggestion, make it the first suggestion from there on?).

What's the easiest and idiomatic way to have Java interop with coroutines. I'd like to call suspend functions somehow in Java. We use our own internal futures library so ideally we'd like some way to take a suspend function and expose it as the future type. And it would be nice to not have to write two functions (one suspend and one that exposes the future). Any suggestions? I took a look at examples in the integrations in repository but not sure if they're updated and the most recent recommendations. @elizarov advice?

Is it possible to profile the performance of coroutines using VisualVM? If it can be done how would the coroutines be profiled? I'm mainly interested in memory usage as I ended up with a bit of a scare the other day when the JVM crashed due to running out of memory. Noticed when running VisualVM that the profiled JVM program is using many threads which is concerning considering that the program only uses coroutines. Was expecting the thread count to average around 5 with the total CPU cores mainly affecting the thread count. Actual thread count is averaging around 29.

Is there a more idiomatic way to create a child scope from a parent scope?

val parentScope = CoroutineScope(Dispatchers.Default)
 val childScope = parentScope + Job(parent = scope.coroutineConext[Job]!!)

just a PSA: be careful when you create a new

CoroutineScope

and make sure realize the side effects of breaking structured concurrency. ex:

CoroutineScope(Dispatchers.Default)

explicitly breaks structured concurrency. This has testing implications like passing tests even when exceptions are thrown https://gist.github.com/ZakTaccardi/e77d5983660ce4e8e0fa2f2ef0d582ea

How do you know when to switch to another dispatcher? On android, when performing a network request within a viewModel scope, my understanding okhttp will perform the fetching task on a background thread. So when I get the response, and want to massage the data before sending to the UI, is this work ok for the main thread? Or do I need to switch to another Dispatcher at this point?

viewModelScope.launch {
   val myResponse = fetchMyData() // okhttp moves this work to a background thread
   doSomethingWithMyResponse(myResponse) // Does this need to be done on a background thread using withContext?
}

How safe is it to use AbstractCoroutine class?

Android Studio / Idea gives me nice hints in my code. Sometimes, A function call is marked with a yellow background because it's an «inappropriate blocking call». But why does such a blocking call become inappropriate when I add the suspend marker to my function? Especially in android code, I use coroutines to escape the main thread. So when I launch, I'm off of the main thread because I know my code might block. Can anyone point me to some reading that explains why the call is more inappropriate in a suspend fun?

Is there a well-known thing with suspend functions being annotated with @JvmStatic? (All my troubles went away as soon as I removed the annotation.)

I have been integrating coroutines with the GraphQL-Java async API, which requires certain functions that I implement to return CompletableFutures. For unfortunate reasons, there are certain methods (this is the DataLoader API if you know GraphQL-Java) that I am only allowed to call synchronously from my functions that returns the CompletableFuture, not later. I've implemented this as a wrapper around a suspend function (eg

fetcher

), like this:

return someCoroutineScope.future(start = UNDISPATCHED) { fetcher() }

UNDISPATCHED has exactly the semantics I need — the code in

fetcher

can safely call the "only before suspending" DataLoader methods as long as it does so before suspending. This works in the sense that it allows me to write code that works. However, if I make an error and call a DataLoader method after having suspended, my code just hangs (because that's the symptom I'm trying to avoid). I'd prefer that it failed fast. (Well, I'd really prefer that none of this junk was necessary, there's an issue on the library about fixing it... but trying to work around today.) So my question is: If I'm running code inside the context of a

CoroutineScope.future(start = UNDISPATCHED) { ... }

, is there any way to know if I'm still in the undispatched part of the execution? (Happy to track this state in a boolean or something if there's a way to get a hook on dispatch.)

Scope question wrt channels and a "servlet style" setup.. We have a system that creates instance of our class(think servlet) and has lifecycles for that instance. In the "do get" method, I create a produce channel(Bar), and then proceed to consume from it. I might swap to flows, but in this case, is this all setup right wrt the structured concurrency? When I close the second instance, the channels seem to shutdown correctly. The only thing that feels like it's missing is a server level context(that would parent the

instanceScope

)

Hello guys, just wanna know your opinion on this launch. The this.cancel() will cancel the coroutine? Do I need to track the scope and finish it when the class is destroyed? Might main go here is launch and forget.

CoroutineScope(Default).launch {
            //do on background
            formDataDTO?.ad?.characteristics?.find { it.id == GENERATION }?.let { cha ->
                withContext(Main) {
                    //do on ui thread
                    characteristicBinder.bind(cha)
                }
            }
            this.cancel("has done his job")
        }

Hi, I'm trying to find the best primitive to use for my use case, maybe in coroutines? maybe in rx? something else perhaps..? I have some state that I'd like to represent as some kind of Flow. I'd like to have multiple subscribers, a default value when subscribing (the default value would be computed at the subscribe time). I need to be able to add more data to that flow at runtime (like you can do with a

ConflatedBroadcastChannel

) and ideally, one subscriber could unsubscribe without disrupting the other subscribers.

I'm currently using a

ConflatedBroadcastChannel

but I need to make sure everyone subscribes at the same time, before any data comes to the channel for all the initial data to be there, for everyone, which is not exactly what I want

Is there a way to make an (inline) function do something iff it is called from a suspend function? Or to, like, overload a function to have a from-suspend and not-from-suspend version? What I'm trying to do is basically have a function we can call for slf4j MDC like

withLoggingField("key", "value") {
  // maybe do other stuff, maybe this next line is actually in some other function called by this block
  <http://log.info|log.info>(...)
}

which, if we're not in a suspend function just does an MDC.put and resets it, but if we are in a suspend function uses kotlinx.coroutines.slf4j.MDCContext to make the MDC context thread-local work

why does idea mark

jackson.readValue(string, class)

as Inappropiate blocking call in a suspend method?

I'm not sure on the proper syntax to get this to work. The error is that a suspend function can only be ran inside a coroutine scope. How do I make "runSomething" be a coroutine scope which is one the launch inside starts? runSomething { someSuspendFunction() } fun runSomething(action: () -> Unit) { viewModelScope.launch { try { busy=true action() } finally { busy=false } } }

One thing that was nice about rxjava is that it forced you to consider the error path. I’m finding that this is a big problem with coroutines especially since kotlin doesn’t believe in checked exceptions. When making a network call one should always handle the error path, with coroutines it’s easy to forget since it’s purely optional. Any ideas how to remind developer to always wrap network calls in try/catch and handle the ioexceptions?

How to implement Dispatchers.Main and bind the main thread into it?

How can I assign a job (local variable) to a coroutineScope?

I have the following code

protected suspend fun stdout(message: String, emit: suspend (WfResult) -> Unit) {
    emit(WfResult.Stdout(message))
}

protected suspend fun stderr(message: String, emit: suspend (WfResult) -> Unit) {
    emit(WfResult.Stderr(message))
}

open fun main() : Flow<Any>  = flow {
   stdout("someString" , ::emit)
 }

How do I make my own flow so I can use it like this (or how do I extend flow to add more methods):

open fun main() : Flow<Any>  = myFlow {

   stdout("someString")
   stderr("someString")
 }

Is it possible to turn on the stack trace recovery on Android? I'm finding it very difficult to understand the issue based on what's printed by default.

👋🏻 Hey everyone, I'm looking to create a

CoroutineScope

which I want to run in the background while it's waiting for updates to be sent via a Channel, but once the update is sent, I want the result to be executed on the main thread. In that case, would I want to use

Dispatchers.Main.immediate

, something like this

CoroutineScope(Dispatchers.Main.immediate + Job())

? I think I'm getting a little hung up as I'm not sure when I would have a Scope that doesn't use that dispatcher (since I can launch a suspending function with the scope, but telling it to switch to a different dispatcher, say

IO

).

Quick ? on

BroadcastChannel<T>.asFlow()

The docs state: `

If the flow consumer fails with an exception, subscription is cancelled.

I have a

broadcastChannel.asFlow().onCompletion{ printSomething } .map {  codeThatThrowsException }.catch { print MessageAboutError } . collect { collector}

The onCompletion fires, then the catch. Afterwards, the collect never sees a new message.. I just added the onComplete to validate. I expected the

.catch

to handle the error so the flow doesn't get canceled. I can easily fix the

codeThatThrowsException

but want to catch unexpected errors in the future. It looks like the downstream catch doesn't get a chance to handle the error prior to the upstream channel getting unsubscribed from

I'm trying to use Roman Elizarov's deep recursion implementation with coroutines (https://medium.com/@elizarov/deep-recursion-with-coroutines-7c53e15993e3), but am having a bit of an issue with calling methods from within the recursion function. I'm attempting to use the implementation without modifying it (in the hopes that it is added to the stdlib some day. Say I have this setup.

private suspend fun DeepRecursiveScope<Any?, Unit>.ppAny(obj: Any?) {
    DeepRecursiveFunction<Any?, Unit> { obj ->
        ppObj(obj)
    }
}

private suspend fun DeepRecursiveScope<Any?, Unit>.ppObj(obj: Any) {
    ppAny(obj)
}

I need to declare both as suspend and both as extension functions in order for them to be callable from within the DeepRecursiveScope. But

ppAny

is not callable from within

ppObj

, with the error

[NON_LOCAL_SUSPENSION_POINT] Suspension functions can be called only within coroutine body

Now, I understand the issue (I think), in that I need to be calling from within a context with,

with(this@DeepRecursiveScopeImpl)

, but I do not have access to

DeepRecursiveScopeImpl

. Can I use another scope for this? If not, how can I get access to this scope to call my method with? Am I doing stuff completely wrong here???

Is there a way to debug

Mutex

deadlocks caused by recursion?