Anyone of you can give me other approach of this:

class ContentPresenter(
        private val useCase: ContentUseCase,
        private val contextPool: CoroutineContextProvider = CoroutineContextProvider()
) {
....

open class CoroutineContextProvider {
    open val Main: CoroutineContext by lazy { UI }
    open val IO: CoroutineContext by lazy { CommonPool }
}

Then inside of each presenter I'm doing

launch(contextPool.Main) {
            val content = withContext(<http://contextPool.IO|contextPool.IO>) {
                repository.requestContent()
            }
            view.displayContent(content)
        }

On every presenter I'm passing the

CoroutineContextProvider

in order to test it with

unit testing

then for testing I have the

TestCouroutineContextProvider

which I use

Unconfined

, the thing is, do you guys are you using something like

AbstractPresenter

where you have there the

Job

and then you only from the presenter call it? I'd like to abstract it a little bit, and then could use my presenter without passing the

CoroutineContextProvider

via parameter, and then use it also in

unit testing

. Could you help me out with this?

Folks, quick help. Is this a right way?

slackIncomingHook.sendText(SlackMessage("Attempting RTS for son's of length: ${stockOutwardNoteIds.size}, warehouseId: ${warehouse.id()}"))
stockOutwardNoteIds.map {
         async {
                           warehousing.packStockOutwardNote(warehouse.id(), it)
                           freshRTSHelper.markRTSForFreshOrder(warehouseId = warehouse.id(), stockOutwardNoteId = it)
                    }
         }.awaitAll()

What is the best practice to use Coroutines (my use case is Fetch API +

await()

on the returned

Promise

) in JavaScript callbacks like

onclient =  { }

? cc @Vsevolod Tolstopyatov [JB]

Should I use Coroutines or Flow for Twitter4j?

This

open class CoroutineContextProvider {
    open val Main: CoroutineContext by lazy { UI }
    open val IO: CoroutineContext by lazy { CommonPool }
}

Is the same as :

open class CoroutineContextProvider {
    open val main: CoroutineContext by lazy { Dispatchers.Main }
    open val IO: CoroutineContext by lazy { <http://Dispatchers.IO|Dispatchers.IO> }
}

I was using experimental coroutines and now I have to update them

I'm trying to dig into the internal implementation of JVM coroutines. How a coroutine block (instantiated as

SuspendLambda

resumes its execution after last suspension point when

invokeSuspend

is called again? Is the magic field

L$0

inside

SuspendLambda

responsible for this? I'm working with a dead-simple snippet to understand the inner workings:

fun main() {
  val i = runBlocking {
    delay(1000)
    1
  }
  println(i)
}

is it possible to put something in a

coroutineContext

with a

finalize

that can detect an abandoned coroutine? Maybe print a warning?

are Coroutines suitable for a resource-level-lock of some kind? Trying to implement a resource locking system where any of those resources can be modified concurrently, but if that resource is already locked then the Coroutine should suspend until the lock is available. This is what I have so far, but I feel like this is a really naive way of doing things. Any suggestions on how to improve this? I'm managing the Set from a single thread since I'm fairly sure that a MutableSet (LinkedHashSet) is not thread safe

Hello, we have a gps location service that we create using dagger and we are getting an exception that

_location

channel was closed and we are not sure how can this happen since

_location

is a private member and there is no call to

_location.close()

. What are we missing? Dagger

@Provides
    @Singleton
    fun gpsLocationService(): GpsLocationService = GpsLocationService(
        context = context,
        coroutineContext = Dispatchers.Default
    )

Service

class GpsLocationService(
    private val context: Context,
    override val coroutineContext: CoroutineContext
) : CoroutineScope {

    private val logger = Logger.get(this::class)

    private val _location = BroadcastChannel<Location>(1)

    private val interval: Millis = GpsUpdateInterval.High.interval

    private val fusedLocationClient = LocationServices.getFusedLocationProviderClient(context)

    private val locationCallback: LocationCallback = object : LocationCallback() {
        override fun onLocationResult(locationResult: LocationResult) {
            if (GpsHelper.isGpsEnabled(context)) {
                logger.trace("Location acquired (${locationResult.lastLocation}).")
                _location.sendBlocking(locationResult.lastLocation)
            } else {
                logger.trace("GPS is disabled, ignoring location change.")
            }
        }
    }

    suspend fun requestLocation(): Location? = _location.openSubscription().consume {
        val location = withTimeoutOrNull(15.seconds) { receiveOrNull() }

        location
    }
}

in kotlin-js how do i launch a coroutine from normal code without using GlobalScope ? using that exceptions get lost apparently ? is there a better thing ?

What's the recommended way to refer to a conflated channel:

private val channel = ConflatedBroadcastChannel<T>

or

private val channel = BroadcastChannel<T>(Channel.CONFLATED)

?

I want to write a function that takes a suspend lambda and returns a custom future type, like

CoroutineScope.future

or

rxSingle

functions from the integrations libraries. Those functions are all implemented with

AbstractCoroutine

, which is marked as internal. Is there a recommended way to implement that type of functionality without internal classes, or is there no way around internal APIs a the moment?

Huh, what's happening here?

@Test
  fun experiment() = runBlockingTest {
    PublishProcessor.create<Unit>()
      .replay(1)
      .refCount()
      .asFlow()
      .first()
  }

This fails with

java.lang.IllegalStateException: This job has not completed yet

oO Can someone explan this?

Is there an issue I can track for improvements to the dispatchers scheduling behavior? Still looks like it very aggressively wakes threads and burns a ton of CPU when just a couple of threads could handle the workload (don't see anything here for a while now https://github.com/Kotlin/kotlinx.coroutines/issues/840)

@Yauhen Pelkin You probably just need to use

combine

.

Hi. I have to clear db when exit fragment. I am creating a coroutine to clear db table inside viewmodel onCleared. As you can see i am not using viewModelScope because my table will not be cleared and coroutine will be cancelled. Do you think its fine to use CoroutineScope here?

Hello, how i can do something like that

suspend fun test() {
async {  someWork()  }
}

How call async inside suspend fun?

To keep things simple until we have a need for something more complex, we run controller code as blocking (using ktor). With that in mind, if I'm calling into a suspend function in a 3rd party library that makes network calls, is this a reasonable construct to use?

runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) {
    try {
        thirdParty.doNetworkStuff()
    } catch (e: Exception) {
        log.warn("recoverable error", e)
    }
}

Why in this code the exception is printed only once: https://pl.kotl.in/TZrPEdnso If I change

launch

to

GlobalScope.launch

the exception is printed twice: https://pl.kotl.in/13sfoRVtk Shouldn't the unhandled exception cancel the job in both cases?

Is the example posted at https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-dispatchers/-unconfined.html correct? How can it print

1 3 2

given that

withContext

returns the

T

from the block, which would be the

Unit

from executing

println(2)

first?

Hi there. I have a

Flow

emitting

1

,

2

,

3

. Is there a way to transform this flow to emit:

[1]

,

[1,2]

,

[1,2,3]

. I know there’s this

fold

operator, but it’s terminal. In my case I need a value to be emitted on each accumulation step. Is there any appropriate operator out-of-the-box?

Hi Folks, I am trying to implement the Structured concurrency using Coroutines.... I fire 2 network calls parallel they work fine....but I have a use Case that I need to cancel one of the them as soon as I get failure response for one of them . Any suggestions appreciated Thanks

Flow/Coroutines question: I'm trying to migrate from RxJava to Flow for Android apps. I often write apps where in some starting point of the app (onCreate or the ViewModel init function for example) I subscribe to a bunch of observables. So, a typical

init

of a view model of mine may look like this:

init {
    myObservable
        .filter { whatever() }
        .map { whatever() }
        .subscribe { pushToMySubject() }

    myOtherObservable
        .filter { whatever() }
        .map { whatever() }
        .subscribe { pushToMyOtherSubject() }

    myThirdObservable
        .filter { whatever() }
        .map { whatever() }
        .subscribe { pushToMyThirdSubject() }
}

But when using flow, we can only call

collect

(which if I'm not mistaken is analagous to

subscribe

in RxJava) from within a coroutine context, which makes sense. But, as far as I can tell, calling

collect

blocks that coroutine context (my terminology might be garbage here) so you can't really have the above setup because we wouldn't execute the last two subscribes until the first observable completes. So two questions: 1. Is my understanding correct? 2. What's the preferred path to handle this sort of scenario?

If i start a coroutine with CoroutineScope inside ViewModel (which is doing some db operation but this operation doesn’t return anything. So coroutine starts performs db operation and completes). During this operation if my viewmodel onCleared() is called but i don’t cancel coroutine will my viewmodel leak?

I want to select between 2

Deferred<V>

but only fail if both fail. Would this work/is there a better way to do it?

What’s the difference between these two syntaxes?

val scope = CoroutineScope(errorHandler)
scope.launch {
    throw SomeSillyException()
}

val scope = CoroutineScope()
scope.launch(errorHandler) {
    throw SomeSillyException()
}

anyone have suggestions for getting started with coroutines for an rxjava user?

I assume you would just call

collect { }

in the unit test and then verify that the correct item comes through, however this throws

IllegalStateException: This job has not completed yet

So I was thinking about parallel mapping. So basically having an extension function like this:

fun <T, R> Iterable<T>.pmap(
    exec: ExecutorService,
    transform: (T) -> R
): List<R>

which uses an instance of ExecutorService to do all the transformations in parallel. An implementation of this is given e.g. here: https://stackoverflow.com/questions/34697828/parallel-operations-on-kotlin-collections/35638609#35638609 The problem is that I want to continue to work on the Rs as soon as they’re done, so List<R> is the wrong collection class (because a list will not be returned until all transformations are done). First I was thinking Sequence<R> would be the anwser for this. But now I think it isn’t. Firstly I could even make it work because I can’t simply modify to code to return a Sequence<R):

= sequence{
    ....
    for (item in this) {
        exec.submit { yield transform(item) }
    }
    ....
}

is actually a compile-time error because I can’t call yield from a “sub-lambda”). But also a Sequence is supposed to be lazy, so you don’t start computing the next value until it’s requested. So what is the correct collection-like class?? Is this what coroutines’

Flow

is for? I guess I could also use a Java

Stream

, but that just seams so unidiomatic. UPDATE: So I tried the flow-builder but it runs into the same problem as with Sequence:

fun <T, R> pFlowMap(
    items: Iterable<T>,
    exec: ExecutorService,
    transform: (T) -> R): Flow<R> = flow {

    for (item in items) {
        exec.submit { emit(transform(item)) }
    }

    exec.shutdown()
    exec.awaitTermination(1, TimeUnit.DAYS)
}

The

emit

is flagged with “Suspension functions can be called only within coroutine body”. So I tried wrapping that into a `runBlocking`:

runBlocking {
        for (item in items) {
            exec.submit { emit(transform(item)) }
        }
    }

but that dosn’t change a thing.

Is there a good interop model where you can subscribe to Rx2 Observable & then unsubscribe when the scope is cancelled?