I have a comment an the coroutine channel API. Currently the

Channel

constructor takes a

capacity

argument of type Int.

capacity

is also used to define the type of the channel by the use of magic constants such as RENDEVOUS=0, BUFFERED=-2, UNLIMITED=Int.MAX_VALUE.

val c1 = Channel<Int>(capacity = 15)
val c2 = Channel<Int>(capacity = RENDEVOUS) // dislike

Kotlin has the great feature of sealed classes. Wouldn't it be better to have an API like:

// Limited(capacity: Int) and object Rendevous belong to sealed class ChannelType
val c1 = Channel<Int>(type = Limited(15))
val c2 = Channel<Int>(type = Rendevous) // Rendevous is an object here

anybody else rarely able to get a breakpoint to work inside a suspending function?

Hi. I have problems limiting the thread pool in coroutines: I have my simple helper function:

fun <T, R> runConcurrently(nThreads: Int, params: Collection<T>, op: suspend (T) -> R): List<R> {
    return runBlocking {
        Executors.newFixedThreadPool(nThreads).asCoroutineDispatcher().use { pool ->
            params.map { p -> async(pool) { op(p) } }
        }.awaitAll()
    }
}

and it works well when I DON'T use any yielding call inside my

op()

, e.g:

runConcurrently(3, (1..8).toList()) { param ->
    counter.increase()
    val sleepTime = 1000L / param
    println("Started $param in thread ${Thread.currentThread().name}, counter is $counter, sleepTime is $sleepTime")

    Thread.sleep(sleepTime)
    println("Stopping $param in thread ${Thread.currentThread().name}")
    counter.decrease()
}

But as soon as I call any suspending function inside, e.g. change

Thread.sleep()

to

delay()

, it switches rest of execution to DefaultExecutor, and my pool is not limiting anymore. What am I doing wrong? And what is current best way to limit thread pool? I know newFixedThreadPool is deprecated, but there is no alternatives afaik.

is there are way to close a Channel (

channelFlow

or

callbackFlow

) without sending a error to the consumers?

I mean, when a I close my

channelFlow

even I don't passing some Exception, internally the flow will throw a

CloseSendChannelException

I'm using

java.util.stream.Stream.consumeAsFlow()

to process a stream as a flow. Sometimes, it appears that the

finally { stream.close() }

block is not getting called (possibly on certain cancellation / error conditions?) (confirmed by duplicating the StreamFlow class and adding a log statement), although if I add an

onCompletion {}

to the returned flow, that block does get called even when the finally gets skipped. Anything that can explain this behavior? Will probably move back to using the

onCompletion

to close the Stream in the meantime though.

In Spring WebFlux, when dealing with blocking IO, I assume the correct dispatcher to use is Dispatchers.IO. However, when using non-blocking IO, should we be using Default or Unconfined? I’m particularly thinking about async blocks like below.

@GetMapping("/things")
    suspend fun getThings() = coroutineScope {
        val thing = async {
            suspendingCall()
        }

        val anotherThing = async {
            suspendingCallAgain()
        }

        listOf(thing.await(), anotherThing.await())
    }

Hey! I have a variable that I need Synchronized access to from two functions: a

suspend

fun and a non-suspending function. I cannot just do

@Synchronized

on the blocking one and use

Mutex

on the suspend-one as these don't exclude each other, such that there is still the posibility of two concurrent accesses. It seems as tough

Mutex

is not usable from non-blocking code, and

@Synchronized

doesn't help with `suspend fun`s. what's the best approach here?

Whats the best way to integrate a Broadcast Channel and Flow together?So that you have a hot flow, that collects anything that the broadcast channel emits

How do I synchronize the dispatchAction function below so that it only starts sending items once there is a downstream collector? onStart unfortunately get called right before collect.

class StateMachine(val scope: CoroutineScope) {  

  private val inputActions: BroadcastChannel<Rankings.Action> = BroadcastChannel(Channel.BUFFERED)
  private var isInitialized = atomic(false)

  init {
    scope.launch {
      inputActions.asFlow()
        .onStart { isInitialized.value = true //onStart hook is called just before collect gets called }
        .collect { action ->
          //handle action
        }
      // CANNOT CALL FUNCTION HERE BECAUSE `collect` SUSPENDS
    }
  }

  fun dispatchAction(action: Rankings.Action) = scope.launch {
    while(!isInitialized.value) {
    }
    inputActions.send(action)
  }
}

I want to be able to attach a resettable timeout on a job where basically I give a timeout and a channel and whenever the channel is sent a value then the timeout is reset. Is there a common practice for something like this? I was thinking of writing an extension function on Job but im not sure how i wanna approach it.

Hi! I'm rather new to working with coroutines and I have a question regarding testing them. I have a service similar to this

fun handleDelete(extIds: List<String>) {
        val ids = someClient1.getInternalIds(extIds) // get deleteable IDs

        try {
            runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) { // use IO dispatcher as someClient2 makes a call to some remote service. 
                ids.forEach{ launch { someClient2.deleteId(it) } }
            }
        } catch (e: Exception) {
            handleException(e)
        }
    }

and I want to test it. My team uses Mockito and not Mockk, if that makes any difference. Maybe the test does not make sense, but as I am not very familiar with coroutines I want to test that these are run correctly in paralell and that they exit the runBlocking block after they're all done. So I was thinking of mocking the responses from

someClient2

so that say i have

extIds = listof("123", "456")

and the calls take respectively 500ms and 750ms, I want to check that in total this function does not use much more time than ~750ms and not (500+750)=1250ms. Is this possible using Mockito? And is it actually a useful test? 🤔

This message was deleted.

I hate flatmapping singles, sooo hurry up 😄

How to implement timer (looping call) in Kotlin coroutine like

rxJava

Single.timer(INTERVAL_SECONDS, TimeUnit.SECONDS)
      .flatMap{ // call API }
      .repeat()
      .subscribeOn(<http://Schedules.io|Schedules.io>())

I heard someone mentioned using

flow

. It look likes

while(true){
    // call 
    // delay(intervalMili)
}

So I think it’s not looking good .

Hello, World! Let's say I have a fun to which I want to pass a lambda - and in certain cases it would make sense to pass it a

suspend

lambda. I think there's 2 way I can do this: • have 2 different functions: something like

handle(onSuccess: () -> Unit)

and

handleSuspend(onSuccess: suspend () -> Unit)

• just have the suspend version and it can also accept a regular (non suspend) lambda My question is is there a performance penalty (or otherwise) in doing the second option?

why can't initializers be

suspend

?

What could be the best approch to call a suspension function inside a coroutine exception handler?

val handler = CoroutineExceptionHandler { context, exception ->
	Log.d("exception", exception.message ?: "", exception)
	suspensionFunc() // <--- This part, such as for calling a network requests.
}

What i thought of are: 1. Wrap the context with a

CoroutineScope()

and then launch with it. 2. Call using

runBlocking{}

, but isn't best idea cause it could affect app performance specially if ran on Main threads 3. Use

GlobalScope()

4. Use

GlobalScope()

but launch to IO dispatcher.

This will create 1 job for every element of the list, regardless of how large the list is, which will attempt to then run in parallel. May or may not be the behaviour you want.

Do you know how could I restrain the number of threads? I would like to execute a task (fetch & sync data) with only 5 threads. Here what I have done so far, but it’s not working for me.

val entryCoroutineContext = newFixedThreadPoolContext(5, "sync_entry_fixed")
        CoroutineScope(entryCoroutineContext).launch {
            val syncEntries = syncEntryDao.findAll()

            syncEntries.forEach { syncEntry ->
                Timber.d("Got this data to sync: $syncEntry") // All data are shown here I exepct just 5 to 5
                syncEntry.locked = true
                syncEntryDao.update(syncEntry)

                val response = webServices.syncEntriesCoroutine(clientName, syncEntry.date, syncEntry.blueAppId, syncEntry.offset, syncEntry.limit)
                if (response.isSuccessful) {
                    ...
                }
            }

I need some advice: I have done a project that provides an FSM implementation and a DSL for defining the FSM. As part of the DSL you can provide lambdas for the actions the FSM will invoke. Will it be best to provide the user with a choice up front as to whether all actions will be suspend functions or is there an elegant way that I may not have considered to have a lambda that may or may not be a suspend function and the decision on how to use is is made later? The primary repo can be found here: https://github.com/open-jumpco/kfsm

Whats the best way to efficiency suspend coroutine indefinitely?

delay(Long.MAX_VALUE)

or

Mutex(true).run { lock() }

or something else?

I am using kotlin version 1.1.3-2 .Are coroutines supported on it?

I updated to kotlin 1.3.61 and seeing errors like: Cannot access built-in declaration 'kotlin.Unit'. Ensure that you have a dependency on the Kotlin standard library . Cannot access built-in declaration 'kotlin.collections.MutableSet'. Ensure that you have a dependency on the Kotlin standard library

I'm making a flow that produces a list of files from Google Cloud Storage through running paginated executions. The actual RPCs should be in Dispatchers.IO (I think!). I'm not sure if I'm supposed to use withContext for this or flowOn. What I have is:

fun listFiles(bucket: String, prefix: String): Flow<Blob> = flow {
		var page = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
			gcs.list(bucket, BlobListOption.prefix(prefix))
		}
		while (true) {
			emit(page)
			if (!page.hasNextPage()) {
				break
			}
			page = withContext(<http://Dispatchers.IO|Dispatchers.IO>) { page.nextPage }
		}
	}.flatMapConcat { page ->
		page.values.asFlow()
	}

but I don't know if that is right

Guys. My question is about Clean Architecture and Flow. How we used it with

RxJava

perviously: presentation layer knows nothing about

RxJava

, we only pass callback from

UseCases

with Success or Error. But now we switch to

Flow

a here’s what I need to understand: I don’t want a

ViewModel

to know anything about the

Flow

, so I pass a

job

to a

ViewModel

to unsubscribe from it and callbacks according to state and use

mainScope()

function in every

UseCase

, so I don’t use a

ViewModelScope

. I need to know your thoughts about it. By doing this I’m sure that if I need to switch in future to other framework than

coroutines flow

I would do it pretty smoothly.

Is there any way to combine 2 calls like this :

val time = measureTimeMillis {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    println("The answer is ${one.await() + two.await()}")
}
println("Completed in $time ms")

because if for instance one.await() crashes what will happen? or if both crashes, is there any way to handle these errors? Or in case I just want to continue with the execution instead of adding the values just printing them, is there any way to print the first one (the one that did not crash) and the others doesn't (if it crashes)?

I have a

Flow<Product>

which emits one by one, how I make it to collect those in list so I could receive a list in

collect()

I am uncertain when to use suspend fun verses CoroutineScope extension function, is there a documentation that says why to prefer one over the other?