Is there an easy way to await the first result of two Deferred and cancel the other?

How do you make a long running, blocking function cancellable? E.g. with this third-party class:

class Worker {
  fun doWork()
  fun cancelWork()
}

My attempt below (untested) doesn't look very good, I wonder if there's any better way.

coroutineScope {
  val job = launch {
    while (isActive) {
      delay(50)
    }
  }
  job.invokeOnCompletion { 
    if (it is CancellationException) {
      worker.cancelWork()
    }
  }
  worker.doWork()
  job.cancel()
}

How do you folks architect around the fact that shared flows (correctly) need scope as a parameter?. Should only classes that own their scope use sharing operators? In rx I used to do this

class AppInForeground {
	val appInForeground: Observable<Boolean> by lazy {
		Lifecycle.wrapAsObservable()
		   .distinctUntilChanged()
		   .publish()
		   .replay(1)
	}
}

naive Flow implementation

class AppInForeground {
	fun appInForeground(scope: CoroutineScope) {
		if (_appInForeground == null) {
			synchornized(this) {
				if(_appInForeground == null) {
					_appInForeground = Lifecycle.wrapAsFlow()
						.distinctUntilChanged()
						.shareIn(scope, SharingStarted.Lazily, 1)
				}

			}
		}
		return _appInForeground!!
	}
}

just looks stupid, however the main issue is that callers should not determine the scope, right? I think idiomatic way is this

class AppInForeground : HasLifecycle {
	private val scope = SupervisorScope(..)

	private val _appInForeground = MutableStateFlow(false)
	val appInForeground: Flow<Boolean> get() = _appInForeground

	override fun create() {
		scope.launch {
			Lifecycle.wrapAsFlow()
				.distinctUntilChanged()
				.collect {
					_appInForeground.value = it
				}
		}
	}

	override fun destroy() {
		scope.cancel()
	}
}

Works, however its a bit less efficient, i.e. it subscribers when the AppInForeground class si instantiated, also whole lot more ceremony... Is there a way more close to the rx way? or should I just deal with it?

Does hibernate work alongside coroutines?

I'm trying to create a cold

Flow

where a lot of updates can happen at once and I want the collector to only process the last one. The producer and the consumer are both on the UI thread. The desired behavior is when the producer gets out of their loop and goes back to waiting for some event, then the consumer gets the latest state. What I have now is a

callbackFlow

that is

conflated

.

val flow = callbackFlow<Any> {
    event.register { offer(someValue) }
    awaitClose { event.unregister() }
}.conflate()

val job = scope.launch() {
   flow.collect { }

The scope is using Android's

MainScope

. And the event is sending events on the main thread. When I was using a

MutableStateFlow

this behaves as I wanted, but with the conflated

callbackFlow

the caller is getting the first and last event, once all the events have been offerred.

Hello, I'm trying to solve the following problem: Multiple Coroutines (on different threads) may send a request to an API. The API may respond, that a login is required. In this case the first coroutine should initiate the login sequence. All other coroutines should suspend until the sequence is executed. I currently implemented this as follows:

val loginSequence = SharedMethod<String>()

suspend fun request() {
    try {
        loginSequence
            .awaitIfNeeded()

        return doRequest(...)
    } catch(e: LoginRequiredException) {
        return loginSequence.awaitOrBlock(
            onBlockReturn = {
                return@awaitOrBlock executeLoginSequence(e)
            },
            onAwaitReturn = {
                return@awaitOrBlock doRequest(
                    ...
                )
            })
    }
}

with the implementation of this class with mutexes:

class SharedMethod<T> {
    private val mutex: Mutex = Mutex()

    suspend fun awaitOrBlock(
        onBlockReturn: suspend () -> T,
        onAwaitReturn: suspend () -> T
    ): T = coroutineScope {
        val isLocked = mutex.tryLock()

        if (!isLocked) {
            awaitIfNeeded()

            return@coroutineScope onAwaitReturn()
        }

        try {
            return@coroutineScope onBlockReturn()
        } finally {
            mutex.unlock()
        }
    }

    suspend fun awaitIfNeeded() {
        while (mutex.isLocked) {
            delay(1000)
        }
    }
}

Is there a better way to implement this behaviour? Especially the "while (mutex.isLocked) {...}" seems pretty ugly. Is this the right place for using mutexes? Happy to hear your thoughts. :)

How do I get usable stack traces for exceptions in coroutines? I'm using a crash reporting service, but all the exceptions I'm getting lose all context, since they only have the function the exception happened in, and that's it, since the next element of the stack trace is

BaseContinuationImpl.resumeWith

and the rest of coroutines machinery, not the "actual" calling method.

Is there a way to get the job a SharedFlow is launched in? Basically, I want to do something like this

// a method ...
val (sharedFlow, job) =  flow.shareIn(scope, SharingStarted.Eagerly)
jobs.add(job)
return sharedFlow

Given a collection of

Deferred<X>

- what’s the best way to retrieve their results as they come in - and at any point bail after receiving a result that you’re looking for while canceling the others?

Hi, I need to insert a initial value to a stream that sometimes doesn't emit, however if it emits within a window of time, take the real value, not the manual initial one TLDR; debounce first value - I have this

private fun <T> Flow<T>.onStartDebounced(initialValue: T, debounceMillis: Long): Flow<T> {
    return onStart { emit(initialValue) }
        .withIndex()
        .debounce { if (it.index == 0) debounceMillis else 0L }
        .map { it.value }
}

Works, but im a Rx guy, is there something more idiomatic / performant (so I can actually learn something)?

What is the standard way to refactor a lazy property to a suspend function that uses a cached value on subsequent runs?

CompletableDeferred

or something else?

// from this
val someProperty: Int by lazy {
    // refactoring to perform some suspendable logic
}

// to this
suspend fun someProperty(): Int {
    // only should be executed max once (and then cached)
}

Is there a way to keep track of the children Jobs launched within a given

CoroutineScope

? As if the

CoroutineScope

were a message queue and Jobs were messages in it: Would there be a way to track if the

CoroutineScope

is currently handling Jobs or is sitting idle? What I’m trying to accomplish: I’d like to build an idle/busy signal (possibly as a

Flow<Boolean>

) that will tell me whether a

CoroutineScope

is currently running any coroutines or not.

MutableSharedFlow()

by default has no buffer (

replay = 0

,

extraBufferCapacity = 0

, i.e.

buffer = replay + extraBufferCapacity = 0

). Does that mean that if I have many emitters that call

emit

, that they would suspend (in order) until the subscribers of the shared flow process the emissions one by one? So, in other words: has the buffer moved away from the shared flow, that intrinsically has no buffer space, to the coroutine scope(s) that now contain various suspended child coroutines trying to

emit

?

Hi everyone, I wonder if someone could clarify something for me. When I first learned about coroutines, I remember being taught that one should prefer to use

launch

from your existing context and then move it elsewhere using

withContext

instead of calling

launch

and specifying dispatcher.

scope.launch { withContext(<http://Dispatchers.IO|Dispatchers.IO>) { work() } }

instead of

scope.launch(<http://Dispatchers.IO|Dispatchers.IO>) { work() }

. I remember this being described as preferential since there was some performance penalty when launching in a different context that did not occur when moving a running coroutine using

withContext

. I’m trying to find guidance on this again and I can’t find anything about it. Did I misunderstand something when I was first learning?

About this documentation: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-mutable-shared-flow/emit.html If you call

MutableSharedFlow<Int>().emit(0)

(from a suspending function), then the

emit

call does not suspend. The documentation says that on buffer overflow the call should suspend. To me it seems that there is no buffer, so

emit(0)

causes a buffer overflow, so the

emit

call should suspend. Instead, it returns quickly! Am I misinterpreting the following?

suspending on buffer overflow

Or can the documentation be improved?

How to test

MutableSharedFlow

with

replay=0

? I'm using turbine for testing Kotlin Flows and I managed to write tests for

MutableStateFlow

, but the same emitted values doesn't work for SharedFlows.

Hello there! I made some measures and coroutines loose to parallelStream when they deal with parallel processing of collection elements. It’s sad 😔 Kotlin v1.3.72

How can I create a timer? I've done it like this

scope.launch{
 var foo = true
 while(foo){
 delay(1_000)
     //do some actions here
     //with these actions I check if the time is correct and if goes inside the if i put the foo to false so it leaves the while
     if(whatever) {
       foo=false
       //call one method here
     }
 }
}

Is this a correct way? Is there a better way with flow?

I found this :

val timer = (0..MyInterval) 
 .asSequence()
 .asFlow()
 .onEach { delay(1_000) }

timer.collect { //do something }

But how do I stop the timer as I do with if(whatever) foo = false?

I'm looking for a way to update a UI from a flow at a given interval. I was using sample() which works but the first emit is delayed. Is there a way to emit the first item without delay using sample() or some other function?

I'm looking to re-emit the last search result when refresh() is called. What I have below works, but is there a nicer way of doing it? (More specifically, I found the apply block for the refreshChannel was necessary for the resultStream to emit at all, but it doesn't feel right).

private val searchQueryStream = MutableStateFlow<String>("")

private val refreshChannel = Channel<Unit>(Channel.CONFLATED).apply { offer(Unit) }

val resultStream: Flow<PagingData<TenorGifTileData>> =
    searchQueryStream
        .combine(refreshChannel.receiveAsFlow()) { queryStream, _ -> queryStream }
        .filter { searchQuery ->
            searchQuery.query.isNotEmpty()
        }
        .flatMapLatest { searchQuery ->
            repository.resultsForSearchQuery(searchQuery)
        }

fun refresh() {
    if (!refreshChannel.isClosedForSend) {
        refreshChannel.offer(Unit)
    }
}

you make suspending what you want to be cancelable marking blocking code thats not cancelable with suspend does nothing however if you mark it suspend, then by convention it should be main-safe, so yea apply dispatcher then

How can I make sure to start collecting to a

MutableSharedFlow

before emitting something ?

val mutableFlow = MutableSharedFlow<Int>()
      val sharedFlow = mutableFlow.asSharedFlow()

      // start listening to the sharedFlow
      launch {
        sharedFlow
            .collect {
              // by the time we start collecting, the first item is gone already
              // and nothing is received
              println("$it")
            }
      }
      // emit something 
      mutableFlow.emit(1)

Adding a delay before emitting works but doesn't sound great and I don't see a way to enforce a before-after relationship there

Hi, is there any plan to release v1.5.0(stable)? Kotlin v1.5 has been released, and the coroutines library is the only thing that blocks me from updating now.

[RESOLVED] What's the difference between casting to

StateFlow

vs

asStateFlow

? 🧵

is it possible to complete MutableStateFlow? I just started playing with coroutines / flow so I'm not really sure whether it's even meaningful question to ask 🙂

I am a bit new to coroutines/flow and wondering if this is a good design or not. I have a server that can receive long running job to process. Each job must be run sequentially. So I have something like this:

class JobsMgr {
    private val _jobsToProcess = Channel<Job>(Channel.UNLIMITED)
    private val _jobsProcessed: Flow<Job> = _jobsToProcess.consumeAsFlow().map { job -> /* processing */ }.flowOn(Dispatchers.Default)

  init {
       // I find this very ugly.. is there a better way?
        Executors.newSingleThreadExecutor().execute {
            runBlocking {
                _jobsProcessed.collect {
                    delay(10 * 1000)
                    _lock.withLock { _jobs.remove(it.jobId) }
                }
            }
        }
  }

fun enqueueJob(...) { 
//...
               runBlocking {
                _jobsToProcess.send(job)
            }
}

}

It is running in a spring boot application so I do not have control over the main loop. This is why I created a process to collect and remove the completed jobs. Is this how Channel/Flow is supposed to be used? And is there a better way to "start" the collection. That seems like ugly code to me but not sure how to do better.

Hi! I see this piece of code in a library. is the function

suspendCancellableCoroutine <Nothing> {}

needed ? For me is just an empty call

class ProduceStateScopeImpl<T>(
    state: MutableState<T>,
    override val coroutineContext: CoroutineContext
) : ProduceStateScope<T>, MutableState<T> by state {

    override suspend fun awaitDispose(onDispose: () -> Unit): Nothing {
        try {
            suspendCancellableCoroutine<Nothing> { }
        } finally {
            onDispose()
        }
    }
}

Hi All! 👋 I'm wondering if this is OK