cancel()

JobCancellationException

kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled; job="coroutine#2444":StandaloneCoroutine{Cancelling}@6b6879d8
	at kotlinx.coroutines.JobSupport.cancel(JobSupport.kt:1605)
	at kotlinx.coroutines.Job$DefaultImpls.cancel$default(Job.kt:189)

1.5.2

1.6.0

fun randomFunction() {
    val channel = Channel<String>()
    val lambda = suspend {
        channel.send("abc")
    }

    runBlocking(randomContext) {
        lambda()

        println(channel.receive())
    }
}

inline fun <T, R> Flow<T>.concurrentMap(
    concurrency: Int,
    @BuilderInference crossinline block: suspend (T) -> R
) = flow<R> {
    val scope = CoroutineScope(Executors.newFixedThreadPool(concurrency).asCoroutineDispatcher())
    val deferredResults = mutableListOf<Deferred<R>>()
    collect { value ->
        deferredResults.add(scope.async { block(value) })
    }
    deferredResults.awaitAll().forEach { newValue -> emit(newValue) }
    scope.cancel()
}

class UserRepository(
    private val userDataSource: UserDataSource,
    private val coroutineScope: CoroutineScope = CoroutineScope(Dispatchers.Default)
) {
    private val userMutex = Mutex()
    private var user: User? = null

    suspend fun fetch(userId: String): User? {
        println("Fetch user")
        println("context = " + coroutineContext) // prints Main

        if (user === null) {
            val result = withContext(coroutineScope.coroutineContext) {
                println("Switch context")
                println("context = " + kotlin.coroutines.coroutineContext) // prints Dispatched
                adsSdkConfigurationDataSource.getUser(userId)
            }

            publisherConfigurationMutex.withLock {
                println("Mutation")
                println("context = " + coroutineContext) // prints Main
                publisherConfigurationApiModel = result
            }
        }

        return publisherConfigurationMutex.withLock {
            publisherConfigurationApiModel
        }
    }
}

Instantiate UserRepository
context = [StandaloneCoroutine{Active}@2272df8, MainDispatcher]
Is UserRepository frozen? false

Fetch user
context = [StandaloneCoroutine{Active}@2272df8, MainDispatcher]

Switch context
context = [DispatchedCoroutine{Active}@3921848, WorkerCoroutineDispatcherImpl@2225538]

Mutation
context = [StandaloneCoroutine{Active}@2272df8, MainDispatcher]
Uncaught Kotlin exception: kotlin.native.concurrent.InvalidMutabilityException: mutation attempt of frozen com.me.UserRepository@227a228

UserRepository

override fun sumSelectedFees(): Single<Float> = rxSingle(workDispatcher) {
    dao.findDeliveryModeShippingOptions(true)
        .fold(0f) { total, shippingOption -> total + shippingOption.feeValue }
}

@Test
fun `should sum all inserted fees for delivery`() = runTest {
    val dataSource = dataSource = DeliveryModesDataSourceImpl(dao, UnconfinedTestDispatcher())
    dataSource.sumSelectedFees()
        .test()
        .assertValue(0f)
}

.test()

UnconfinedTestDispatcher

Flux.just(…).flatMap { someAsnyStuff(it) }.subscribe()

myFlow.flatMapMerge { someSuspendingStuff(it) }.collect()

flow { getSubscriptionFlow().collect { emit(it) } }

incoming.filterIsInstance<WampRawTransportPacket.Handshake>().first()

withTimeoutOrNull(connectionConfig.handshakeTimeout) {
    incoming.filterIsInstance<WampRawTransportPacket.Handshake>().first()
} ?: throw WampTimeout(
    remoteAddress,
    "Timed out waiting ${connectionConfig.handshakeTimeout}ms for a handshake back!"
)

Callable

and/or

simple

simple

suspend fun flowsWithCombine() {
        val myFlow = flowOf("some value")
        val lettersFlow = flowOf("A", "D1", "C",  "E", "F", "G", "H").onEach { delay(500) }
        myFlow.map { value->
            val newValue = lettersFlow.first() 
            println("before $newValue")
            if(newValue == "D1") {
                return@map "D1"
            }else {
                delay(1000)
                val newValue = lettersFlow.first() // here i still get A only though i expect newValue as D1
                 println("after $newValue")
                if(newValue == "D1")  return@map "D1" else return@map value
            }
        }.collect {
            println(it) // This should print D1 
        }
    }

LaunchedEffect(key1 = true) {        sharedViewModel.getAllTasks()    }

val allTasks by sharedViewModel.allTasks.collectAsState()

import kotlinx.coroutines.*

fun main() = runBlocking {
    val average = GlobalScope.calculateAverageAsync(emptyList())

    try {
        println(average.await())
    } catch (ex: ArithmeticException) {
        println("Calculation error")
    }
}

fun CoroutineScope.calculateAverageAsync(list: List<Int>): Deferred<Int> = async {
    list.sum() / list.size
}

Calculation error

Process finished with exit code 0

calculateAverageAsync

runBlocking's scope

import kotlinx.coroutines.*

fun main() = runBlocking {
    val average = this.calculateAverageAsync(emptyList())

    try {
        println(average.await())
    } catch (ex: ArithmeticException) {
        println("Calculation error")
    }
}

fun CoroutineScope.calculateAverageAsync(list: List<Int>): Deferred<Int> = async {
    list.sum() / list.size
}

Calculation error
Exception in thread "main" java.lang.ArithmeticException: / by zero
	at MainKt$calculateAverageAsync$1.invokeSuspend(main.kt:14)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
	at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:279)
	at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:85)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:59)
	at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:38)
	at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
	at MainKt.main(main.kt:3)
	at MainKt.main(main.kt)

Process finished with exit code 1

runBlocking

ArithmeticException

runBlocking

class Test<Value> {
    val listeners = ArrayList<suspend (Value) -> (Unit)>()
    val queue = ArrayDeque<() -> (Unit)>()

    suspend fun fire(value: Value) = suspendCoroutine<Unit> { continuation ->
        synchronized(queue) {
            if (queue.isEmpty()) {
                for (listener in listeners)
                    listener(value)
                continuation.resume(Unit)
                while (queue.isNotEmpty())
                    queue.removeLast()()
            } else {
                queue.add {
                    for (listener in listeners)
                        listener(value)
                    continuation.resume(Unit)
                }
                COROUTINE_SUSPENDED
            }
        }
    }
    fun listen(listener: suspend (Value) -> (Unit)) {
        listeners += listener
    }
}

test.listen { if (it <= 0) test.fire(50) }

<http://Dispatchers.IO|Dispatchers.IO>

CoroutineScope

CoroutineExceptionHandler

flow.take(1).collect {
   println(it)
}
flow.take(4).collect {
    println(it)
}

internal interface ExclusiveObject {
    val mutex: Mutex
}

internal suspend inline fun <R> ExclusiveObject.lockByCoroutineJob(block: () -> R): R {
    return if(mutex.holdsLock(coroutineContext.job)) block()
    else mutex.withLock(coroutineContext.job) {
        block()
    }
}

map

parallel

map

suspend fun <I, O> Iterable<I>.pMap(f: suspend (I) -> O): List<O> = coroutineScope {
    map {
        async { f(it) }
    }.awaitAll()
}
val aP = (1..1000000).toList()

@OptIn(ExperimentalTime::class)
suspend fun main(args: Array<String>) {

    val timeP = measureTime {
        aP.parallelStream().map { it.plus(6) }
        aP.parallelStream().filter { it == 800000 }
    }
    println(timeP)

    val time = measureTime {
        aP.map { it.plus(6) }
        aP.filter { it == 800000 }
    }
    println(time)

}

pMap

parallelStream

Store

suspend fun dispatch(action: Action) {
        val currentState = stateFlow.value

        _actionFlow.tryEmit(action)

        supervisorScope {
            sideEffects.forEach { sideEffect ->
                launch {
                    sideEffect(currentState, action)?.let { dispatch(it) }
                }
            }
        }
}

@Test
    fun should_cancel_side_effect() = runTest {
        launch {
            val store = createStore(this) {
                registerSideEffect sideEffect { _, action ->
                    if (action !is TestAction.Add) return@sideEffect null
                    delay(1_000)
                    return@sideEffect TestAction.Remove(3)
                }
            }

            store.stateFlow.test {
                CoroutineScope(Job()).launch childScope@{
                    store.dispatch(TestAction.Add(3))
                    cancel()
                }.join()

                assertEquals(TestCounterState(0), awaitItem())

                this@launch.cancel()
            }
        }.join()
    }

TestAction.Remove(3)

cancel()

coroutineScope

coroutineScope

.map

StateFlow

StateFlow

supervisorScope{
    list.map {
        async { doSomeApiCall(it) }
    }.awaitAll()
}

supervisorScope {
    list.map {
        async { doSomeApiCall(it) }
    }
}.awaitAll()