I'm confused as how to to plug a suspending function into a flow I notice the flowOf builder, but that requires value, so that means its evaluated beforehand I'd expect something like this

flowOf {
   someSuspendFunction()
}

am I missing something, or is it left to clients to create such extension comming from rx, I have it mapped in by brain as such

Observable.just == flowOf(value)
Observable.fromCallable == flowOf { .. }

if all my

Dispatchers.Default

threads (

Runtime.getRuntime().availableProcessors()

) are blocked under

runBlocking

, a

GlobalScope.future(<http://Dispatcher.IO|Dispatcher.IO>)

also fails to launch; why is this the case?

So.. is it safe to use

runCatching

in coroutines for control flow? I gather it will catch CancelationException as well. Won't that break stuff?

I updated my question and posted it to stack overflow: https://stackoverflow.com/questions/66784930/how-to-find-first-desired-result-from-kotlin-coroutines-deferred-server

Is this a known issue? https://pl.kotl.in/ILirg4-hM Using Result as flow type throws ClassCastException.

I’m having trouble wrapping my head around a paragraph in the Coroutines Basics doc:

But what if the extracted function contains a coroutine builder which is invoked on the current scope? In this case, the

suspend

modifier on the extracted function is not enough. Making

doWorld

an extension method on

CoroutineScope

is one of the solutions, but it may not always be applicable as it does not make the API clearer. The idiomatic solution is to have either an explicit

CoroutineScope

as a field in a class containing the target function or an implicit one when the outer class implements

CoroutineScope

. As a last resort, CoroutineScope(coroutineContext) can be used, but such an approach is structurally unsafe because you no longer have control on the scope of execution of this method. Only private APIs can use this builder.

Particularly, what does it mean that having an explicit CoroutineScope as a field on a class containing the target function is preferable to calling CoroutineScope(…)? How else would you create an explicit CoroutineScope as a field, other than

val scope = CoroutineScope(…)

?

Hey everyone, I’m having some issues finding some way to do something similar to DispatchQueue in Swift What I want to do is launch one coroutine off the main thread with the first queue element and when it’s done, launch the next in queue. I could implement something with async and all that, but I was wondering if there’s anything that already exists to support this behaviour?

Hi! Does anyone know how I should be able to compare a return value with

COROUTINE_SUSPENDED

? I'm building an AOP-based logging interceptor and need to check if the return value of a reflective methodcall is

COROUTINE_SUSPENDED

🙂 Currently doing this by checking

if (result.toString() == "COROUTINE_SUSPENDED")

but it feels like I should be able to access the symbol directly..

it still seems unclear whether to use

callbackFlow

vs

channelFlow

. It seems the latter respects operator fusion, but the prior does not?

If I were to inject CoroutineScope everywhere, do you bother to make sure nobody calls cancel on you? Is there some base interface so I dont need to compose myself?

I've been holding off from coroutines a while now. Every time I tried adding them, I'd run into severe headwinds. Today I tried again, and basically just converted my executor service like this:

-    executorService: ExecutorService = Executors.newCachedThreadPool(Executors.defaultThreadFactory())
+    executorService: ExecutorService = Executors.newCachedThreadPool(Executors.defaultThreadFactory()).asCoroutineDispatcher().executor as ExecutorService,

and bam! Performance up by 40% on my heaviest function.

Hi, I have a

coroutineScope

and within this scope I want to cancel it. How can I prevent cancel cooperation of this scope, so that cancellation isn't propagated?

Hi all, I have a weird behaviour regarding coroutines on Android and I hope that you can help me: I have a suspending function that performs some UI logic

suspend fun showSnackbar() {
    val snackbar = Snackbar.make(...)
    snackbar.show()
    delay(4000)
    snackbar.setAction("...", null) // crash here
    snackbar.dismiss()
}

The functions is called via

launch(Dispatchers.Main) { showSnackbar() }

. Please don't ask why I have to

null

the action. It is required to prevent a false positive memory leak warning. Anyway, the call

snackbar.setAction()

crashes the app with

android.view.ViewRootImpl$CalledFromWrongThreadException: Only the original thread that created a view hierarchy can touch its views.

and I wonder if

delay()

might change the Dispatcher?? 🤔 Thanks for your help!

Can someone explain me why the second child is not cancelled?

coroutineScope {
		launch {
			if (condition) cancel() // does not cancel the coroutine of flow
		}
		
		launch {
			dataSource.channel.receiveAsFlow() {
				.collect { doSomething() }
		}
	}

I would expect that cancelling first coroutine would cancel second coroutine but it doesn't. Any ideas?

Is there an operator for

SharedFlow

that cancels the flow automatically after all subscribers have left? Something like

refCount()

in RxJava.

Is it possible to receive from a

Channel

without removing the element?

why is this

fun intervalFlow(delayMillis: Long): Flow<Int> {
    return flow {
        withContext(Dispatchers.Default) {
            var i = 0
            emit(i)
            while (true) {
                delay(delayMillis)
                emit(++i)
            }
        }
    }
}

wrong -- I mean I read the article, but at runtime it doesnt crash, and printing threads in each step, seems correct

Ok so I think the answer to this is yes but is it unsafe to have a

ViewModel

collect from a

Flow<T>

given to it by a

View

,

Activity

or

Fragment

? I.E.

// MyView.kt
val events: Flow<Event> = flowOf(Event1, Event2)

viewModel.handleEvents(events)

Is there a testing library other than turbine, which would work with MutableStateFlow, i.e. sideffecting type of code? (toList hangs) in rx it would looks like this

@Test fun `doSomething should move state to FINAL`() {
    val viewModel = createViewModel(initialState = State.INITIAL)

    val testSubscriber = viewModel.state.test()
    viewModel.doSomething()

    testSubscriber.assertValues(State.INITIAL, State.FINAL)
}

I know this is a recurrent question and Im a bit sorry for asking again but: Q: What is currently the best an simplest way to ensure that a mutable variable is only changed by one coroutine simultaneously? A:

MutableStateFlow

If I run into an NPE on that line, how would I fix it?

Hey everybody. I was hoping I could get some help in how to best approach this situation… I have a suspending function that returns some data. I may need to suspend the coroutine that called the function while something else is done asynchronously, and then resume the coroutine, eventually returning data from that function.

suspend fun foo(): Bar {
    // I call another suspend function here returning an intermediate value.
    if (error) {
        suspendCancellableCoroutine<Bar> { continuation ->
            val listener = object : Listener {
                override fun callback() {
                    // I need to call this suspend function again.
                    // However, here is where I get an error.
                    continuation.resume(foo())
                }
            }
            // I call other functions that will eventually call the listener.
        }
    }
}

The error that I get is obvious -

Suspension functions can be called only within coroutine body

. However, I’m not really quite sure on how to resume the coroutine properly. Any help is much appreciated. Thank you.

Does anyone know of any good articles on replacing RxRelay with Stateflow/SharedFlows?

Hi Everyone! I have a question that straddles the Android/coroutine slack channels, but I think the heart of the question is about structured concurrency. I'm working on an Android app that has a ViewModel that references a repository. The repository exposes a method that makes two concurrent API calls and returns the result of both after both have finished. What I have now is something like this:

suspend fun fetchTwoThings(): TheResultsModel =
            coroutineScope {
                val job1 = getStuff1(this) // both of these functions use the passed in coroutine scope like scope.async(<http://Dispatchers.IO|Dispatchers.IO>) { ... API call here }
                val job2 = getStuff2(this)
                awaitAll(job1, job2)

                return@coroutineScope TheResultsModel(
                        thing1 = job1.await(),
                        thing2 = job2.await()
            }

Is there any downside to using the

coroutineScope

like this in the repository? Alternatively, I could pass in the

viewModelScope

. Maybe that's ok since the ViewModel class instantiates the repo as a member variable. The lifetime of the repository is already tied to the ViewModel class. Yet another option is to expose

getStuff1

and

getStuff2

to the ViewModel, and manage the coroutine there. I'm not a huge fan of that since I might want to call

fetchTwoOtherThings

from other places and would have to duplicate that code. What are your thoughts? How can I do this better?

Is

Dispatchers.Unconfined

equal to rx

Schedulers.trampoline()

? if so, why does my code not run blockingly if I replace all dispatcher references with the

Unconfined

one? That's what I used to do for testing in rx Is it not supposed to? To not impose any threading, i.e. run on native unit test thread? (Im aware of TestDispatcher, that works as expected)

Has anybody had ever written timeout tests for Retrofit with MockWebServer + coroutines? I’m getting

This job has not completed yet

on the following test:

@Test
    fun requestWithTimeout_serverRespondsInTime_returnsResponse() = coroutinesRule.testDispatcher.runBlockingTest {
        val body = "{ field: Test }"
        val response = MockResponse()
            .setResponseCode(HttpURLConnection.HTTP_OK)
            .setBody(body)
            .throttleBody(body.length.toLong() / 2, 9, TimeUnit.SECONDS)

        mockWebServer.enqueue(response)

        val actual = service.requestWithTimeout(20_000)

        assertEquals(ResponseTest("Test"), actual)
    }

I have a remote call (

suspend

function via Retrofit) that is part of my (Android app's ) data syncing. I am attempting to ensure that the code block with this call completes, even if the calling coroutine (started in a

viewModelScope

) is cancelled. My initial thought was to wrap the function in

appScope.async { //code }.await()

, where

appScope

is an application-level scope. Android Studio has suggested that I just use

withContext(appScope.coroutineContext) { //code }

. Are these really equivalent in terms of coroutine lifecycles? I had thought that

withContext

created a child coroutine (with a different, but related

CoroutineContext

) that would be cancelled if the context creating it is cancelled, whereas

appScope.async()

creates a whole new coroutine with its own lifecycle tied to

appScope

and not the original context. Also welcoming input on my solution to the initial problem.

Hello! I’m new to Kotlin and checking the documentation, I found some statement for this code:

fun main() = runBlocking<Unit> {
    val counter = counterActor() // create the actor withContext(Dispatchers.Default) {
        massiveRun { 
            counter.send(IncCounter)
        } 
}
    // send a message to get a counter value from an actor
    val response = CompletableDeferred<Int>() 
    counter.send(GetCounter(response)) 
    println("Counter = ${response.await()}")     
    counter.close() // shutdown the actor
}

Actor is more efficient than locking under load, because in this case it always has work to do and it does not have to switch to a different context at all. How to understand it?And how is it efficient than locks?

If I have a Stateflow is there a difference between

emit(newValue)

and

value = newValue

like there was for post/value in Livedata? I'm not seeing any documentation, I think we're supposed to use Emit, but using value = appears to be working in my code?