Why does refCount behavior require to specify scope? Scope is implicit via last subscriber unsubscription

How to design a CoroutineScope that active when user login and cancel when user logout?

So i have a coroutine that consumes a

ReceiveChannel

.

MainScope

is a basic scope tied to

Dispatchers.Main

-- For some reason, despite calling

cancel()

from the calling viewmodel's

onCleared

function, This job continues to run after back pressing and reentering the app. If i change the scope to

viewmodelScope

the coroutine is cancelled appropriately. Can anyone shed some light on this?

fun run(params: Params, onEachResult: (any: Any) -> Unit) {   

      mainScope.job = mainScope.launch {
            run(params).consumeEach(onEachResult)
      }

   }

   fun cancel() {
        mainScope.job.cancel()
   }

    internal class MainScope(private val mainContext: CoroutineContext) : CoroutineScope {

        var job: Job = Job()
        private val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
            showError(throwable)
        }

        override val coroutineContext: CoroutineContext = mainContext + job + exceptionHandler

        fun showError(t: Throwable) {
            Timber.e(t)
            throw t
        }
    }

Is this a good way to create a dispatcher which runs coroutines on a high priority thread?

import java.util.concurrent.Executors

import kotlinx.coroutines.asCoroutineDispatcher
import kotlin.concurrent.thread

fun highPriorityContext() = Executors.newSingleThreadExecutor { runnable ->
    thread(priority = Thread.MAX_PRIORITY) { runnable.run() }
}.asCoroutineDispatcher()

Can I suspend get the value by

SharedFlow

(replay is 1) like

Channel::receive

? for example:

suspend fun generateSomething() : String {
    val result = _channel.receive() // Change Channel to SharedFlow
    return "Result is $result"
}

Hello, got a question on coroutines Flow. Is there a way of replicating RxJava 

timeout

 operator on Kotlin Flow? Namely, for my use case, I want to emit values from a Flow downstream, until a timeout happened (timeout duration should reset after every new emission), then stop emitting (no error) ? The closest impl I see to what I need is the 

debounce

 operator, with the diff that I need to close / complete the flow when timeout happens.

I found this article from 2019 talking about the best practice of injecting dispatchers so that tests work well. Is this still the state of things? Just confirming because it sounds a little weird to me. Is there no way to change all dispatchers provided by

Dispatchers

to a test dispatcher when testing? https://craigrussell.io/2019/11/unit-testing-coroutine-suspend-functions-using-testcoroutinedispatcher/

Hey, why this test fails? I want to implement a “single instance job” which means there is should be only one active job with specific name in a scope

@Test
    fun test() {
        val job = GlobalScope.launch(CoroutineName("my_name")) {
            delay(1000)
        }

        assertEquals("my_name", job[CoroutineName]?.name)
    }

In my Project I am using Ktor to download *many Files*_(each almost 4mb_) I have a Coroutine Scope(IO) at App Level (_i.e., act_ive throught application Lifecycle.) Now as of now , I am Just *Blindly Launching coroutines(like 150 of them)* in this scope which download Files. Instead, I want to alter it to act as described Below: 1. Making this scope to only run a certain number of courtines parallelly say 4. 2. launching in this scope remains same, launch 150 or more coroutines all together and let the scope handle parallel downloads 3. A function , returning number of active/pending coroutines in this scope. Any help in pointing me in the right direction to correctly and efficiently implement this is appreciated!

Hello everyone.... Am I going crazy? In my Android ViewModel I have something like this:

viewModelScope.launch(<http://Dispatchers.IO|Dispatchers.IO>) { someFunction() }
....
fun someFunction(){}

It's compiling!!! But whyyyy? Shouldn't someFunction() be suspendable??

JVM: Hi, has anyone ever implemented a suspend function like

suspend fun <http://java.io|java.io>.File.readTextAsync(charset: Charset)

or

suspend fun java.nio.file.Path.readTextAsync(charset: Charset)

that is cancellable?

Good morning/day/evening. I need a solution for the following case: There is a cache, and I need to get objects from different threads (rest controller). If there is no object in cache I need to fetch this object via network. So the problem - only single thread should execute the network request, but others should wait until object is fetched and cached. This is the solution I have in mind. Is it okay?

What are the ways to fix this problem ?

Hi, I'm learning coroutines (coming from Rx world) and I'm trying to find a good way to refresh my flows on ViewModel... So far what I've been thinking is creating a shared flow to represent refresh events, something like:

private val refreshFlow =
    MutableSharedFlow<Unit>(replay = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST).apply {
      tryEmit(Unit)
    }

init {
  fetchNews()
}

fun refreshData() {
  refreshFlow.tryEmit(Unit)
}

private fun fetchNews() {
  viewModelScope.launch {
    refreshFlow
        .flatMapLatest { fetchLatestNewsFlow() }
        ...
        .collect()
  }

Is this ok? Are there any better solutions?

Hi, I'm struggling with an specific implementation of my use case. I have a simple channel

private val packetChannel: Channel<ResponsePacket> by lazy { Channel(Channel.UNLIMITED) }

that receives

ResponsePacket

which is just a data class which holds a

ByteArray

. After making a request to the API that produces the packets, I offer it via the channel (packets are only offered on request and communication is sequential, so request waits until packet is offered). I would like to consume these packets via flow but some packets are different which require different actions on collection, see screenshot. I'm clueless how to achieve this. For case 1 I would have to request 3 packets and while packet A is received I would parse it but then would have to wait until packet B + C is received and parsed and then merge the results of these and emit the result or otherwise case 2. That's the part I can't get my head around. Some help is much appreciated 🙏

How do I run N network calls parallel-ly and wait for their response? Data {1, 2, 3, ... N} -> Response { Response1, Response2, Response3...ResponseN} I want to run the tasks parallel-ly and wait for their response separately without losing which response belongs to which data param.

Hello, is there a something like

withTimeout(time) { operation }

but without canceling the operation? What I want to achieve: If an operation takes longer than

N

millis I want to have a callback about this and I want the operation to continue working. Something like:

withTime(
    time = 500L, 
    block = { operation }, 
    onTimeOver = {  }
)

Hello everyone! Would :

private val _event = MutableSharedFlow<S>(extraBufferCapacity = 1)
    val event: SharedFlow<S> get() = _event.asSharedFlow()

be a correct alternative to

Channels

for single shots events? Or would events would still be dropped if there are no subscribers

Hi. Trying to dig into the docs & tutorials on https://kotlinlang.org/docs/coroutines-basic-jvm.html and https://play.kotlinlang.org/hands-on/Introduction%20to%20Coroutines%20and%20Channels/01_Introduction for the n-th time to try to get the grip on them 😉 The doc says that the following code will return an arbitrary number and I agree with it; we don’t wait for the coroutines to finish. However it does not for me - I’ve run it 20 times - each resulting in the sum being `500000500000`… (java 15, kotlin 1.4.31, coroutines 1.4.3, macOS) Any tips/insight as to why?

package samples

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.launch
import java.util.concurrent.atomic.AtomicLong

fun main() {
    val c = AtomicLong()
        for (i in 1..1_000_000L)
        GlobalScope.launch {
            c.addAndGet(i)
        }
        println(c.get())
}

PS! if this is better suited as question in #getting-started please tell me and I’ll move it there.

What's the elegant way to have something like

combine(flow1, flow2, ..., flowN, transform)

that does not wait for all flows to emit the first value? I'd happily receive a smaller array if some flow does not have a value yet.

Hi guys I'm trying to learn channels went through docs, I got lost after the fan-out section, and after watching talk I'm totally a gone case 😛😵, can anyone share some uses cases of channels? examples or real-life use cases?

in my understanding, the first code fragment should behavior the same as the latter one, coz launch w/o explicit receiver adds the new created coroutine as a child of current coroutine, so cancel the collectLatest block will cause the child coroutine to cancel, which should leads to the same result as the latter code fragment

does collecting two flows inside the same coroutineScope influence each other when the coroutine context is a

SupervisorJob

?

Should I use

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

or

Dispatchers.Default

for drawing with canvas on a bunch of bitmaps (GIF frames)?

I am importing:

val commonTest by getting {
    dependencies {
        implementation(kotlin("test-common"))
        implementation(kotlin("test-annotations-common"))
        implementation ("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.4.3")
    }
}

and I am getting this:

org.gradle.internal.resolve.ArtifactNotFoundException: Could not find kotlinx-coroutines-test-1.4.3-samplessources.jar (org.jetbrains.kotlinx:kotlinx-coroutines-test:1.4.3).

Why am I getting an unresolved reference here:

What's the difference between executing some operations in a

coroutineScope { }

vs. just executing them in the outer scope? Assuming the outer context has a regular

Job

.

Is there an equivalent for

while (!packetChannel.isEmpty) {
	packetChannel.receive()
}

in flows when I convert the channel to a flow:

packetChannel.receiveAsFlow()

Edit: I would like to consume the channel as a flow, but instead of collect all values then parse I would like to collect one value, parse it, wait for next one then emit final result

(reposting this question since i did it late at night and it didn’t get any responses): I’ve built a sharding library and i’m trying to add coroutine functionality to it. In the following snippet i’m hoping that it’s doing what I’m thinking it does (returns the first 

true

 result that it finds):

override fun emailExists(email: String): Boolean {
        val results = shards.asyncAll { userDao.emailExists(email) }
        return runBlocking {
            results.map { it.await() }
                .firstOrNull { it }
        } ?: false
    }

the shards.asyncAll method is:

fun <T> async(
        shardId: Long,
        context: CoroutineContext = EmptyCoroutineContext,
        start: CoroutineStart = CoroutineStart.DEFAULT,
        block: suspend CoroutineScope.() -> T): Deferred<T> {
        return scope.async(context, start) {
            selectShard(shardId)
            block()
        }
    }

    fun <T> asyncAll(
        shardIds: Collection<Long> = this.shardIds,
        context: CoroutineContext = EmptyCoroutineContext,
        start: CoroutineStart = CoroutineStart.DEFAULT,
        block: suspend CoroutineScope.() -> T): List<Deferred<T>> {
        return shardIds.map { async(it, context, start, block) }
    }

So first question is if that code above (

emailExists

) will work, and the second question is: how can i change it so that the first 

Deferred<Boolean>

 result that returns that is 

true

 returns from the method (rather than it testing them in order as they come back - I’d like it so that the very first one that comes back with a result of 

true

 triggers a return)