I wonder what is the use case of SupervisorJob, what is the case that you don’t care about your children job might throw an exception?

Hello everyone! I'm trying to use a

RendezvousChannel

in the place of

PublishSubject

from RxJava in a project, but I need a way to access the current element in the channel without removing it. Something like a 'peek' operation on a

LinkedBlockingQueue

. Now a

RendezvousChannel

has a method called

poll()

which does give me the current element (or null) but it also removes it from the channel. How can I perform a peek operation on a channel then?

Is there a

collect

variant for

Flow

that allows to access the previous value, or null for initial value?

@elizarov In the article you write that one should not use

withContext

inside a

flow{}

, basically. Can there be a realistic situation when such use is permitted? Maybe if

emit

is outside of

withContext

, or there are several `nested

withContext

that work in a way that protect the caller's context?

Hello everybody. I was wondering if there is a way to

emit

values into a

flow

outside its building block. For example sending values when the user clicks a button or if this use-case would be better covered with something else instead of a

Flow

.

@elizarov Anyway to have some feedback on https://github.com/Kotlin/kotlinx.coroutines/issues/1092 ? Just to be sure it's not forget due to the left label.

Is there any documentation somewhere on kotlinx.coroutines.io?

Hello guys, any example about how to handle complex concurrency with `actor`s?

Hey all ! A quality check question - I have produce some data via

flow

in interactors, such as

flow {
        emit(Result.State.Loading())
        emit(
            dataSourceFactory.getGrid(
                params.path,
                params.page,
                params.displaySize
            )
        )
        emit(Result.State.Loaded())
}

An interactor is producing it’s own “state”, but I also want to support multiple interactors with single state. I came with an idea to merge them async in this way:

fun CoroutineScope.merge(vararg results: Flow<SimpleResult<Any>>) = flow {
    emit(Result.State.Loading())

    val tasks = mutableListOf<Deferred<Unit>>()

    for (result in results) {
        tasks.add(
            async(<http://Dispatchers.IO|Dispatchers.IO>) {
                result
                    .filterNot { it is Result.State }
                    .collect {
                        emit(it)
                    }
            }
        )
    }

    tasks.awaitAll()
    emit(Result.State.Loaded())
}

I’m wondering if this is a robust solution. This

CoroutineScope

is handled properly for closing etc., but what do you think ? Any suggestions to merge flows together while holding ability to ignore some emits ?

How to properly initialize such kind of patterns? Currently depending on the state of the device and load, the repeat call can trigger creation of tons of threads even if there's no tasks yet. Generating a lot of unecessary load at init time.

class XXX  : CoroutineScope {

	override val coroutineContext = <http://Dispatchers.IO|Dispatchers.IO> + SupervisorJob()

	internal val tasks = Channel<Task>(UNLIMITED)

    fun startWorkers() {
        repeat(150) {
            launch {
                for (task in tasks) executeTask(task)
            }
        }
    }
}

import kotlinx.coroutines.*

val job: Job = Job()
val scope = CoroutineScope(Dispatchers.Default + job)

fun throwsAsync(): Deferred<Unit> = scope.async {
  throw Exception()
}

fun loadData() = scope.launch {
  throwsAsync().await()
}

suspend fun main() {
  loadData().join()
  scope.launch { println("worked") }.join()
  joinAll()
  println("done")
}

Why doesn't this crash?

i am trying to wrap my head around scopes, still. most examples online are simplistic uses of

runBlocking

... use case here is that i need to, once a day run a bunch of operations as coroutines. currently this is a java scheduled executor, but i want to switch to coroutines for the actual running of the code.

example use case: saving a file to disk. Does using suspended function + continuation make sense here?

is there any point to doing something like

lauch { val result = async { doStuff() }.await() }

, assuming

doStuff()

is a suspend fun? or would it make more sense to simply call

doStuff()

inside the launch block without

async

?

So far I have

runBlockingTest {
            whenever(model.loadData(false)).thenThrow(DeprecatedApiException())

I'm having an hard time trying to figure out how merging of scope and context works and would appreciate some details 🙂

class XXX  : CoroutineScope {

	override val coroutineContext = <http://Dispatchers.IO|Dispatchers.IO> + SupervisorJob()

	val aDispatcher =  ThreadPoolExecutor(...).asCoroutineDispatcher()

	internal val tasks = Channel<Task>(UNLIMITED)

    fun startWorkers() {
        repeat(150) {
            launch(aDispatcher) {
				val initialTask = try {
                    tasks.receive()
                } catch (e: Exception) {
                    return@launch
                }
                withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
					// Correctly run on <http://Dispatchers.IO|Dispatchers.IO>
				}
			}
        }
    }
}

class XXX  : CoroutineScope {

	override val coroutineContext = <http://Dispatchers.IO|Dispatchers.IO> + SupervisorJob()

	val aDispatcher =  ThreadPoolExecutor(...).asCoroutineDispatcher()

	internal val tasks = Channel<Task>(UNLIMITED)

    fun startWorkers() {
        repeat(150) {
            launch(aDispatcher) {
				val initialTask = try {
                    tasks.receive()
                } catch (e: Exception) {
                    return@launch
                }
                withContext(coroutineContext) {
					// Incorrectly runs on aDispatcher instead of <http://Dispatchers.IO|Dispatchers.IO> that is used for coroutineContext
				}
			}
        }
    }
}

I am doing Coroutines-Rx2 interop. According to https://github.com/Kotlin/kotlinx.coroutines/blob/master/reactive/kotlinx-coroutines-rx2/src/RxConvert.kt

@ExperimentalCoroutinesApi
public fun <T : Any> Deferred<T>.asSingle(context: CoroutineContext): Single<T> = GlobalScope.rxSingle(context) {
    this@asSingle.await()
}

calling

myDeferred.asSingle(coroutineContext)

will immediately call

await()

on the deferred, which effectively starts coroutine execution. Shouldn’t this only happen when Single will be subscribed on?

Do you think passing a

CompletableDeferred

down a

Flow

or a

Channel

(exposed as a

ReceiveChannel

) is a good idea? The use case is to allow to observe a state (a sealed class instance), and allow an action depending on that state.

Is there a particular reason why

AtomicInt

from kotlinx:atomicfu doesn't declare these:

operator fun plusAssign

(+=)

operator fun minusAssign

(-=)

operator fun inc

(++)

operator fun dec

(--) In Java, these couldn't be atomic, but since you can implement them yourself, they would be. For

inc

and

dec

I guess it might be because they would have to return AtomicInt, instead of Int. Asking here because I didn't see an atomicfu channel.

📣 📣 📣

kotlinx.coroutines

version 1.3.0-M1 is here! A lot of changes in Flow: * Core

Flow

interfaces and operators are graduated from preview status to experimental 😒uspend: * Serious performance improvements of Flow * More accurate context preservation invariant that should prevent most of the concurrency and contract violation bugs *

Flow

interface cannot be implemented directly, only via

AbstractFlow

to preserve Flow’s invariant * Separate

buffer

operator composable with buffered operator (

flowOn

,

flowChannel

etc.). No more

bufferSize

constant in operators! * New operators and a lot more! General changes: * Scalable state-of-the-art

Semaphore

implementation * Channels API improved: performance, experimental API and consistent exception handling *

withContext

checks cancellation status on exit to make reasoning about sequential concurrent code easier * JS dispatcher performance * Various API and performance improvements and bug fixes Full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.3.0-M1

It seems replacement for

flowViaChannel(Channel.CONFLATED) { … }

is

channelFlow<T> { … }.buffer(Channel.CONFLATED)

. Isn't that less efficient as values pass through 2 operators?

Is

kotlinx-coroutines-test

planned for multiplatform?

What is the equivalent of SingletonHolder when using coroutines: https://medium.com/@BladeCoder/kotlin-singletons-with-argument-194ef06edd9e

open class SingletonHolder<out T: Any, in A>(creator: (A) -> T) {
    private var creator: ((A) -> T)? = creator
    @Volatile private var instance: T? = null

    fun getInstance(arg: A): T {
        val i = instance
        if (i != null) {
            return i
        }

        return synchronized(this) {
            val i2 = instance
            if (i2 != null) {
                i2
            } else {
                val created = creator!!(arg)
                instance = created
                creator = null
                created
            }
        }
    }
}

Never use this anti-pattern is the solution

Could anyone explain to me why this code throws an exception (on Android)? The

collect

and

cancel

functions are both called on the main thread. I tried this with both version 1.2.1 and 1.3.0-M1

var cancelled = false
val job = GlobalScope.launch(Dispatchers.Main) {
    val flow = flow {
        while (true) {
            emit(Unit)
        }
    }
    flow.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)
        .collect {
            if (cancelled) { // main thread
                throw IllegalStateException()
            }
        }
}

GlobalScope.launch(Dispatchers.Main) {
    delay(1000)
    job.cancel() // main thread
    cancelled = true
}

Does something like a hot

Flow

make sense, or is

Flow

dedicated to only cold asynchronous streams while `Channel`s are the best choice for hot streams ?

I am trying to create an InvocableFlow, the idea is being able to send values without outside the declaration site of the flow while still exposing a flow Api, would this implementation actually work? Is there any flow in my approach? :

internal class InvocableFlow<T>(private val scope: CoroutineScope) {
    val flow = flow {
        for (value in channel.openSubscription()) {
            emit(value)
        }
    }

    private val channel = BroadcastChannel<T>(BUFFERED)

    fun invoke(data: T) {
        scope.launch {
            channel.send(data)
        }
    }
}

Is there a reason why there aren’t delay functions for channels, and only for flows? https://github.com/Kotlin/kotlinx.coroutines/blob/master/kotlinx-coroutines-core/common/src/flow/operators/Delay.kt Is it just that it isn’t as efficient to do so with hot streams?