Lets say I have a sealed class

A

with 2 subclasses

B

and

C

. Can I combine a

Flow<B>

and a

Flow<C>

to a

Flow<A>

?

Hello everyone! I was wondering if someone has tried to implement something like this:

fun <T> loopFlow(initial: T, f: suspend (T) -> Flow<T>): Flow<T>

The idea being that the

initial

value is emitted once and then

f

is applied recursively. And then finish the flow when cancelled or when an empty flow is produced (duh). I’ll post in thread what I’ve tried so far without much luck…

Hi, I have a function that I want to suspend until I resume it manually from another function:

private suspend fun sendPacketWithAcknowledgement(packet: RequestPacket) {
        try {
            withTimeout(PACKET_CONFIRMATION_TIMEOUT) {
                sendPacket(packet.bytes)
            }
            // Wait here until ack packet received
            // mutex.lock() // I thought this will suspend the code here and on mutex.unlock() it will resume here
        } catch (e: TimeoutCancellationException) {
            // do sth.
        }
    }

On caller site:

sendPacketWithAcknowledgement(StartCommunication()) // should be sequential
isCommunicationEstablished = true

The caller code should behave sequential, so that

isCommunicationEstablished = true

is only processed, when the ack packet is received and the suspension is resumed manually. I thought mutex would fit here but I guess I didn't use it the right away. It's a bluetooth classic scenario btw. So does someone know how I can achieve this behavior?

What does getMyData().collect do where getMyData is a flow object?

Is there a way to wait for an element from a flow with some timeout? Like to emit either 1) whatever is sent on the wrapped flow or 2) something else if the wrapped flow doesn't produce an element within some time period?

can i poll all items of a buffered channel as a list ? my intent is to ingest events into a buffered channel and at a set rate bulk insert all that are currently in the buffered channel is that the correct way to handle this or is there cleaner constructs provided by coroutines that i am missing ? one of the requirements is that inserting events should cause backpressure

Would you please help me with the following? I want to test that the buffer size of my

MutableSharedFlow<T>(replay = 0, extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)

equals

1

(in reality it's variable). I can't figure out how to get this working in a unit test: every time I emit something it is immediately collected (I use Turbine to

test

the flow). I basically want to emit 2 items and verify that only the last one is collected, the oldest is dropped.

how can i combine a ticker channel and a flow so that on either element in flow or a tick from the channel it emits to the combined flow? is combine the correct operator there?

I have a question regarding this article: https://medium.com/androiddevelopers/coroutines-patterns-for-work-that-shouldnt-be-cancelled-e26c40f142ad My question: If the repository method would not contain

doSomeOtherWork

, we could also switch to the `applicationScope`/`externalScope` in the ViewModel directly, right?

Is there a proper way to "merge" multiple jobs from flow

launchIn

to allow simple cancellation without cancelling the scope / job that is used in the

launchIn

call? Currently saving each jobs separately and cancelling them one by one.

Is there something wrong with the design of this method? It's both a

suspend fun

but also returns a Flow. And the first line can't be caught like this because it's not inside the Flow.

suspend fun getChatUserFlow(uid: String): Flow<ChatUser?> {
        val results = userCollection.whereEqualTo(FieldPath.documentId(), uid).get().await()
        return if (!results.isEmpty) {
            results.documents[0].reference.asFlow<ChatUser>()
        } else {
            flowOf(null)
        }
    }

I am trying to write a suspend function that can listen to cancellation of continuation, and invoke a suspended callback. Here is what I am looking at:

suspend fun <T> PreparedStatement.withCancellation(callback: suspend () -> T): T {
    val stmt = this
    suspendCancellableCoroutine { cont ->
        cont.invokeOnCancellation { 
            stmt.cancel()
        }
    }

    return callback()
}

Exceptions that happen inside a callback used inside a

callbackFlow

can not be caught with the

catch

operator?

Is there a way to register a cancellation listener on a coroutine scope without doing

suspendCancellableCoroutine

so that if the scope gets cancelled I can close my DB statements

What happens if you combine 2 Flows and one of them throws an exception? Will it be caught in a

catch

operator I add on the resulting (combined) Flow?

Is their a need to wrap a DAO Room coroutine function with a withcontext(Dispatcher.io) block. Example

Query()

suspend fun getUser():User

Repository

suspend fun getUserFromDb():User {

return withcontext (<http://dispatcher.IO|dispatcher.IO>) {return data.getUser()}

}

How can an initial value be loaded from a

suspend

function using something similar to

.scan(..)

with a

SharedFlow

?

.scan(loadInitialValueAsynchronously()) { .. }

I’m effectively trying to replace an

actor

with a

SharedFlow

.

This message was deleted.

Just a simple confirmation. The collect of the stateflow in following code will ran in whatever scope the created flow is collected and will be properly cancelled when the collecting flow is cancelled?

fun <T> StateFlow<T>.withPrevious(): Flow<Pair<T?, T>> {
    return flow {
        var previous: T? = null
        this@withPrevious.onEach {
            emit(Pair(previous, it))
            previous = it
        }.collect()
    }
}

fun main() = runBlocking {

    val httpClient = HttpClient {
        install(Logging) {
            level = LogLevel.ALL
        }

        @OptIn(KtorExperimentalAPI::class)
        install(WebSockets)

        expectSuccess = false

        install(JsonFeature) {
            @OptIn(KtorExperimentalAPI::class)
            acceptContentTypes = listOf(ContentType("application", "json-rpc"))
            serializer = KotlinxSerializer()
        }
    }
    val wss =httpClient.webSocketSession(method = HttpMethod.Get, host = "127.0.0.1", port = 6800, path = "/jsonrpc")
    wss.close()
}

this does not stop the program. what am I doing wrong?

Trying out a few ways to throttle API GET requests. The `debounce`/`sample` operators for 

Flow

 don’t work as desired, so what I need is more of a 

throttleFirst

 . Tried with using 

async

 and returning 

null

 when the request should be throttled:

class ApiClient {

    private var deferredLoad: Deferred<Result>? = null

    suspend fun load(throttleMillis: Long = 300): Result? {
        return coroutineScope {
            if (deferredLoad?.isCompleted != false) {
                async {
                    val result = /** make GET API call **/
                    delay(throttleMillis)
                    result
                }
                .apply { deferredLoad = this }
                .await()
            } else {
                null
            }
        }
    }
}

Is this approach recommended? Finding it a little weird to hold on to a 

var

 of 

Deferred

 , thinking it might need a 

Mutex

. Wondering if there’s a more elegant approach that others have found?

Hello. Does anyone know why, when using

.asSingle()

on a

Deferred<T>

from the

kotlinx-coroutines-rx2

library, why the

coroutineContext

I pass into

.asSingle()

can't contain a Job?

when adapting a callback API to suspending is there any point in using

suspendCancellableCoroutine

instead of

suspendCoroutine

when there is no way to propagate cancellation to the adapted API and no action to run via

CancellableContinuation::invokeOnCancellation

?

Hey guys, not sure if this is big enough for slack, how can I run a suspending coroutine inside another suspending coroutine? I have this callback function in my repository

override suspend fun saveConfession(text: String): SaveConfessionResponse = suspendCoroutine {cont ->
        val confessionToSave = ConfessionDataModel(text)
        val call = newConfessionService.saveConfession(confessionToSave)
        call.enqueue(object : Callback<SaveConfessionResponse>{
            override fun onResponse(call: Call<SaveConfessionResponse>, response: Response<SaveConfessionResponse>) {
               cont.resume(response.body()!!)
            }

            override fun onFailure(call: Call<SaveConfessionResponse>, t: Throwable) {
                cont.resumeWithException(t)
            }
        })
    }

And I am trying to call it from my useCase object

   suspend fun execute(text: String): Result = suspendCoroutine{ cont->
        try {
            val response = confessionRepository.saveConfession(text)
            when (response.status) {
                200 -> Result.Success(response.id, 200)
                400 -> Result.Error(null)
                500 -> Result.Error(null)
                else -> Result.Error(null)
            }
        } catch (e: Exception) {
            return Result.Error(e)
        }
    }

/ But i cannot call saveConfession outside of coroutine body, can I somehow get the scope that the execute function was called from?

New to coroutines and flow but does anyone have an example of doing an edittext debounce in flow? The few examples from my search look a LOT lengthier than I think it should be compared to rxjava. Any good resource anyone has found in the past?

Hello, can anyone help me understand the risks associated with using

ExperimentalCoroutinesApi

functions in prod for a library? Should I surface that to my users? If I understand correctly, it means it will be binary compatible (unlike

FlowPreview

). So the only issue is that at some point I might need to change code. But the AAR will be fine using a future version of the library?

what is the best way to simulate an arrayblockingqueue/channel ? MutableSharedFlow? I want to have a place to put items in such that only 1 of my 64 coroutines receive them - simple use case - items come in via a blocking http server [i.e. will need runBlocking to call any suspend methods] - I want to add it to the queue and have the coroutines in the background fetch them in chunks if possible

Support suspend test functions Why's it still not implemented in stdlib 🤔 There should be a runBlocking or runBlockingTest, for testing coroutine based code in commonTest module...

Not to reinvent the wheel, what Flow operators are you all using to accomplish offline logic (return from local persistence if available, or hit api, cache in local persistence and return).

Hello, I'm writing a worker using kotlin, I receive messages from a channel (from a queuing system) - those messages trigger the processing of tasks that I do not manage (user can add its own). To run n job in parallel, I have n coroutines constantly pulling from this channel and running the jobs. I did some tests with CPU intensive jobs (with plain not suspended java) and apparently it works - jobs are processed in parallel - but I do not understand how/why ?! Do you have an explanation ?