Hello everyone, when calling ReceiveChannel<T>::receiveCatching i get an object of type ChannelResult<ChannelResult<T>> back. But when i call ReceiveChannel<T>::receive i get an object of type T. Is this a bug or did i miss something in the API? Thanks in advance!

Is there any performance or concurrency difference in using a MutableSharedFlow (with replay 1, buffer 1 and drop oldest) over a MutableStateFlow?

MutableSharedFlow<Something?>(
    replay = 1,
    extraBufferCapacity = 1,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)

vs

MutableStateFlow<Something?>(null)

Does the GlobalScope.launch function not work with the linuxX64 target? Confusing error messages keep on appearing, eg:

kotlin.IllegalStateException: There is no event loop. Use runBlocking { ... } to start one.
    at kfun:kotlinx.coroutines.takeEventLoop#internal (0x312759)
    at kfun:kotlinx.coroutines.DefaultExecutor#dispatch(kotlin.coroutines.CoroutineContext;kotlinx.coroutines.Runnable){} (0x312570)
    at kfun:kotlinx.coroutines.internal#resumeCancellableWith@kotlin.coroutines.Continuation<0:0>(kotlin.Result<0:0>;kotlin.Function1<kotlin.Throwable,kotlin.Unit>?){0§<kotlin.Any?>} (0x30d572)
    at kfun:kotlinx.coroutines.intrinsics#startCoroutineCancellable$default@kotlin.coroutines.SuspendFunction1<0:0,0:1>(0:0;kotlin.coroutines.Continuation<0:1>;kotlin.Function1<kotlin.Throwable,kotlin.Unit>?;kotlin.Int){0§<kotlin.Any?>;1§<kotlin.Any?>} (0x3118a4)
    at kfun:kotlinx.coroutines.AbstractCoroutine#start(kotlinx.coroutines.CoroutineStart;0:0;kotlin.coroutines.SuspendFunction1<0:0,1:0>){0§<kotlin.Any?>} (0x2f8f69)
    at kfun:kotlinx.coroutines#launch$default@kotlinx.coroutines.CoroutineScope(kotlin.coroutines.CoroutineContext?;kotlinx.coroutines.CoroutineStart?;kotlin.coroutines.SuspendFunction1<kotlinx.coroutines.CoroutineScope,kotlin.Unit>;kotlin.Int){}kotlinx.coroutines.Job (0x2f9d37)
    at _636f6d2e636861706d616e3a626f6c742d646174612d636f6e73756d6572_kncfun86 (0x333419)
    at MQTTClient_run (0x34a3c3)
    at  (0x7f7e4de366db)
    at clone (0x7f7e4d947a3f)
    at  ((nil))

I want to run IO bound tasks (network orientated) that run on a separate thread. The main thread with the linuxX64 target is mainly used for tasks that need to block (eg printing messages to the console). Linux unlike JS (the JavaScript platform) for example allows for multiple threads to run at once.

I have this code

class Executor {
    //    private val scope = CoroutineScope(Job() + Executors.newFixedThreadPool(10).asCoroutineDispatcher())
    private val context = Executors.newFixedThreadPool(10).asCoroutineDispatcher()

    suspend fun submit(value: Any) {
        withContext(context) {
            logger.debug { "Successfully received: $value" }
        }
    }

    fun stop() {
        // TODO
    }
}

How can I cancel the submitted jobs? And fail submission of a new job if the context is "cancelled"

short of writing my own extension function, is there any more concise way of writing

flow {
    emit(suspendFunction())
}

I know I could use

flowOf(…)

but the call site is not in a suspend function already

Hi, is there good `suspend`ing equivalent for 

ReadWriteLock

 in a library anywhere?

Is it possible to obtain a

CoroutineContext

from

Thread.currentThread()

? I don't control the current thread instance, but I want to specifically switch to it using e.g.

withContext(/*..*/) {}

.

override suspend fun getProducts()=flow<State<List<ProductListFromEntity>>> {
            try {
                val it = productDao.getProducts()
                if (it.isEmpty()) {
                    loadProducts()
                } else {
                   emit(State.success(it))
                }
            }catch(e:Exception){
                emit(State.error(e.localizedMessage.toString()))
                e.printStackTrace()
            }
}



//Dao function
 @Query("SELECT product.productId, product.productCode, product.name FROM Product product")
    abstract suspend fun getProducts(): List<ProductListFromEntity>
    

//viewModel
suspend fun getProducts() {
        productRepository.getProducts().collectLatest {
            when (it) {
                is State.Error -> {
                    _productListStates.value =
                        ProductListStates.Error(it.message, UIComponentType.Dialog())
                }
                is State.Success -> {
                    parseGivenProductList(it.data)
                }
            }
        }
    }

i am calling this function from viewmodel (android), But it is always throwing error saying,

Child of the scoped flow was cancelled

without any stacktrace. Am i doing anything wrong here...

Hello! if i have a coroutine task and i call a join on it

task.join()

and this join never returns than i do have a memory leak, don’t i?

This message was deleted.

Hi all - is there a

Flow<T>.flowOn(dispatcher)

equivalent that that affects downstream? so that i can do

results
  .map { result -> format(result) }
  .flowOn(<http://Dispatchers.IO|Dispatchers.IO>)
  .collectOn(Dispatchers.Main)
  .collect { presenter.display(it) }

instead of

withContext(Dispatchers.Main) { 
  results
    .map { result -> format(result) }
    .flowOn(<http://Dispatchers.IO|Dispatchers.IO>)
    .collect {
      presenter.display(it)
    }
}

Hi folks. I'm trying to upgrade from coroutines v1.3.9 to v1.5 in a Kotlin-based JVM library, but I'm struggling because of a change introduced in coroutines v1.4.0 (which doesn't seem mentioned on the release notes). I have a method where I'm using

BroadcastChannel

, but when I upgrade to coroutines 1.4.0 (or 1.5.0), the consumer's lambda function doesn't get called (even though my debugging shows that

ackChannel.sendBlocking(...)

is indeed called). Here's how it's consumed:

ackChannel
                .asFlow()
                .onEach {
                    <http://logger.info|logger.info>("onEach called with $it")
                    ackObserver.onNext(it.toCargoDeliveryAck())
                }
                .onCompletion { ackObserver.onCompleted() }
                .collect()

(Full method definition on GitHub) I've also tried replacing the

BroadcastChannel

with

MutableSharedFlow<String>(0, extraBufferCapacity = 1)

(and replacing calls to

ackChannel.sendBlocking()

with

mutableFlow.tryEmit()

), but the exact same issue persists. Any idea what I'm doing wrong? I created a PR where this issue is easily reproducible because it breaks a unit test.

I want to create a Flow that is backed by something like a channel so I can emit values to it from the outside, but I don't figure out what's right here. Maybe "channelFlow()" is the correct thing, but I don't find a tutorial for that. 😕 I want to send URLs to a flow while a user scrolls through a photo gallery to process thumbnails. I don't want to span multiple coroutines that read from disk in parallel as this is a performance problem. Reading from disk should be in sequence, but after that creating thumbnails in memory can be done in parallel. What's a good way to achieve that?

Hi,

fun OkhttpAuthenticator.authenticate() {
	...
	synchronized(this) {
		runBlocking { retrofitService.suspendingRefreshTokens() }<--- 
		...
	}
}

Could this get broken somehow? runBlocking seems to swap to caller thread context after the network call, so the mutex should be released or is it not guaranteed? (Because I see certain bugs in production where the code just stops; and not sure if its runBlocning or retrofit related)

I'm playing with flow and trying to integrate it with RabbitMQ. But I'm running into an issue where I can't emit values to a flow. I've created a simple example of my problem as shown on the image below. How would I solve this issue? If I try to run it in some other coroutine, Flow will just throw an error with the message

Emission from another coroutine is detected.

Does anyone have any articles to share or recommendations on how to best live with both RXJava & coroutines in the same Android project? We’re going to be migrating over time and (understandably) a lot of the documentation assumes you are using all coroutines

Why does

.map

on a

StateFlow

returns a

Flow

?

I want to be able to easily bind StateFlow, in particular as part of Android ViewModels. I'm going to use this code:

interface BindableMutableStateFlow<T> : MutableStateFlow<T> {
    val coroutineScope: CoroutineScope

    fun bind(other: StateFlow<T>)
    fun unbind()
}

and

fun <T> ViewModel.BindableMutableStateFlow(initialValue: T) : BindableMutableStateFlow<T> =
    MutableStateFlow<T>(initialValue).let { _stateFlow ->
        object : BindableMutableStateFlow<T>, MutableStateFlow<T> by _stateFlow {
            override val coroutineScope: CoroutineScope get() = viewModelScope

            private var job: Job? = null
            override fun bind(other: StateFlow<T>) {
                unbind()
                job = viewModelScope.launch {
                    other.collect {
                        value = it
                    }
                }
            }

            override fun unbind() {
                job?.cancel()
                job = null
            }
        }
    }

Am I duplicating or misusing (under-using) any existing Kotlin API? Do you see any problems with this code?

I have a

MutableStateFlow<Map<..., ...>>

what’s the most idiomatic way to mutate the map and update the

StateFlow

? seems like I need to make a copy of the current value and then update it? something like this seems verbose and awkward

stateFlow.value = stateFlow.value.toMutableMap().apply { … modify …}

📣 📣 📣

kotlinx.coroutines

1.5.1 is here! • Bi-directional cancellation in Play Services integration • Behavioural improvements in

CopyableThrowable

• Fixed annoying but rare crash with

java.lang.ClassCastException: kotlin.coroutines.jvm.internal.CompletedContinuation cannot be cast to kotlinx.coroutines.internal.DispatchedContinuation

• New API reference site: https://kotlin.github.io/kotlinx.coroutines/ • Various bug fixes and improvements Full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.5.1

Is there a convenient way to make a

channelFlow

hot? I would like to open a stream but defer the collecting. What I'm currently doing:

// cold
fun readByteArrayStream(): Flow<ByteArray> = channelFlow { .. }

// hot
val bytesChannel: Channel<ByteArray> by lazy { Channel(Channel.UNLIMITED) }

suspend fun openByteArrayStream() {
        readByteArrayStream().onEach { bytesChannel.trySend(it) }.collect()
    }

Why doesn't this deadlock?

val stream = MutableSharedFlow<Unit>()
val unused = stream.buffer(Channel.RENDEZVOUS).produceIn(GlobalScope)
stream.emit(Unit)

Maybe already asked - I run test with `

runBlockingTest

And it swallows exception without failing test

when calling a

suspend fun

exposed from a library like

Retrofit

or

Room

or

SqlDelight

- is there a need to change the dispatcher to

IO

or since they are code generating a

suspend fun

is it safe to assume that its ‘main-safe’?

How can i trigger a non-blocking refresh inside a flow?

getDataFlow()
  .onStart {
     coroutineScope {
         launch { repository.refresh() }
     } // this wait refresh to be finished
  }

the above code in onStart wait the refresh to finish according to

coroutineScope

documentation, I need the

repository.refresh()

to run in parallel and just be canceled with the flow

CoroutineScope(Dispatchers.IO).launch{ print("This is before delay") delay(1000) print("This is after delay") } why does it not print after delay text

I've observed that when an async-built-coroutine is nested within a launch-built-coroutine, the async-coroutine's job isn't added to the launch-coroutine's job's children. Is this always true?

I wrote an article about dispatchers - how do they work and when do we use each of them 🙂 https://kt.academy/article/cc-dispatchers

sorry about a possibly stupid question but is this:

val x = (async { foo() }).await()

ever different from just

val x = foo()

?