### runningReduce example Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/running-reduce.html Produces a list of intermediate reduction results, starting with the first element. The output is [1, 3, 6, 10]. ```kotlin flowOf(1, 2, 3, 4).runningReduce { acc, value -> acc + value }.toList() ``` -------------------------------- ### DebugProbes usage example with dump output Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-debug/index.html Runnable example from test/Example.kt that installs DebugProbes, launches a coroutine that computes two values with delays, and dumps running coroutines. The printed output shows the dump format for all coroutines and for a specific deferred job. ```kotlin suspend fun computeValue(): String = coroutineScope { val one = async { computeOne() } val two = async { computeTwo() } combineResults(one, two) } suspend fun combineResults(one: Deferred, two: Deferred): String = one.await() + two.await() suspend fun computeOne(): String { delay(5000) return "4" } suspend fun computeTwo(): String { delay(5000) return "2" } fun main() = runBlocking { DebugProbes.install() val deferred = async { computeValue() } // Delay for some time delay(1000) // Dump running coroutines DebugProbes.dumpCoroutines() println("\nDumping only deferred") DebugProbes.printJob(deferred) } ``` ```text Coroutines dump 2018/11/12 21:44:02 Coroutine "coroutine#2":DeferredCoroutine{Active}@289d1c02, state: SUSPENDED at kotlinx.coroutines.DeferredCoroutine.await$suspendImpl(Builders.common.kt:99) at ExampleKt.combineResults(Example.kt:11) at ExampleKt$computeValue$2.invokeSuspend(Example.kt:7) at ExampleKt$main$1$deferred$1.invokeSuspend(Example.kt:25) ... More coroutines here ... Dumping only deferred "coroutine#2":DeferredCoroutine{Active}, continuation is SUSPENDED at line kotlinx.coroutines.DeferredCoroutine.await$suspendImpl(Builders.common.kt:99) "coroutine#3":DeferredCoroutine{Active}, continuation is SUSPENDED at line ExampleKt.computeOne(Example.kt:14) "coroutine#4":DeferredCoroutine{Active}, continuation is SUSPENDED at line ExampleKt.computeTwo(Example.kt:19) ``` -------------------------------- ### install Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-debug/kotlinx.coroutines.debug/-debug-probes Installs debug probes. ```APIDOC ## Function: install ### Description Installs a DebugProbes instead of no-op stdlib probes by redefining debug probes class using the same class loader as one loaded DebugProbes class. ### Signature `fun install()` ### Usage Call this function to install debug probes. ``` -------------------------------- ### install Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-debug/kotlinx.coroutines.debug/-debug-probes/index.html Installs DebugProbes instead of no-op stdlib probes by redefining debug probes class using the same class loader as one loaded DebugProbes class. ```APIDOC ## install ### Description Installs DebugProbes instead of no-op stdlib probes by redefining debug probes class using the same class loader as one loaded DebugProbes class. ### Function Signature `fun install()` ### Usage Example ```kotlin DebugProbes.install() ``` ``` -------------------------------- ### Job.start() Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job/start.html Documentation for the Job.start() method, which starts the coroutine associated with the job if it has not been started yet. ```APIDOC ## Job.start() ### Description Starts the coroutine related to this job if it was not started yet. Returns `true` if this invocation actually started the coroutine, or `false` if it was already started or completed. ### Method `abstract fun start(): Boolean` ### Return Value - **Boolean** - `true` if this invocation actually started the coroutine, `false` if it was already started or completed. ``` -------------------------------- ### Alternative: launch directly instead of LAZY start Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-l-a-z-y Alternative to using LAZY with launch and then unconditionally starting it: create the coroutine directly at the place where it is started, avoiding the LAZY start. ```kotlin // instead of `val job = scope.launch(start = CoroutineStart.LAZY) { }; job.start()`, do scope.launch { } ``` -------------------------------- ### onStart Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow Invokes an action before the flow starts to be collected. ```APIDOC ## onStart ### Description Returns a flow that invokes the given action **before** this flow starts to be collected. ### Function Signature `fun Flow.onStart(action: suspend FlowCollector.() -> Unit): Flow` ### Parameters - **action** (suspend FlowCollector.() -> Unit) - Required - The action to invoke before collection starts. ### Returns - **Flow** - A flow that invokes the action before collection. ``` -------------------------------- ### scan example with flowOf Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/scan.html Example showing how to use scan on a flow of integers, accumulating into a list and emitting intermediate results. The output is [[], [1], [1, 2], [1, 2, 3]]. ```kotlin flowOf(1, 2, 3).scan(emptyList()) { acc, value -> acc + value }.toList() ``` -------------------------------- ### start Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-job Starts the coroutine if it has not been started yet. ```APIDOC ## start ### Description Starts coroutine related to this job (if any) if it was not started yet. The result is `true` if this invocation actually started coroutine or `false` if it was already started or completed. ### Method start ### Signature `abstract fun start(): Boolean` ### Parameters None ### Returns - **Boolean** - `true` if the coroutine was started by this invocation, `false` otherwise. ``` -------------------------------- ### CoroutineStart.DEFAULT Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-d-e-f-a-u-l-t/index.html Documentation for the CoroutineStart.DEFAULT constant, including its behavior and usage examples. ```APIDOC ## CoroutineStart.DEFAULT ### Description Immediately schedules the coroutine for execution according to its context. This is usually the default option. DEFAULT uses the default dispatch procedure described in the CoroutineDispatcher documentation. If the coroutine's Job is cancelled before it started executing, then it will not start its execution at all and will be considered cancelled. ### Usage This is the default start mode for coroutine builders like `launch` and `async`. It schedules the coroutine for execution according to its context. ### Examples #### Example 1: Starting a new coroutine that goes through a dispatch ```kotlin runBlocking { println("1. About to start a new coroutine.") // Dispatch the job to execute later. // The parent coroutine's dispatcher is inherited by default. // In this case, it's the single thread backing `runBlocking`. launch { // CoroutineStart.DEFAULT is launch's default start mode println("3. When the thread is available, we start the coroutine") } println("2. The thread keeps doing other work after launching the coroutine") } ``` #### Example 2: Starting a new coroutine that doesn't go through a dispatch initially ```kotlin runBlocking { println("1. About to start a coroutine not needing a dispatch.") // Dispatch the job to execute. // `Dispatchers.Unconfined` is explicitly chosen. launch(Dispatchers.Unconfined) { // CoroutineStart.DEFAULT is the launch's default start mode println("2. The body will be executed immediately") delay(50.milliseconds) // give up the thread to the outer coroutine println("4. When the thread is next available, this coroutine proceeds further") } println("3. After the initial suspension, the thread does other work.") } ``` #### Example 3: Cancelling coroutines before they start executing ```kotlin runBlocking { // dispatch the job to execute on this thread later launch { // CoroutineStart.DEFAULT is the launch's default start mode println("This code will never execute") } cancel() // cancels the current coroutine scope and its children launch(Dispatchers.Unconfined) { println("This code will never execute") } println("This code will execute.") } ``` ``` -------------------------------- ### rxSingle Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-rx2/index.html Creates a cold Single that starts the coroutine on subscription. ```APIDOC ## rxSingle ### Description Creates a cold Single that starts the coroutine on subscription. ### Result Single ### Scope CoroutineScope ### Parameters - **block** (suspend CoroutineScope.() -> T) - Required - The coroutine body that returns a value. ### Returns A Single that emits the value when subscribed. ``` -------------------------------- ### Example of usage: computeValue and DebugProbes Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-debug Shows how to use DebugProbes.install() and dumpCoroutines() to inspect running coroutines. The example defines suspend functions that compute values asynchronously and prints a coroutine dump after a delay. ```kotlin suspend fun computeValue(): String = coroutineScope { val one = async { computeOne() } val two = async { computeTwo() } combineResults(one, two) } suspend fun combineResults(one: Deferred, two: Deferred): String = one.await() + two.await() suspend fun computeOne(): String { delay(5000) return "4" } suspend fun computeTwo(): String { delay(5000) return "2" } fun main() = runBlocking { DebugProbes.install() val deferred = async { computeValue() } // Delay for some time delay(1000) // Dump running coroutines DebugProbes.dumpCoroutines() println("\nDumping only deferred") DebugProbes.printJob(deferred) } ``` -------------------------------- ### runTest with TestResult return Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-test/index.html Advanced form of runTest that allows initializing test state before returning the runTest result. Useful when more setup is needed before the test coroutine starts. ```kotlin @Test fun testFoo(): TestResult { // initialize some test state return runTest { // code under test } } ``` -------------------------------- ### Accessing the test coroutine scope with TestScope Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-test/index.html Demonstrates manually creating a TestScope before the test starts, for example to inject it via dependency injection in @BeforeTest. It sets the Main dispatcher to a StandardTestDispatcher using the scope's scheduler and resets it in @AfterTest, then runs the test body with scope.runTest. ```kotlin val scope = TestScope() @BeforeTest fun setUp() { Dispatchers.setMain(StandardTestDispatcher(scope.testScheduler)) TestSubject.setScope(scope) } @AfterTest fun tearDown() { Dispatchers.resetMain() TestSubject.resetScope() } @Test fun testSubject() = scope.runTest { // the receiver here is `testScope` } ``` -------------------------------- ### associateWith usage example Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/associate-with.html Collects a flow of words into a map where each word maps to its length. The example prints the keys and values to show the resulting map order and content. ```kotlin val words = flowOf("a", "abc", "ab", "def", "abcd") val withLength = words.associateWith { it.length } println(withLength.keys) // [a, abc, ab, def, abcd] println(withLength.values) // [1, 3, 2, 3, 4] ``` -------------------------------- ### Possible output of ChannelIterator example Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/-channel-iterator Possible output from the example usage, showing that elements are distributed among consumers. ```text Consumer A got 1 Consumer A got 2 Consumer B got 3 ``` -------------------------------- ### CoroutineStart.UNDISPATCHED Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-u-n-d-i-s-p-a-t-c-h-e-d/index.html Documentation for the CoroutineStart.UNDISPATCHED option, including its behavior, usage, and examples. ```APIDOC ## CoroutineStart.UNDISPATCHED ### Description Immediately executes the coroutine until its first suspension point in the current thread. Resumptions from later suspensions use the actual dispatcher from the coroutine's context. The coroutine will not form an event loop. ### Usage Use `CoroutineStart.UNDISPATCHED` as the `start` parameter in coroutine builders like `launch` and `async`. ### Parameters - `start` (CoroutineStart) - Required - Set to `CoroutineStart.UNDISPATCHED` to start the coroutine undispatched. ### Behavior - Executes the coroutine immediately in the current thread until the first suspension point. - Resumptions from later suspensions use the actual dispatcher from the coroutine's context. - Even if the coroutine is already cancelled, its code will still start running (similar to ATOMIC). - Does not form an event loop. ### Example ```kotlin var tasks = 0 repeat(3) { launch(start = CoroutineStart.UNDISPATCHED) { tasks++ try { println("Waiting for a reply...") delay(50.milliseconds) println("Got a reply!") } finally { tasks-- } } } // Because of UNDISPATCHED, // we know that the tasks already ran to their first suspension point, // so this number is non-zero initially. while (tasks > 0) { println("currently active: $tasks") delay(10.milliseconds) } ``` ``` -------------------------------- ### rxMaybe Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-rx2/index.html Creates a cold Maybe that starts the coroutine on subscription. ```APIDOC ## rxMaybe ### Description Creates a cold Maybe that starts the coroutine on subscription. ### Result Maybe ### Scope CoroutineScope ### Parameters - **block** (suspend CoroutineScope.() -> T?) - Required - The coroutine body that returns a value or null. ### Returns A Maybe that emits the value or completes when subscribed. ``` -------------------------------- ### toList example with produce Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/to-list.html Example showing how to use toList to collect elements from a channel created with produce. The channel is closed when the coroutine body finishes, and toList returns a list equal to the original values. ```kotlin val values = listOf(1, 5, 2, 9, 3, 3, 1) // start a new coroutine that creates a channel, // sends elements to it, and closes it // once the coroutine's body finishes val channel = produce { values.forEach { send(it) } } check(channel.toList() == values) ``` -------------------------------- ### Flow.sample(Duration) example Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/sample.html Example of using Flow.sample with a Duration period. Emits the latest value every 200ms from a flow that emits 0-9 with 110ms delays. Output is 1, 3, 5, 7, 9. Note that the latest element is not emitted if it does not fit into the sampling window. ```kotlin flow { repeat(10) { emit(it) delay(110.milliseconds) } }.sample(200.milliseconds) ``` ```text 1, 3, 5, 7, 9 ``` -------------------------------- ### kotlinx.coroutines.reactive.publish Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-reactive Creates a cold reactive publisher that starts the coroutine on subscribe. ```APIDOC ## kotlinx.coroutines.reactive.publish ### Description Creates a cold reactive publisher that starts the coroutine on subscribe. ### Method Coroutine builder ### Parameters - **scope** (ProducerScope) - The scope of the coroutine. ### Returns - **Publisher** - A cold reactive publisher. ``` -------------------------------- ### Override Dispatchers.Main with setMain Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-test/index.html Shows how to override Dispatchers.Main with a single-threaded context for testing UI code. Use Dispatchers.setMain in setup and Dispatchers.resetMain in teardown to restore the original dispatcher. ```kotlin class SomeTest { private val mainThreadSurrogate = newSingleThreadContext("UI thread") @Before fun setUp() { Dispatchers.setMain(mainThreadSurrogate) } @After fun tearDown() { Dispatchers.resetMain() // reset the main dispatcher to the original Main dispatcher mainThreadSurrogate.close() } @Test fun testSomeUI() = runBlocking { launch(Dispatchers.Main) { // Will be launched in the mainThreadSurrogate dispatcher // ... } } } ``` -------------------------------- ### Using multiple test dispatchers with shared scheduler Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-test/index.html Shows how to create multiple StandardTestDispatcher instances that share the same TestCoroutineScheduler via the testScheduler property, so they share virtual time. The example launches coroutines on different dispatchers and prints the virtual currentTime at each delay point. ```kotlin @Test fun testWithMultipleDispatchers() = runTest { val scheduler = testScheduler // the scheduler used for this test val dispatcher1 = StandardTestDispatcher(scheduler, name = "IO dispatcher") val dispatcher2 = StandardTestDispatcher(scheduler, name = "Background dispatcher") launch(dispatcher1) { delay(1_000) println("1. $currentTime") // 1000 delay(200) println("2. $currentTime") // 1200 delay(2_000) println("4. $currentTime") // 3200 } val deferred = async(dispatcher2) { delay(3_000) println("3. $currentTime") // 3000 delay(500) println("5. $currentTime") // 3500 } deferred.await() } ``` -------------------------------- ### runTest basic and explicit return forms Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-test/index.html Use runTest when testing coroutine-based code. The first form with expression body is recommended. The second form, returning TestResult, is needed for more advanced scenarios where you need to initialize state before starting the test, or on platforms that require the result to be returned. In both cases, the test coroutine is automatically skipped through delays and canceled after 60 seconds. ```kotlin @Test fun testFoo() = runTest { // code under test } ``` ```kotlin @Test fun testFoo(): TestResult { // initialize some test state return runTest { // code under test } } ``` -------------------------------- ### Lazy start without initial dispatch Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-l-a-z-y Shows that a LAZY coroutine with Dispatchers.Unconfined starts on the thread that calls start() and runs in-place. The example uses a separate coroutine on Dispatchers.Default to call job.start(), illustrating that the lazy coroutine executes immediately on that thread. ```kotlin // Example of lazily starting a new coroutine that doesn't go through a dispatch initially runBlocking { println("1. About to lazily start a new coroutine.") // Create a job to execute on `Dispatchers.Unconfined` later. val lazyJob = launch(Dispatchers.Unconfined, start = CoroutineStart.LAZY) { println("3. The coroutine starts on the thread that called `join`.") } // We start the job on another thread for illustrative purposes launch(Dispatchers.Default) { println("2. We start the lazyJob.") job.start() // runs lazyJob's code in-place println("4. Only now does the `start` call return.") } } ``` -------------------------------- ### ATOMIC Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-a-t-o-m-i-c ATOMIC is a coroutine start option that schedules the coroutine for execution atomically, ensuring it starts even if cancelled before execution. This example demonstrates using ATOMIC to guarantee mutex release despite cancellation. ```APIDOC ## ATOMIC ### Description Atomically (i.e., in a non-cancellable way) schedules the coroutine for execution according to its context. This is similar to DEFAULT, but the coroutine is guaranteed to start executing even if it was cancelled. This only affects the behavior until the body of the coroutine starts executing; inside the body, cancellation will work as usual. ### Usage Example ```kotlin val mutex = Mutex() mutex.lock() // lock the mutex outside the coroutine // ... // initial portion of the work, protected by the mutex val job = launch(start = CoroutineStart.ATOMIC) { // the work must continue in a coroutine, but still under the mutex println("Coroutine running!") try { // this `try` block will be entered in any case because of ATOMIC println("Starting task...") delay(10.milliseconds) // throws due to cancellation println("Finished task.") } finally { mutex.unlock() // correctly release the mutex } } job.cancelAndJoin() // we immediately cancel the coroutine. mutex.withLock { println("The lock has been returned correctly!") } ``` ### Cancellation Behavior Example ```kotlin runBlocking { println("1. Atomically starting a coroutine that goes through a dispatch.") launch(start = CoroutineStart.ATOMIC) { check(!isActive) // attempting to suspend later will throw println("4. The coroutine was cancelled (isActive = $isActive), but starts anyway.") try { delay(10.milliseconds) // will throw: the coroutine is cancelled println("This code will never run.") } catch (e: CancellationException) { println("5. Cancellation at later points still works.") throw e } } println("2. Cancelling this coroutine and all of its children.") cancel() launch(Dispatchers.Unconfined, start = CoroutineStart.ATOMIC) { check(!isActive) // attempting to suspend will throw println("3. An undispatched coroutine starts.") } ensureActive() // we can even crash the current coroutine. } ``` ### Notes This is a **delicate** API. The coroutine starts execution even if its Job is cancelled before starting. However, the resources used within a coroutine may rely on the cancellation mechanism, and cannot be used after the Job cancellation. For instance, in Android development, updating a UI element is not allowed if the coroutine's scope, which is tied to the element's lifecycle, has been cancelled. ``` -------------------------------- ### async Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core Returns a single value with the future result. It starts a coroutine and returns a Deferred that can be awaited to get the result. ```APIDOC ## async ### Description Returns a single value with the future result. ### Result Deferred ### Scope CoroutineScope ### Parameters - **block** (suspend CoroutineScope.() -> T) - Required - The coroutine code that produces a result. ### Returns Deferred - A deferred value representing the future result. ``` -------------------------------- ### onStart Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow Returns a flow that invokes the given action **before** this flow starts to be collected. ```APIDOC ## [FUNCTION] Flow.onStart ### Description Returns a flow that invokes the given action **before** this flow starts to be collected. ### Signature `fun Flow.onStart(action: suspend FlowCollector.() -> Unit): Flow` ### Parameters - **action** (suspend FlowCollector.() -> Unit) - Required - The action to invoke before collection starts. ### Returns Flow - A flow that invokes the action before collection. ``` -------------------------------- ### mapLatest cancellation example Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/map-latest.html Shows how mapLatest cancels the previous transform when a new value is emitted. The flow emits "a" and "b" with a 100ms delay, and the transform delays 200ms, so only "Computed b" is emitted while both "Started computing a" and "Started computing b" are printed. ```kotlin flow { emit("a") delay(100) emit("b") }.mapLatest { value -> println("Started computing $value") delay(200) "Computed $value" } ``` -------------------------------- ### Java API methods for image loading and combining Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-guava/index.html Java API methods based on Guava that are used in the example. The first starts asynchronous image loading, the second synchronously combines two images. ```java public ListenableFuture loadImageAsync(String name); // starts async image loading public Image combineImages(Image image1, Image image2); // synchronously combines two images using some algorithm ``` -------------------------------- ### sample Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.time Adapter method for kotlinx.coroutines.flow.sample using java.time.Duration. ```APIDOC ## sample ### Description "java.time" adapter method for kotlinx.coroutines.flow.sample. ### Signature ```kotlin @FlowPreview fun Flow.sample(period: Duration): Flow ``` ### Parameters - **period** (Duration) - The sampling period. ### Returns A Flow that emits the most recent value at each sampling period. ``` -------------------------------- ### launchIn with onEach, onCompletion, and catch Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/launch-in.html Shows the typical usage of launchIn with onEach, onCompletion, and catch to process values and handle exceptions. The returned job is not used; the provided scope handles cancellation. ```kotlin flow .onEach { value -> updateUi(value) } .onCompletion { cause -> updateUi(if (cause == null) "Done" else "Failed") } .catch { cause -> LOG.error("Exception: $cause") } .launchIn(uiScope) ``` -------------------------------- ### SharedFlow Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/index.html A hot Flow that shares emitted values among all its collectors in a broadcast fashion, so that all collectors get all emitted values. A shared flow is called hot because its active instance exists independently of the presence of collectors. This is opposed to a regular Flow, such as defined by the `flow { ... }` function, which is cold and is started separately for each collector. ```APIDOC ## SharedFlow ### Description A hot Flow that shares emitted values among all its collectors in a broadcast fashion, so that all collectors get all emitted values. A shared flow is called hot because its active instance exists independently of the presence of collectors. This is opposed to a regular Flow, such as defined by the `flow { ... }` function, which is cold and is started separately for each collector. ### Declaration ```kotlin interface SharedFlow : Flow ``` ### Type Parameters - **T** - The type of values emitted by the flow. ### Inheritors - `Flow` ``` -------------------------------- ### Provide an initial value with onStart before shareIn Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/share-in.html Emits a special initial value to signal that the upstream is still loading, then shares eagerly with a replay of one message. ```kotlin backendMessages .onStart { emit(UpstreamIsStartingMessage) } .shareIn(scope, SharingStarted.Eagerly, 1) // replay one most recent message ``` -------------------------------- ### Structured concurrency motivating example Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-scope Shows parallel downloads with async and a UI update with launch. If one download fails, the other is cancelled and the whole operation fails. coroutineScope waits for all coroutines to complete or cancel. ```kotlin suspend fun downloadFile(url: String): ByteArray { return withContext(Dispatchers.IO) { // this code will execute on a thread for blocking work val file = byteArrayOf() // download the file file } } suspend fun downloadAndCompareTwoFiles() { coroutineScope { val file1 = async { // if this fails, everything else quickly fails too downloadFile("http://example.com/file1") } val file2 = async { downloadFile("http://example.com/file2") } launch(Dispatchers.Main) { // create a separate coroutine on the UI thread if (file1.await().contentEquals(file2.await())) { uiShow("Files are equal") } else { uiShow("Files are not equal") } } } // this line will only run once all the coroutines created above // finish their work or get cancelled } ``` -------------------------------- ### Using yield() to share a single thread between coroutines Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/yield.html Shows how yield() lets two coroutines share a single-threaded dispatcher, alternating after each stage. Without yield, the first coroutine would run all five stages before the second starts. The example uses Dispatchers.Default.limitedParallelism(1) and two launch blocks that call yield() after each updateProgressBar call. ```kotlin fun updateProgressBar(value: Int, marker: String) { print(marker) } val singleThreadedDispatcher = Dispatchers.Default.limitedParallelism(1) withContext(singleThreadedDispatcher) { launch { repeat(5) { updateProgressBar(it, "A") yield() } } launch { repeat(5) { updateProgressBar(it, "B") yield() } } } ``` -------------------------------- ### CoroutineStart.ATOMIC Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-a-t-o-m-i-c/index.html Documentation for the CoroutineStart.ATOMIC option, including its behavior, usage, and an example. ```APIDOC ## CoroutineStart.ATOMIC ### Description Atomically (i.e., in a non-cancellable way) schedules the coroutine for execution according to its context. This is similar to DEFAULT, but the coroutine is guaranteed to start executing even if it was cancelled. This only affects the behavior until the body of the coroutine starts executing; inside the body, cancellation will work as usual. ### Usage Use `CoroutineStart.ATOMIC` as the `start` parameter in coroutine builders like `launch`. ### Example ```kotlin val mutex = Mutex() mutex.lock() // lock the mutex outside the coroutine // ... // initial portion of the work, protected by the mutex val job = launch(start = CoroutineStart.ATOMIC) { // the work must continue in a coroutine, but still under the mutex println("Coroutine running!") try { // this `try` block will be entered in any case because of ATOMIC println("Starting task...") delay(10.milliseconds) // throws due to cancellation println("Finished task.") } finally { mutex.unlock() // correctly release the mutex } } job.cancelAndJoin() // we immediately cancel the coroutine. mutex.withLock { println("The lock has been returned correctly!") } ``` ### Notes - This is a **delicate** API. The coroutine starts execution even if its Job is cancelled before starting. However, the resources used within a coroutine may rely on the cancellation mechanism, and cannot be used after the Job cancellation. - Unlike UNDISPATCHED, ATOMIC performs the full dispatch procedure just as DEFAULT does. - Use ATOMIC in cases where you want to be certain that some code eventually runs and uses a specific dispatcher to do that. ``` -------------------------------- ### Emit an element before flow collection with onStart Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/on-start.html Use onStart to emit an additional element before the upstream flow starts. The action is called before the upstream flow is collected, so emissions from the upstream flow that happen inside or immediately after this action are not guaranteed to be collected when used with a SharedFlow. The receiver is a FlowCollector, allowing emit calls. The example prints Begin, a, b, c. ```kotlin flowOf("a", "b", "c") .onStart { emit("Begin") } .collect { println(it) } // prints Begin, a, b, c ``` -------------------------------- ### Cancelling coroutines before they start Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-d-e-f-a-u-l-t Shows that if the coroutine's Job is cancelled before it starts executing, it will not start at all and will be considered cancelled. Both a dispatched and an unconfined launch are cancelled. ```kotlin // Example of cancelling coroutines before they start executing. runBlocking { // dispatch the job to execute on this thread later launch { // CoroutineStart.DEFAULT is the launch's default start mode println("This code will never execute") } cancel() // cancels the current coroutine scope and its children launch(Dispatchers.Unconfined) { println("This code will never execute") } println("This code will execute.") } ``` -------------------------------- ### produce usage example: producer-consumer with channel Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.channels/produce.html Shows how to use produce to generate random integers into a channel with a buffer of 16, and launch four consumers that process the integers. The producer stops when it randomly generates -1, and consumers cancel the channel when they find the square root of 9801. Note that this is an experimental API. ```kotlin /* Generate random integers until we find the square root of 9801. To calculate whether the given number is that square root, use several coroutines that separately process these integers. Alternatively, we may randomly give up during value generation. `produce` is used to generate the integers and put them into a channel, from which the square-computing coroutines take them. */ val parentScope = CoroutineScope(SupervisorJob()) val channel = parentScope.produce( Dispatchers.IO, capacity = 16 // buffer of size 16 ) { // this code will run on Dispatchers.IO while (true) { val request = run { // simulate waiting for the next request delay(5.milliseconds) val randomInt = Random.nextInt(-1, 100) if (randomInt == -1) { // external termination request received println("Producer: no longer accepting requests") return@produce } println("Producer: sending a request ($randomInt)") randomInt } send(request) } } // Launch consumers repeat(4) { launch(Dispatchers.Default) { for (request in channel) { // simulate processing a request delay(25.milliseconds) println("Consumer $it: received a request ($request)") if (request * request == 9801) { println("Consumer $it found the square root of 9801!") /* the work is done, the producer may finish. the internal termination request will cancel the producer on the next suspension point. */ channel.cancel() } } } } ``` -------------------------------- ### fun CoroutineScope.future( context: CoroutineContext = EmptyCoroutineContext, start: CoroutineStart = CoroutineStart.DEFAULT, block: suspend CoroutineScope.() -> T): ListenableFuture Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-guava/kotlinx.coroutines.guava/future.html Starts a coroutine and returns a ListenableFuture pointing to its result. The coroutine is started immediately, and cancellation is propagated bidirectionally. ```APIDOC ## fun CoroutineScope.future( context: CoroutineContext = EmptyCoroutineContext, start: CoroutineStart = CoroutineStart.DEFAULT, block: suspend CoroutineScope.() -> T): ListenableFuture ### Description Starts a coroutine and returns a ListenableFuture pointing to its result. The coroutine is started immediately, and cancellation is propagated bidirectionally. ### Parameters - **context** (CoroutineContext) - Optional - Added overlaying CoroutineScope.coroutineContext to form the new context. - **start** (CoroutineStart) - Optional - Coroutine start option. The default value is CoroutineStart.DEFAULT. - **block** (suspend CoroutineScope.() -> T) - Required - The code to execute. ### Returns - **ListenableFuture** - A future that completes with the result of the block, or with a cancellation exception if cancelled. ``` -------------------------------- ### Flow.all example with predicate and flowOf Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/all.html Shows a flow that emits 0..9 then throws, and a flowOf(1,2,3). The first `all` call returns false because 6..9 do not match `it <= 5`; the second returns true because all elements match. ```kotlin val myFlow = flow { repeat(10) { emit(it) } throw RuntimeException("You still didn't find the required number? I gave you ten!") } println(myFlow.all { it <= 5 }) // false println(flowOf(1, 2, 3).all { it <= 5 }) // true ``` -------------------------------- ### Cancelling atomically started coroutines Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-a-t-o-m-i-c/index.html Demonstrates the behavior of ATOMIC when cancellation occurs. The coroutine starts even if cancelled, but later suspension points throw CancellationException. Includes an undispatched coroutine that also starts despite cancellation. ```kotlin // Example of cancelling atomically started coroutines runBlocking { println("1. Atomically starting a coroutine that goes through a dispatch.") launch(start = CoroutineStart.ATOMIC) { check(!isActive) // attempting to suspend later will throw println("4. The coroutine was cancelled (isActive = $isActive), but starts anyway.") try { delay(10.milliseconds) // will throw: the coroutine is cancelled println("This code will never run.") } catch (e: CancellationException) { println("5. Cancellation at later points still works.") throw e } } println("2. Cancelling this coroutine and all of its children.") cancel() launch(Dispatchers.Unconfined, start = CoroutineStart.ATOMIC) { check(!isActive) // attempting to suspend will throw println("3. An undispatched coroutine starts.") } ensureActive() // we can even crash the current coroutine. } ``` -------------------------------- ### tryAcquire() Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.sync/-semaphore/try-acquire.html Documentation for the Semaphore.tryAcquire() function. ```APIDOC ## tryAcquire() ### Description Tries to acquire a permit from this semaphore without suspension. ### Method Function ### Signature `abstract fun tryAcquire(): Boolean` ### Parameters None ### Return - **Boolean** - `true` if a permit was acquired, `false` otherwise. ``` -------------------------------- ### advanceUntilIdle() Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-test/kotlinx.coroutines.test/-test-coroutine-scheduler/index.html Runs all enqueued tasks, advancing virtual time as needed until no tasks remain. ```APIDOC ## advanceUntilIdle() ### Description Runs the enqueued tasks in the specified order, advancing the virtual time as needed until there are no more tasks associated with the dispatchers linked to this scheduler. ### Function `fun advanceUntilIdle()` ### Parameters None. ``` -------------------------------- ### async Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/async.html Launches a new coroutine and returns a Deferred representing the result of the block. The coroutine is started lazily if start is set to CoroutineStart.LAZY. ```APIDOC ## async ### Description Launches a new coroutine and returns a Deferred representing the result of the block. The coroutine is started lazily if start is set to CoroutineStart.LAZY. ### Function Signature fun CoroutineScope.async(context: CoroutineContext = EmptyCoroutineContext, start: CoroutineStart = CoroutineStart.DEFAULT, block: suspend CoroutineScope.() -> T): Deferred ### Parameters - **context** (CoroutineContext) - Optional - additional to CoroutineScope.coroutineContext context of the coroutine. - **start** (CoroutineStart) - Optional - coroutine start option. The default value is CoroutineStart.DEFAULT. - **block** (suspend CoroutineScope.() -> T) - Required - the coroutine code which will be invoked in the child coroutine. ### Returns - **Deferred** - A Deferred representing the result of the block. ### Communication with the coroutine - **Deferred.await** - Suspends the current coroutine until the result is available. Returns the result of the block. Throws the exception if the coroutine fails, including CancellationException if cancelled. - **Deferred.cancel** - Cancels the coroutine. - **Deferred.join** - Suspends until completion without blocking the current thread or accessing the result. Succeeds even if the coroutine was cancelled or failed. - **Deferred.cancelAndJoin** - Convenience function that combines cancellation and joining. - **Deferred.start** - Starts the coroutine explicitly if it was started with CoroutineStart.LAZY. Awaiting or joining also causes the coroutine to start. ``` -------------------------------- ### Cancellation does not prevent UNDISPATCHED start Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-u-n-d-i-s-p-a-t-c-h-e-d Demonstrates that a cancelled scope still starts an UNDISPATCHED child, which runs until its first suspension point. ```kotlin // Cancellation does not prevent the coroutine from being started runBlocking { println("1. First, we cancel this scope.") cancel() println("2. Now, we start a new UNDISPATCHED child.") launch(start = CoroutineStart.UNDISPATCHED) { check(!isActive) // the child is already cancelled println("3. We entered the coroutine despite being cancelled.") } println("4. Execution of the outer coroutine only continues later.") } ``` -------------------------------- ### Resume with resource cleanup using onCancellation Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-cancellable-continuation/resume.html Shows how to pass a resource to `resume` and close it in the `onCancellation` handler if the value is not used. The handler must be fast, non-blocking, and thread-safe. ```kotlin continuation.resume(resourceToResumeWith) { _, resourceToClose, _ -> resourceToClose.close() } ``` -------------------------------- ### flatMapLatest flow example Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flat-map-latest.html Example of flatMapLatest switching to a new flow on each emission. The previous flow is cancelled when a new value arrives, so the output is 'a b b_last'. ```kotlin flow { emit("a") delay(100) emit("b") }.flatMapLatest { value -> flow { emit(value) delay(200) emit(value + "_last") } } ``` -------------------------------- ### Lazy start with dispatch via join Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-coroutine-start/-l-a-z-y Demonstrates that a LAZY coroutine does not start until join() is called. The coroutine is created with Dispatchers.Default but does not execute until job.join() is invoked, after which it dispatches and runs. ```kotlin // Example of lazily starting a new coroutine that goes through a dispatch runBlocking { println("1. About to start a new coroutine.") // Create a job to execute on `Dispatchers.Default` later. val job = launch(Dispatchers.Default, start = CoroutineStart.LAZY) { println("3. Only now does the coroutine start.") } delay(10.milliseconds) // try to give the coroutine some time to run println("2. The coroutine still has not started. Now, we join it.") job.join() } ``` -------------------------------- ### Flow.sample(Long) example Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/sample.html Example of using Flow.sample with a Long period in milliseconds. Emits the latest value every 200ms from a flow that emits 0-9 with 110ms delays. Output is 1, 3, 5, 7, 9. Note that the latest element is not emitted if it does not fit into the sampling window. ```kotlin flow { repeat(10) { emit(it) delay(110) } }.sample(200) ``` ```text 1, 3, 5, 7, 9 ``` -------------------------------- ### coroutineScope example: download and compare two files Source: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/coroutine-scope.html Shows how to use coroutineScope to run concurrent subtasks (async) and combine their results. If any child fails, the scope fails and rethrows the exception, but does not cancel the calling coroutine. Also demonstrates the prompt cancellation guarantee. ```kotlin // If the current coroutine is cancelled, `firstFile`, `secondFile`, // and `await()` get cancelled. suspend fun downloadAndCompareTwoFiles() = coroutineScope { val firstFile = async { // If this fails, `secondFile` and `await()` get cancelled, // and `downloadAndCompareTwoFiles` rethrows the exception, // but does not cancel the calling coroutine, // giving it a chance to recover downloadFile("1.txt") } val secondFile = async { downloadFile("2.txt") } firstFile.await().contentEquals(secondFile.await()) } ```