Is it possible to check whether sending to Channel will suspend without actually sending anything? I know, that there is

isFull()

function on SendChannel, but that is depracated without replacement

I’m having trouble to really understand how one should test coroutines/flow. I have this dummy code :

@get:Rule
val coroutineTestRule = CoroutineTestRule()

class StateFlowTestClass(
    coroutineDispatcher: CoroutineDispatcher = Dispatchers.Main
) {
    private val _state = MutableStateFlow(0)
    val state: StateFlow<Int> = _state

    init {
        CoroutineScope(coroutineDispatcher).launch {
            blockingTask()
        }
    }

    private suspend fun blockingTask() {
        delay(10.seconds)
        _state.value = 1
    }
}

@Test
fun `test on StateFlow`() = coroutineTestRule.runBlockingTest {
    advanceTimeBy(11.seconds.toLongMilliseconds())
    val value = StateFlowTestClass(TestCoroutineDispatcher()).state.value
    Assert.assertEquals(1, value)
}

when I run test on StateFlow I get the following error :

java.lang.AssertionError: Expected:1, Actual:0

. How am I suppose to fix this? How about when there is no delay but just a bunch of code taking time to execute, is there a way to wait it to be done?

so i know i cannot block on javascript.. but does anyone have a snippet for kotlin-js that does approximately the same thing as runBlocking and returning the result ? something like

val derferred = async {  ... }
while(!deferred.isCompleted) {
   delay(10)
}
return deferred.getCompleted()

except since i am not in a coroutine i cannot use delay outside of it..

Hello, have anyone tried to implement something like RxJava PublishSubject with Coroutines? If so, could you please show some examples or gists?

Hello. Let's say I want to process data from the

ReceiveChannel

and then pass the channel somewhere else in my app. Is the following approach a good solution (in sense of resources consumption and performance)? 1. Transform the channel to a

Flow

with

consumeAsFlow()

2. Use a bunch for different flow operations (

map

,

mapNotNull

etc.) 3. Transform

Flow

back to the

ReceiveChannel

with

produceIn()

Is there a

shareIn

alt which evaluates lazily? i.e.,

aFlow = flow(1,2,3).onEach{ println("A $it") }
shared = aFlow.lazyCache()
shared.take(1).forEach{ println("B $it")}
shared.take(2).forEach{ println("C $it")}

would print “A1 B1 A2 C2" … or something to that effect

came up with something that accomplishes my goals well enough

if I use launch {} with a multithreaded coroutines dispatcher, will the whole launch block run in the same thread, or will the thread change possibly after return from a suspend method?

Are coroutines in kotlin js backed by javascript generators at any point? Just curious

I still struggle with fully understanding scopes & contexts. Let’s say I have two scopes and would like to perform some suspendable work that is canceled if either of the two scopes is canceled. Is this sufficient?

withContext(scope1.coroutineContext) {
    withContext(scope2.coroutineContext) {
        doSomeWork()
    }
}

Similarly with `Job`:

val context2 = scope1.coroutineContext + SupervisorJob()

withContext(scope1.coroutineContext) {
     withContext(context2) {
         doSomeWork()
     }
}

Would the cancellation of one context affect the other context? Is that the proper way to do that?

do kotlin flows provide a mean of batching items into collections? e.g.

flow.batch(size=20).collect { items -> /* do some batched operation e.g. a database insert */ }

👋 I want to convert the

Stream

returned by

Files.walk(Path)

to a

Flow

. I could use the extension function

consumeAsFlow()

but

Files.walk(Path)

blocks the calling thread. So, what's the correct way to do it? I implemented this but it doesn't feel right:

flow<Path> {
    Files.walk(path)
        .forEach { path -> runBlocking { emit(path) } }
}
.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

And, as side question, is there any library that implements the filesystem api as correct suspend functions?

image.png

I have an

Actor

, and I'd like to be able to remove/cancel an operation when a new one comes in, depending on the progress of the previous operation.

really stupid question, but i can’t find an answer by googling… Are there any guarantees on order of execution when using a single threaded coroutine dispatcher? e.g.:

val dispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
val scope = CoroutineScope(dispatcher)

for (i in 1..5) {
    scope.launch {
        println(i)
    }
}

How come

Main.immediate

isn't the default dispatcher of

MainScope()

? Is there any disadvantage in using

immediate

?

I have this function

suspend fun foo(progressLambda: (Long) -> Unit)

and I want to convert this to a function returning a

StateFlow<State>

where

State

is a sealed class providing

Progress

,

Success

etc. Initially I thought of using

callbackFlow

but that seems to be more for functions that run asynchronously. Is there a more appropriate way?

I'm emitting a state and I want it to be cleared after receiving by an observer, I'm currently using

StateFlow

and after changing the state to desired object, I immediately set it back to null. This already sounding a bad practice so could someone guide me here with which Coroutine Flow/Channel should I use?

Hello everyone and Happy New Year 🎉 I have a question about SharedFlow. According to documentation

A default implementation of a shared flow that is created with MutableSharedFlow() constructor function without parameters has no replay cache nor additional buffer. emit call to such a shared flow suspends until all subscribers receive the emitted value and returns immediately if there are no subscribers. Thus, tryEmit call succeeds and returns true only if there are no subscribers (in which case the emitted value is immediately lost).

That means that we might lose some emitted values. To tackle this problem I created the following function:

/**
 * This function emits value when the receiver shared flow gets the first subscriber
 * It helps to avoid loosing events while view is not subscribed to shared flow
 */
suspend fun <T> MutableSharedFlow<T>.emitLazily(value: T) {
    subscriptionCount.first { it > 0 }
    emit(value)
}

The question is will the 

first

 function throw an 

NoSuchElementException

 in this case? As far as I understand, it should not do it 😄) Because, the shared flow never completes and we can not convert it into a completing one without adding subscriber. It will be extremely helpful to hear your thoughts))

I want to perform an event every 10 minutes so I'm following this code. https://stackoverflow.com/questions/54827455/how-to-implement-timer-with-kotlin-coroutines

val tickerChannel = ticker(delayMillis = 10 * 60_000, initialDelayMillis = 0)

repeat(10) {
    tickerChannel.receive()
    println("TIMER SUCCESS")
}

I want it to repeat forever though. I'm probably missing something dumb, but can I signify forever somehow, or do I have to use Integer.MAX_VALUE?

If i assign the listener to a file level variable outside the method, and clear it inside awaitClose, it does work

Are there non-blocking suspend functions in the standard library that replace these blocking ones FileInputStream.read() FileOutputStream.write()?

I understand that flows are sequential. So even if there's a delay it doesn't disrupt the ordering. But what if I want to disrupt the ordering? Details in 🧵

I want to ask about thread safety of state flow imperative reading/writing

private val state = MutableStateFlow(IDLE)

fun doSomething() {
    if (state.value == LOADING) return
    state.value = LOADING
    scope.launch {
        try {
            actuallyDoSomething()
            state.value = SUCCESS
        } catch(ex: Exception) {
            state.value = ERROR
        }
    }
}

i.e. this is not guaranteed to be only single execution, right? I need to make reading & writing of LOADING synchronized, correct?

Is there an operator to combine the latest values from a

List<Flow<T>>

? Something that would have this signature

(List<Flow<T>>) -> Flow<List<T>>

? I’m updating a RecyclerView using

AsyncListDiffer.submitIList()

.

If one was to model an unidirectional data-flow, which primitive would you use to represent the events that the UI can create back to the Presenter/ViewModel? I’m currently using a

SendChannel

backed by a

RENDEZVOUS

implementation, but wondering if

MutableSharedFlow

(with

replay=0

) would be better (i assume due to less overhead?!)

I wonder if anybody has played with integrating Flow in iOS Combine - perhaps something like flow.observeAsPublisher() or similar shortcut…

New Blogpost: “7 Common Mistakes you might be making when using Kotlin Coroutines” => https://www.lukaslechner.com/7-common-mistakes-you-might-be-making-when-using-kotlin-coroutines/

Hey, I'm trying out Ktor's WebSocket client library. It returns incoming messages from a web socket as a 

SharedFlow

 - I'm trying to figure out how to keep the web socket session (

DefaultClientWebSocketSession

) running in the background so that I can collect the 

SharedFlow

 messages and display them on screen as they stream in. I tried using 

coroutineScope.launch{}

  but it seems like the web socket session would stop as soon as the coroutine scope block ended. Not sure what to do to make sure the process runs forever in the background

Does anybody know how you can create a

MutableStateFlow

that allows duplicate events to be send. Basically mitigate the default

distinctUntilChanged

behavior?