kotlinlang #coroutines

As can be seen in the screen recording when I call getAuthenticationData on my repository it should call the asFlow() extension to add the listener to the query and hence reflow future updates. But it seems like that collect and setting of the data is happening before I can get the asFlows to add the listener which then prevents my app from getting an updated value. Does anyone know what might be wrong?

Screen Recording 2021-03-11 at 11.06.22 PM.mov

Clocks

03/13/2021, 6:01 AM

I've been working on kotlinx-gtk to calm my mind, and I created a few tests relating to concurrency and such. https://github.com/Doomsdayrs/kotlinx-gtk/blob/main/src/nativeTest/kotlin/Main.kt Refer to the

ViewModel

I have created in the file above,

Copy code

startSharedDataTest	-> expected execution 
stopSharedDataTest	-> kotlinx.coroutines.CoroutinesInternalError: Fatal exception in coroutines machinery for CancellableContinuation(Shareable[used]){Completed}@1399e08

Second is related to UI execution via dialog tests:

Copy code

Default						->	x11 MT needed
Main 						-> kotlinx.coroutines.CoroutinesInternalError: Fatal exception in coroutines machinery for CancellableContinuation(Shareable[used]){Completed}@82c0c8
Unconfined 					-> Expected event
Unconfinedx2 				-> Expected event
Unconfined via supervisor 	-> Expected event
IO 							-> x11 MT needed
UI 							-> x11 MT needed (need to have execute on the main loop for a gtk application, not on a newSingleThreadContext )
Empty 						-> x11 MT needed

Can someone explain to me: What is happening with the

sharedDataTest

Why

Main

context leads to a fatal error How would I create a context that executes on the main loop for GTK? (I know android has their own dispatches that let you run on the actual UI thread of android)

Javier

03/13/2021, 1:06 PM

Anyone knows about this?

Kefas

03/15/2021, 12:42 PM

Is there any alternative to

flatMapLatest

that is not experimental?

ursus

03/17/2021, 4:40 AM

Is there a rx `publish { observsble -> ... }`equivalent? I found useful for creating pseudo operators like

publish { o -> merge(o.take(1)... , o.skip(1)....) }

baxter

03/22/2021, 6:56 AM

I've seen a few questions about having a

timeout()

operator for flow (similar to Rx), so I created a PR with this operator implemented. https://github.com/Kotlin/kotlinx.coroutines/pull/2597 This operator will only timeout if the upstream fails to emit after the set period of time, but ignores the time spent collecting.

Natsuki(开元米粉实力代购)

03/22/2021, 7:51 AM

Hello everyone, does anyone could help me understand why

collectLatest

wont cancel coroutine launched in its block, saying

Copy code

flow {
        emit(1)
        delay(100L)
        emit(2)
        delay(100L)
        emit(3)
        delay(100L)
    }.collectLatest { value ->
        launch {
            println("start: $value")
            while(true) {
                delay(200L)
                println("loop: $value")
            }
        }
    }

will print

loop:1

and

loop:2

and

loop:3

forever (which i think should not be the desired behavior) while

Copy code

flow {
        emit(1)
        delay(100L)
        emit(2)
        delay(100L)
        emit(3)
        delay(100L)
    }.collectLatest { value ->
        coroutineScope {
            println("start: $value")
            while(true) {
                delay(200L)
                println("loop: $value")
            }
        }
    }

will always print

loop: 3

after a seconds later

Simon Lin

03/23/2021, 2:46 AM

Can I receive most recently two value when a new value emit? for example: emit: 1, 2, 3, 4, 5, 6 downstream receive: [initValue, 1] [1, 2] [2, 3] [3, 4] [4, 5] [5, 6] Slack Conversation

Jakob K

03/23/2021, 7:19 PM

I am getting a strange error when calling

asCompletableFuture()

on a

Deferred<T>

instance:

Copy code

Exception in thread "DefaultDispatcher-worker-2" [20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: kotlinx.coroutines.CoroutinesInternalError: Fatal exception in coroutines machinery for CancellableContinuation(DispatchedContinuation[<http://Dispatchers.IO|Dispatchers.IO>, Continuation at net.axay.fabrik.commands.BrigardierWrapperKt$simpleSuggests$1$1.invokeSuspend(BrigardierWrapper.kt:78)@6fab03fa]){Cancelled}@7a8f83b6. Please read KDoc to 'handleFatalException' method and report this incident to maintainers
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlinx.coroutines.DispatchedTask.handleFatalException(DispatchedTask.kt:144)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:115)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:571)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:750)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:678)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:665)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: Caused by: java.lang.ClassCastException: class kotlin.coroutines.jvm.internal.CompletedContinuation cannot be cast to class kotlinx.coroutines.internal.DispatchedContinuation (kotlin.coroutines.jvm.internal.CompletedContinuation and kotlinx.coroutines.internal.DispatchedContinuation are in unnamed module of loader 'app')
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlinx.coroutines.CoroutineDispatcher.releaseInterceptedContinuation(CoroutineDispatcher.kt:104)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlin.coroutines.jvm.internal.ContinuationImpl.releaseIntercepted(ContinuationImpl.kt:118)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:39)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:104)
[20:05:24] [DefaultDispatcher-worker-2/INFO] (Minecraft) [STDERR]: 	... 4 more

The image shows the code where the error appears. Could this have something to do with

inline

crossinline

or the lambda (or the combination of them all)? Is this an internal bug? The cast that fails can be seen in the second image I attached. Does anyone have an idea why this error even appears?

Alex Vasilkov

03/25/2021, 5:40 PM

Hm, I’m confused. Does anybody know what is the difference between kotlinx.coroutines.CancellationException (from kotlinx-coroutines-core-jvm-1.4.2) and kotlin.coroutines.cancellation.CancellationException (from kotlin-stdlib-1.4.31)? Which one should I import? I know it’s the same typealias for JVM, but what if I would like to support Kotlin Native later?

Patrick Ramsey

03/26/2021, 12:39 AM

Apologies from this structured concurrency newbie!

ursus

03/30/2021, 2:36 AM

tldr; doSomething mutates viewmodel state

ursus

03/30/2021, 3:37 AM

Copy code

fun <T> Flow<T>.test(scope: CoroutineScope): TestCollector<T> {
    return TestCollector(scope, this)
}

Copy code

class TestCollector<T>(
    scope: CoroutineScope,
    flow: Flow<T>
) {
    private val values = mutableListOf<T>()

Copy code

init {
       scope.launch {
           flow.collect {
               values += it
           }
       }
    }

Copy code

fun assertValues(vararg values: T) {
        assertEquals(values.toList(), this.values)
    }
}

quick and dirty of how it worked in rx -- see anything wrong with it?

🧵 3

Susheel

03/30/2021, 6:15 PM

I'm still new to coroutines and I'm trying to understand what line is doing here... https://github.com/ktorio/ktor/blob/master/ktor-client/ktor-client-core/common/src/io/ktor/client/statement/HttpResponse.kt#L81.

Simon Lin

04/12/2021, 6:00 AM

findViewTreeLifecycleOwner

is a useful func to me. Thank you. ---- I want to replace handler in custom view, e.g. auto delay scroll in custom ViewPager2. is it a good practices using coroutines in custom view in this case?

xenomachina

04/13/2021, 11:22 PM

Is there an equivalent to AutoCloseable with a suspending

close

(and

use

) method?

🦗 1

uli

04/14/2021, 8:50 AM

Should I file an issue against refactoring or against the compiler?

ursus

04/19/2021, 5:42 PM

Hi I get this warning Task myappfeaturesothercompileDebugKotlin
w: Opt-in requirement marker kotlinx.coroutines.ExperimentalCoroutinesApi is unresolved. Please make sure it's present in the module dependencies even though I add this in root build.gradle

Copy code

allprojects {
    tasks.withType(org.jetbrains.kotlin.gradle.dsl.KotlinJvmCompile).configureEach { task ->
        task.kotlinOptions {
            jvmTarget = "1.8"
            freeCompilerArgs += "-Xopt-in=kotlin.ExperimentalStdlibApi"
            freeCompilerArgs += "-Xopt-in=kotlinx.coroutines.ExperimentalCoroutinesApi"
        }
    }
}

Why is that? Is it maybe because the featuresother module is

com.android.library

louiscad

04/21/2021, 11:16 AM

Hello, I'd like to test Kotlin 1.5.0-RC but I'm unable to because I use

delay(Duration)

. When can we expect a native-mt and 1.5.0-RC compatible release of kotlinx.coroutines?

Kristian Frøhlich

04/21/2021, 12:43 PM

Copy code

fun CoroutineScope.doFireAndForget(arg: Argument) = launch {
        delay(5000)
        fireAndForgetSuspendFunction(arg)

    }

Kuba Petržílka

04/21/2021, 4:43 PM

Hi, does anybody have an idea how to implement the following simple "result monad comprehension" using coroutines, in a better way than what I have come up with so far:?

Copy code

sealed class Result<out T, out E>
class Success<out T>(val value: T) : Result<T, Nothing>()
class Failure<out E>(val error: E) : Result<Nothing, E>()

...

class AccumulatedResultContext<U, E> {
    var lastPartialResult: Result<U, E>? = null
    var continuation: CancellableContinuation<*>? = null

    suspend operator fun <T, F: E> Result<T, F>.not(): T = when(this) {
        is Success -> this.value
        is Failure -> suspendCancellableCoroutine {
            lastPartialResult = this
            continuation = it
        }
    }
}

fun <T, E> accumulateResult(
    block: suspend AccumulatedResultContext<T, E>.() -> Result<T, E>
): Result<T, E> =
    runBlocking {
        val context = AccumulatedResultContext<T, E>()
        launch(start = CoroutineStart.UNDISPATCHED) {
            context.lastPartialResult = block(context)
        }
        if (context.continuation != null) context.continuation!!.cancel()
        context.lastPartialResult!!
    }

...

Copy code

fun mapToUserRecord(
       userSource: UserSoruce
    ): Result<UserRecord.Persisted, ConversionError> = accumulateResult {
        val id = userSource.id
           
        val name = ! validate(userSource.name)

        val description = ! validate(userSource.description)

        Success(UserRecord.Persisted(id, name, description))
    }

Especially, I d like to avoid using the experimental

CoroutineStart.UNDISPATCHED

but it is required when I start nesting

accumulateResult { }

because the execution order of nested coroutines is not guaranteed Thanks for any suggestions

Alexander Karkossa

04/21/2021, 6:31 PM

Could someone tell me, why Android Studio Canary 14 / Jetpack Compose / kotlin 1.4.32 doesn't find kotlin.coroutines.suspendCoroutine anymore?

Orhan Tozan

04/29/2021, 11:46 AM

In my crash reports,

Copy code

Fatal Exception: kotlinx.coroutines.flow.internal.AbortFlowException
Flow was aborted, no more elements needed

keeps showing up, with no stack tracec or line numbers where it occurred. Anyone have an idea where/when this exception could occur? I'm on 1.4.3

Andrew Gazelka

05/02/2021, 10:11 PM

so this

try

doesn’t seem to be working in catching cancellations (the program still spits out an error) when

newContext.job

is cancelled … any reason why and how I can fix this?

ursus

05/02/2021, 11:55 PM

Hi, I have this flow

Copy code

stream
	.flatMapLatest {
		flow {
			if (!blockingGetWritten()) {
				someSuspendFunction()
				blockingSetWritten(true)
			}
		}
	}

when upstream emits during the inner if statement, lets say just after

someSuspendFunction

but the

blockingSetWritten

has started executing, so it will finish regardless, because its blocking that however means, the new upstream emit might see

blockingGetWritten

still as false thefore i neet to wrap the if statement in a

Mutex

right? (or will the upstream somehow wait till

blockingSetWritten

completes -- I presume not) // Oh, If I actually cause upstream to change during the
blockingSetWritten
the flatMapLatest does not emit right away, it waits for the blocking function to complete, is this a coincidence or by design? I'd not expect that, since its on IO dispatcher anyways

Pablo

05/06/2021, 2:17 PM

Because I'm trying to find a flow example to do this with Flow

Pablo

05/07/2021, 11:18 AM

Guys if I have a mapper like Mapper<I,O> that is very heavy should I do this

fun map

suspend? Or running it with a

withContext(<http://Dispatchers.IO|Dispatchers.IO>)

withContext(Dispatchers.Undefined)

would be enough?

➕ 1