I have an actor, i dispatch a message to it that contains a suspending lambda, that lambas operation dispatches more messages to the actor, but they never seem to arrive. is that because the actor is still busy processing the first message which is waiting for the suspending lambda to complete?

Am I the only one who thinks it is strange you cannot close

MutableSharedFlow

? Seems that cancelling all listeners (so parent scope) is the only way for this program to close 😛 https://pl.kotl.in/EzyFeqARz

It seems

SharingStarted.WhileSubscribed()

isn't working on 1.6.0-native-mt on iOS

Can anyone recommend any resources that cover the internal implementation of courotines? I've been using suspendCoroutineUninterceptedOrReturn in my networking library. When I made it I looked into what suspending functions compiled down to, so I know that at least at some point in the past a suspend function was basically a function that takes a continuation and has a big switch statement over continuation.label which it uses to determine where it will resume from each time it gets called. Even if this is still true I don't fully understand how the continuation itself is created and how confined vs unconfined works etc. The actual issue I'm having is understanding where exceptions end up going since it seems to be based on what scope is being used or something along those lines.

If I have a

CoroutineScope

and then I have dynamically created another scope, how could I add this scope to the first scope, so that second is cancelled along with the first?

Is there a nice way to convert replaying Flow into a non replaying? A naive implementation using

drop(1)

makes subscriber loose first emitted item:

val flow = MutableSharedFlow<Int>(replay = 1)
  fun get(replayLast: Boolean): Flow<Int> = flow.drop(1) // naive impl

  launch { get(replayLast = false).collect { println(it} } } // doesn't receive 1
  launch { delay(1000); flow.emit(1) }

Is having two separate flows: one with replay and one witout - the only option?

Hi im attempting to use

private val mutableDisplayedPageNoState = MutableSharedFlow<Int>(replay = 0, extraBufferCapacity = 3, onBufferOverflow = BufferOverflow.DROP_OLDEST)

Is it possible to test actors (or more generally infinity loops) with a TestScope without creating a separate Job for an actor's owner (with the following cancellation before runTest completes)?

I want

scanAudios()

and

scanVideos()

to execute first before other stuff in this function.

private suspend fun getFolderDataList(): List<FolderData> {
        thisScope.async() {
            scanAudios()
            scanVideos()
        }.await()

        val audioFolderData = viewModel.getAudioUris().map { uri ->
            FolderData("Audio", uri)
        }

        val videoFolderData = viewModel.getVideoUris().map { uri ->
            FolderData("Video", uri)
        }

        return audioFolderData.plus(videoFolderData)
    }

Am I doing it correctly?

I have recently read an article which suggests that SupervisorJob shouldn't be used and CoroutineScope with proper try/catch (which is more mallable for your usecase) should be used instead (see SupervisorJob section there). On one hand I find this is true, but I still keep thinking that SupervisorJob is useful in scenarios where you specifically manage some task-like coroutine executions. What do you think?

combine(
    flowOne(),
    flowTwo(),
    flowThree()
) { _, _, flowThree->
    flowThree
}.stateIn(
    viewModelScope + ioDispatcher,
    SharingStarted.WhileSubscribed(NETWORK_RESULT_SHARING_TIMEOUT),
    Result.Loading
)

Should 

flowThree

 emit values here when collected? FlowThree should emit every time flowOne() or flowTwo() emit values.

flow.flatMapLatest { flowValue -> 
    stateFlowDependingOnFlowValue(flow1Value,scope) // internally does something like `.stateIn(scope)`
}

Given this code, every time

flatMapLatest

is called, a new coroutine is spawned in the scope. Is there a way to use a scope, which is only active within

flatMapLatest

, so old calculations would be cancelled?

there was some handy function to suspend coroutine forever until cancellation will be triggered. can't find it right away, does anyone remember?

Kind of a design question that is enabled by coroutines semantics:

/**
   * This function returns once a shake was detected
   */
  suspend fun detect() {
    val sensorManager = context.getSystemService(Context.SENSOR_SERVICE) as SensorManager?
      ?: awaitCancellation()
    val shakeDetected = CompletableDeferred<Unit>()
    val listener = SeismicShakeDetector.Listener {
      shakeDetected.complete(Unit)
    }
    val shakeDetector = SeismicShakeDetector(listener)
    try {
      shakeDetector.start(sensorManager)
      shakeDetected.await()
    } finally {
      shakeDetector.stop()
    }
  }

I’ve written this function that suspend until a shake was detected. Do you think such a usage of coroutines makes sense? It feels kind of unintuitive in terms of ergonomics of a regular function

Is it ok, even if not recommended, to use Kotlin 1.6.10 and Coroutines 1.5.2 together?

Hello there! I’m looking for a right operator to implement the following logic: I need to make a request to load an object. Basically, it’s a class with some fields. Then I need to subscribe to the changes relative to this object. These are granular changes — transformations applied to some of the fields of this object.

fun loadObject(id: Id) : Flow<Object>
fun subscribe(id: Id) : Flow<List<SubscriptionEvent>>

First, I tried to use

combine

operator in the following manner:

loadObject(id).combine(subscribe(id)) {  obj, events ->
        var result = obj
        events.forEach { event ->
            result = when (event) {
                is Init -> {
                    obj
                }
                is Amend -> {
                    result.amend(event.diff)
                }
                is Set -> {
                    event.data
                }
                is SubscriptionEvent.Unset -> {
                    result.unset(event.keys)
                }
            }
        }
        result
    }

But this doesn’t work, obviously, since in the flow, there is always initial object (emitted only once after loading), which is combined with latest events / transformations, whereas what I need is something like

scan

operator, where I could first initialize (load) my object, then start consuming events, which lead to the transformation of this object. Observers will receive something like: “object in initial state, after loaded”, “object with the first sets of transformations applied”, “object with the first and second set of transformations applied”. Is there something like:

transformations.scan(initialObjectFlow) { latestVersionOfTheObject, events -> 
// apply new events / transformations
// emit new version of the object 
 }

Thank you!

Is it safe to combine a flow with something derived from itself? Stupid example:

val data: Flow<String?> = ....
val hasData: Flow<Boolean> = data.map { it != null }
val combined = data.combine(hasData) { data, hasData -> 
    require(data == null || hasData)
}

Note that hasData does not suspend, and let's assume flow will be collected on a single thread dispatcher.

Use case: I want a flow of events from a callback based API. But whenever there is a new value incoming, I need to call a suspending function (may make a network request). However this callback is not suspending, so I am not in a suspending context. More details in thread 🧵

Is there anyway to set

NSThread.stackSize

when using coroutines? Is NSThread even used? I'm trying to workaround a stack overflow in kotlinx.serialization. But how can I increase Worker stack size? Is it possible? https://youtrack.jetbrains.com/issue/KT-50870

This section in the coroutines documentation mention that all suspending functions inside

kotlinx.coroutines

is cancellable. At first I thought at any suspension point, if the current context is cancelled it would not enter, but this wording makes it sound like it’s not like that. Does this mean that potentially, if one made a series of suspending functions that are all not manually cooperating with cancellation (say all of them call blocking code, and/or other suspending functions that only call blocking code) cancellation would not happen and the suspending functions would keep alive?

Hello guys! 🙂 I’m in the process of migrating to the new test api with

runTest

and it’s been an extremely frustrating process so far 😞 I have a very simple view model that doesn’t do a whole lot, something like :

fun registerStuff() {
    viewModelScope.launch {
        registerStuffUseCase.execute()
    }
}

And a very simple test (with mockk)

@Before
    fun setup() {
        Dispatchers.setMain(StandardTestDispatcher())
    }

    @Test
    fun `SHOULD fetch stuff WHEN called`() =
        runTest {
            // arrange
            val registerStuffUseCase = mockk<RegisterStuffUseCase>(relaxed = true)
            val viewModel = ViewModel(registerStuffUseCase)

            // act
            viewModel.registerStuff()

            // assert
            coVerify(exactly = 1) { registerStuffUseCase.execute() }
        }

And it just doesn’t work. 100% of the time, verify is called before the use case is executed and the test fails. * unless * I add a

advanceUntillIdle()

after calling the VM... Which I can do but that’s like a hundred tests to update and I’m not even sure I’m doing things properly... Anything obvious I’m missing?

I was wondering. Is this:

flow.flatMapLatest {
  flow {
    emit("a")
    emit("b")
    emitAll(otherFlow())
  }
}

the same as this?

flow.transformLatest {
  emit("a")
  emit("b")
  emitAll(otherFlow())
}

They seem to do the same on paper (minus the extra flow object), however, is there anything that using

flatMapLatest

will do that

transformLatest

doesn’t? 🙂

Hi all, how to catch exception thrown by

callbackFlow

?

fun connect():Flow<Item> = callbackFlow {
    check(1 == 0) // This is expected to throw an IllegalStateException
}

I have the following Presenter class and test setup for it:

class Presenter(
    val api: Api,
    coroutineContext: CoroutineContext
) {

    val coroutineScope = CoroutineScope(coroutineContext)

    fun onAttachView(view View) {
        coroutineScope.launch {
            api.getDetail()
        }
    }
}

Test:

import io.mockk.*

class PresenterTest {

    val testCoroutineScope = TestCoroutineScope()

    val presenter = Presenter(
        api = mockk(),
        coroutineContext = testCoroutineScope.coroutineContext
    )
    val view = mockk()

    @AfterEach
    fun tearDown() {
        testCoroutineScope.cleanupTestCoroutines()
    }

    @Test
    fun `when view attached then call API`() {
        presenter.test().attachView(view)

        coVerify { api.getData() }
    }
}

How do I migrate it to coroutine-test 1.6.0?

Hi, I can't find how to remove a subscriber to a StateFlow since collect never terminates. I just want to stop listening once a certain state is emitted. Thanks

scope.launch { stateFlow.collect { if (it is State) doWork(); /* how do i unsubscribe here? */ } }

Hello! Does Kotlin Flow has collectLatest function but without cancelling current block?

Is there anything particularly special with vmScope.coroutineContext? When I use CoroutineScope(job + vmScope.coroutineContext) to launch coroutine and use job.cancel() it doesn't cancel the coroutine. This is an example https://pl.kotl.in/e0WxVRCBt which is able to cancel with another context in place of vmScope.coroutineContext

How can you emit (infinite) values before the real flow emits? I want to combine two flows: flow b into flow a. Flow a should not wait until the first emission of flow b, but should emit

null

until b emits values. After b emits some values, the last value of b should be used (combine).

This message was deleted.