sealed interface FooResult
data object FooCancelled : FooResult
data class FooSuccess : FooResult

suspend fun foo(): FooResult = TODO()

val deferred = async { foo() }
deferred.cancel()
val result = deferred.await()

CancellationException

FooCancelled

deferred.await()

val foo = MutableStateFlow(0)
val bar = flow { ... }
foo.zip(bar) { ... } // I don't want the resulting flow to be limited by how often foo is updated.

Flow

callbackFlow

close()

trySend(...)

kotlin-coroutines-debug

first()

object MyObject {
    var x = 1
    init {
        GlobalScope.launch {
            x++
        }
    }
}

suspend fun main() = coroutineScope {
    println(MyClass.x.toString()) // output: 1
    delay(100L)
    println(MyClass.x.toString()) // output: 2
}

MyClass.x.toString()

DebugProbes.install()

MutableSharedFlow

class TriggerFlow : Flow<Unit> {
    private val _trigger = MutableSharedFlow<Unit>(
        replay = 1
    )
    
    init {
        _trigger.tryEmit(Unit)
    }
    
    suspend fun trigger() {
        _trigger.emit(Unit)
    }
    
    override suspend fun collect(collector: FlowCollector<Unit>) {
        _trigger.collect(collector)
    }
}

tryEmit

internal fun ScheduledAdItem.adItemStatusAsFlow() = callbackFlow {
    trySendBlocking(adItemStatus)
    val statusListener = AdItemStatusListener { _, status -> trySendBlocking(status) }
    addStatusListener(statusListener)
    awaitClose { removeStatusListener(statusListener) }
}

trySendBlocking

runBlocking

internal fun ScheduledAdItem.adItemStatusAsFlow() = callbackFlow {
    send(adItemStatus)
    val statusListener = AdItemStatusListener { _, status -> launch { send(status) }}
    addStatusListener(statusListener)
    awaitClose { removeStatusListener(statusListener) }
}

delay()

runTest

withTimeout

withContext(Dispatchers.Default.limitedParallelism(1))

Flow.timeout

flowOn

MDCContext

MDC

ThreadContextElement

CopyableThreadContextElement

capacity

onBufferOverflow

channelFlow

suspend fun receiveOutgoingTransactionsAsFlow(capacity: Int): Flow<EventTransaction> = channelFlow(capacity = capacity /* this parameter doesn't exist */) {
        launch {
            sharedFlow.collect { item ->
                trySend(item).getOrElse {
                    close(Error("Private buffer of size $capacity overflowed."))
                }
            }
        }
    }

capacity

Dispatchers.Main

javafx

MainCoroutineDispatcher

launch {
  Foo().use {
    cancel()
    ensureActive()
  }
}.join()
class Foo : AutoCloseable {
  override fun close() {
    error("Foo")
  }
}

use

close

CancellationException

error

use

use

addSuppressed

CancellationException

try-finally

use

use

CancellationException

mono{something().await()}

mono

GlobalScope

class ComponentHealthIndicator(
  private val conn: Component
) : AbstractReactiveHealthIndicator("Component health check failed") {
  override fun doHealthCheck(builder: Health.Builder) : Mono<Health> = mono {
    val ping = conn.ping().await()
    if (isHealthy(ping)) {
      builder.up().build()
    } else {
      builder.down().build()
    }
  }
}

launchIn

scope.launch(dispatcher) { ... }

someFlow
    .onEach { ... }
    .launchIn(scope) // <- I can't set a dispatcher here

fun <T> Flow<T>.launchIn(scope: CoroutineScope, coroutineDispatcher: CoroutineDispatcher): Job = scope.launch(coroutineDispatcher) {
    collect()
}

// Usage
someFlow
    .onEach { ... }
    .launchIn(scope, dispatcher)

@Configuration
class ConcurrencyConfig {
    @Bean
    fun executors(properties: ConcurrencyProperties): CustomExecutors {
        val taskDecorator = MDCTaskDecorator()
        return CustomExecutors(
            Default = createExecutor(properties.default, taskDecorator),
            Io = createExecutor(properties.io, taskDecorator, useVirtual = true)
        )
    }
    
    @Bean
    fun dispatchers(executors: CustomExecutors): CustomDispatchers {
        return CustomDispatchers(
            Default = executors.Default.asCoroutineDispatcher(),
            Io = executors.Io.asCoroutineDispatcher()
        )
    }
}

counter

1.10.2

1.8.0

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach

var counter = 0L
val mutableSharedFlow = MutableSharedFlow<Long>()

var cs: CoroutineScope? = null
fun start() {
    cs?.cancel()
    cs = CoroutineScope(Dispatchers.IO + SupervisorJob())

    mutableSharedFlow.onEach {
        delay(50)
        println(it)
    }.launchIn(cs!!)

    cs!!.launch {
        while (currentCoroutineContext().isActive) {
            mutableSharedFlow.emit(counter)
            counter++
        }
    }
}

suspend fun main() {
    while (true) {
        start()
        println("restart, counter = $counter")

        delay(3000)
    }
}

....
47
48
49
50
51
52
53
54
restart, counter = 56
restart, counter = 56
restart, counter = 56
restart, counter = 56
restart, counter = 56
... // proper behavior (emissions happening) is never restored after this point

CompletionStage

CompletableFuture

CoroutineContext