Need Help - Android App - Coroutines background: Create secure socket server to allow maximum possible socket client to connect. as to build connection with socket client, socket server need continues loop to accept connection. along way to communicate with each client need to open stream input and output to listen and send messages, this also need continues loop to read messages. Technically: I have tried using Thread but at this level handle thread and expect performance is hard, so come to the solution to use Coroutine. I have also tried Ktor for coroutine socket but don't find proper way(or I don't understand) how we want to build SSL Not a Coroutine master just learn and implemented. Problem: observed its uses higher CPU after some sort of time and UI start freezing, sometimes it hanging the phone. Also, continuously getting below kind of warning logs:

zygote: Explicit concurrent copying GC freed 35008(4MB) AllocSpace objects, 5(164KB) LOS objects, 89% free, 2MB/20MB, paused 211us total 67.413ms

zygote: Increasing code cache capacity to 512KB

Choreographer: Skipped 48 frames!  The application may be doing too much work on its main thread.

zygote: Do partial code cache collection, code=53KB, data=54KB
zygote: After code cache collection, code=53KB, data=54KB
zygote: Increasing code cache capacity to 256KB

(after some optimization it lowered but still having the issue) I need help to select the right approach understand the issue insightful in way to resolve and use correctly.

I have a custom scope built like this

CoroutineScope(Dispatchers.Default + CoroutineExceptionHandler { _, e ->
                    log.error("Uncaught error in scope", e)
                })

But it looks like when an error occurs, or rather after a few errors, coroutines basically stop like if the scope was dead or something. Eveyrthing seems stuck. Does calling the CoroutineExceptionHandler cancel the scope’s job or something? Do I need to specify a

+ SupervisorJob()

when building the scope? I don’t really understand what’s happening

If I want to replace livedata with StateFlow, should I be using

collect

or

onEach

to subscribe to the updates on Stateflow

open class Listener<T>(element: T) {

var value: T = element

protected set(value) {

field = value

notifyObservers(field)

}

private val observers: ConcurrentHashMap<UUID, Observer<T>> = ConcurrentHashMap()

private fun notifyObservers(newValue: T) {

observers.forEach { it.value.onChange(newValue) }

}

fun addListener(observer: Observer<T>): Subscription {

var uuid = UUID.randomUUID()

while(observers.contains(uuid)) {

Thread.sleep(1)

uuid = UUID.randomUUID()

}

val subscription = Subscription(uuid, this)

observers[uuid] = observer

return subscription

}

fun unsubscribeListener(uuid: UUID) = observers.remove(uuid)

fun unsubscribeAll(): Int {

val num = observers.size

observers.clear()

return num

}

}

(DefaultDispatcher-worker-5 @coroutine#5) Subscribing 10000 listeners took 177ms

(DefaultDispatcher-worker-4 @coroutine#4) Subscribing 10000 listeners took 178ms

(DefaultDispatcher-worker-1 @coroutine#3) Subscribing 10000 listeners took 178ms

(DefaultDispatcher-worker-3 @coroutine#2) Subscribing 10000 listeners took 178ms

(DefaultDispatcher-worker-6 @coroutine#6) Subscribing 10000 listeners took 178ms

(main @coroutine#1) Unsubscribing all 49962 listeners took 0ms

Is there a way to use have a

MutableStateFlow<T : Any>

where the initial value is not set immediately? I am migrating from RxJava2

BehaviorSubject<T : Any>

. The value is not initialized until some process finishes. Observers of the value are transient. I would rather it not be a

T?

where I filter out the initial null value. Thanks!

Say you have a fragment that’s observing a stateflow in the viewmodel. How do you know what event triggered that change in state? Is the only way to know by setting breakpoints in the viewmodel everywhere where the stateflow gets updated?

This test fails:

@Test
  fun `emitting too fast`() {
    val scope = CoroutineScope(Dispatchers.Default)
    val actionFlow = MutableSharedFlow<String>(extraBufferCapacity = 10)
    var success = false

    actionFlow
      .onEach { success = true }
      .launchIn(scope) // 1

    actionFlow.tryEmit("HELLO") // 2
    

    runBlocking {
      withTimeout(1000) {
        while (!success) {
          yield()
        }
      }
      success shouldBe true
    }
  }

I understand why, I think, it's because the 'subscription' to actionFlow is launched (

1

) and therefore hasn't happened yet by the time I try to emit (

2

). Therefore, because `SharedFlow`s don't buffer or anything by default, the emission of "HELLO" is just dropped. My question is, what's the best way to make sure

1

is ready before I do

2

. I can imagine there's a way to use

SharedFlow.subscriptionCount

but that seems inelegant. Any better way?

Is there an

onEach

variant but with

onEachLatest

instead? I have a flow that has an

onEach

operator on its chain, and in its body, there is a suspendCancellableCoroutine, and I want it to be cancelled when a new value is emitted, I tried looking around but couldn't find any.

I have an

object

with private construct-once properties (e.g.

private val x = SomeConstructor()

) and public

suspend

functions that access these properties. The suspending funs all use

withContext(myContext) {}

to ensure they access the properties from the correct thread. How can I ensure that the properties are instantiated on that same thread? Would it be enough to use

private val x by lazy { SomeConstructor() }

as long as

x

is only accessed from within

withContext(myContext)

? Will that guarantee that

x

is constructed on the thread(s) backing

myContext

?

What would I use to basically get a channel or flow that can potentially send to itself? Basically async tail recursion where if the x condition is satisfied, you send channel item back, otherwise process the current item

I am having difficulty getting the difference in behavior between

runBlocking

and

coroutineScope

, which I put into this question. Could someone help me with it, please?

is there a way to do something like this for all modules in a project?

// ignore these warnings
kotlin.target.compilations.all {
    allKotlinSourceSets.forEach { sourceSet ->
        sourceSet.languageSettings.apply {
            useExperimentalAnnotation("kotlinx.coroutines.FlowPreview")
            useExperimentalAnnotation("kotlinx.coroutines.ExperimentalCoroutinesApi")
            useExperimentalAnnotation("io.ktor.util.KtorExperimentalAPI")
        }
    }
}

How would I transform a flow with multiple parallel coroutines? I saw .buffer() option, but that just does the transform on a single parallel coroutine to collector

If I want to make a server request, but also, at the same time subscribe to further updates of that data from a StateFlow (driven separately by something i.e. a push message telling the app the data is fresh), what would the best way to do that? Something to keep in mind is I need to pass exceptions from the server request though as well. I haven’t found an elegant way to do this without it getting messy.

is lint lying or am doing it in a wrong way? 🤔

Which Flow method do you happen to use more than the other? 1️⃣

flatMapConcat { }

2️⃣

flatMapMerge { }

3️⃣ i dont remember using any of them

A common pattern I've seen: Holding onto a job because you want to cancel the existing job if it's already in progress. Is there a better way to do this?

What’s the best way to test multiplatform coroutines that have

delay

in them?

Having difficulties with suspendCancellableCoroutine. Could someone please help me to figure out what I missed? Caused by java.lang.IllegalStateException: Already resumed, but proposed with update kotlin.Unit at kotlinx.coroutines.CancellableContinuationImpl.alreadyResumedError(CancellableContinuationImpl.kt:447) at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl(CancellableContinuationImpl.kt:412) at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl$default(CancellableContinuationImpl.kt:388) at kotlinx.coroutines.CancellableContinuationImpl.resumeWith(CancellableContinuationImpl.kt:293) at $wrapStatusListener$$inlined$suspendCancellableCoroutine$lambda$1.onSuccess(Misc.kt:88) at java.lang.reflect.Method.invoke(Method.java) at com.twilio.chat.internal.RethrowingForwarder$RethrowingProxy.invoke(RethrowingForwarder.java:121) at java.lang.reflect.Proxy.invoke(Proxy.java:1006) at $Proxy11.onSuccess() at com.twilio.chat.internal.StatusListenerForwarder$1.run(StatusListenerForwarder.java:33) at android.os.Handler.handleCallback(Handler.java:938) at android.os.Handler.dispatchMessage(Handler.java:99) at android.os.Looper.loop(Looper.java:223) at android.app.ActivityThread.main(ActivityThread.java:7660) at java.lang.reflect.Method.invoke(Method.java) at com.android.internal.os.RuntimeInit$MethodAndArgsCaller.run(RuntimeInit.java:592) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:947)

suspend fun <T> wrapStatusListener(value: T, block: (StatusListener) -> Unit): T {
    return suspendCancellableCoroutine { continuation: CancellableContinuation<T> ->
        block(object : StatusListener() {
            override fun onSuccess() {
                continuation.resume(value)
            }

            override fun onError(errorInfo: ErrorInfo) {
                continuation.resumeWithException(
                       Exception(
                        code = errorInfo.code,
                        status = errorInfo.status,
                        errorMessage = errorInfo.message
                    )
                )

            }
        })
    }
}

Android specific question re: coroutines ... is there any issue with doing something like this?

class MyViewModel(): ViewModel() {
    private var job: Job = viewModelScope.launch {
        // Some long running async task, like polling a server or something
    }

    fun onSomeUIEvent() {
        job.cancel()
    }
}

Does kotlin have an equivalent of twisted’s

deferToThread()

? IE, a neat, packaged way to spawn a thread/run some blocking work on a thread pool, and monitor it from a coroutine which resumes the caller when the blocking code finishes?

📣 📣 📣

kotlinx.coroutines

1.4.3 and 1.4.3-native-mt are here! •

CoroutineStart.UNDISPATCHED

promoted to stable API • Better support of

ThreadContextElement

and its integrations • Proper support of

onUndeliveredElement

in unlimited channels • Various performance improvements and bug fixes Full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.4.3

However, the replacement

onReceiveOrNull

is already imported, but, not being used.

Hello! I asked this over in #getting-started and it was suggsted that ask ask over here.

Not sure if this is the right place to ask, but is there a reason why for flows there is an imperative interface for generating flows, but no imperative interface for consuming flows, like you can with channels or iterators? Or did I just not find it? I understand the benefits in readability etc. of using flow combinators where possible, but it seems to me there are certain parsing-type problems which are just not easily / readably expressable with the current flow combinators. Here https://pl.kotl.in/G07SkQwGw is a playground where I've implemented something which preserves the cold-by-default behaviour of flows while allowing you to consume and transform flows imperatively. Code sample:

// f1 is a flow of somewhat random numbers, no deeper meaning
    val f1 : Flow<Int> = flowOf(1, 1, 2, 3, 5, 8, 3, 1, 4, 5, 9, 4, 3, 7, 0, 7, 7, 4, 1, 5)
    /*
    Let's imagine that we are told f1 consists of "packets", which start with a length field followed by as many numbers
    as indicated in the length field. We want to transform f1 into a flow where each element is the sum of the numbers
    in the "payload" of one packet.
    */
    runBlocking {
        println("package sums:")
        f1.transformImperatively<Int, Int> {
            while (true) {
                if (hasNext()) {
                    emit(sum(next()))
                } else {
                    break
                }
            }
        }.collect(::println)
    }

code for sum:

suspend fun (FlowIterator<Int>).sum( count : Int ) : Int {
    var s : Int = 0
    for (i in 1..count) {
        if( hasNext() ) {
            s += next()
        } else {
            break
        }
    }
    return s
}

It's not finished, more a proof-of-concept. (In particular, I haven't had time yet to try and understand flow cancellation, or the context preservation properties expected of flows, so don't expect it to behave correctly in that regard yet. Though I would very much expect it to be possible to make it behave correctly in that sense as well.) I would be happy about any kind pointers to discussions about similar things, reasons why something like this isn't or shouldn't be in kotlinx.coroutines, reasons why it might break down later on, etc.

Question: Why would this be preferred

private val _myFlow = MutableStateFlow<String>("")
val myFlow: SharedFlow<String> = _myFlow.asStateFlow()

over

private val _myFlow = MutableStateFlow<String>("")
val myFlow: SharedFlow<String> = _myFlow

? A MutableStateFlow is a StateFlow; why the extra call to

asStateFlow()

?

are there any plans for a serial queue style scope that doesn't require a dedicated thread? similar to how GCD works.

I'm having trouble grasping something basic about coroutines (even though I understand concurrency in other programming languages). Why is it necessary for responsiveness to have both

launch

(which presumably puts something on another thread) and

suspend

functions, which enable that thread to be used for another coroutine? I'm working in the context of Android, where it's vital that the main thread not get blocked. Why isn't launching a database access, for example, sufficient to guarantee main thread responsiveness? Is the issue that there are a small number of worker threads so no single database access should monopolize one?