kotlinx.coroutines Documentation

Technical Documentation: Kotlin/kotlinx.coroutines

ℹ️ Provenance: Multi-Tier Ingestion · CodeWiki Reference

Overview (`README.md`) Copy Section

kotlinx.coroutines

Library support for Kotlin coroutines with multiplatform support.
This is a companion version for the Kotlin 2.2.20 release.

suspend fun main() = coroutineScope {
    launch { 
       delay(1.seconds)
       println("Kotlin Coroutines World!") 
    }
    println("Hello")
}

Play with coroutines online here

Modules Copy Chapter

• core — common coroutines across all platforms:
  • launch and async coroutine builders returning Job and Deferred light-weight futures with cancellation support;
  • Dispatchers object with Main dispatcher for Android/Swing/JavaFx (which require the corresponding artifacts in runtime) and Darwin (included out of the box), and Default dispatcher for background coroutines;
  • delay and yield top-level suspending functions;
  • Flow — cold asynchronous stream with flow builder and comprehensive operator set (filter, map, etc);
  • Channel, Mutex, and Semaphore communication and synchronization primitives;
  • coroutineScope, supervisorScope, withContext, and withTimeout scope builders;
  • MainScope() for Android and UI applications;
  • SupervisorJob() and CoroutineExceptionHandler for supervision of coroutines hierarchies;
  • select expression support and more.
• core/jvm — additional core features available on Kotlin/JVM:
  • Dispatchers.IO dispatcher for blocking coroutines;
  • Executor.asCoroutineDispatcher extension, custom thread pools, and more;
  • Integrations with CompletableFuture and JVM-specific extensions.
• core/web — additional core features available on Kotlin/JS and Kotlin/Wasm/JS:
  • Integration with Promise via Promise.await and promise builder.
• core/js — additional core features available on Kotlin/JS:
  • Integration with Window via Window.asCoroutineDispatcher, etc.
• test — test utilities for coroutines:
  • Dispatchers.setMain to override Dispatchers.Main in tests;
  • runTest and TestScope to test suspending functions and coroutines.
• debug — debug utilities for coroutines:
  • DebugProbes API to probe, keep track of, print and dump active coroutines;
  • CoroutinesTimeout test rule to automatically dump coroutines on test timeout.
  • Automatic integration with BlockHound.
• reactive — modules that provide builders and iteration support for various reactive streams libraries:
  • Reactive Streams (Publisher.collect, Publisher.awaitSingle, kotlinx.coroutines.reactive.publish, etc),
  • Flow (JDK 9) (the same interface as for Reactive Streams),
  • RxJava 2.x (rxFlowable, rxSingle, etc), and
  • RxJava 3.x (rxFlowable, rxSingle, etc), and
  • Project Reactor (flux, mono, etc).
• ui — modules that provide the Main dispatcher for various single-threaded UI libraries:
  • Android, JavaFX, and Swing.
• integration — modules that provide integration with various asynchronous callback- and future-based libraries:
  • Guava ListenableFuture.await, and Google Play Services Task.await;
  • SLF4J MDC integration via MDCContext.

Documentation Copy Chapter

• Presentations and videos:
  • Kotlin Coroutines in Practice (Roman Elizarov at KotlinConf 2018, slides)
  • Deep Dive into Coroutines (Roman Elizarov at KotlinConf 2017, slides)
  • History of Structured Concurrency in Coroutines (Roman Elizarov at Hydra 2019, slides)
• Guides and manuals:
  • Guide to kotlinx.coroutines by example (read it first)
  • Guide to UI programming with coroutines
  • Debugging capabilities in kotlinx.coroutines
• Compatibility policy and experimental annotations
• Change log for kotlinx.coroutines
• Coroutines design document (KEEP)
• Full kotlinx.coroutines API reference

Using in your projects Copy Chapter

Maven Copy Section

Add dependencies (you can also add other modules that you need):

<dependency>
    <groupId>org.jetbrains.kotlinx</groupId>
    <artifactId>kotlinx-coroutines-core</artifactId>
    <version>1.11.0</version>
</dependency>

And make sure that you use the latest Kotlin version:

<properties>
    <kotlin.version>2.2.20</kotlin.version>
</properties>

Gradle Copy Section

Add dependencies (you can also add other modules that you need):

dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0")
}

And make sure that you use the latest Kotlin version:

plugins {
    // For build.gradle.kts (Kotlin DSL)
    kotlin("jvm") version "2.2.20"
    
    // For build.gradle (Groovy DSL)
    id "org.jetbrains.kotlin.jvm" version "2.2.20"
}

Make sure that you have mavenCentral() in the list of repositories:

repositories {
    mavenCentral()
}

Android Copy Section

Add kotlinx-coroutines-android
module as a dependency when using kotlinx.coroutines on Android:

implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.11.0")

This gives you access to the Android Dispatchers.Main
coroutine dispatcher and also makes sure that in case of a crashed coroutine with an unhandled exception that
this exception is logged before crashing the Android application, similarly to the way uncaught exceptions in
threads are handled by the Android runtime.

R8 and ProGuard

R8 and ProGuard rules are bundled into the kotlinx-coroutines-android module.
For more details see "Optimization" section for Android.

Avoiding including the debug infrastructure in the resulting APK

The kotlinx-coroutines-core artifact contains a resource file that is not required for the coroutines to operate
normally and is only used by the debugger. To exclude it at no loss of functionality, add the following snippet to the
android block in your Gradle file for the application subproject:

packagingOptions {
    resources.excludes += "DebugProbesKt.bin"
}

Multiplatform Copy Section

Core modules of kotlinx.coroutines are also available for
Kotlin/JS and Kotlin/Native.

In common code that should get compiled for different platforms, you can add a dependency to kotlinx-coroutines-core right to the commonMain source set:

commonMain {
    dependencies {
        implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0")
    }
}

Platform-specific dependencies are recommended to be used only for non-multiplatform projects that are compiled only for target platform.

JS

Kotlin/JS version of kotlinx.coroutines is published as
kotlinx-coroutines-core-js
(follow the link to get the dependency declaration snippet).

Native

Kotlin/Native version of kotlinx.coroutines is published as
kotlinx-coroutines-core-$platform where $platform is
the target Kotlin/Native platform.
Targets are provided in accordance with official K/N target support.

Building and Contributing Copy Chapter

See Contributing Guidelines.

Overview (`dokka-templates/README.md`) Copy Section

Customize Dokka's HTML.

To customize Dokka's HTML output, place a file in this folder.
Dokka will find a template file there. If the file is not found, a default one will be used.
This folder is defined by the templatesDir property.

Overview (`integration-testing/README.md`) Copy Section

Integration tests

This is a supplementary project that provides integration tests.

The tests are the following:

• mavenTest depends on the published artifacts and tests artifacts binary content for absence of atomicfu in the classpath.
• jvmCoreTest miscellaneous tests that check the behaviour of kotlinx-coroutines-core dependency in a smoke manner.
• coreAgentTest checks that kotlinx-coroutines-core can be run as a Java agent.
• debugAgentTest checks that the coroutine debugger can be run as a Java agent.
• debugDynamicAgentTest checks that kotlinx-coroutines-debug agent can self-attach dynamically to JVM as a standalone dependency.
• debugDynamicAgentJpmsTest checks that kotlinx-coroutines-debug agent can self-attach dynamically to JVM as a standalone dependency (with JPMS)
• smokeTest builds the multiplatform test project that depends on coroutines.
• java8Test checks that some APIs built with Java 9+ can be used with Java 8.

The integration-testing project is expected to be in a subdirectory of the main kotlinx.coroutines project.

To run all the available tests: ./gradlew publishAllPublicationsToBuildLocalRepository + cd integration-testing + ./gradlew check.

Overview (`integration/README.md`) Copy Section

Coroutines integration

This directory contains modules that provide integration with various asynchronous callback- and future-based libraries.
Module names below correspond to the artifact names in Maven/Gradle.

Modules Copy Chapter

• kotlinx-coroutines-guava -- integration with Guava ListenableFuture.
• kotlinx-coroutines-slf4j -- integration with SLF4J MDC.
• kotlinx-coroutines-play-services -- integration with Google Play Services Tasks API.

Overview (`kotlinx-coroutines-core/README.md`) Copy Section

Module kotlinx-coroutines-core

Core primitives to work with coroutines.

Coroutine builder functions:

Coroutine dispatchers implementing CoroutineDispatcher:

More context elements:

Synchronization primitives for coroutines:

Top-level suspending functions:

Cancellation support for user-defined suspending functions is available with suspendCancellableCoroutine
helper function.
The NonCancellable job object is provided to suppress cancellation inside the
withContext(NonCancellable) {...} block of code.

Ways to construct asynchronous streams of values:

A cold stream is some process of generating values, and this process is performed separately for each subscriber.
A hot stream uses the same source of values independently of whether there are subscribers.

A select expression waits for the result of multiple suspending functions simultaneously:

Package kotlinx.coroutines

General-purpose coroutine builders, contexts, and helper functions.

Package kotlinx.coroutines.sync

Synchronization primitives (mutex and semaphore).

Package kotlinx.coroutines.channels

Channels — non-blocking primitives for communicating a stream of elements between coroutines.

Package kotlinx.coroutines.flow

Flow — asynchronous cold and hot streams of elements.

Package kotlinx.coroutines.selects

Select — expressions that perform multiple suspending operations simultaneously until one of them succeeds.

Package kotlinx.coroutines.intrinsics

Low-level primitives for finer-grained control of coroutines.

Package kotlinx.coroutines.future

JDK 8's CompletableFuture support.

Package kotlinx.coroutines.stream

JDK 8's Stream support.

Package kotlinx.coroutines.time

JDK 8's Duration support via additional overloads for existing time-based operators.

Overview (`kotlinx-coroutines-debug/README.md`) Copy Section

Module kotlinx-coroutines-debug

Debugging facilities for kotlinx.coroutines on JVM.

Overview Copy Section

This module provides a debug JVM agent that allows to track and trace existing coroutines.
The main entry point to debug facilities is DebugProbes API.
Call to DebugProbes.install installs debug agent via ByteBuddy and starts spying on coroutines when they are created, suspended and resumed.

After that, you can use DebugProbes.dumpCoroutines to print all active (suspended or running) coroutines, including their state, creation and
suspension stacktraces.
Additionally, it is possible to process the list of such coroutines via DebugProbes.dumpCoroutinesInfo or dump isolated parts
of coroutines hierarchy referenced by a Job or CoroutineScope instances using DebugProbes.printJob and DebugProbes.printScope respectively.

This module also provides an automatic BlockHound integration
that detects when a blocking operation was called in a coroutine context that prohibits it. In order to use it,
please follow the BlockHound quick start guide.

Using in your project Copy Section

Add kotlinx-coroutines-debug to your project test dependencies:

dependencies {
    testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-debug:1.4.0'
}

Using in unit tests Copy Section

For JUnit4 debug module provides special test rule, CoroutinesTimeout, for installing debug probes
and to dump coroutines on timeout to simplify tests debugging.

Its usage is better demonstrated by the example (runnable code is here):

class TestRuleExample {
    @get:Rule
    public val timeout = CoroutinesTimeout.seconds(1)

    private suspend fun someFunctionDeepInTheStack() {
        withContext(Dispatchers.IO) {
            delay(Long.MAX_VALUE) // Hang method
        }  
    }

    @Test
    fun hangingTest() = runBlocking {
        val job = launch {
            someFunctionDeepInTheStack()
        }
        job.join() // Join will hang
    }
}

After 1 second, test will fail with TestTimeoutException and all coroutines (runBlocking and launch) and their
stacktraces will be dumped to the console.

Using as JVM agent Copy Section

Debug module can also be used as a standalone JVM agent to enable debug probes on the application startup.
You can run your application with an additional argument: -javaagent:kotlinx-coroutines-debug-1.11.0.jar.
Additionally, on Linux and Mac OS X you can use kill -5 $pid command in order to force your application to print all alive coroutines.
When used as Java agent, "kotlinx.coroutines.debug.enable.creation.stack.trace" system property can be used to control
DebugProbes.enableCreationStackTraces along with agent startup.

Using in production environment Copy Section

It is possible to run an application in production environments with debug probes in order to monitor its
state and improve its observability.
For that, it is strongly recommended not to enable DebugProbes.enableCreationStackTraces, as enabling it makes
the performance overhead of the debug probes non-negligible.
With creation stack-traces disabled, the typical overhead of enabled debug probes is a single-digit percentage of the total
application throughput.

Example of usage Copy Section

Capabilities of this module can be demonstrated by the following example
(runnable code is here):

suspend fun computeValue(): String = coroutineScope {
    val one = async { computeOne() }
    val two = async { computeTwo() }
    combineResults(one, two)
}

suspend fun combineResults(one: Deferred<String>, two: Deferred<String>): 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)
}

Printed result will be:

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)

Status of the API Copy Section

API is experimental, and it is not guaranteed it won't be changed (while it is marked as @ExperimentalCoroutinesApi).
Like the rest of experimental API, DebugProbes is carefully designed, tested and ready to use in both test and production
environments. It is marked as experimental to leave us the room to enrich the output data in a potentially backwards incompatible manner
to further improve diagnostics and debugging experience.

The output format of DebugProbes can be changed in the future and it is not recommended to rely on the string representation
of the dump programmatically.

Debug agent and Android Copy Section

Android runtime does not support Instrument API necessary for kotlinx-coroutines-debug to function, triggering java.lang.NoClassDefFoundError: Failed resolution of: Ljava/lang/management/ManagementFactory;,
and it is not possible to use coroutine debugger along with Android emulator.

Build failures due to duplicate resource files

Building an Android project that depends on kotlinx-coroutines-debug (usually introduced by being a transitive
dependency of kotlinx-coroutines-test) may fail with DuplicateRelativeFileException for META-INF/AL2.0,
META-INF/LGPL2.1, or win32-x86/attach_hotspot_windows.dll when trying to merge the Android resource.

The problem is that Android merges the resources of all its dependencies into a single directory and complains about
conflicts, but:
kotlinx-coroutines-debug transitively depends on JNA and JNA-platform, byte-buddy and byte-buddy-agent, all of them include license files in their
META-INF directories. Trying to merge these files leads to conflicts, which means that any Android project that
depends on JNA and JNA-platform will experience build failures.

One possible workaround for these issues is to add the following to the android block in your gradle file for the
application subproject:

packagingOptions {
         // for JNA and JNA-platform
         exclude "META-INF/AL2.0"
         exclude "META-INF/LGPL2.1"
         // for byte-buddy
         exclude "META-INF/licenses/ASM"
         pickFirst "win32-x86-64/attach_hotspot_windows.dll"
         pickFirst "win32-x86/attach_hotspot_windows.dll"
     }

This will cause the resource merge algorithm to exclude the problematic license files altogether and only leave a single
copy of the files needed for byte-buddy-agent to work.

Alternatively, avoid depending on kotlinx-coroutines-debug. In particular, if the only reason why this library a
dependency of your project is that kotlinx-coroutines-test in turn depends on it, you may change your dependency on
kotlinx.coroutines.test to exclude kotlinx-coroutines-debug. For example, you could replace

androidTestImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:$coroutines_version")

with

androidTestImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:$coroutines_version") {
    exclude group: "org.jetbrains.kotlinx", module: "kotlinx-coroutines-debug"
}

Overview (`kotlinx-coroutines-test/README.md`) Copy Section

Module kotlinx-coroutines-test

Test utilities for kotlinx.coroutines.

Overview Copy Chapter

This package provides utilities for efficiently testing coroutines.

Provided TestDispatcher implementations:

Using in your project Copy Chapter

Add kotlinx-coroutines-test to your project test dependencies:

dependencies {
    testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-test:1.11.0'
}

Do not depend on this project in your main sources, all utilities here are intended and designed to be used only from tests.

Dispatchers.Main Delegation Copy Chapter

Dispatchers.setMain will override the Main dispatcher in test scenarios.
This is helpful when one wants to execute a test in situations where the platform Main dispatcher is not available,
or to replace Dispatchers.Main with a testing dispatcher.

On the JVM,
the ServiceLoader mechanism is responsible
for overwriting Dispatchers.Main with a testable implementation, which by default will delegate its calls to the real
Main dispatcher, if any.

The Main implementation can be overridden using Dispatchers.setMain method with any CoroutineDispatcher
implementation, e.g.:

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
            // ...
        }
    }
}

Calling setMain or resetMain immediately changes the Main dispatcher globally.

If Main is overridden with a TestDispatcher, then its TestCoroutineScheduler is used when new TestDispatcher or
TestScope instances are created without TestCoroutineScheduler being passed as an argument.

runTest Copy Chapter

runTest is the way to test code that involves coroutines. suspend functions can be called inside it.

IMPORTANT: in order to work with on Kotlin/JS, the result of runTest must be immediately return-ed from each test.
The typical invocation of runTest thus looks like this:

@Test
fun testFoo() = runTest {
    // code under test
}

In more advanced scenarios, it's possible instead to use the following form:

@Test
fun testFoo(): TestResult {
    // initialize some test state
    return runTest {
        // code under test
    }
}

runTest is similar to running the code with runBlocking on Kotlin/JVM and Kotlin/Native, or launching a new promise
on Kotlin/JS. The main differences are the following:

• The calls to delay are automatically skipped, preserving the relative execution order of the tasks. This way,
  it's possible to make tests finish more-or-less immediately.
• The execution times out after 60 seconds, cancelling the test coroutine to prevent tests from hanging forever
  and eating up the CI resources.
• Controlling the virtual time: in case just skipping delays is not sufficient, it's possible to more carefully
  guide the execution, advancing the virtual time by a duration, draining the queue of the awaiting tasks, or running
  the tasks scheduled at the present moment.
• Handling uncaught exceptions spawned in the child coroutines by throwing them at the end of the test.
• Waiting for asynchronous callbacks.
  Sometimes, especially when working with third-party code, it's impossible to mock all the dispatchers in use.
  runTest will handle the situations where some code runs in dispatchers not integrated with the test module.

Timeout Copy Chapter

Test automatically time out after 60 seconds. For example, this test will fail with a timeout exception:

@Test
fun testHanging() = runTest {
    CompletableDeferred<Unit>().await() // will hang forever
}

In case the test is expected to take longer than 60 seconds, the timeout can be increased by passing the timeout
parameter:

@Test
fun testTakingALongTime() = runTest(timeout = 30.seconds) {
    val result = withContext(Dispatchers.Default) {
        delay(20.seconds) // this delay is not in the test dispatcher and will not be skipped
        3
    }
    assertEquals(3, result)
}

Delay-skipping Copy Chapter

To test regular suspend functions, which may have a delay, just run them inside the runTest block.

@Test
fun testFoo() = runTest { // a coroutine with an extra test control
    val actual = foo()
    // ...
}

suspend fun foo() {
    delay(1_000) // when run in `runTest`, will finish immediately instead of delaying
    // ...
}

launch and async Copy Chapter

The coroutine dispatcher used for tests is single-threaded, meaning that the child coroutines of the runTest block
will run on the thread that started the test, and will never run in parallel.

If several coroutines are waiting to be executed next, the one scheduled after the smallest delay will be chosen.
The virtual time will automatically advance to the point of its resumption.

@Test
fun testWithMultipleDelays() = runTest {
    launch {
        delay(1_000)
        println("1. $currentTime") // 1000
        delay(200)
        println("2. $currentTime") // 1200
        delay(2_000)
        println("4. $currentTime") // 3200
    }
    val deferred = async {
        delay(3_000)
        println("3. $currentTime") // 3000
        delay(500)
        println("5. $currentTime") // 3500
    }
    deferred.await()
}

Controlling the virtual time Copy Chapter

Inside runTest, the execution is scheduled by TestCoroutineScheduler, which is a virtual time scheduler.
The scheduler has several special methods that allow controlling the virtual time:

• currentTime gets the current virtual time.
• runCurrent() runs the tasks that are scheduled at this point of virtual time.
• advanceUntilIdle() runs all enqueued tasks until there are no more.
• advanceTimeBy(timeDelta) runs the enqueued tasks until the current virtual time advances by timeDelta.
• timeSource returns a TimeSource that uses the virtual time.

@Test
fun testFoo() = runTest {
    launch {
        val workDuration = testScheduler.timeSource.measureTime {
            println(1)   // executes during runCurrent()
            delay(1_000) // suspends until time is advanced by at least 1_000
            println(2)   // executes during advanceTimeBy(2_000)
            delay(500)   // suspends until the time is advanced by another 500 ms
            println(3)   // also executes during advanceTimeBy(2_000)
            delay(5_000) // will suspend by another 4_500 ms
            println(4)   // executes during advanceUntilIdle()
        }
        assertEquals(6500.milliseconds, workDuration) // the work took 6_500 ms of virtual time
    }
    // the child coroutine has not run yet
    testScheduler.runCurrent()
    // the child coroutine has called println(1), and is suspended on delay(1_000)
    testScheduler.advanceTimeBy(2.seconds) // progress time, this will cause two calls to `delay` to resume
    // the child coroutine has called println(2) and println(3) and suspends for another 4_500 virtual milliseconds
    testScheduler.advanceUntilIdle() // will run the child coroutine to completion
    assertEquals(6500, currentTime) // the child coroutine finished at virtual time of 6_500 milliseconds
}

Using multiple test dispatchers Copy Chapter

The virtual time is controlled by an entity called the TestCoroutineScheduler, which behaves as the shared source of
virtual time.

Several dispatchers can be created that use the same TestCoroutineScheduler, in which case they will share their
knowledge of the virtual time.

To access the scheduler used for this test, use the TestScope.testScheduler property.

@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()
    }

Note: if Dispatchers.Main is replaced by a TestDispatcher, runTest will automatically use its scheduler.
This is done so that there is no need to go through the ceremony of passing the correct scheduler to runTest.

Accessing the test coroutine scope Copy Chapter

Structured concurrency ties coroutines to scopes in which they are launched.
TestScope is a special coroutine scope designed for testing coroutines, and a new one is automatically created
for runTest and used as the receiver for the test body.

However, it can be convenient to access a CoroutineScope before the test has started, for example, to perform mocking
of some
parts of the system in @BeforeTest via dependency injection.
In these cases, it is possible to manually create TestScope, the scope for the test coroutines, in advance,
before the test begins.

TestScope on its own does not automatically run the code launched in it.
In addition, it is stateful in order to keep track of executing coroutines and uncaught exceptions.
Therefore, it is important to ensure that TestScope.runTest is called eventually.

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`
}

Running background work Copy Chapter

Sometimes, the fact that runTest waits for all the coroutines to finish is undesired.
For example, the system under test may need to receive data from coroutines that always run in the background.
Emulating such coroutines by launching them from the test body is not sufficient, because runTest will wait for them
to finish, which they never typically do.

For these cases, there is a special coroutine scope: TestScope.backgroundScope.
Coroutines launched in it will be cancelled at the end of the test.

@Test
fun testExampleBackgroundJob() = runTest {
  val channel = Channel<Int>()
  backgroundScope.launch {
    var i = 0
    while (true) {
      channel.send(i++)
    }
  }
  repeat(100) {
    assertEquals(it, channel.receive())
  }
}

Eagerly entering launch and async blocks Copy Chapter

Some tests only test functionality and don't particularly care about the precise order in which coroutines are
dispatched.
In these cases, it can be cumbersome to always call runCurrent or yield to observe the effects of the coroutines
after they are launched.

If runTest executes with an UnconfinedTestDispatcher, the child coroutines launched at the top level are entered
eagerly, that is, they don't go through a dispatch until the first suspension.

@Test
fun testEagerlyEnteringChildCoroutines() = runTest(UnconfinedTestDispatcher()) {
    var entered = false
    val deferred = CompletableDeferred<Unit>()
    var completed = false
    launch {
        entered = true
        deferred.await()
        completed = true
    }
    assertTrue(entered) // `entered = true` already executed.
    assertFalse(completed) // however, the child coroutine then suspended, so it is enqueued.
    deferred.complete(Unit) // resume the coroutine.
    assertTrue(completed) // now the child coroutine is immediately completed.
}

If this behavior is desirable, but some parts of the test still require accurate dispatching, for example, to ensure
that the code executes on the correct thread, then simply launch a new coroutine with the StandardTestDispatcher.

@Test
fun testEagerlyEnteringSomeChildCoroutines() = runTest(UnconfinedTestDispatcher()) {
    var entered1 = false
    launch {
        entered1 = true
    }
    assertTrue(entered1) // `entered1 = true` already executed

    var entered2 = false
    launch(StandardTestDispatcher(testScheduler)) {
        // this block and every coroutine launched inside it will explicitly go through the needed dispatches
        entered2 = true
    }
    assertFalse(entered2)
    runCurrent() // need to explicitly run the dispatched continuation
    assertTrue(entered2)
}

Using withTimeout inside runTest Copy Section

Timeouts are also susceptible to time control, so the code below will immediately finish.

@Test
fun testFooWithTimeout() = runTest {
    assertFailsWith<TimeoutCancellationException> {
        withTimeout(1_000) {
            delay(999)
            delay(2)
            println("this won't be reached")
        }
    }
}

Virtual time support with other dispatchers Copy Chapter

Calls to withContext(Dispatchers.IO), withContext(Dispatchers.Default) ,and withContext(Dispatchers.Main) are
common in coroutines-based code bases. Unfortunately, just executing code in a test will not lead to these dispatchers
using the virtual time source, so delays will not be skipped in them.

suspend fun veryExpensiveFunction() = withContext(Dispatchers.Default) {
    delay(1_000)
    1
}

fun testExpensiveFunction() = runTest {
    val result = veryExpensiveFunction() // will take a whole real-time second to execute
    // the virtual time at this point is still 0
}

Tests should, when possible, replace these dispatchers with a TestDispatcher if the withContext calls delay in the
function under test. For example, veryExpensiveFunction above should allow mocking with a TestDispatcher using
either dependency injection, a service locator, or a default parameter, if it is to be used with virtual time.

Status of the API Copy Section

Many parts of the API is experimental, and it is may change before migrating out of experimental (while it is marked as
@ExperimentalCoroutinesApi).
Changes during experimental may have deprecation applied when possible, but it is not
advised to use the API in stable code before it leaves experimental due to possible breaking changes.

If you have any suggestions for improvements to this experimental API please share them on the
issue tracker.

Overview (`reactive/README.md`) Copy Section

Coroutines for reactive streams

This directory contains modules with utilities for various reactive stream libraries.
Module name below corresponds to the artifact name in Maven/Gradle.

Modules Copy Chapter

• kotlinx-coroutines-reactive -- utilities for Reactive Streams
• kotlinx-coroutines-reactor -- utilities for Reactor
• kotlinx-coroutines-rx2 -- utilities for RxJava 2.x
• kotlinx-coroutines-rx3 -- utilities for RxJava 3.x

Overview (`ui/README.md`) Copy Section

Coroutines for UI

This directory contains modules for coroutine programming with various single-threaded UI libraries.
After adding dependency to the UI library, corresponding UI dispatcher will be available via Dispatchers.Main.
Module name below corresponds to the artifact name in Maven/Gradle.

Modules Copy Chapter

• kotlinx-coroutines-android -- Dispatchers.Main context for Android applications.
• kotlinx-coroutines-javafx -- Dispatchers.JavaFx context for JavaFX UI applications.
• kotlinx-coroutines-swing -- Dispatchers.Swing context for Swing UI applications.

Overview (`integration/kotlinx-coroutines-guava/README.md`) Copy Section

Module kotlinx-coroutines-guava

Integration with Guava ListenableFuture.

Coroutine builders:

Extension functions:

Example Copy Chapter

Given the following functions defined in some Java API based on Guava:

public ListenableFuture<Image> loadImageAsync(String name); // starts async image loading
public Image combineImages(Image image1, Image image2); // synchronously combines two images using some algorithm

We can consume this API from Kotlin coroutine to load two images and combine then asynchronously.
The resulting function returns ListenableFuture for ease of use back from Guava-based Java code.

fun combineImagesAsync(name1: String, name2: String): ListenableFuture<Image> = future {
    val future1 = loadImageAsync(name1) // start loading first image
    val future2 = loadImageAsync(name2) // start loading second image
    combineImages(future1.await(), future2.await()) // wait for both, combine, and return result
}

Note that this module should be used only for integration with existing Java APIs based on ListenableFuture.
Writing pure-Kotlin code that uses ListenableFuture is highly not recommended, since the resulting APIs based
on the futures are quite error-prone. See the discussion on
Asynchronous Programming Styles
for details on general problems pertaining to any future-based API and keep in mind that ListenableFuture exposes
a blocking method
get
that makes it especially bad choice for coroutine-based Kotlin code.

Package kotlinx.coroutines.future

Integration with Guava ListenableFuture.

Overview (`integration/kotlinx-coroutines-jdk8/README.md`) Copy Section

Stub module

Stub module for backwards compatibility. Since 1.7.0, this module was merged with core.

Overview (`integration/kotlinx-coroutines-play-services/README.md`) Copy Section

Module kotlinx-coroutines-play-services

Integration with Google Play Services Tasks API.

Extension functions:

Example Copy Chapter

Using Firebase APIs becomes simple:

FirebaseAuth.getInstance().signInAnonymously().await()
val snapshot = try {
    FirebaseFirestore.getInstance().document("users/$id").get().await() // Cancellable await
} catch (e: FirebaseFirestoreException) {
    // Handle exception
    return@async
}

// Do stuff

If the Task supports cancellation via passing a CancellationToken, pass the corresponding CancellationTokenSource to asDeferred or await to support bi-directional cancellation:

val cancellationTokenSource = CancellationTokenSource()
val currentLocationTask = fusedLocationProviderClient.getCurrentLocation(PRIORITY_HIGH_ACCURACY, cancellationTokenSource.token)
val currentLocation = currentLocationTask.await(cancellationTokenSource) // cancelling `await` also cancels `currentLocationTask`, and vice versa

Overview (`integration/kotlinx-coroutines-slf4j/README.md`) Copy Section

Module kotlinx-coroutines-slf4j

Integration with SLF4J MDC.

Example Copy Chapter

Add MDCContext to the coroutine context so that the SLF4J MDC context is captured and passed into the coroutine.

MDC.put("kotlin", "rocks") // put a value into the MDC context

launch(MDCContext()) {
   logger.info { "..." }   // the MDC context will contain the mapping here
}

Package kotlinx.coroutines.slf4j

Integration with SLF4J MDC.

Overview (`kotlinx-coroutines-core/benchmarks/README.md`) Copy Section

kotlinx-coroutines-core benchmarks Copy Chapter

Multiplatform benchmarks for kotlinx-coroutines-core.

This source-set contains benchmarks that leverage internal API (e.g. suspendCancellableCoroutineReusable) or
that are multiplatform (-> only supported with kotlinx-benchmarks which is less convenient than jmh plugin).
For JVM-only non-internal benchmarks, consider using benchmarks top-level project.

Usage Copy Section

// JVM only
./gradlew :kotlinx-coroutines-core:jvmBenchmarkBenchmarkJar
java -jar kotlinx-coroutines-core/build/benchmarks/jvmBenchmark/jars/kotlinx-coroutines-core-jvmBenchmark-jmh-*-JMH.jar

// Native, OS X
./gradlew :kotlinx-coroutines-core:macosArm64BenchmarkBenchmark

// Figure out what to use
./gradlew :kotlinx-coroutines-core:tasks | grep -i bench

Overview (`kotlinx-coroutines-core/common/README.md`) Copy Section

Module kotlinx-coroutines-core

Core primitives to work with coroutines available on all platforms.

Coroutine builder functions:

Coroutine dispatchers implementing CoroutineDispatcher:

More context elements:

Synchronization primitives for coroutines:

Top-level suspending functions:

Cancellation support for user-defined suspending functions is available with suspendCancellableCoroutine
helper function. NonCancellable job object is provided to suppress cancellation with
withContext(NonCancellable) {...} block of code.

Select expression waits for the result of multiple suspending functions simultaneously:

This module provides debugging facilities for coroutines (run JVM with -ea or -Dkotlinx.coroutines.debug options)
and newCoroutineContext function to write user-defined coroutine builders that work with these
debugging facilities. See DEBUG_PROPERTY_NAME for more details.

Package kotlinx.coroutines

General-purpose coroutine builders, contexts, and helper functions.

Package kotlinx.coroutines.flow

Flow -- primitive to work with asynchronous and event-based streams of data.

Package kotlinx.coroutines.sync

Synchronization primitives (mutex).

Package kotlinx.coroutines.channels

Channels -- non-blocking primitives for communicating a stream of elements between coroutines.

Package kotlinx.coroutines.selects

Select expression to perform multiple suspending operations simultaneously until one of them succeeds.

Package kotlinx.coroutines.intrinsics

Low-level primitives for finer-grained control of coroutines.

Overview (`kotlinx-coroutines-test/npm/README.md`) Copy Section

kotlinx-coroutines-test

Testing support for kotlinx-coroutines in
Kotlin/JS.

Overview (`reactive/kotlinx-coroutines-jdk9/README.md`) Copy Section

Module kotlinx-coroutines-jdk9

Utilities for Java Flow.

Implemented as a collection of thin wrappers over kotlinx-coroutines-reactive,
an equivalent package for the Reactive Streams.

Package kotlinx.coroutines.jdk9

Utilities for Java Flow.

Overview (`reactive/kotlinx-coroutines-reactive/README.md`) Copy Section

Module kotlinx-coroutines-reactive

Utilities for Reactive Streams.

Coroutine builders:

Integration with Flow:

If these adapters are used along with kotlinx-coroutines-reactor in the classpath, then Reactor's Context is properly
propagated as coroutine context element (ReactorContext) and vice versa.

Suspending extension functions and suspending iteration:

Package kotlinx.coroutines.reactive

Utilities for Reactive Streams.

Overview (`reactive/kotlinx-coroutines-reactor/README.md`) Copy Section

Module kotlinx-coroutines-reactor

Utilities for Reactor.

Coroutine builders:

Note that Mono and Flux are subclasses of Reactive Streams'
Publisher and extensions for it are covered by the
kotlinx-coroutines-reactive module.

Integration with Flow:

This adapter is integrated with Reactor's Context and coroutines' ReactorContext.

Conversion functions:

Package kotlinx.coroutines.reactor

Utilities for Reactor.

Overview (`reactive/kotlinx-coroutines-rx2/README.md`) Copy Section

Module kotlinx-coroutines-rx2

Utilities for RxJava 2.x.

Coroutine builders:

Integration with Flow:

Suspending extension functions and suspending iteration:

Note that Flowable is a subclass of Reactive Streams
Publisher and extensions for it are covered by
kotlinx-coroutines-reactive module.

Conversion functions:

Package kotlinx.coroutines.rx2

Utilities for RxJava 2.x.

Overview (`reactive/kotlinx-coroutines-rx3/README.md`) Copy Section

Module kotlinx-coroutines-rx3

Utilities for RxJava 3.x.

Coroutine builders:

Integration with Flow:

Suspending extension functions and suspending iteration:

Note that Flowable is a subclass of Reactive Streams
Publisher and extensions for it are covered by
kotlinx-coroutines-reactive module.

Conversion functions:

Package kotlinx.coroutines.rx3

Utilities for RxJava 3.x.

Overview (`ui/kotlinx-coroutines-android/README.md`) Copy Section

Module kotlinx-coroutines-android

Provides Dispatchers.Main context for Android applications.

Read Guide to UI programming with coroutines
for tutorial on this module.

Optimization

R8 and ProGuard rules are bundled into this module.
R8 is a replacement for ProGuard in Android ecosystem, it is enabled by default since Android gradle plugin 3.4.0
(3.3.0-beta also had it enabled).
For best results it is recommended to use a recent version of R8, which produces a smaller binary.

When optimizations are enabled with R8 version 1.6.0 or later
the following debugging features are permanently turned off to reduce the size of the resulting binary:

• Debugging mode
• Stacktrace recovery
• The internal assertions in the library are also permanently removed.

You can examine the corresponding rules in this
coroutines.pro file.

Package kotlinx.coroutines.android

Provides Dispatchers.Main context for Android applications.

Overview (`ui/kotlinx-coroutines-javafx/README.md`) Copy Section

Module kotlinx-coroutines-javafx

Provides Dispatchers.JavaFx context and Dispatchers.Main implementation for JavaFX UI applications.

Read Guide to UI programming with coroutines
for tutorial on this module.

Package kotlinx.coroutines.javafx

Provides Dispatchers.JavaFx context and Dispatchers.Main implementation for JavaFX UI applications.

Overview (`ui/kotlinx-coroutines-swing/README.md`) Copy Section

Module kotlinx-coroutines-swing

Provides Dispatchers.Swing context and Dispatchers.Main implementation for Swing UI applications.

Read Guide to UI programming with coroutines
for tutorial on this module.

Package kotlinx.coroutines.swing

Provides Dispatchers.Swing context and Dispatchers.Main implementation for Swing UI applications.

Overview (`benchmarks/src/jmh/kotlin/benchmarks/flow/scrabble/README.md`) Copy Section

Reactive scrabble benchmarks Copy Chapter

This package contains reactive scrabble benchmarks.

Reactive Scrabble benchmarks were originally developed by José Paumard and are available under Apache 2.0,
Flow version is adaptation of this work.
All Rx and Reactive benchmarks are based on (or copied from) David Karnok work.

Benchmark classes Copy Section

The package (split into two sourcesets, kotlin and java), contains different benchmarks with different purposes

• RxJava2PlaysScrabble and RxJava2PlaysScrabbleOpt are copied as is and used for comparison. The infrastructure (e.g. FlowableSplit)
  is copied from akarnokd-misc in order for the latter benchmark to work.
  This is the original benchmark for RxJava.
• ReactorPlaysScrabble is an original benchmark for Reactor, but rewritten into Kotlin.
  It is disabled by default and had the only purpose -- verify that Kotlin version performs as the original Java version
  (which could have been different due to lambdas translation, implicit boxing, etc.). It is disabled because
  it has almost no difference compared to RxJava benchmark.
• FlowPlaysScrabbleBase is a scrabble benchmark rewritten on top of the Flow API without using any optimizations or tricky internals.
• FlowPlaysScrabbleOpt is an optimized version of benchmark that follows the same guidelines as RxJava2PlaysScrabbleOpt: it still is
  lazy, reactive and uses only Flow abstraction.
• SequencePlaysScrabble is a version of benchmark built on top of Sequence without suspensions, used as a lower bound.
• SaneFlowPlaysScrabble is a SequencePlaysScrabble that produces Flow.
  This benchmark is not identical (in terms of functions pipelining) to FlowPlaysScrabbleOpt, but rather is used as a lower bound of Flow performance
  on this particular task.

Results Copy Section

Benchmark results for throughput mode, Java 1.8.0_172
running on Intel(R) Core(TM) i9-9880H CPU @ 2.30GHz
under Darwin Kernel Version 18.7.0.
Full command: java -jar benchmarks.jar -f 2 -jvmArgsPrepend "-XX:+UseParallelGC" '.*Scrabble.*'.

Benchmark                     Mode  Cnt   Score   Error  Units  

FlowPlaysScrabbleBase.play    avgt   14  62.480 ± 1.018  ms/op
FlowPlaysScrabbleOpt.play     avgt   14  13.958 ± 0.278  ms/op

RxJava2PlaysScrabble.play     avgt   14  88.456 ± 0.950  ms/op
RxJava2PlaysScrabbleOpt.play  avgt   14  23.653 ± 0.379  ms/op

SaneFlowPlaysScrabble.play    avgt   14  13.608 ± 0.332  ms/op
SequencePlaysScrabble.play    avgt   14   9.824 ± 0.190  ms/op

Basics (`docs/basics.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/coroutines-basics.html page.

To edit the documentation, open the topics/coroutines-basics.md page.

Cancellation And Timeouts (`docs/cancellation-and-timeouts.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/coroutines-cancellation.html page.

To edit the documentation, open the topics/coroutines-cancellation.md page.

Channels (`docs/channels.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/channels.html page.

To edit the documentation, open the topics/channels.md page.

Compatibility (`docs/compatibility.md`) Copy Section

The documentation has been moved to the topics/compatibility.md.

Composing Suspending Functions (`docs/composing-suspending-functions.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/composing-suspending-functions.html page.

To edit the documentation, open the topics/composing-suspending-functions.md page.

Coroutine Context And Dispatchers (`docs/coroutine-context-and-dispatchers.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/coroutine-context-and-dispatchers.html page.

To edit the documentation, open the topics/coroutine-context-and-dispatchers.md page.

Coroutines Guide (`docs/coroutines-guide.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/coroutines-guide.html page.

To edit the documentation, open the topics/coroutines-guide.md page.

Debugging (`docs/debugging.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/coroutines-debugging.html page.

Exception Handling (`docs/exception-handling.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/exception-handling.html page.

To edit the documentation, open the topics/exception-handling.md page.

Flow (`docs/flow.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/coroutines-flow.html page.

To edit the documentation, open the topics/coroutines-flow.md page.

Select Expression (`docs/select-expression.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/select-expression.html page.

To edit the documentation, open the topics/select-expression.md page.

Shared Mutable State And Concurrency (`docs/shared-mutable-state-and-concurrency.md`) Copy Section

The documentation has been moved to the https://kotlinlang.org/docs/shared-mutable-state-and-concurrency.html page.

To edit the documentation, open the topics/shared-mutable-state-and-concurrency.md page.

Topics Channels (`docs/topics/channels.md`) Copy Section

https://github.com/Kotlin/kotlinx.coroutines/edit/master/docs/topics/

Deferred values provide a convenient way to transfer a single value between coroutines.
Channels provide a way to transfer a stream of values.

Channel basics Copy Chapter

A Channel is conceptually very similar to BlockingQueue. One key difference is that
instead of a blocking put operation it has a suspending send, and instead of
a blocking take operation it has a suspending receive.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
//sampleStart
    val channel = Channel<Int>()
    launch {
        // this might be heavy CPU-consuming computation or async logic, 
        // we'll just send five squares
        for (x in 1..5) channel.send(x * x)
    }
    // here we print five received integers:
    repeat(5) { println(channel.receive()) }
    println("Done!")
//sampleEnd
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The output of this code is:

1
4
9
16
25
Done!

Closing and iteration over channels Copy Chapter

Unlike a queue, a channel can be closed to indicate that no more elements are coming.
On the receiver side it is convenient to use a regular for loop to receive elements
from the channel.

Conceptually, a close is like sending a special close token to the channel.
The iteration stops as soon as this close token is received, so there is a guarantee
that all previously sent elements before the close are received:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
//sampleStart
    val channel = Channel<Int>()
    launch {
        for (x in 1..5) channel.send(x * x)
        channel.close() // we're done sending
    }
    // here we print received values using `for` loop (until the channel is closed)
    for (y in channel) println(y)
    println("Done!")
//sampleEnd
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

Building channel producers Copy Chapter

The pattern where a coroutine is producing a sequence of elements is quite common.
This is a part of producer-consumer pattern that is often found in concurrent code.
You could abstract such a producer into a function that takes channel as its parameter, but this goes contrary
to common sense that results must be returned from functions.

There is a convenient coroutine builder named produce that makes it easy to do it right on producer side,
and an extension function consumeEach, that replaces a for loop on the consumer side:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

//sampleStart
fun CoroutineScope.produceSquares(): ReceiveChannel<Int> = produce {
    for (x in 1..5) send(x * x)
}

fun main() = runBlocking {
    val squares = produceSquares()
    squares.consumeEach { println(it) }
    println("Done!")
//sampleEnd
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

Pipelines Copy Chapter

A pipeline is a pattern where one coroutine is producing, possibly infinite, stream of values:

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) send(x++) // infinite stream of integers starting from 1
}

And another coroutine or coroutines are consuming that stream, doing some processing, and producing some other results.
In the example below, the numbers are just squared:

fun CoroutineScope.square(numbers: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (x in numbers) send(x * x)
}

The main code starts and connects the whole pipeline:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
//sampleStart
    val numbers = produceNumbers() // produces integers from 1 and on
    val squares = square(numbers) // squares integers
    repeat(5) {
        println(squares.receive()) // print first five
    }
    println("Done!") // we are done
    coroutineContext.cancelChildren() // cancel children coroutines
//sampleEnd
}

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) send(x++) // infinite stream of integers starting from 1
}

fun CoroutineScope.square(numbers: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (x in numbers) send(x * x)
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

All functions that create coroutines are defined as extensions on CoroutineScope,
so that we can rely on structured concurrency to make
sure that we don't have lingering global coroutines in our application.

{style="note"}

Prime numbers with pipeline Copy Chapter

Let's take pipelines to the extreme with an example that generates prime numbers using a pipeline
of coroutines. We start with an infinite sequence of numbers.

fun CoroutineScope.numbersFrom(start: Int) = produce<Int> {
    var x = start
    while (true) send(x++) // infinite stream of integers from start
}

The following pipeline stage filters an incoming stream of numbers, removing all the numbers
that are divisible by the given prime number:

fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime: Int) = produce<Int> {
    for (x in numbers) if (x % prime != 0) send(x)
}

Now we build our pipeline by starting a stream of numbers from 2, taking a prime number from the current channel,
and launching new pipeline stage for each prime number found:

numbersFrom(2) -> filter(2) -> filter(3) -> filter(5) -> filter(7) ...

The following example prints the first ten prime numbers,
running the whole pipeline in the context of the main thread. Since all the coroutines are launched in
the scope of the main runBlocking coroutine
we don't have to keep an explicit list of all the coroutines we have started.
We use cancelChildren
extension function to cancel all the children coroutines after we have printed
the first ten prime numbers.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
//sampleStart
    var cur = numbersFrom(2)
    repeat(10) {
        val prime = cur.receive()
        println(prime)
        cur = filter(cur, prime)
    }
    coroutineContext.cancelChildren() // cancel all children to let main finish
//sampleEnd    
}

fun CoroutineScope.numbersFrom(start: Int) = produce<Int> {
    var x = start
    while (true) send(x++) // infinite stream of integers from start
}

fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime: Int) = produce<Int> {
    for (x in numbers) if (x % prime != 0) send(x)
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The output of this code is:

2
3
5
7
11
13
17
19
23
29

Note that you can build the same pipeline using
iterator
coroutine builder from the standard library.
Replace produce with iterator, send with yield, receive with next,
ReceiveChannel with Iterator, and get rid of the coroutine scope. You will not need runBlocking either.
However, the benefit of a pipeline that uses channels as shown above is that it can actually use
multiple CPU cores if you run it in Dispatchers.Default context.

Anyway, this is an extremely impractical way to find prime numbers. In practice, pipelines do involve some
other suspending invocations (like asynchronous calls to remote services) and these pipelines cannot be
built using sequence/iterator, because they do not allow arbitrary suspension, unlike
produce, which is fully asynchronous.

Fan-out Copy Chapter

Multiple coroutines may receive from the same channel, distributing work between themselves.
Let us start with a producer coroutine that is periodically producing integers
(ten numbers per second):

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1 // start from 1
    while (true) {
        send(x++) // produce next
        delay(100) // wait 0.1s
    }
}

Then we can have several processor coroutines. In this example, they just print their id and
received number:

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
    for (msg in channel) {
        println("Processor #$id received $msg")
    }    
}

Now let us launch five processors and let them work for almost a second. See what happens:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking<Unit> {
//sampleStart
    val producer = produceNumbers()
    repeat(5) { launchProcessor(it, producer) }
    delay(950)
    producer.cancel() // cancel producer coroutine and thus kill them all
//sampleEnd
}

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1 // start from 1
    while (true) {
        send(x++) // produce next
        delay(100) // wait 0.1s
    }
}

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
    for (msg in channel) {
        println("Processor #$id received $msg")
    }    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The output will be similar to the following one, albeit the processor ids that receive
each specific integer may be different:

Processor #2 received 1
Processor #4 received 2
Processor #0 received 3
Processor #1 received 4
Processor #3 received 5
Processor #2 received 6
Processor #4 received 7
Processor #0 received 8
Processor #1 received 9
Processor #3 received 10

Note that cancelling a producer coroutine closes its channel, thus eventually terminating iteration
over the channel that processor coroutines are doing.

Also, pay attention to how we explicitly iterate over channel with for loop to perform fan-out in launchProcessor code.
Unlike consumeEach, this for loop pattern is perfectly safe to use from multiple coroutines. If one of the processor
coroutines fails, then others would still be processing the channel, while a processor that is written via consumeEach
always consumes (cancels) the underlying channel on its normal or abnormal completion.

Fan-in Copy Chapter

Multiple coroutines may send to the same channel.
For example, let us have a channel of strings, and a suspending function that
repeatedly sends a specified string to this channel with a specified delay:

suspend fun sendString(channel: SendChannel<String>, s: String, time: Long) {
    while (true) {
        delay(time)
        channel.send(s)
    }
}

Now, let us see what happens if we launch a couple of coroutines sending strings
(in this example we launch them in the context of the main thread as main coroutine's children):

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
//sampleStart
    val channel = Channel<String>()
    launch { sendString(channel, "foo", 200L) }
    launch { sendString(channel, "BAR!", 500L) }
    repeat(6) { // receive first six
        println(channel.receive())
    }
    coroutineContext.cancelChildren() // cancel all children to let main finish
//sampleEnd
}

suspend fun sendString(channel: SendChannel<String>, s: String, time: Long) {
    while (true) {
        delay(time)
        channel.send(s)
    }
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The output is:

foo
foo
BAR!
foo
foo
BAR!

Buffered channels Copy Chapter

The channels shown so far had no buffer. Unbuffered channels transfer elements when sender and receiver
meet each other (aka rendezvous). If send is invoked first, then it is suspended until receive is invoked,
if receive is invoked first, it is suspended until send is invoked.

Both Channel() factory function and produce builder take an optional capacity parameter to
specify buffer size. Buffer allows senders to send multiple elements before suspending,
similar to the BlockingQueue with a specified capacity, which blocks when buffer is full.

Take a look at the behavior of the following code:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking<Unit> {
//sampleStart
    val channel = Channel<Int>(4) // create buffered channel
    val sender = launch { // launch sender coroutine
        repeat(10) {
            println("Sending $it") // print before sending each element
            channel.send(it) // will suspend when buffer is full
        }
    }
    // don't receive anything... just wait....
    delay(1000)
    sender.cancel() // cancel sender coroutine
//sampleEnd    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

It prints "sending" five times using a buffered channel with capacity of four:

Sending 0
Sending 1
Sending 2
Sending 3
Sending 4

The first four elements are added to the buffer and the sender suspends when trying to send the fifth one.

Channels are fair Copy Chapter

Send and receive operations to channels are fair with respect to the order of their invocation from
multiple coroutines. They are served in first-in first-out order, e.g. the first coroutine to invoke receive
gets the element. In the following example two coroutines "ping" and "pong" are
receiving the "ball" object from the shared "table" channel.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

//sampleStart
data class Ball(var hits: Int)

fun main() = runBlocking {
    val table = Channel<Ball>() // a shared table
    launch { player("ping", table) }
    launch { player("pong", table) }
    table.send(Ball(0)) // serve the ball
    delay(1000) // delay 1 second
    coroutineContext.cancelChildren() // game over, cancel them
}

suspend fun player(name: String, table: Channel<Ball>) {
    for (ball in table) { // receive the ball in a loop
        ball.hits++
        println("$name $ball")
        delay(300) // wait a bit
        table.send(ball) // send the ball back
    }
}
//sampleEnd

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The "ping" coroutine is started first, so it is the first one to receive the ball. Even though "ping"
coroutine immediately starts receiving the ball again after sending it back to the table, the ball gets
received by the "pong" coroutine, because it was already waiting for it:

ping Ball(hits=1)
pong Ball(hits=2)
ping Ball(hits=3)
pong Ball(hits=4)

Note that sometimes channels may produce executions that look unfair due to the nature of the executor
that is being used. See this issue for details.

Ticker channels Copy Chapter

Ticker channel is a special rendezvous channel that produces Unit every time given delay passes since last consumption from this channel.
Though it may seem to be useless standalone, it is a useful building block to create complex time-based produce
pipelines and operators that do windowing and other time-dependent processing.
Ticker channel can be used in select to perform "on tick" action.

To create such channel use a factory method ticker.
To indicate that no further elements are needed use ReceiveChannel.cancel method on it.

Now let's see how it works in practice:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

//sampleStart
fun main() = runBlocking<Unit> {
    val tickerChannel = ticker(delayMillis = 200, initialDelayMillis = 0) // create a ticker channel
    var nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
    println("Initial element is available immediately: $nextElement") // no initial delay

    nextElement = withTimeoutOrNull(100) { tickerChannel.receive() } // all subsequent elements have 200ms delay
    println("Next element is not ready in 100 ms: $nextElement")

    nextElement = withTimeoutOrNull(120) { tickerChannel.receive() }
    println("Next element is ready in 200 ms: $nextElement")

    // Emulate large consumption delays
    println("Consumer pauses for 300ms")
    delay(300)
    // Next element is available immediately
    nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
    println("Next element is available immediately after large consumer delay: $nextElement")
    // Note that the pause between `receive` calls is taken into account and next element arrives faster
    nextElement = withTimeoutOrNull(120) { tickerChannel.receive() }
    println("Next element is ready in 100ms after consumer pause in 300ms: $nextElement")

    tickerChannel.cancel() // indicate that no more elements are needed
}
//sampleEnd

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

It prints following lines:

Initial element is available immediately: kotlin.Unit
Next element is not ready in 100 ms: null
Next element is ready in 200 ms: kotlin.Unit
Consumer pauses for 300ms
Next element is available immediately after large consumer delay: kotlin.Unit
Next element is ready in 100ms after consumer pause in 300ms: kotlin.Unit

Note that ticker is aware of possible consumer pauses and, by default, adjusts next produced element
delay if a pause occurs, trying to maintain a fixed rate of produced elements.

Optionally, a mode parameter equal to TickerMode.FIXED_DELAY can be specified to maintain a fixed
delay between elements.

Topics Compatibility (`docs/topics/compatibility.md`) Copy Section

• Compatibility
• Public API types
  • Experimental API
  • Flow preview API
  • Obsolete API
  • Internal API
  • Stable API
  • Deprecation cycle
• Using annotated API
  • Programmatically
  • Gradle
  • Maven

Compatibility Copy Chapter

This document describes the compatibility policy of kotlinx.coroutines library since version 1.0.0 and semantics of compatibility-specific annotations.

Public API types Copy Chapter

kotlinx.coroutines public API comes in five flavours: stable, experimental, obsolete, internal and deprecated.
All public API except stable is marked with the corresponding annotation.

Experimental API Copy Section

Experimental API is marked with @ExperimentalCoroutinesApi annotation.
API is marked experimental when its design has potential open questions which may eventually lead to
either semantics changes of the API or its deprecation.

By default, most of the new API is marked as experimental and becomes stable in one of the next major releases if no new issues arise.
Otherwise, either semantics is fixed without changes in ABI or API goes through deprecation cycle.

When using experimental API may be dangerous:

• You are writing a library which depends on kotlinx.coroutines and want to use experimental coroutines API in a stable library API.
  It may lead to undesired consequences when end users of your library update their kotlinx.coroutines version where experimental API
  has slightly different semantics.
• You want to build core infrastructure of the application around experimental API.

Flow preview API Copy Section

All Flow-related API is marked with @FlowPreview annotation.
This annotation indicates that Flow API is in preview status.
We provide no compatibility guarantees between releases for preview features, including binary, source and semantics compatibility.

When using preview API may be dangerous:

• You are writing a library/framework and want to use Flow API in a stable release or in a stable API.
• You want to use Flow in the core infrastructure of your application.
• You want to use Flow as "write-and-forget" solution and cannot afford additional maintenance cost when
  it comes to kotlinx.coroutines updates.

Obsolete API Copy Section

Obsolete API is marked with @ObsoleteCoroutinesApi annotation.
Obsolete API is similar to experimental, but already known to have serious design flaws and its potential replacement,
but replacement is not yet implemented.

The semantics of this API won't be changed, but it will go through a deprecation cycle as soon as the replacement is ready.

Internal API Copy Section

Internal API is marked with @InternalCoroutinesApi or is part of kotlinx.coroutines.internal package.
This API has no guarantees on its stability, can and will be changed and/or removed in the future releases.
If you can't avoid using internal API, please report it to issue tracker.

Stable API Copy Section

Stable API is guaranteed to preserve its ABI and documented semantics. If at some point unfixable design flaws will be discovered,
this API will go through a deprecation cycle and remain binary compatible as long as possible.

Deprecation cycle Copy Section

When some API is deprecated, it goes through multiple stages and there is at least one major release between stages.

• Feature is deprecated with compilation warning. Most of the time, proper replacement
  (and corresponding replaceWith declaration) is provided to automatically migrate deprecated usages with a help of IntelliJ IDEA.
• Deprecation level is increased to error or hidden. It is no longer possible to compile new code against deprecated API,
  though it is still present in the ABI.
• API is completely removed. While we give our best efforts not to do so and have no plans of removing any API, we still are leaving
  this option in case of unforeseen problems such as security holes.

Using annotated API Copy Chapter

All API annotations are kotlin.Experimental.
It is done in order to produce compilation warning about using experimental or obsolete API.
Warnings can be disabled either programmatically for a specific call site or globally for the whole module.

Programmatically Copy Section

For a specific call-site, warning can be disabled by using OptIn annotation:

@OptIn(ExperimentalCoroutinesApi::class) // Disables warning about experimental coroutines API 
fun experimentalApiUsage() {
    someKotlinxCoroutinesExperimentalMethod()
}

Gradle Copy Section

For the Gradle project, a warning can be disabled by passing a compiler flag in your build.gradle file:

tasks.withType(org.jetbrains.kotlin.gradle.tasks.AbstractKotlinCompile).all {
    kotlinOptions.freeCompilerArgs += ["-Xuse-experimental=kotlinx.coroutines.ExperimentalCoroutinesApi"]
}

Maven Copy Section

For the Maven project, a warning can be disabled by passing a compiler flag in your pom.xml file:

<plugin>
    <artifactId>kotlin-maven-plugin</artifactId>
    <groupId>org.jetbrains.kotlin</groupId>
    ... your configuration ...
    <configuration>
        <args>
            <arg>-Xuse-experimental=kotlinx.coroutines.ExperimentalCoroutinesApi</arg>
        </args>
    </configuration>
</plugin>

Topics Composing Suspending Functions (`docs/topics/composing-suspending-functions.md`) Copy Section

https://github.com/Kotlin/kotlinx.coroutines/edit/master/docs/topics/

This section covers various approaches to composition of suspending functions.

Sequential by default Copy Chapter

Assume that we have two suspending functions defined elsewhere that do something useful like some kind of
remote service call or computation. We just pretend they are useful, but actually each one just
delays for a second for the purpose of this example:

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

What do we do if we need them to be invoked sequentially — first doSomethingUsefulOne and then
doSomethingUsefulTwo, and compute the sum of their results?
In practice, we do this if we use the result of the first function to make a decision on whether we need
to invoke the second one or to decide on how to invoke it.

We use a normal sequential invocation, because the code in the coroutine, just like in the regular
code, is sequential by default. The following example demonstrates it by measuring the total
time it takes to execute both suspending functions:

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
//sampleStart
    val time = measureTimeMillis {
        val one = doSomethingUsefulOne()
        val two = doSomethingUsefulTwo()
        println("The answer is ${one + two}")
    }
    println("Completed in $time ms")
//sampleEnd    
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

It produces something like this:

The answer is 42
Completed in 2017 ms

Concurrent using async Copy Chapter

What if there are no dependencies between invocations of doSomethingUsefulOne and doSomethingUsefulTwo and
we want to get the answer faster, by doing both concurrently? This is where async comes to help.

Conceptually, async is just like launch. It starts a separate coroutine which is a light-weight thread
that works concurrently with all the other coroutines. The difference is that launch returns a Job and
does not carry any resulting value, while async returns a Deferred — a light-weight non-blocking future
that represents a promise to provide a result later. You can use .await() on a deferred value to get its eventual result,
but Deferred is also a Job, so you can cancel it if needed.

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
//sampleStart
    val time = measureTimeMillis {
        val one = async { doSomethingUsefulOne() }
        val two = async { doSomethingUsefulTwo() }
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")
//sampleEnd    
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

It produces something like this:

The answer is 42
Completed in 1017 ms

This is twice as fast, because the two coroutines execute concurrently.
Note that concurrency with coroutines is always explicit.

Lazily started async Copy Chapter

Optionally, async can be made lazy by setting its start parameter to CoroutineStart.LAZY.
In this mode it only starts the coroutine when its result is required by
await, or if its Job's start function
is invoked. Run the following example:

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
//sampleStart
    val time = measureTimeMillis {
        val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
        val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
        // some computation
        one.start() // start the first one
        two.start() // start the second one
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")
//sampleEnd    
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

It produces something like this:

The answer is 42
Completed in 1017 ms

So, here the two coroutines are defined but not executed as in the previous example, but the control is given to
the programmer on when exactly to start the execution by calling start. We first
start one, then start two, and then await for the individual coroutines to finish.

Note that if we just call await in println without first calling start on individual
coroutines, this will lead to sequential behavior, since await starts the coroutine
execution and waits for its finish, which is not the intended use-case for laziness.
The use-case for async(start = CoroutineStart.LAZY) is a replacement for the
standard lazy function in cases
when computation of the value involves suspending functions.

Async-style functions Copy Chapter

This programming style with async functions is provided here only for illustration, because it is a popular style
in other programming languages. Using this style with Kotlin coroutines is strongly discouraged for the
reasons explained below.

{style="note"}

We can define async-style functions that invoke doSomethingUsefulOne and doSomethingUsefulTwo
asynchronously using the async coroutine builder using a GlobalScope reference to
opt-out of the structured concurrency.
We name such functions with the
"...Async" suffix to highlight the fact that they only start asynchronous computation and one needs
to use the resulting deferred value to get the result.

GlobalScope is a delicate API that can backfire in non-trivial ways, one of which will be explained
below, so you must explicitly opt-in into using GlobalScope with @OptIn(DelicateCoroutinesApi::class).

{style="note"}

// The result type of somethingUsefulOneAsync is Deferred<Int>
@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulOneAsync() = GlobalScope.async {
    doSomethingUsefulOne()
}

// The result type of somethingUsefulTwoAsync is Deferred<Int>
@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulTwoAsync() = GlobalScope.async {
    doSomethingUsefulTwo()
}

Note that these xxxAsync functions are not suspending functions. They can be used from anywhere.
However, their use always implies asynchronous (here meaning concurrent) execution of their action
with the invoking code.

The following example shows their use outside of coroutine:

import kotlinx.coroutines.*
import kotlin.system.*

//sampleStart
// note that we don't have `runBlocking` to the right of `main` in this example
fun main() {
    val time = measureTimeMillis {
        // we can initiate async actions outside of a coroutine
        val one = somethingUsefulOneAsync()
        val two = somethingUsefulTwoAsync()
        // but waiting for a result must involve either suspending or blocking.
        // here we use `runBlocking { ... }` to block the main thread while waiting for the result
        runBlocking {
            println("The answer is ${one.await() + two.await()}")
        }
    }
    println("Completed in $time ms")
}
//sampleEnd

@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulOneAsync() = GlobalScope.async {
    doSomethingUsefulOne()
}

@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulTwoAsync() = GlobalScope.async {
    doSomethingUsefulTwo()
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

Consider what happens if between the val one = somethingUsefulOneAsync() line and one.await() expression there is some logic
error in the code, and the program throws an exception, and the operation that was being performed by the program aborts.
Normally, a global error-handler could catch this exception, log and report the error for developers, but the program
could otherwise continue doing other operations. However, here we have somethingUsefulOneAsync still running in the background,
even though the operation that initiated it was aborted. This problem does not happen with structured
concurrency, as shown in the section below.

Structured concurrency with async Copy Chapter

Let's refactor the Concurrent using async example into a function that runs
doSomethingUsefulOne and doSomethingUsefulTwo concurrently and returns their combined results.
Since async is a CoroutineScope extension,
we'll use the coroutineScope function to provide the necessary scope:

suspend fun concurrentSum(): Int = coroutineScope {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    one.await() + two.await()
}

This way, if something goes wrong inside the code of the concurrentSum function, and it throws an exception,
all the coroutines that were launched in its scope will be cancelled.

import kotlinx.coroutines.*
import kotlin.system.*

fun main() = runBlocking<Unit> {
//sampleStart
    val time = measureTimeMillis {
        println("The answer is ${concurrentSum()}")
    }
    println("Completed in $time ms")
//sampleEnd    
}

suspend fun concurrentSum(): Int = coroutineScope {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    one.await() + two.await()
}

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

We still have concurrent execution of both operations, as evident from the output of the above main function:

The answer is 42
Completed in 1017 ms

Cancellation is always propagated through coroutines hierarchy:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    try {
        failedConcurrentSum()
    } catch(e: ArithmeticException) {
        println("Computation failed with ArithmeticException")
    }
}

suspend fun failedConcurrentSum(): Int = coroutineScope {
    val one = async<Int> { 
        try {
            delay(Long.MAX_VALUE) // Emulates very long computation
            42
        } finally {
            println("First child was cancelled")
        }
    }
    val two = async<Int> { 
        println("Second child throws an exception")
        throw ArithmeticException()
    }
    one.await() + two.await()
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

Note how both the first async and the awaiting parent are cancelled on failure of one of the children
(namely, two):

Second child throws an exception
First child was cancelled
Computation failed with ArithmeticException

Topics Coroutine Context And Dispatchers (`docs/topics/coroutine-context-and-dispatchers.md`) Copy Section

https://github.com/Kotlin/kotlinx.coroutines/edit/master/docs/topics/

Coroutines always execute in some context represented by a value of the
CoroutineContext
type, defined in the Kotlin standard library.

The coroutine context is a set of various elements. The main elements are the Job of the coroutine,
which we've seen before, and its dispatcher, which is covered in this section.

Dispatchers and threads Copy Chapter

The coroutine context includes a coroutine dispatcher (see CoroutineDispatcher) that determines what thread or threads
the corresponding coroutine uses for its execution. The coroutine dispatcher can confine coroutine execution
to a specific thread, dispatch it to a thread pool, or let it run unconfined.

All coroutine builders like launch and async accept an optional
CoroutineContext
parameter that can be used to explicitly specify the dispatcher for the new coroutine and other context elements.

Try the following example:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    launch { // context of the parent, main runBlocking coroutine
        println("main runBlocking      : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(Dispatchers.Unconfined) { // not confined -- will work with main thread
        println("Unconfined            : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(Dispatchers.Default) { // will get dispatched to DefaultDispatcher 
        println("Default               : I'm working in thread ${Thread.currentThread().name}")
    }
    launch(newSingleThreadContext("MyOwnThread")) { // will get its own new thread
        println("newSingleThreadContext: I'm working in thread ${Thread.currentThread().name}")
    }
//sampleEnd    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

It produces the following output (maybe in different order):

Unconfined            : I'm working in thread main
Default               : I'm working in thread DefaultDispatcher-worker-1
newSingleThreadContext: I'm working in thread MyOwnThread
main runBlocking      : I'm working in thread main

When launch { ... } is used without parameters, it inherits the context (and thus dispatcher)
from the CoroutineScope it is being launched from. In this case, it inherits the
context of the main runBlocking coroutine which runs in the main thread.

Dispatchers.Unconfined is a special dispatcher that also appears to run in the main thread, but it is,
in fact, a different mechanism that is explained later.

The default dispatcher is used when no other dispatcher is explicitly specified in the scope.
It is represented by Dispatchers.Default and uses a shared background pool of threads.

newSingleThreadContext creates a thread for the coroutine to run.
A dedicated thread is a very expensive resource.
In a real application it must be either released, when no longer needed, using the close
function, or stored in a top-level variable and reused throughout the application.

Unconfined vs confined dispatcher Copy Chapter

The Dispatchers.Unconfined coroutine dispatcher starts a coroutine in the caller thread, but only until the
first suspension point. After suspension it resumes the coroutine in the thread that is fully determined by the
suspending function that was invoked. The unconfined dispatcher is appropriate for coroutines which neither
consume CPU time nor update any shared data (like UI) confined to a specific thread.

On the other side, the dispatcher is inherited from the outer CoroutineScope by default.
The default dispatcher for the runBlocking coroutine, in particular,
is confined to the invoker thread, so inheriting it has the effect of confining execution to
this thread with predictable FIFO scheduling.

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    launch(Dispatchers.Unconfined) { // not confined -- will work with main thread
        println("Unconfined      : I'm working in thread ${Thread.currentThread().name}")
        delay(500)
        println("Unconfined      : After delay in thread ${Thread.currentThread().name}")
    }
    launch { // context of the parent, main runBlocking coroutine
        println("main runBlocking: I'm working in thread ${Thread.currentThread().name}")
        delay(1000)
        println("main runBlocking: After delay in thread ${Thread.currentThread().name}")
    }
//sampleEnd    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

Produces the output:

Unconfined      : I'm working in thread main
main runBlocking: I'm working in thread main
Unconfined      : After delay in thread kotlinx.coroutines.DefaultExecutor
main runBlocking: After delay in thread main

So, the coroutine with the context inherited from runBlocking {...} continues to execute
in the main thread, while the unconfined one resumes in the default executor thread that the delay
function is using.

The unconfined dispatcher is an advanced mechanism that can be helpful in certain corner cases where
dispatching of a coroutine for its execution later is not needed or produces undesirable side-effects,
because some operation in a coroutine must be performed right away.
The unconfined dispatcher should not be used in general code.

{style="note"}

Debugging coroutines and threads Copy Chapter

Coroutines can suspend on one thread and resume on another thread.
Even with a single-threaded dispatcher it might be hard to
figure out what the coroutine was doing, where, and when if you don't have special tooling.

Debugging with IDEA Copy Section

The Coroutine Debugger of the Kotlin plugin simplifies debugging coroutines in IntelliJ IDEA.

Debugging works for versions 1.3.8 or later of kotlinx-coroutines-core.

{style="note"}

The Debug tool window contains the Coroutines tab. In this tab, you can find information about both currently running and suspended coroutines.
The coroutines are grouped by the dispatcher they are running on.

{width=700}

With the coroutine debugger, you can:

• Check the state of each coroutine.
• See the values of local and captured variables for both running and suspended coroutines.
• See a full coroutine creation stack, as well as a call stack inside the coroutine. The stack includes all frames with
  variable values, even those that would be lost during standard debugging.
• Get a full report that contains the state of each coroutine and its stack. To obtain it, right-click inside the Coroutines tab, and then click Get Coroutines Dump.

To start coroutine debugging, you just need to set breakpoints and run the application in debug mode.

Learn more about coroutines debugging in the tutorial.

Debugging using logging Copy Section

Another approach to debugging applications with
threads without Coroutine Debugger is to print the thread name in the log file on each log statement. This feature is universally supported
by logging frameworks. When using coroutines, the thread name alone does not give much of a context, so
kotlinx.coroutines includes debugging facilities to make it easier.

Run the following code with -Dkotlinx.coroutines.debug JVM option:

import kotlinx.coroutines.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main() = runBlocking<Unit> {
//sampleStart
    val a = async {
        log("I'm computing a piece of the answer")
        6
    }
    val b = async {
        log("I'm computing another piece of the answer")
        7
    }
    log("The answer is ${a.await() * b.await()}")
//sampleEnd    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

There are three coroutines. The main coroutine (#1) inside runBlocking
and two coroutines computing the deferred values a (#2) and b (#3).
They are all executing in the context of runBlocking and are confined to the main thread.
The output of this code is:

[main @coroutine#2] I'm computing a piece of the answer
[main @coroutine#3] I'm computing another piece of the answer
[main @coroutine#1] The answer is 42

The log function prints the name of the thread in square brackets, and you can see that it is the main
thread with the identifier of the currently executing coroutine appended to it. This identifier
is consecutively assigned to all created coroutines when the debugging mode is on.

Debugging mode is also turned on when JVM is run with -ea option.
You can read more about debugging facilities in the documentation of the DEBUG_PROPERTY_NAME property.

{style="note"}

Jumping between threads Copy Chapter

Run the following code with the -Dkotlinx.coroutines.debug JVM option (see debug):

import kotlinx.coroutines.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main() {
    newSingleThreadContext("Ctx1").use { ctx1 ->
        newSingleThreadContext("Ctx2").use { ctx2 ->
            runBlocking(ctx1) {
                log("Started in ctx1")
                withContext(ctx2) {
                    log("Working in ctx2")
                }
                log("Back to ctx1")
            }
        }
    }
}

You can get the full code here.

{style="note"}

The example above demonstrates new techniques in coroutine usage.

The first technique shows how to use runBlocking with a specified context.
The second technique involves calling withContext,
which may suspend the current coroutine and switch to a new context—provided the new context differs from the existing one.
Specifically, if you specify a different CoroutineDispatcher, extra dispatches are required:
the block is scheduled on the new dispatcher, and once it finishes, execution returns to the original dispatcher.

As a result, the output of the above code is:

[Ctx1 @coroutine#1] Started in ctx1
[Ctx2 @coroutine#1] Working in ctx2
[Ctx1 @coroutine#1] Back to ctx1

The example above uses the use function from the Kotlin standard library
to properly release thread resources created by newSingleThreadContext when they're no longer needed.

Job in the context Copy Chapter

The coroutine's Job is part of its context, and can be retrieved from it
using the coroutineContext[Job] expression:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    println("My job is ${coroutineContext[Job]}")
//sampleEnd    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

In debug mode, it outputs something like this:

My job is "coroutine#1":BlockingCoroutine{Active}@6d311334

Note that isActive in CoroutineScope is just a convenient shortcut for
coroutineContext[Job]?.isActive == true.

Children of a coroutine Copy Chapter

When a coroutine is launched in the CoroutineScope of another coroutine,
it inherits its context via CoroutineScope.coroutineContext and
the Job of the new coroutine becomes
a child of the parent coroutine's job. When the parent coroutine is cancelled, all its children
are recursively cancelled, too.

However, this parent-child relation can be explicitly overridden in one of two ways:

1. When a different scope is explicitly specified when launching a coroutine (for example, GlobalScope.launch),
   it does not inherit a Job from the parent scope.
2. When a different Job object is passed as the context for the new coroutine (as shown in the example below),
   it overrides the Job of the parent scope.

In both cases, the launched coroutine is not tied to the scope it was launched from and operates independently.

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    // launch a coroutine to process some kind of incoming request
    val request = launch {
        // it spawns two other jobs
        launch(Job()) { 
            println("job1: I run in my own Job and execute independently!")
            delay(1000)
            println("job1: I am not affected by cancellation of the request")
        }
        // and the other inherits the parent context
        launch {
            delay(100)
            println("job2: I am a child of the request coroutine")
            delay(1000)
            println("job2: I will not execute this line if my parent request is cancelled")
        }
    }
    delay(500)
    request.cancel() // cancel processing of the request
    println("main: Who has survived request cancellation?")
    delay(1000) // delay the main thread for a second to see what happens
//sampleEnd
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The output of this code is:

job1: I run in my own Job and execute independently!
job2: I am a child of the request coroutine
main: Who has survived request cancellation?
job1: I am not affected by cancellation of the request

Parental responsibilities Copy Chapter

A parent coroutine always waits for the completion of all its children.
A parent does not have to explicitly track
all the children it launches, and it does not have to use Job.join to wait for them at the end:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    // launch a coroutine to process some kind of incoming request
    val request = launch {
        repeat(3) { i -> // launch a few children jobs
            launch  {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, 600ms
                println("Coroutine $i is done")
            }
        }
        println("request: I'm done and I don't explicitly join my children that are still active")
    }
    request.join() // wait for completion of the request, including all its children
    println("Now processing of the request is complete")
//sampleEnd
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The result is going to be:

request: I'm done and I don't explicitly join my children that are still active
Coroutine 0 is done
Coroutine 1 is done
Coroutine 2 is done
Now processing of the request is complete

Naming coroutines for debugging Copy Chapter

Automatically assigned ids are good when coroutines log often and you just need to correlate log records
coming from the same coroutine. However, when a coroutine is tied to the processing of a specific request
or doing some specific background task, it is better to name it explicitly for debugging purposes.
The CoroutineName context element serves the same purpose as the thread name. It is included in the thread name that
is executing this coroutine when the debugging mode is turned on.

The following example demonstrates this concept:

import kotlinx.coroutines.*

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

fun main() = runBlocking(CoroutineName("main")) {
//sampleStart
    log("Started main coroutine")
    // run two background value computations
    val v1 = async(CoroutineName("v1coroutine")) {
        delay(500)
        log("Computing v1")
        6
    }
    val v2 = async(CoroutineName("v2coroutine")) {
        delay(1000)
        log("Computing v2")
        7
    }
    log("The answer for v1 * v2 = ${v1.await() * v2.await()}")
//sampleEnd    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The output it produces with -Dkotlinx.coroutines.debug JVM option is similar to:

[main @main#1] Started main coroutine
[main @v1coroutine#2] Computing v1
[main @v2coroutine#3] Computing v2
[main @main#1] The answer for v1 * v2 = 42

Combining context elements Copy Chapter

Sometimes we need to define multiple elements for a coroutine context. We can use the + operator for that.
For example, we can launch a coroutine with an explicitly specified dispatcher and an explicitly specified
name at the same time:

import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
//sampleStart
    launch(Dispatchers.Default + CoroutineName("test")) {
        println("I'm working in thread ${Thread.currentThread().name}")
    }
//sampleEnd    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The output of this code with the -Dkotlinx.coroutines.debug JVM option is:

I'm working in thread DefaultDispatcher-worker-1 @test#2

Coroutine scope Copy Chapter

Let us put our knowledge about contexts, children, and jobs together.
Assume that our application has an object with a lifecycle, but that object is not a coroutine.
For example,
we are writing an Android application,
and launching various coroutines in the context of an Android activity
to perform asynchronous operations to fetch and update data,
do animations, etc. These coroutines must be cancelled when the activity is destroyed
to avoid memory leaks.
We, of course, can manipulate contexts and jobs manually to tie the lifecycles of the activity
and its coroutines, but kotlinx.coroutines provides an abstraction encapsulating that: CoroutineScope.
You should be already familiar with the coroutine scope as all coroutine builders are declared as extensions on it.

We manage the lifecycles of our coroutines by creating an instance of CoroutineScope tied to
the lifecycle of our activity. A CoroutineScope instance can be created by the CoroutineScope() or MainScope()
factory functions. The former creates a general-purpose scope, while the latter creates a scope for UI applications and uses
Dispatchers.Main as the default dispatcher:

class Activity {
    private val mainScope = MainScope()
    
    fun destroy() {
        mainScope.cancel()
    }
    // to be continued ...

Now, we can launch coroutines in the scope of this Activity using the defined mainScope.
For the demo, we launch ten coroutines that delay for a different time:

// class Activity continues
    fun doSomething() {
        // launch ten coroutines for a demo, each working for a different time
        repeat(10) { i ->
            mainScope.launch {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
                println("Coroutine $i is done")
            }
        }
    }
} // class Activity ends

In our main function we create the activity, call our test doSomething function, and destroy the activity after 500ms.
This cancels all the coroutines that were launched from doSomething.
We can see that because after the destruction
of the activity, no more messages are printed, even if we wait a little longer.

import kotlinx.coroutines.*

class Activity {
    private val mainScope = CoroutineScope(Dispatchers.Default) // use Default for test purposes
    
    fun destroy() {
        mainScope.cancel()
    }

    fun doSomething() {
        // launch ten coroutines for a demo, each working for a different time
        repeat(10) { i ->
            mainScope.launch {
                delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
                println("Coroutine $i is done")
            }
        }
    }
} // class Activity ends

fun main() = runBlocking<Unit> {
//sampleStart
    val activity = Activity()
    activity.doSomething() // run test function
    println("Launched coroutines")
    delay(500L) // delay for half a second
    println("Destroying activity!")
    activity.destroy() // cancels all coroutines
    delay(1000) // visually confirm that they don't work
//sampleEnd    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

The output of this example is:

Launched coroutines
Coroutine 0 is done
Coroutine 1 is done
Destroying activity!

As you can see, only the first two coroutines print a message and the others are cancelled
by a single invocation of mainScope.cancel() in Activity.destroy().

Note that Android has first-party support for coroutine scope in all entities with the lifecycle.
See the corresponding documentation.

{style="note"}

Thread-local data Copy Section

Sometimes it is convenient to be able to pass some thread-local data to or between coroutines.
However, since they are not bound to any particular thread, this will likely lead to boilerplate if done manually.

For ThreadLocal,
the asContextElement extension function is here for the rescue. It creates an additional context element
which keeps the value of the given ThreadLocal and restores it every time the coroutine switches its context.

It is easy to demonstrate it in action:

import kotlinx.coroutines.*

val threadLocal = ThreadLocal<String?>() // declare thread-local variable

fun main() = runBlocking<Unit> {
//sampleStart
    threadLocal.set("main")
    println("Pre-main, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
    val job = launch(Dispatchers.Default + threadLocal.asContextElement(value = "launch")) {
        println("Launch start, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
        yield()
        println("After yield, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
    }
    job.join()
    println("Post-main, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
//sampleEnd    
}

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

You can get the full code here.

{style="note"}

In this example, we launch a new coroutine in a background thread pool using Dispatchers.Default, so
it works on different threads from the thread pool, but it still has the value of the thread local variable
that we specified using threadLocal.asContextElement(value = "launch"),
no matter which thread the coroutine is executed on.
Thus, the output (with debug) is:

Pre-main, current thread: Thread[main @coroutine#1,5,main], thread local value: 'main'
Launch start, current thread: Thread[DefaultDispatcher-worker-1 @coroutine#2,5,main], thread local value: 'launch'
After yield, current thread: Thread[DefaultDispatcher-worker-2 @coroutine#2,5,main], thread local value: 'launch'
Post-main, current thread: Thread[main @coroutine#1,5,main], thread local value: 'main'

It's easy to forget to set the corresponding context element. The thread-local variable accessed from the coroutine may
then have an unexpected value if the thread running the coroutine is different.
To avoid such situations, it is recommended to use the ensurePresent method
and fail-fast on improper usages.

ThreadLocal has first-class support and can be used with any primitive kotlinx.coroutines provides.
It has one key limitation, though: when a thread-local is mutated, a new value is not propagated to the coroutine caller
(because a context element cannot track all ThreadLocal object accesses), and the updated value is lost on the next suspension.
Use withContext to update the value of the thread-local in a coroutine, see asContextElement for more details.

Alternatively, a value can be stored in a mutable box like class Counter(var i: Int), which is, in turn,
stored in a thread-local variable.
However, in this case, you are fully responsible to synchronize
potentially concurrent modifications to the variable in this mutable box.

For advanced usage, for example, for integration with logging MDC, transactional contexts or any other libraries
that internally use thread-locals for passing data, see the documentation of the ThreadContextElement interface
that should be implemented.

Topics Coroutines And Channels (`docs/topics/coroutines-and-channels.md`) Copy Section

https://github.com/Kotlin/kotlinx.coroutines/edit/master/docs/topics/

An upcoming update will revise this tutorial. In the meantime, for an up-to-date guide to getting started with coroutines, see Coroutines basics.

{style="note"}

In this tutorial, you'll learn how to use coroutines in IntelliJ IDEA to perform network requests without blocking the
underlying thread or callbacks.

No prior knowledge of coroutines is required, but you're expected to be familiar with basic Kotlin syntax.

{style="tip"}

You'll learn:

• Why and how to use suspending functions to perform network requests.
• How to send requests concurrently using coroutines.
• How to share information between different coroutines using channels.

For network requests, you'll need the Retrofit library, but the approach shown in
this tutorial works similarly for any other libraries that support coroutines.

You can find solutions for all of the tasks on the solutions branch of the project's repository.

{style="tip"}

Before you start Copy Chapter

1. Download and install the latest version of IntelliJ IDEA.

2. Clone the project template by choosing Get from VCS on the
   Welcome screen or selecting File | New | Project from Version Control.

   You can also clone it from the command line:

   bash Copy

   git clone https://github.com/kotlin-hands-on/intro-coroutines

Generate a GitHub developer token Copy Section

You'll be using the GitHub API in your project. To get access, provide your GitHub account name and either a password or a
token. If you have two-factor authentication enabled, a token will be enough.

Generate a new GitHub token to use the GitHub API with your account:

1. Specify the name of your token, for example, coroutines-tutorial:

   {width=700}

2. Do not select any scopes. Click Generate token at the bottom of the page.

3. Copy the generated token.

Run the code Copy Section

The program loads the contributors for all of the repositories under the given organization (named "kotlin" by default).
Later you'll add logic to sort the users by the number of their contributions.

1. Open the src/contributors/main.kt file and run the main() function. You'll see the following window:

   {width=500}

   If the font is too small, adjust it by changing the value of setDefaultFontSize(18f) in the main() function.

2. Provide your GitHub username and token (or password) in the corresponding fields.

3. Make sure that the BLOCKING option is selected in the Variant dropdown menu.

4. Click Load contributors. The UI should freeze for some time and then show the list of contributors.

5. Open the program output to ensure the data has been loaded. The list of contributors is logged after each successful request.

There are different ways of implementing this logic: by using blocking requests
or callbacks. You'll compare these solutions with one that uses coroutines and see how
channels can be used to share information between different coroutines.

Blocking requests Copy Chapter

You will use the Retrofit library to perform HTTP requests to GitHub. It allows
requesting the list of repositories under the given organization and the list of contributors for each repository:

interface GitHubService {
    @GET("orgs/{org}/repos?per_page=100")
    fun getOrgReposCall(
        @Path("org") org: String
    ): Call<List<Repo>>

    @GET("repos/{owner}/{repo}/contributors?per_page=100")
    fun getRepoContributorsCall(
        @Path("owner") owner: String,
        @Path("repo") repo: String
    ): Call<List<User>>
}

This API is used by the loadContributorsBlocking() function to fetch the list of contributors for the given organization.

1. Open src/tasks/Request1Blocking.kt to see its implementation:

   kotlin Copy

   fun loadContributorsBlocking(
       service: GitHubService,
       req: RequestData
   ): List<User> {
       val repos = service
           .getOrgReposCall(req.org)   // #1
           .execute()                  // #2
           .also { logRepos(req, it) } // #3
           .body() ?: emptyList()      // #4
   
       return repos.flatMap { repo ->
           service
               .getRepoContributorsCall(req.org, repo.name) // #1
               .execute()                                   // #2
               .also { logUsers(repo, it) }                 // #3
               .bodyList()                                  // #4
       }.aggregate()
   }

   • At first, you get a list of the repositories under the given organization and store it in the repos list. Then for
     each repository, the list of contributors is requested, and all of the lists are merged into one final list of
     contributors.
   • getOrgReposCall() and getRepoContributorsCall() both return an instance of the *Call class (#1). At this point,
     no request is sent.
   • *Call.execute() is then invoked to perform the request (#2). execute() is a synchronous call that blocks the
     underlying thread.
   • When you get the response, the result is logged by calling the specific logRepos() and logUsers() functions (#3).
     If the HTTP response contains an error, this error will be logged here.
   • Finally, get the response's body, which contains the data you need. For this tutorial, you'll use an empty list as a
     result in case there is an error, and you'll log the corresponding error (#4).

2. To avoid repeating .body() ?: emptyList(), an extension function bodyList() is declared:

   kotlin Copy

   fun <T> Response<List<T>>.bodyList(): List<T> {
       return body() ?: emptyList()
   }

3. Run the program again and take a look at the system output in IntelliJ IDEA. It should have something like this:

   text Copy

   1770 [AWT-EventQueue-0] INFO  Contributors - kotlin: loaded 40 repos
   2025 [AWT-EventQueue-0] INFO  Contributors - kotlin-examples: loaded 23 contributors
   2229 [AWT-EventQueue-0] INFO  Contributors - kotlin-koans: loaded 45 contributors
   ...

   • The first item on each line is the number of milliseconds that have passed since the program started, then the thread
     name in square brackets. You can see from which thread the loading request is called.
   • The final item on each line is the actual message: how many repositories or contributors were loaded.

   This log output demonstrates that all of the results were logged from the main thread. When you run the code with a BLOCKING
   option, the window freezes and doesn't react to input until the loading is finished. All of the requests are executed from
   the same thread as the one called loadContributorsBlocking() is from, which is the main UI thread (in Swing, it's an AWT
   event dispatching thread). This main thread becomes blocked, and that's why the UI is frozen:

   {width=700}

   After the list of contributors has loaded, the result is updated.

4. In src/contributors/Contributors.kt, find the loadContributors() function responsible for choosing how
   the contributors are loaded and look at how loadContributorsBlocking() is called:

   kotlin Copy

   when (getSelectedVariant()) {
       BLOCKING -> { // Blocking UI thread
           val users = loadContributorsBlocking(service, req)
           updateResults(users, startTime)
       }
   }

   • The updateResults() call goes right after the loadContributorsBlocking() call.
   • updateResults() updates the UI, so it must always be called from the UI thread.
   • Since loadContributorsBlocking() is also called from the UI thread, the UI thread becomes blocked and the UI is
     frozen.

Task 1 Copy Section

The first task helps you familiarize yourself with the task domain. Currently, each contributor's name is repeated
several times, once for every project they have taken part in. Implement the aggregate() function combining the users
so that each contributor is added only once. The User.contributions property should contain the total number of
contributions of the given user to all the projects. The resulting list should be sorted in descending order according
to the number of contributions.

Open src/tasks/Aggregation.kt and implement the List.aggregate() function. Users should be sorted by the total
number of their contributions.

The corresponding test file test/tasks/AggregationKtTest.kt shows an example of the expected result.

You can jump between the source code and the test class automatically by using the IntelliJ IDEA shortcut
Ctrl+Shift+T / ⇧ ⌘ T.

{style="tip"}

After implementing this task, the resulting list for the "kotlin" organization should be similar to the following:

{width=500}

Solution for task 1 {initial-collapse-state="collapsed" collapsible="true"}

1. To group users by login, use groupBy(),
   which returns a map from a login to all occurrences of the user with this login in different repositories.

2. For each map entry, count the total number of contributions for each user and create a new instance of the User class
   by the given name and total of contributions.

3. Sort the resulting list in descending order:

   kotlin Copy

   fun List<User>.aggregate(): List<User> =
       groupBy { it.login }
           .map { (login, group) -> User(login, group.sumOf { it.contributions }) }
           .sortedByDescending { it.contributions }

An alternative solution is to use the groupingBy()
function instead of groupBy().

Callbacks Copy Chapter

The previous solution works, but it blocks the thread and therefore freezes the UI. A traditional approach that avoids this
is to use callbacks.

Instead of calling the code that should be invoked right after the operation is completed, you can extract it
into a separate callback, often a lambda, and pass that lambda to the caller in order for it to be called later.

To make the UI responsive, you can either move the whole computation to a separate thread or switch to the Retrofit API
which uses callbacks instead of blocking calls.

Use a background thread Copy Section

1. Open src/tasks/Request2Background.kt and see its implementation. First, the whole computation is moved to a different
   thread. The thread() function starts a new thread:

   kotlin Copy

   thread {
       loadContributorsBlocking(service, req)
   }

   Now that all of the loading has been moved to a separate thread, the main thread is free and can be occupied by other
   tasks:

   {width=700}

2. The signature of the loadContributorsBackground() function changes. It takes an updateResults()
   callback as the last argument to call it after all the loading completes:

   kotlin Copy

   fun loadContributorsBackground(
       service: GitHubService, req: RequestData,
       updateResults: (List<User>) -> Unit
   )

3. Now when the loadContributorsBackground() is called, the updateResults() call goes in the callback, not immediately
   afterward as it did before:

   kotlin Copy

   loadContributorsBackground(service, req) { users ->
       SwingUtilities.invokeLater {
           updateResults(users, startTime)
       }
   }

   By calling SwingUtilities.invokeLater, you ensure that the updateResults() call, which updates the results,
   happens on the main UI thread (AWT event dispatching thread).

However, if you try to load the contributors via the BACKGROUND option, you can see that the list is updated but
nothing changes in UI.

Task 2 Copy Section

Fix the loadContributorsBackground() function in src/tasks/Request2Background.kt so that the resulting list is shown
in the UI.

Solution for task 2 {initial-collapse-state="collapsed" collapsible="true"}

If you try to load the contributors, you can see in the log that the contributors are loaded but the result isn't displayed.
To fix this, call updateResults() on the resulting list of users:

thread {
    updateResults(loadContributorsBlocking(service, req))
}

Make sure to call the logic passed in the callback explicitly. Otherwise, nothing will happen.

Use the Retrofit callback API Copy Section

In the previous solution, the whole loading logic is moved to the background thread, but that still isn't the best use of
resources. All of the loading requests go sequentially and the thread is blocked while waiting for the loading result,
while it could have been occupied by other tasks. Specifically, the thread could start loading another request to
receive the entire result earlier.

Handling the data for each repository should then be divided into two parts: loading and processing the
resulting response. The second processing part should be extracted into a callback.

The loading for each repository can then be started before the result for the previous repository is received (and the
corresponding callback is called):

{width=700}

The Retrofit callback API can help achieve this. The Call.enqueue() function starts an HTTP request and takes a
callback as an argument. In this callback, you need to specify what needs to be done after each request.

Open src/tasks/Request3Callbacks.kt and see the implementation of loadContributorsCallbacks() that uses this API:

fun loadContributorsCallbacks(
    service: GitHubService, req: RequestData,
    updateResults: (List<User>) -> Unit
) {
    service.getOrgReposCall(req.org).onResponse { responseRepos ->  // #1
        logRepos(req, responseRepos)
        val repos = responseRepos.bodyList()

        val allUsers = mutableListOf<User>()
        for (repo in repos) {
            service.getRepoContributorsCall(req.org, repo.name)
                .onResponse { responseUsers ->  // #2
                    logUsers(repo, responseUsers)
                    val users = responseUsers.bodyList()
                    allUsers += users
                }
            }
        }
        // TODO: Why doesn't this code work? How to fix that?
        updateResults(allUsers.aggregate())
    }

• For convenience, this code fragment uses the onResponse() extension function declared in the same file. It takes a
  lambda as an argument rather than an object expression.
• The logic for handling the responses is extracted into callbacks: the corresponding lambdas start at lines #1 and #2.

However, the provided solution doesn't work. If you run the program and load contributors by choosing the CALLBACKS
option, you'll see that nothing is shown. However, the test from Request3CallbacksKtTest immediately returns the result
that it successfully passed.

Think about why the given code doesn't work as expected and try to fix it, or see the solutions below.

Task 3 (optional) Copy Section

Rewrite the code in the src/tasks/Request3Callbacks.kt file so that the loaded list of contributors is shown.

The first attempted solution for task 3 {initial-collapse-state="collapsed" collapsible="true"}

In the current solution, many requests are started concurrently, which decreases the total loading time. However,
the result isn't loaded. This is because the updateResults() callback is called right after all of the loading requests are started,
before the allUsers list has been filled with the data.

You could try to fix this with a change like the following:

val allUsers = mutableListOf<User>()
for ((index, repo) in repos.withIndex()) {   // #1
    service.getRepoContributorsCall(req.org, repo.name)
        .onResponse { responseUsers ->
            logUsers(repo, responseUsers)
            val users = responseUsers.bodyList()
            allUsers += users
            if (index == repos.lastIndex) {    // #2
                updateResults(allUsers.aggregate())
            }
        }
}

• First, you iterate over the list of repos with an index (#1).
• Then, from each callback, you check whether it's the last iteration (#2).
• And if that's the case, the result is updated.

However, this code also fails to achieve our objective. Try to find the answer yourself, or see the solution below.

The second attempted solution for task 3 {initial-collapse-state="collapsed" collapsible="true"}

Since the loading requests are started concurrently, there's no guarantee that the result for the last one comes last. The
results can come in any order.

Thus, if you compare the current index with the lastIndex as a condition for completion, you risk losing the results for
some repos.

If the request that processes the last repo returns faster than some prior requests (which is likely to happen), all of the
results for requests that take more time will be lost.

One way to fix this is to introduce an index and check whether all of the repositories have already been processed:

val allUsers = Collections.synchronizedList(mutableListOf<User>())
val numberOfProcessed = AtomicInteger()
for (repo in repos) {
    service.getRepoContributorsCall(req.org, repo.name)
        .onResponse { responseUsers ->
            logUsers(repo, responseUsers)
            val users = responseUsers.bodyList()
            allUsers += users
            if (numberOfProcessed.incrementAndGet() == repos.size) {
                updateResults(allUsers.aggregate())
            }
        }
}

This code uses a synchronized version of the list and AtomicInteger() because, in general, there's no guarantee that
different callbacks that process getRepoContributors() requests will always be called from the same thread.

The third attempted solution for task 3 {initial-collapse-state="collapsed" collapsible="true"}

An even better solution is to use the CountDownLatch class. It stores a counter initialized with the number of
repositories. This counter is decremented after processing each repository. It then waits until the latch is counted
down to zero before updating the results:

val countDownLatch = CountDownLatch(repos.size)
for (repo in repos) {
    service.getRepoContributorsCall(req.org, repo.name)
        .onResponse { responseUsers ->
            // processing repository
            countDownLatch.countDown()
        }
}
countDownLatch.await()
updateResults(allUsers.aggregate())

The result is then updated from the main thread. This is more direct than delegating the logic to the child threads.

After reviewing these three attempts at a solution, you can see that writing correct code with callbacks is non-trivial
and error-prone, especially when several underlying threads and synchronization occur.

As an additional exercise, you can implement the same logic using a reactive approach with the RxJava library. All of the
necessary dependencies and solutions for using RxJava can be found in a separate rx branch. It is also possible to
complete this tutorial and implement or check the proposed Rx versions for a proper comparison.

{style="tip"}

Suspending functions Copy Chapter

You can implement the same logic using suspending functions. Instead of returning Call<List<Repo>>, define the API
call as a suspending function as follows:

interface GitHubService {
    @GET("orgs/{org}/repos?per_page=100")
    suspend fun getOrgRepos(
        @Path("org") org: String
    ): List<Repo>
}

• getOrgRepos() is defined as a suspend function. When you use a suspending function to perform a request, the
  underlying thread isn't blocked. More details about how this works will come in later sections.
• getOrgRepos() returns the result directly instead of returning a Call. If the result is unsuccessful, an
  exception is thrown.

Alternatively, Retrofit allows returning the result wrapped in Response. In this case, the result body is
provided, and it is possible to check for errors manually. This tutorial uses the versions that return Response.

In src/contributors/GitHubService.kt, add the following declarations to the GitHubService interface:

interface GitHubService {
    // getOrgReposCall & getRepoContributorsCall declarations

    @GET("orgs/{org}/repos?per_page=100")
    suspend fun getOrgRepos(
        @Path("org") org: String
    ): Response<List<Repo>>

    @GET("repos/{owner}/{repo}/contributors?per_page=100")
    suspend fun getRepoContributors(
        @Path("owner") owner: String,
        @Path("repo") repo: String
    ): Response<List<User>>
}

Task 4 Copy Section

Your task is to change the code of the function that loads contributors to make use of two new suspending functions,
getOrgRepos() and getRepoContributors(). The new loadContributorsSuspend() function is marked as suspend to use the
new API.

Suspending functions can't be called everywhere. Calling a suspending function from loadContributorsBlocking() will
result in an error with the message "Suspend function 'getOrgRepos' should be called only from a coroutine or another
suspend function".

{style="note"}

1. Copy the implementation of loadContributorsBlocking() that is defined in src/tasks/Request1Blocking.kt
   into the loadContributorsSuspend() that is defined in src/tasks/Request4Suspend.kt.
2. Modify the code so that the new suspending functions are used instead of the ones that return Calls.
3. Run the program by choosing the SUSPEND option and ensure that the UI is still responsive while the GitHub requests
   are performed.

Solution for task 4 {initial-collapse-state="collapsed" collapsible="true"}

Replace .getOrgReposCall(req.org).execute() with .getOrgRepos(req.org) and repeat the same replacement for the
second "contributors" request:

suspend fun loadContributorsSuspend(service: GitHubService, req: RequestData): List<User> {
    val repos = service
        .getOrgRepos(req.org)
        .also { logRepos(req, it) }
        .bodyList()

    return repos.flatMap { repo ->
        service.getRepoContributors(req.org, repo.name)
            .also { logUsers(repo, it) }
            .bodyList()
    }.aggregate()
}

• loadContributorsSuspend() should be defined as a suspend function.
• You no longer need to call execute, which returned the Response before, because now the API functions return
  the Response directly. Note that this detail is specific to the Retrofit library. With other libraries, the API will be different,
  but the concept is the same.

Coroutines Copy Chapter

The code with suspending functions looks similar to the "blocking" version. The major difference from the blocking version
is that instead of blocking the thread, the coroutine is suspended:

block -> suspend
thread -> coroutine

Coroutines are often called lightweight threads because you can run code on coroutines, similar to how you run code on
threads. The operations that were blocking before (and had to be avoided) can now suspend the coroutine instead.

{style="note"}

Starting a new coroutine Copy Section

If you look at how loadContributorsSuspend() is used in src/contributors/Contributors.kt, you can see that it's
called inside launch. launch is a library function that takes a lambda as an argument:

launch {
    val users = loadContributorsSuspend(req)
    updateResults(users, startTime)
}

Here launch starts a new computation that is responsible for loading the data and showing the results. The computation
is suspendable – when performing network requests, it is suspended and releases the underlying thread.
When the network request returns the result, the computation is resumed.

Such a suspendable computation is called a coroutine. So, in this case, launch starts a new coroutine responsible
for loading data and showing the results.

Coroutines run on top of threads and can be suspended. When a coroutine is suspended, the
corresponding computation is paused, removed from the thread, and stored in memory. Meanwhile, the thread is free to be
occupied by other tasks:

{width=700}

When the computation is ready to be continued, it is returned to a thread (not necessarily the same one).

In the loadContributorsSuspend() example, each "contributors" request now waits for the result using the suspension
mechanism. First, the new request is sent. Then, while waiting for the response, the whole "load contributors" coroutine
that was started by the launch function is suspended.

The coroutine resumes only after the corresponding response is received:

{width=700}

While the response is waiting to be received, the thread is free to be occupied by other tasks. The UI stays responsive,
despite all the requests taking place on the main UI thread:

1. Run the program using the SUSPEND option. The log confirms that all of the requests are sent from the main UI thread:

   text Copy

   2538 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin: loaded 30 repos
   2729 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - ts2kt: loaded 11 contributors
   3029 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin-koans: loaded 45 contributors
   ...
   11252 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin-coroutines-workshop: loaded 1 contributors

2. The log can show you which coroutine the corresponding code is running on. To enable it, open Run | Edit configurations
   and add the -Dkotlinx.coroutines.debug VM option:

   {width=500}

   The coroutine name will be attached to the thread name while main() is run with this option. You can also
   modify the template for running all of the Kotlin files and enable this option by default.

Now all of the code runs on one coroutine, the "load contributors" coroutine mentioned above, denoted as @coroutine#1.
While waiting for the result, you shouldn't reuse the thread for sending other requests because the code is
written sequentially. The new request is sent only when the previous result is received.

Suspending functions treat the thread fairly and don't block it for "waiting". However, this doesn't yet bring any concurrency
into the picture.

Concurrency Copy Chapter

Kotlin coroutines are much less resource-intensive than threads.
Each time you want to start a new computation asynchronously, you can create a new coroutine instead.

To start a new coroutine, use one of the main coroutine builders: launch, async, or runBlocking. Different
libraries can define additional coroutine builders.

async starts a new coroutine and returns a Deferred object. Deferred represents a concept known by other names
such as Future or Promise. It stores a computation, but it defers the moment you get the final result;
it promises the result sometime in the future.

The main difference between async and launch is that launch is used to start a computation that isn't expected to
return a specific result. launch returns a Job that represents the coroutine. It is possible to wait until it completes
by calling Job.join().

Deferred is a generic type that extends Job. An async call can return a Deferred or a Deferred<CustomType>,
depending on what the lambda returns (the last expression inside the lambda is the result).

To get the result of a coroutine, you can call await() on the Deferred instance. While waiting for the result,
the coroutine that this await() is called from is suspended:

import kotlinx.coroutines.*

fun main() = runBlocking {
    val deferred: Deferred<Int> = async {
        loadData()
    }
    println("waiting...")
    println(deferred.await())
}

suspend fun loadData(): Int {
    println("loading...")
    delay(1000L)
    println("loaded!")
    return 42
}

runBlocking is used as a bridge between regular and suspending functions, or between the blocking and non-blocking worlds. It works
as an adaptor for starting the top-level main coroutine. It is intended primarily to be used in main() functions and
tests.

Watch this video for a better understanding of coroutines.

{style="tip"}

If there is a list of deferred objects, you can call awaitAll() to await the results of all of them:

import kotlinx.coroutines.*

fun main() = runBlocking {
    val deferreds: List<Deferred<Int>> = (1..3).map {
        async {
            delay(1000L * it)
            println("Loading $it")
            it
        }
    }
    val sum = deferreds.awaitAll().sum()
    println("$sum")
}

When each "contributors" request is started in a new coroutine, all of the requests are started asynchronously. A new request
can be sent before the result for the previous one is received:

{width=700}

The total loading time is approximately the same as in the CALLBACKS version, but it doesn't need any callbacks.
What's more, async explicitly emphasizes which parts run concurrently in the code.

Task 5 Copy Section

In the Request5Concurrent.kt file, implement a loadContributorsConcurrent() function by using the
previous loadContributorsSuspend() function.

Tip for task 5 {initial-collapse-state="collapsed" collapsible="true"}

You can only start a new coroutine inside a coroutine scope. Copy the content
from loadContributorsSuspend() to the coroutineScope call so that you can call async functions there:

suspend fun loadContributorsConcurrent(
    service: GitHubService,
    req: RequestData
): List<User> = coroutineScope {
    // ...
}

Base your solution on the following scheme:

val deferreds: List<Deferred<List<User>>> = repos.map { repo ->
    async {
        // load contributors for each repo
    }
}
deferreds.awaitAll() // List<List<User>>

Solution for task 5 {initial-collapse-state="collapsed" collapsible="true"}

Wrap each "contributors" request with async to create as many coroutines as there are repositories. async
returns Deferred<List>. This is not an issue because creating new coroutines is not very resource-intensive, so you can
create as many as you need.

1. You can no longer use flatMap because the map result is now a list of Deferred objects, not a list of lists.
   awaitAll() returns List<List>, so call flatten().aggregate() to get the result:

   kotlin Copy

   suspend fun loadContributorsConcurrent(
       service: GitHubService, 
       req: RequestData
   ): List<User> = coroutineScope {
       val repos = service
           .getOrgRepos(req.org)
           .also { logRepos(req, it) }
           .bodyList()
   
       val deferreds: List<Deferred<List<User>>> = repos.map { repo ->
           async {
               service.getRepoContributors(req.org, repo.name)
                   .also { logUsers(repo, it) }
                   .bodyList()
           }
       }
       deferreds.awaitAll().flatten().aggregate()
   }

2. Run the code and check the log. All of the coroutines still run on the main UI thread because
   multithreading hasn't been employed yet, but you can already see the benefits of running coroutines concurrently.

3. To change this code to run "contributors" coroutines on different threads from the common thread pool,
   specify Dispatchers.Default as the context argument for the async function:

   kotlin Copy

   async(Dispatchers.Default) { }

   • CoroutineDispatcher determines what thread or threads the corresponding coroutine should be run on. If you don't
     specify one as an argument, async will use the dispatcher from the outer scope.
   • Dispatchers.Default represents a shared pool of threads on the JVM. This pool provides a means for parallel execution.
     It consists of as many threads as there are CPU cores available, but it will still have two threads if there's only one core.

4. Modify the code in the loadContributorsConcurrent() function to start new coroutines on different threads from the
   common thread pool. Also, add additional logging before sending the request:

   kotlin Copy

   async(Dispatchers.Default) {
       log("starting loading for ${repo.name}")
       service.getRepoContributors(req.org, repo.name)
           .also { logUsers(repo, it) }
           .bodyList()
   }

5. Run the program once again. In the log, you can see that each coroutine can be started on one thread from the
   thread pool and resumed on another:

   text Copy

   1946 [DefaultDispatcher-worker-2 @coroutine#4] INFO  Contributors - starting loading for kotlin-koans
   1946 [DefaultDispatcher-worker-3 @coroutine#5] INFO  Contributors - starting loading for dokka
   1946 [DefaultDispatcher-worker-1 @coroutine#3] INFO  Contributors - starting loading for ts2kt
   ...
   2178 [DefaultDispatcher-worker-1 @coroutine#4] INFO  Contributors - kotlin-koans: loaded 45 contributors
   2569 [DefaultDispatcher-worker-1 @coroutine#5] INFO  Contributors - dokka: loaded 36 contributors
   2821 [DefaultDispatcher-worker-2 @coroutine#3] INFO  Contributors - ts2kt: loaded 11 contributors

   For instance, in this log excerpt, coroutine#4 is started on the worker-2 thread and continued on the
   worker-1 thread.

In src/contributors/Contributors.kt, check the implementation of the CONCURRENT option:

1. To run the coroutine only on the main UI thread, specify Dispatchers.Main as an argument:

   kotlin Copy

   launch(Dispatchers.Main) {
       updateResults()
   }

   • If the main thread is busy when you start a new coroutine on it,
     the coroutine becomes suspended and scheduled for execution on this thread. The coroutine will only resume when the
     thread becomes free.
   • It's considered good practice to use the dispatcher from the outer scope rather than explicitly specifying it on each
     end-point. If you define loadContributorsConcurrent() without passing Dispatchers.Default as an
     argument, you can call this function in any context: with a Default dispatcher, with
     the main UI thread, or with a custom dispatcher.
   • As you'll see later, when calling loadContributorsConcurrent() from tests, you can call it in the context
     with TestDispatcher, which simplifies testing. That makes this solution much more flexible.

2. To specify the dispatcher on the caller side, apply the following change to the project while
   letting loadContributorsConcurrent start coroutines in the inherited context:

   kotlin Copy

   launch(Dispatchers.Default) {
       val users = loadContributorsConcurrent(service, req)
       withContext(Dispatchers.Main) {
           updateResults(users, startTime)
       }
   }

   • updateResults() should be called on the main UI thread, so you call it with the context of Dispatchers.Main.
   • withContext() calls the given code with the specified coroutine context, is suspended until it completes, and returns
     the result. An alternative but more verbose way to express this would be to start a new coroutine and explicitly
     wait (by suspending) until it completes: launch(context) { ... }.join().

3. Run the code and ensure that the coroutines are executed on the threads from the thread pool.

Structured concurrency Copy Chapter

• The coroutine scope is responsible for the structure and parent-child relationships between different coroutines. New
  coroutines usually need to be started inside a scope.
• The coroutine context stores additional technical information used to run a given coroutine, like the coroutine custom
  name, or the dispatcher specifying the threads the coroutine should be scheduled on.

When launch, async, or runBlocking are used to start a new coroutine, they automatically create the corresponding
scope. All of these functions take a lambda with a receiver as an argument, and CoroutineScope is the implicit receiver type:

launch { /* this: CoroutineScope */ }

• New coroutines can only be started inside a scope.
• launch and async are declared as extensions to CoroutineScope, so an implicit or explicit receiver must always
  be passed when you call them.
• The coroutine started by runBlocking is the only exception because runBlocking is defined as a top-level function.
  But because it blocks the current thread, it's intended primarily to be used in main() functions and tests as a bridge
  function.

A new coroutine inside runBlocking, launch, or async is started automatically inside the scope:

import kotlinx.coroutines.*

fun main() = runBlocking { /* this: CoroutineScope */
    launch { /* ... */ }
    // the same as:   
    this.launch { /* ... */ }
}

When you call launch inside runBlocking, it's called as an extension to the implicit receiver of
the CoroutineScope type. Alternatively, you could explicitly write this.launch.

The nested coroutine (started by launch in this example) can be considered as a child of the outer coroutine (started
by runBlocking). This "parent-child" relationship works through scopes; the child coroutine is started from the scope
corresponding to the parent coroutine.

It's possible to create a new scope without starting a new coroutine, by using the coroutineScope function.
To start new coroutines in a structured way inside a suspend function without access to the outer scope, you can create
a new coroutine scope that automatically becomes a child of the outer scope that this suspend function is called from.
loadContributorsConcurrent() is a good example.

You can also start a new coroutine from the global scope using GlobalScope.async or GlobalScope.launch.
This will create a top-level "independent" coroutine.

The mechanism behind the structure of the coroutines is called structured concurrency. It provides the following
benefits over global scopes:

• The scope is generally responsible for child coroutines, whose lifetime is attached to the lifetime of the scope.
• The scope can automatically cancel child coroutines if something goes wrong or a user changes their mind and decides
  to revoke the operation.
• The scope automatically waits for the completion of all child coroutines.
  Therefore, if the scope corresponds to a coroutine, the parent coroutine does not complete until all the coroutines
  launched in its scope have completed.

When using GlobalScope.async, there is no structure that binds several coroutines to a smaller scope.
Coroutines started from the global scope are all independent – their lifetime is limited only by the lifetime of the
whole application. It's possible to store a reference to the coroutine started from the global scope and wait for its
completion or cancel it explicitly, but that won't happen automatically as it would with structured concurrency.

Canceling the loading of contributors Copy Section

Create two versions of the function that loads the list of contributors. Compare how both versions behave when you try to
cancel the parent coroutine. The first version will use coroutineScope to start all of the child coroutines,
whereas the second will use GlobalScope.

1. In Request5Concurrent.kt, add a 3-second delay to the loadContributorsConcurrent() function:

   kotlin Copy

   suspend fun loadContributorsConcurrent(
       service: GitHubService, 
       req: RequestData
   ): List<User> = coroutineScope {
       // ...
       async {
           log("starting loading for ${repo.name}")
           delay(3000)
           // load repo contributors
       }
       // ...
   }

   The delay affects all of the coroutines that send requests, so that there's enough time to cancel the loading
   after the coroutines are started but before the requests are sent.

2. Create the second version of the loading function: copy the implementation of loadContributorsConcurrent() to
   loadContributorsNotCancellable() in Request5NotCancellable.kt and then remove the creation of a new coroutineScope.

3. The async calls now fail to resolve, so start them by using GlobalScope.async:

   kotlin Copy

   suspend fun loadContributorsNotCancellable(
       service: GitHubService,
       req: RequestData
   ): List<User> {   // #1
       // ...
       GlobalScope.async {   // #2
           log("starting loading for ${repo.name}")
           // load repo contributors
       }
       // ...
       return deferreds.awaitAll().flatten().aggregate()  // #3
   }

   • The function now returns the result directly, not as the last expression inside the lambda (lines #1 and #3).
   • All of the "contributors" coroutines are started inside the GlobalScope, not as children of the coroutine scope
     (line #2).

4. Run the program and choose the CONCURRENT option to load the contributors.

5. Wait until all of the "contributors" coroutines are started, and then click Cancel. The log shows no new results,
   which means that all of the requests were indeed canceled:

   text Copy

   2896 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin: loaded 40 repos
   2901 [DefaultDispatcher-worker-2 @coroutine#4] INFO  Contributors - starting loading for kotlin-koans
   ...
   2909 [DefaultDispatcher-worker-5 @coroutine#36] INFO  Contributors - starting loading for mpp-example
   /* click on 'cancel' */
   /* no requests are sent */

6. Repeat step 5, but this time choose the NOT_CANCELLABLE option:

   text Copy

   2570 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin: loaded 30 repos
   2579 [DefaultDispatcher-worker-1 @coroutine#4] INFO  Contributors - starting loading for kotlin-koans
   ...
   2586 [DefaultDispatcher-worker-6 @coroutine#36] INFO  Contributors - starting loading for mpp-example
   /* click on 'cancel' */
   /* but all the requests are still sent: */
   6402 [DefaultDispatcher-worker-5 @coroutine#4] INFO  Contributors - kotlin-koans: loaded 45 contributors
   ...
   9555 [DefaultDispatcher-worker-8 @coroutine#36] INFO  Contributors - mpp-example: loaded 8 contributors

   In this case, no coroutines are canceled, and all the requests are still sent.

7. Check how the cancellation is triggered in the "contributors" program. When the Cancel button is clicked,
   the main "loading" coroutine is explicitly canceled and the child coroutines are canceled automatically:

   kotlin Copy

   interface Contributors {
   
       fun loadContributors() {
           // ...
           when (getSelectedVariant()) {
               CONCURRENT -> {
                   launch {
                       val users = loadContributorsConcurrent(service, req)
                       updateResults(users, startTime)
                   }.setUpCancellation()      // #1
               }
           }
       }
   
       private fun Job.setUpCancellation() {
           val loadingJob = this              // #2
   
           // cancel the loading job if the 'cancel' button was clicked:
           val listener = ActionListener {
               loadingJob.cancel()            // #3
               updateLoadingStatus(CANCELED)
           }
           // add a listener to the 'cancel' button:
           addCancelListener(listener)
   
           // update the status and remove the listener
           // after the loading job is completed
       }
   }

The launch function returns an instance of Job. Job stores a reference to the "loading coroutine", which loads
all of the data and updates the results. You can call the setUpCancellation() extension function on it (line #1),
passing an instance of Job as a receiver.

Another way you could express this would be to explicitly write:

val job = launch { }
job.setUpCancellation()

• For readability, you could refer to the setUpCancellation() function receiver inside the function with the
  new loadingJob variable (line #2).
• Then you could add a listener to the Cancel button so that when it's clicked, the loadingJob is canceled (line #3).

With structured concurrency, you only need to cancel the parent coroutine and this automatically propagates cancellation
to all of the child coroutines.

Using the outer scope's context Copy Section

When you start new coroutines inside the given scope, it's much easier to ensure that all of them run with the same
context. It is also much easier to replace the context if needed.

Now it's time to learn how using the dispatcher from the outer scope works. The new scope created by
the coroutineScope or by the coroutine builders always inherits the context from the outer scope. In this case, the
outer scope is the scope the suspend loadContributorsConcurrent() function was called from:

launch(Dispatchers.Default) {  // outer scope
    val users = loadContributorsConcurrent(service, req)
    // ...
}

All of the nested coroutines are automatically started with the inherited context. The dispatcher is a part of this
context. That's why all of the coroutines started by async are started with the context of the default dispatcher:

suspend fun loadContributorsConcurrent(
    service: GitHubService, req: RequestData
): List<User> = coroutineScope {
    // this scope inherits the context from the outer scope
    // ...
    async {   // nested coroutine started with the inherited context
        // ...
    }
    // ...
}

With structured concurrency, you can specify the major context elements (like dispatcher) once, when creating the
top-level coroutine. All the nested coroutines then inherit the context and modify it only if needed.

When you write code with coroutines for UI applications, for example Android ones, it's a common practice to
use CoroutineDispatchers.Main by default for the top coroutine and then to explicitly put a different dispatcher when
you need to run the code on a different thread.

{style="tip"}

Showing progress Copy Chapter

Despite the information for some repositories being loaded rather quickly, the user only sees the resulting list after all of
the data has been loaded. Until then, the loader icon runs showing the progress, but there's no information about the current
state or what contributors are already loaded.

You can show the intermediate results earlier and display all of the contributors after loading the data for each of the
repositories:

{width=500}

To implement this functionality, in the src/tasks/Request6Progress.kt, you'll need to pass the logic updating the UI
as a callback, so that it's called on each intermediate state:

suspend fun loadContributorsProgress(
    service: GitHubService,
    req: RequestData,
    updateResults: suspend (List<User>, completed: Boolean) -> Unit
) {
    // loading the data
    // calling `updateResults()` on intermediate states
}

On the call site in Contributors.kt, the callback is passed to update the results from the Main thread for
the PROGRESS option:

launch(Dispatchers.Default) {
    loadContributorsProgress(service, req) { users, completed ->
        withContext(Dispatchers.Main) {
            updateResults(users, startTime, completed)
        }
    }
}

• The updateResults() parameter is declared as suspend in loadContributorsProgress(). It's necessary to call
  withContext, which is a suspend function inside the corresponding lambda argument.
• updateResults() callback takes an additional Boolean parameter as an argument specifying whether the loading has
  completed and the results are final.

Task 6 Copy Section

In the Request6Progress.kt file, implement the loadContributorsProgress() function that shows the intermediate
progress. Base it on the loadContributorsSuspend() function from Request4Suspend.kt.

• Use a simple version without concurrency; you'll add it later in the next section.
• The intermediate list of contributors should be shown in an "aggregated" state, not just the list of users loaded for
  each repository.
• The total number of contributions for each user should be increased when the data for each new
  repository is loaded.

Solution for task 6 {initial-collapse-state="collapsed" collapsible="true"}

To store the intermediate list of loaded contributors in the "aggregated" state, define an allUsers variable which
stores the list of users, and then update it after contributors for each new repository are loaded:

suspend fun loadContributorsProgress(
    service: GitHubService,
    req: RequestData,
    updateResults: suspend (List<User>, completed: Boolean) -> Unit
) {
    val repos = service
        .getOrgRepos(req.org)
        .also { logRepos(req, it) }
        .bodyList()

    var allUsers = emptyList<User>()
    for ((index, repo) in repos.withIndex()) {
        val users = service.getRepoContributors(req.org, repo.name)
            .also { logUsers(repo, it) }
            .bodyList()

        allUsers = (allUsers + users).aggregate()
        updateResults(allUsers, index == repos.lastIndex)
    }
}

Consecutive vs concurrent

An updateResults() callback is called after each request is completed:

{width=700}

This code doesn't include concurrency. It's sequential, so you don't need synchronization.

The best option would be to send requests concurrently and update the intermediate results after getting the response
for each repository:

{width=700}

To add concurrency, use channels.

Channels Copy Chapter

Writing code with a shared mutable state is quite difficult and error-prone (like in the solution using callbacks).
A simpler way is to share information by communication rather than by using a common mutable state.
Coroutines can communicate with each other through channels.

Channels are communication primitives that allow data to be passed between coroutines. One coroutine can send
some information to a channel, while another can receive that information from it:

A coroutine that sends (produces) information is often called a producer, and a coroutine that receives (consumes)
information is called a consumer. One or multiple coroutines can send information to the same channel, and one or multiple
coroutines can receive data from it:

When many coroutines receive information from the same channel, each element is handled only once by one of the
consumers. Once an element is handled, it is immediately removed from the channel.

You can think of a channel as similar to a collection of elements, or more precisely, a queue, in which elements are added
to one end and received from the other. However, there's an important difference: unlike collections, even in their
synchronized versions, a channel can suspend send() and receive() operations. This happens when the channel is empty
or full. The channel can be full if the channel size has an upper bound.

Channel is represented by three different interfaces: SendChannel, ReceiveChannel, and Channel, with the latter
extending the first two. You usually create a channel and give it to producers as a SendChannel instance so that only
they can send information to the channel.
You give a channel to consumers as a ReceiveChannel instance so that only they can receive from it. Both send
and receive methods are declared as suspend:

interface SendChannel<in E> {
    suspend fun send(element: E)
    fun close(): Boolean
}

interface ReceiveChannel<out E> {
    suspend fun receive(): E
}

interface Channel<E> : SendChannel<E>, ReceiveChannel<E>

The producer can close a channel to indicate that no more elements are coming.

Several types of channels are defined in the library. They differ in how many elements they can internally store and
whether the send() call can be suspended or not.
For all of the channel types, the receive() call behaves similarly: it receives an element if the channel is not empty;
otherwise, it is suspended.

When you create a channel, specify its type or the buffer size (if you need a buffered one):

val rendezvousChannel = Channel<String>()
val bufferedChannel = Channel<String>(10)
val conflatedChannel = Channel<String>(CONFLATED)
val unlimitedChannel = Channel<String>(UNLIMITED)

By default, a "Rendezvous" channel is created.

In the following task, you'll create a "Rendezvous" channel, two producer coroutines, and a consumer coroutine:

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    launch {
        channel.send("A1")
        channel.send("A2")
        log("A done")
    }
    launch {
        channel.send("B1")
        log("B done")
    }
    launch {
        repeat(3) {
            val x = channel.receive()
            log(x)
        }
    }
}

fun log(message: Any?) {
    println("[${Thread.currentThread().name}] $message")
}

Watch this video for a better understanding of channels.

{style="tip"}

Task 7 Copy Section

In src/tasks/Request7Channels.kt, implement the function loadContributorsChannels() that requests all of the GitHub
contributors concurrently and shows intermediate progress at the same time.

Use the previous functions, loadContributorsConcurrent() from Request5Concurrent.kt
and loadContributorsProgress() from Request6Progress.kt.

Tip for task 7 {initial-collapse-state="collapsed" collapsible="true"}

Different coroutines that concurrently receive contributor lists for different repositories can send all of the received
results to the same channel:

val channel = Channel<List<User>>()
for (repo in repos) {
    launch {
        val users = TODO()
        // ...
        channel.send(users)
    }
}

Then the elements from this channel can be received one by one and processed:

repeat(repos.size) {
    val users = channel.receive()
    // ...
}

Since the receive() calls are sequential, no additional synchronization is needed.

Solution for task 7 {initial-collapse-state="collapsed" collapsible="true"}

As with the loadContributorsProgress() function, you can create an allUsers variable to store the intermediate
states of the "all contributors" list.
Each new list received from the channel is added to the list of all users. You aggregate the result and update the state
using the updateResults callback:

suspend fun loadContributorsChannels(
    service: GitHubService,
    req: RequestData,
    updateResults: suspend (List<User>, completed: Boolean) -> Unit
) = coroutineScope {

    val repos = service
        .getOrgRepos(req.org)
        .also { logRepos(req, it) }
        .bodyList()

    val channel = Channel<List<User>>()
    for (repo in repos) {
        launch {
            val users = service.getRepoContributors(req.org, repo.name)
                .also { logUsers(repo, it) }
                .bodyList()
            channel.send(users)
        }
    }
    var allUsers = emptyList<User>()
    repeat(repos.size) {
        val users = channel.receive()
        allUsers = (allUsers + users).aggregate()
        updateResults(allUsers, it == repos.lastIndex)
    }
}

• Results for different repositories are added to the channel as soon as they are ready. At first, when all of the requests
  are sent, and no data is received, the receive() call is suspended. In this case, the whole "load contributors" coroutine
  is suspended.
• Then, when the list of users is sent to the channel, the "load contributors" coroutine resumes, the receive() call
  returns this list, and the results are immediately updated.

You can now run the program and choose the CHANNELS option to load the contributors and see the result.

Although neither coroutines nor channels completely remove the complexity that comes with concurrency,
they make life easier when you need to understand what's going on.

Testing coroutines Copy Chapter

Let's now test all solutions to check that the solution with concurrent coroutines is faster than the solution with
the suspend functions, and check that the solution with channels is faster than the simple "progress" one.

In the following task, you'll compare the total running time of the solutions. You'll mock a GitHub service and make
this service return results after the given timeouts:

repos request - returns an answer within 1000 ms delay
repo-1 - 1000 ms delay
repo-2 - 1200 ms delay
repo-3 - 800 ms delay

The sequential solution with the suspend functions should take around 4000 ms (4000 = 1000 + (1000 + 1200 + 800)).
The concurrent solution should take around 2200 ms (2200 = 1000 + max(1000, 1200, 800)).

For the solutions that show progress, you can also check the intermediate results with timestamps.

The corresponding test data is defined in test/contributors/testData.kt, and the files Request4SuspendKtTest,
Request7ChannelsKtTest, and so on contain the straightforward tests that use mock service calls.

However, there are two problems here:

• These tests take too long to run. Each test takes around 2 to 4 seconds, and you need to wait for the results each
  time. It's not very efficient.
• You can't rely on the exact time the solution runs because it still takes additional time to prepare and run the code.
  You could add a constant, but then the time would differ from machine to machine. The mock service delays
  should be higher than this constant so you can see a difference. If the constant is 0.5 sec, making the delays
  0.1 sec won't be enough.

A better way would be to use special frameworks to test the timing while running the same code several times (which increases
the total time even more), but that is complicated to learn and set up.

To solve these problems and make sure that solutions with provided test delays behave as expected, one faster than the other,
use virtual time with a special test dispatcher. This dispatcher keeps track of the virtual time passed from
the start and runs everything immediately in real time. When you run coroutines on this dispatcher,
the delay will return immediately and advance the virtual time.

Tests that use this mechanism run fast, but you can still check what happens at different moments in virtual time. The
total running time drastically decreases:

{width=700}

To use virtual time, replace the runBlocking invocation with a runTest. runTest takes an
extension lambda to TestScope as an argument.
When you call delay in a suspend function inside this special scope, delay will increase the virtual time instead
of delaying in real time:

@Test
fun testDelayInSuspend() = runTest {
    val realStartTime = System.currentTimeMillis() 
    val virtualStartTime = currentTime
        
    foo()
    println("${System.currentTimeMillis() - realStartTime} ms") // ~ 6 ms
    println("${currentTime - virtualStartTime} ms")             // 1000 ms
}

suspend fun foo() {
    delay(1000)    // auto-advances without delay
    println("foo") // executes eagerly when foo() is called
}

You can check the current virtual time using the currentTime property of TestScope.

The actual running time in this example is several milliseconds, whereas virtual time equals the delay argument, which
is 1000 milliseconds.

To get the full effect of "virtual" delay in child coroutines,
start all of the child coroutines with TestDispatcher. Otherwise, it won't work. This dispatcher is
automatically inherited from the other TestScope, unless you provide a different dispatcher:

@Test
fun testDelayInLaunch() = runTest {
    val realStartTime = System.currentTimeMillis()
    val virtualStartTime = currentTime

    bar()

    println("${System.currentTimeMillis() - realStartTime} ms") // ~ 11 ms
    println("${currentTime - virtualStartTime} ms")             // 1000 ms
}

suspend fun bar() = coroutineScope {
    launch {
        delay(1000)    // auto-advances without delay
        println("bar") // executes eagerly when bar() is called
    }
}

If launch is called with the context of Dispatchers.Default in the example above, the test will fail. You'll get an
exception saying that the job has not been completed yet.

You can test the loadContributorsConcurrent() function this way only if it starts the child coroutines with the
inherited context, without modifying it using the Dispatchers.Default dispatcher.

You can specify the context elements like the dispatcher when calling a function rather than when defining it,
which allows for more flexibility and easier testing.

The testing API that supports virtual time is Experimental and may change in the future.

{style="warning"}

By default, the compiler shows warnings if you use the experimental testing API. To suppress these warnings, annotate
the test function or the whole class containing the tests with @OptIn(ExperimentalCoroutinesApi::class).
Add the compiler argument instructing the compiler that you're using the experimental API:

compileTestKotlin {
    kotlinOptions {
        freeCompilerArgs += "-Xuse-experimental=kotlin.Experimental"
    }
}

In the project corresponding to this tutorial, the compiler argument has already been added to the Gradle script.

Task 8 Copy Section

Refactor the following tests in tests/tasks/ to use virtual time instead of real time:

• Request4SuspendKtTest.kt
• Request5ConcurrentKtTest.kt
• Request6ProgressKtTest.kt
• Request7ChannelsKtTest.kt

Compare the total running times before and after applying your refactoring.

Tip for task 8 {initial-collapse-state="collapsed" collapsible="true"}

1. Replace the runBlocking invocation with runTest, and replace System.currentTimeMillis() with currentTime:

   kotlin Copy

   @Test
   fun test() = runTest {
       val startTime = currentTime
       // action
       val totalTime = currentTime - startTime
       // testing result
   }

2. Uncomment the assertions that check the exact virtual time.

3. Don't forget to add @UseExperimental(ExperimentalCoroutinesApi::class).

Solution for task 8 {initial-collapse-state="collapsed" collapsible="true"}

Here are the solutions for the concurrent and channels cases:

fun testConcurrent() = runTest {
    val startTime = currentTime
    val result = loadContributorsConcurrent(MockGithubService, testRequestData)
    Assert.assertEquals("Wrong result for 'loadContributorsConcurrent'", expectedConcurrentResults.users, result)
    val totalTime = currentTime - startTime

    Assert.assertEquals(
        "The calls run concurrently, so the total virtual time should be 2200 ms: " +
                "1000 for repos request plus max(1000, 1200, 800) = 1200 for concurrent contributors requests)",
        expectedConcurrentResults.timeFromStart, totalTime
    )
}

First, check that the results are available exactly at the expected virtual time, and then check the results
themselves:

fun testChannels() = runTest {
    val startTime = currentTime
    var index = 0
    loadContributorsChannels(MockGithubService, testRequestData) { users, _ ->
        val expected = concurrentProgressResults[index++]
        val time = currentTime - startTime
        Assert.assertEquals(
            "Expected intermediate results after ${expected.timeFromStart} ms:",
            expected.timeFromStart, time
        )
        Assert.assertEquals("Wrong intermediate results after $time:", expected.users, users)
    }
}

The first intermediate result for the last version with channels becomes available sooner than the progress version, and you
can see the difference in tests that use virtual time.

The tests for the remaining "suspend" and "progress" tasks are very similar – you can find them in the project's
solutions branch.

{style="tip"}

What's next Copy Chapter

• Check out the Asynchronous Programming with Kotlin workshop at KotlinConf.
• Find out more about using virtual time and the experimental testing package.

Topics Coroutines Basics (`docs/topics/coroutines-basics.md`) Copy Section

https://github.com/Kotlin/kotlinx.coroutines/edit/master/docs/topics/

To create applications that perform multiple tasks at once, a concept known as concurrency,
Kotlin uses coroutines. A coroutine is a suspendable computation that lets you write concurrent code in a clear, sequential style.
Coroutines can run concurrently with other coroutines and potentially in parallel.

On the JVM and in Kotlin/Native, all concurrent code, such as coroutines, runs on threads, managed by the operating system.
Coroutines can suspend their execution instead of blocking a thread.
This allows one coroutine to suspend while waiting for some data to arrive and another coroutine to run on the same thread, ensuring effective resource utilization.

{width="700"}

For more information about the differences between coroutines and threads, see Comparing coroutines and JVM threads.

Suspending functions Copy Chapter

The most basic building block of coroutines is the suspending function.
It allows a running operation to pause and resume later without affecting the structure of your code.

To declare a suspending function, use the suspend keyword:

suspend fun greet() {
    println("Hello world from a suspending function")
}

You can only call a suspending function from another suspending function.
To call suspending functions at the entry point of a Kotlin application, mark the main() function with the suspend keyword:

suspend fun main() {
    showUserInfo()
}

suspend fun showUserInfo() {
    println("Loading user...")
    greet()
    println("User: John Smith")
}

suspend fun greet() {
    println("Hello world from a suspending function")
}

{kotlin-runnable="true"}

This example doesn't use concurrency yet, but by marking the functions with the suspend keyword,
you allow them to call other suspending functions and run concurrent code inside.

While the suspend keyword is part of the core Kotlin language, most coroutine features
are available through the kotlinx.coroutines library.

Add the kotlinx.coroutines library to your project Copy Chapter

To include the kotlinx.coroutines library in your project, add the corresponding dependency configuration based on your build tool:

// build.gradle.kts
repositories {
    mavenCentral()
}

dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:%coroutinesVersion%")
}

// build.gradle
repositories {
    mavenCentral()
}

dependencies {
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:%coroutinesVersion%'
}

<project>
    <dependencies>
        <dependency>
            <groupId>org.jetbrains.kotlinx</groupId>
            <artifactId>kotlinx-coroutines-core</artifactId>
            <version>%coroutinesVersion%</version>
        </dependency>
    </dependencies>
    ...
</project>

Create your first coroutines Copy Chapter

The examples on this page use the explicit this expression with the coroutine builder functions CoroutineScope.launch() and CoroutineScope.async().
These coroutine builders are extension functions on CoroutineScope, and the this expression refers to the current CoroutineScope as the receiver.

For a practical example, see Extract coroutine builders from the coroutine scope.

{style="note"}

To create a coroutine in Kotlin, you need the following:

• A suspending function.
• A coroutine scope in which it can run, for example inside the withContext() function.
• A coroutine builder like CoroutineScope.launch() to start it.
• A dispatcher to control which threads it uses.

Let's look at an example that uses multiple coroutines in a multithreaded environment:

1. Import the kotlinx.coroutines library:

   kotlin Copy

   import kotlinx.coroutines.*

2. Mark functions that can pause and resume with the suspend keyword:

   kotlin Copy

   suspend fun greet() {
       println("The greet() on the thread: ${Thread.currentThread().name}")
   }
   
   suspend fun main() {}

   While you can mark the main() function as suspend in some projects, it may not be possible when integrating with existing code or using a framework.
   In that case, check the framework's documentation to see if it supports calling suspending functions.
   If not, use runBlocking() to call them by blocking the current thread.

   {style="note"}

3. Add the delay() function to simulate a suspending task, such as fetching data or writing to a database:

   kotlin Copy

   suspend fun greet() {
       println("The greet() on the thread: ${Thread.currentThread().name}")
       delay(1000L)
   }

4. Use withContext(Dispatchers.Default) to define an entry point for multithreaded concurrent code that runs on a shared thread pool:

   kotlin Copy

   suspend fun main() {
       withContext(Dispatchers.Default) {
           // Add the coroutine builders here
       }
   }

   The suspending withContext() function is typically used for context switching, but in this example,
   it also defines a non-blocking entry point for concurrent code.
   It uses the Dispatchers.Default dispatcher to run code on a shared thread pool for multithreaded execution.
   By default, this pool uses up to as many threads as there are CPU cores available at runtime, with a minimum of two threads.

   The coroutines launched inside the withContext() block share the same coroutine scope, which ensures structured concurrency.

   {style="note"}

5. Use a coroutine builder function like CoroutineScope.launch() to start the coroutine:

   kotlin Copy

   suspend fun main() {
       withContext(Dispatchers.Default) { // this: CoroutineScope
           // Starts a coroutine inside the scope with CoroutineScope.launch()
           this.launch { greet() }
           println("The withContext() on the thread: ${Thread.currentThread().name}")
       }
   }

6. Combine these pieces to run multiple coroutines at the same time on a shared pool of threads:

   kotlin Copy

   // Imports the coroutines library
   import kotlinx.coroutines.*
   
   // Imports the kotlin.time.Duration to express duration in seconds
   import kotlin.time.Duration.Companion.seconds
   
   // Defines a suspending function
   suspend fun greet() {
       println("The greet() on the thread: ${Thread.currentThread().name}")
       // Suspends for 1 second and releases the thread
       delay(1.seconds) 
       // The delay() function simulates a suspending API call here
       // You can add suspending API calls here like a network request
   }
   
   suspend fun main() {
       // Runs the code inside this block on a shared thread pool
       withContext(Dispatchers.Default) { // this: CoroutineScope
           this.launch() {
               greet()
           }
   
           // Starts another coroutine
           this.launch() {
               println("The CoroutineScope.launch() on the thread: ${Thread.currentThread().name}")
               delay(1.seconds)
               // The delay function simulates a suspending API call here
               // You can add suspending API calls here like a network request
           }
   
           println("The withContext() on the thread: ${Thread.currentThread().name}")
       }
   }

   {kotlin-runnable="true"}

Try running the example multiple times.
You may notice that the output order and thread names may change each time you run the program, because the OS decides when threads run.

You can display coroutine names next to thread names in the output of your code for additional information.
To do so, pass the -Dkotlinx.coroutines.debug VM option in your build tool or IDE run configuration.

See Debug coroutines for more information.

{style="tip"}

Coroutine scope and structured concurrency Copy Chapter

When you run many coroutines in an application, you need a way to manage them as groups.
Kotlin coroutines rely on a principle called structured concurrency to provide this structure.

According to this principle, coroutines form a tree hierarchy of parent and child tasks with linked lifecycles.
A coroutine's lifecycle is the sequence of states from its creation until completion, failure, or cancellation.

A parent coroutine waits for its children to complete before it finishes.
If the parent coroutine fails or gets canceled, all its child coroutines are recursively canceled too.
Keeping coroutines connected this way makes cancellation and error handling predictable and safe.

To maintain structured concurrency, new coroutines can only be launched in a CoroutineScope that defines and manages their lifecycle.
The CoroutineScope includes the coroutine context, which defines the dispatcher and other execution properties.
When you start a coroutine inside another coroutine, it automatically becomes a child of its parent scope.

Calling a coroutine builder function, such as CoroutineScope.launch() on a CoroutineScope, starts a child coroutine of the coroutine associated with that scope.
Inside the builder's block, the receiver is a nested CoroutineScope, so any coroutines you launch there become its children.

Create a coroutine scope with the `coroutineScope()` function Copy Section

To create a new coroutine scope with the current coroutine context, use the
coroutineScope() function.
This function creates a root coroutine of the coroutine subtree.
It's the direct parent of coroutines launched inside the block and the indirect parent of any coroutines they launch.
coroutineScope() executes the suspending block and waits until the block and any coroutines launched in it complete.

Here's an example:

// Imports the kotlin.time.Duration to express duration in seconds
import kotlin.time.Duration.Companion.seconds

import kotlinx.coroutines.*

// If the coroutine context doesn't specify a dispatcher,
// CoroutineScope.launch() uses Dispatchers.Default
//sampleStart
suspend fun main() {
    // Root of the coroutine subtree
    coroutineScope { // this: CoroutineScope
        this.launch {
            this.launch {
                delay(2.seconds)
                println("Child of the enclosing coroutine completed")
            }
            println("Child coroutine 1 completed")
        }
        this.launch {
            delay(1.seconds)
            println("Child coroutine 2 completed")
        }
    }
    // Runs only after all children in the coroutineScope have completed
    println("Coroutine scope completed")
}
//sampleEnd

{kotlin-runnable="true"}

Since no dispatcher is specified in this example, the CoroutineScope.launch() builder functions in the coroutineScope() block inherit the current context.
If that context doesn't have a specified dispatcher, CoroutineScope.launch() uses Dispatchers.Default, which runs on a shared pool of threads.

Extract coroutine builders from the coroutine scope Copy Section

In some cases, you may want to extract coroutine builder calls, such as CoroutineScope.launch(), into separate functions.

Consider the following example:

suspend fun main() {
    coroutineScope { // this: CoroutineScope
        // Calls CoroutineScope.launch() where CoroutineScope is the receiver
        this.launch { println("1") }
        this.launch { println("2") }
    } 
}

You can also write this.launch without the explicit this expression, as launch.
These examples use explicit this expressions to highlight that it's an extension function on CoroutineScope.

For more information on how lambdas with receivers work in Kotlin, see Function literals with receiver.

{style="tip"}

The coroutineScope() function takes a lambda with a CoroutineScope receiver.
Inside this lambda, the implicit receiver is a CoroutineScope, so builder functions like CoroutineScope.launch() and CoroutineScope.async() resolve as
extension functions on that receiver.

To extract the coroutine builders into another function, that function must declare a CoroutineScope receiver, otherwise a compilation error occurs:

import kotlinx.coroutines.*
//sampleStart
suspend fun main() {
    coroutineScope {
        launchAll()
    }
}

fun CoroutineScope.launchAll() { // this: CoroutineScope
    // Calls .launch() on CoroutineScope
    this.launch { println("1") }
    this.launch { println("2") } 
}
//sampleEnd
/* -- Calling launch without declaring CoroutineScope as the receiver results in a compilation error --

fun launchAll() {
    // Compilation error: this is not defined
    this.launch { println("1") }
    this.launch { println("2") }
}
 */

{kotlin-runnable="true"}

In this example, the launchAll() function doesn't need the suspend keyword, because it only starts the coroutines in the current CoroutineScope and then returns immediately.
Only mark functions with suspend when it makes sense for them to pause and resume before returning.

Coroutine builder functions Copy Chapter

A coroutine builder function is a function that accepts a suspend lambda that defines a coroutine to run.
Here are some examples:

• CoroutineScope.launch()
• CoroutineScope.async()
• runBlocking()
• withContext()
• coroutineScope()

Coroutine builder functions require a CoroutineScope to run in.
This can be an existing scope or one you create with helper functions such as coroutineScope(), runBlocking(), or withContext().
Each builder defines how the coroutine starts and how you interact with its result.

`CoroutineScope.launch()` Copy Section

The CoroutineScope.launch() coroutine builder function is an extension function on CoroutineScope.
It starts a new coroutine without blocking the rest of the scope, inside an existing coroutine scope.

Use CoroutineScope.launch() to run a task alongside other work when the result isn't needed or you don't want to wait for it:

// Imports the kotlin.time.Duration to enable expressing duration in milliseconds
import kotlin.time.Duration.Companion.milliseconds

import kotlinx.coroutines.*

suspend fun main() {
    withContext(Dispatchers.Default) {
        performBackgroundWork()
    }
}

//sampleStart
suspend fun performBackgroundWork() = coroutineScope { // this: CoroutineScope
    // Starts a coroutine that runs without blocking the scope
    this.launch {
        // Suspends to simulate background work
        delay(100.milliseconds)
        println("Sending notification in background")
    }

    // Main coroutine continues while a previous one suspends
    println("Scope continues")
}
//sampleEnd

{kotlin-runnable="true"}

After running this example, you can see that the main() function isn't blocked by CoroutineScope.launch() and keeps running other code while the coroutine works in the background.

The CoroutineScope.launch() function returns a Job handle.
Use this handle to wait for the launched coroutine to complete.
For more information, see Cancellation and timeouts.

{style="tip"}

`CoroutineScope.async()` Copy Section

The CoroutineScope.async() coroutine builder function is an extension function on CoroutineScope.
It starts a concurrent computation inside an existing coroutine scope and returns a Deferred handle that represents an eventual result.
Use the .await() function to suspend the code until the result is ready:

// Imports the kotlin.time.Duration to enable expressing duration in milliseconds
import kotlin.time.Duration.Companion.milliseconds

import kotlinx.coroutines.*

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) { // this: CoroutineScope
    // Starts downloading the first page
    val firstPage = this.async {
        delay(50.milliseconds)
        "First page"
    }

    // Starts downloading the second page in parallel
    val secondPage = this.async {
        delay(100.milliseconds)
        "Second page"
    }

    // Awaits both results and compares them
    val pagesAreEqual = firstPage.await() == secondPage.await()
    println("Pages are equal: $pagesAreEqual")
}
//sampleEnd

{kotlin-runnable="true"}

`runBlocking()` Copy Section

The runBlocking() coroutine builder function creates a coroutine scope and blocks the current thread until
the coroutines launched in that scope finish.

Use runBlocking() only when there is no other option to call suspending code from non-suspending code:

import kotlin.time.Duration.Companion.milliseconds
import kotlinx.coroutines.*

// A third-party interface you can't change
interface Repository {
    fun readItem(): Int
}

object MyRepository : Repository {
    override fun readItem(): Int {
        // Bridges to a suspending function
        return runBlocking {
            myReadItem()
        }
    }
}

suspend fun myReadItem(): Int {
    delay(100.milliseconds)
    return 4
}

Coroutine dispatchers Copy Chapter

A coroutine dispatcher controls which thread or thread pool coroutines use for their execution.
Coroutines aren't always tied to a single thread.
They can pause on one thread and resume on another, depending on the dispatcher.
This lets you run many coroutines at the same time without allocating a separate thread for every coroutine.

Even though coroutines can suspend and resume on different threads,
values written before the coroutine suspends are still guaranteed to be available within the same coroutine when it resumes.

{style="tip"}

A dispatcher works together with the coroutine scope to define when coroutines run and where they run.
While the coroutine scope controls the coroutine's lifecycle, the dispatcher controls which threads are used for execution.

You don't have to specify a dispatcher for every coroutine.
By default, coroutines inherit the dispatcher from their parent scope.
You can specify a dispatcher to run a coroutine in a different context.

If the coroutine context doesn't include a dispatcher, coroutine builders use Dispatchers.Default.

{style="note"}

The kotlinx.coroutines library includes different dispatchers for different use cases.
For example, Dispatchers.Default runs coroutines on a shared pool of threads, performing work in the background,
separate from the main thread. This makes it an ideal choice for CPU-intensive operations like data processing.

To specify a dispatcher for a coroutine builder like CoroutineScope.launch(), pass it as an argument:

suspend fun runWithDispatcher() = coroutineScope { // this: CoroutineScope
    this.launch(Dispatchers.Default) {
        println("Running on ${Thread.currentThread().name}")
    }
}

Alternatively, you can use a withContext() block to run all code in it on a specified dispatcher:

// Imports the kotlin.time.Duration to enable expressing duration in milliseconds
import kotlin.time.Duration.Companion.milliseconds

import kotlinx.coroutines.*

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) { // this: CoroutineScope
    println("Running withContext block on ${Thread.currentThread().name}")

    val one = this.async {
        println("First calculation starting on ${Thread.currentThread().name}")
        val sum = (1L..500_000L).sum()
        delay(200L)
        println("First calculation done on ${Thread.currentThread().name}")
        sum
    }

    val two = this.async {
        println("Second calculation starting on ${Thread.currentThread().name}")
        val sum = (500_001L..1_000_000L).sum()
        println("Second calculation done on ${Thread.currentThread().name}")
        sum
    }

    // Waits for both calculations and prints the result
    println("Combined total: ${one.await() + two.await()}")
}
//sampleEnd

{kotlin-runnable="true"}

To learn more about coroutine dispatchers and their uses, including other dispatchers like Dispatchers.IO and Dispatchers.Main, see Coroutine context and dispatchers.

Comparing coroutines and JVM threads Copy Chapter

While coroutines are suspendable computations that run code concurrently like threads on the JVM, they work differently under the hood.

A thread is managed by the operating system. Threads can run tasks in parallel on multiple CPU cores and represent a standard approach to concurrency on the JVM.
When you create a thread, the operating system allocates memory for its stack and uses the kernel to switch between threads.
This makes threads powerful but also resource-intensive.
Each thread usually needs a few megabytes of memory, and typically the JVM can only handle a few thousand threads at once.

On the other hand, a coroutine isn't bound to a specific thread.
It can suspend on one thread and resume on another, so many coroutines can share the same thread pool.
When a coroutine suspends, the thread isn't blocked and remains free to run other tasks.
This makes coroutines much lighter than threads and allows running millions of them in one process without exhausting system resources.

{width="700"}

Let's look at an example where 50,000 coroutines each wait five seconds and then print a period (.):

import kotlin.time.Duration.Companion.seconds
import kotlinx.coroutines.*

suspend fun main() {
    withContext(Dispatchers.Default) {
        // Launches 50,000 coroutines that each wait five seconds, then print a period
        printPeriods()
    }
}

//sampleStart
suspend fun printPeriods() = coroutineScope { // this: CoroutineScope
    // Launches 50,000 coroutines that each wait five seconds, then print a period
    repeat(50_000) {
        this.launch {
            delay(5.seconds)
            print(".")
        }
    }
}
//sampleEnd

{kotlin-runnable="true" kotlin-min-compiler-version="1.3"}

Now let's look at the same example using JVM threads:

import kotlin.concurrent.thread

fun main() {
    repeat(50_000) {
        thread {
            Thread.sleep(5000L)
            print(".")
        }
    }
}

{kotlin-runnable="true" validate="false"}

Running this version uses much more memory because each thread needs its own memory stack.
For 50,000 threads, that can be up to 100 GB, compared to roughly 500 MB for the same number of coroutines.

Depending on your operating system, JDK version, and settings,
the JVM thread version may throw an out-of-memory error or slow down thread creation to avoid running too many threads at once.

What's next Copy Chapter

• Discover more about combining suspending functions in Composing suspending functions.
• Learn how to cancel coroutines and handle timeouts in Cancellation and timeouts.
• Dive deeper into coroutine execution and thread management in Coroutine context and dispatchers.
• Learn how to return multiple asynchronously computed values in Flows.

Topics Coroutines Cancellation (`docs/topics/coroutines-cancellation.md`) Copy Section

https://github.com/Kotlin/kotlinx.coroutines/edit/master/docs/topics/

Cancellation lets you request to stop a coroutine before it completes.
It stops work that's no longer needed, such as when a user closes a window or navigates away in a user interface while a coroutine is still running.

You can use cancellation to release resources early and to stop a coroutine from accessing objects past their disposal.
You can also use it to stop long-running coroutines that perform repeated work, such as:

• Sending heartbeats.
• Running scheduled tasks.
• Updating a state to reflect the newest reading, such as in a clock UI.

Cancellation works through the Job handle, which represents the lifecycle of a coroutine and its parent-child relationships.
Job allows you to check whether the coroutine is active and allows you to cancel it, along with its children, as defined by structured concurrency.

Cancel coroutines Copy Chapter

A coroutine is canceled when the cancel() function is invoked on its Job handle.
Coroutine builder functions such as
.launch() return a Job. The .async()
function returns a Deferred, which
implements Job and supports the same cancellation behavior.

You can call the cancel() function manually, or it can be invoked automatically through cancellation propagation when a parent coroutine is canceled.

When a coroutine is canceled, it throws a CancellationException the next time it checks for cancellation.
Suspending functions in the kotlinx.coroutines library, such as the delay() function, check for cancellation when they suspend.

You can use the awaitCancellation() function to suspend a coroutine until it's canceled.
It's equivalent to calling delay(Duration.INFINITE).

For more information about how and when coroutines check for cancellation, see Suspension points and cancellation.

{style="tip"}

Here's an example of how to manually cancel a coroutine:

import kotlinx.coroutines.*

suspend fun main() {
//sampleStart
withContext(Dispatchers.Default) {
    // Used as a signal that the coroutine has started running
    val childStarted = CompletableDeferred<Unit>()
    
    val childJob: Job = launch {
        println("The coroutine has started")

        // Completes the CompletableDeferred,
        // signaling that the coroutine has started running
        childStarted.complete(Unit)
        try {
            // Suspends indefinitely
            // This call will never return unless the coroutine is canceled
            awaitCancellation()
        } catch (e: CancellationException) {
            println("The coroutine was canceled: $e")
          
            // Always rethrow cancellation exceptions!
            throw e
        }
        println("This line will never be executed")
    }
  
    // Waits for the coroutine to start before canceling it
    childStarted.await()

    // Cancels the coroutine,
    // so awaitCancellation() throws a CancellationException
    childJob.cancel()
}
// Coroutine builders such as withContext() or coroutineScope()
// wait for all child coroutines to complete,
// even when the children are canceled
println("All coroutines have completed")
//sampleEnd
}

{kotlin-runnable="true" id="manual-cancellation-example"}

In this example, CompletableDeferred is used as a signal that the coroutine has started running.
The coroutine calls complete() when it starts executing, and await() only returns once that CompletableDeferred is completed.
You don't need this check to cancel a coroutine.
It's included here to make the example reproducible by ensuring that the coroutine starts and prints its messages before it's canceled.

Because Deferred implements Job, cancellation works the same way for coroutines created by the async() coroutine builder function:

val deferred = async { /* ... */ }
deferred.cancel()

Catching CancellationException can break the cancellation propagation.
If you must catch it, rethrow it to let the cancellation propagate correctly through the coroutine hierarchy.

For more information, see Coroutine exceptions handling.

{style="warning"}

Cancellation propagation Copy Section

Structured concurrency ensures that canceling a coroutine also cancels all of its children.
This prevents child coroutines from continuing their work after their parent coroutine is canceled.

Here's an example:

import kotlinx.coroutines.*

suspend fun main() {
    withContext(Dispatchers.Default) {
//sampleStart
// Used as a signal that the child coroutines have been launched
val childrenLaunched = CompletableDeferred<Unit>()

// Launches two child coroutines
val parentJob = launch {
    launch {
        println("Child coroutine 1 has started running")
        try {
            awaitCancellation()
        } finally {
            println("Child coroutine 1 has been canceled")
        }
    }
    launch {
        println("Child coroutine 2 has started running")
        try {
            awaitCancellation()
        } finally {
            println("Child coroutine 2 has been canceled")
        }
    }
    // Completes the CompletableDeferred,
    // signaling that the child coroutines have been launched
    childrenLaunched.complete(Unit)
}
// Waits for the parent coroutine to signal that it has launched
// all of its children
childrenLaunched.await()

// Cancels the parent coroutine, which cancels all its children
parentJob.cancel()
//sampleEnd
    }
}

{kotlin-runnable="true" id="cancellation-propagation-example"}

In this example, each child coroutine uses a finally block, so the code inside it runs when the coroutine is canceled.
Here, CompletableDeferred signals that the child coroutines are launched before they're canceled, but it doesn't guarantee that they start running.
If they're canceled first, nothing is printed.

Make coroutines react to cancellation {id="cancellation-is-cooperative"} Copy Chapter

In Kotlin, coroutine cancellation is cooperative.
Coroutines react to cancellation only when they cooperate by suspending or checking for cancellation explicitly.

In this section, you can learn how adding suspension points, such as calls to the yield() function, lets coroutines react to cancellation.

Suspension points and cancellation Copy Section

When a coroutine is canceled, it continues running until it reaches a point in the code where it may suspend, also known as a suspension point.
If the coroutine suspends there, the suspending function checks whether it has been canceled.
If it has, the coroutine stops and throws CancellationException.

A call to a suspend function is a suspension point, but it doesn't always suspend.
For example, when awaiting a Deferred result, the coroutine only suspends if that Deferred isn't completed yet.

Here's an example using common suspending functions that suspend, enabling the coroutine to check and stop when it's canceled:

import kotlinx.coroutines.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.channels.Channel
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration

suspend fun main() {
//sampleStart
withContext(Dispatchers.Default) {
    val childJobs = listOf(
        launch {
            // Suspends until canceled
            awaitCancellation()
        },
        launch {
            // Suspends until canceled
            delay(Duration.INFINITE)
        },
        launch {
            val channel = Channel<Int>()
            // Suspends while waiting for a value that's never sent
            channel.receive()
        },
        launch {
            val deferred = CompletableDeferred<Int>()
            // Suspends while waiting for a value that's never completed
            deferred.await()
        },
        launch {
            val mutex = Mutex(locked = true)
            // Suspends while waiting for a mutex that remains locked indefinitely
            mutex.lock()
        }
    )
    
    // Gives the child coroutines time to start and suspend
    delay(100.milliseconds)
    
    // Cancels all child coroutines
    childJobs.forEach { it.cancel() }
}
println("All child jobs completed!")
//sampleEnd
}

{kotlin-runnable="true" id="suspension-points-example"}

All suspending functions in the kotlinx.coroutines library cooperate with cancellation because they use suspendCancellableCoroutine() internally, which checks for cancellation when the coroutine suspends.
In contrast, custom suspending functions that use suspendCoroutine() don't react to cancellation.

{style="tip"}

The `yield()` suspending function Copy Section

If a coroutine doesn't suspend, other coroutines can't run on the same thread until it completes.
As a result, coroutines that don't suspend run sequentially on that thread.
If a coroutine doesn't suspend for a long time, it also doesn't stop when it's canceled.

In CPU-intensive computations and other code that runs for a long time without suspending, call the yield() function periodically.
This function releases the current thread and gives other coroutines a chance to run on it.
It also ensures that the coroutine checks for cancellation regularly. If the coroutine is canceled, the yield() function throws a CancellationException.

Here's an example:

import kotlinx.coroutines.*

fun main() {
//sampleStart
// runBlocking uses the current thread for running all coroutines
runBlocking {
    val coroutineCount = 5
    repeat(coroutineCount) { coroutineIndex ->
        launch {
            val id = coroutineIndex + 1
            repeat(5) { iterationIndex ->
                val iteration = iterationIndex + 1
                // Suspends temporarily to give other coroutines a chance to run
                // Without this, the coroutines run sequentially
                yield()
                // Prints the coroutine index and iteration index
                println("$id * $iteration = ${id * iteration}")
            }
        }
    }
}
//sampleEnd
}

{kotlin-runnable="true" id="yield-example"}

In this example, each coroutine uses yield() to let other coroutines run between iterations.

Check for cancellation explicitly Copy Section

You can check for cancellation explicitly, which lets long-running code react to cancellation without suspending.
A long-running coroutine that doesn't suspend can prevent other coroutines on the same thread from running until it completes.
Unless this behavior is intentional for your use case, use the yield() function instead.

Depending on the API, the check either returns a Boolean value or throws an exception:

• The isActive property returns false when the coroutine is canceled.
• The ensureActive() function throws a CancellationException when the coroutine is canceled.

Interrupt blocking code when coroutines are canceled Copy Section

On the JVM, some blocking functions, such as Thread.sleep() or BlockingQueue.take(), can block the current thread.
These blocking functions can be interrupted, which stops them prematurely.
However, when you call them from a coroutine, cancellation doesn't interrupt the thread.

To interrupt the thread when canceling a coroutine, wrap the blocking code in the runInterruptible() function:

import kotlinx.coroutines.*

suspend fun main() {
//sampleStart
withContext(Dispatchers.Default) {
    val childStarted = CompletableDeferred<Unit>()
    val childJob = launch {
        try {
            // Cancellation triggers a thread interruption
            runInterruptible {
                childStarted.complete(Unit)
                try {
                    // Blocks the current thread for a very long time
                    Thread.sleep(Long.MAX_VALUE)
                } catch (e: InterruptedException) {
                    println("Thread interrupted (Java): $e")
                    throw e
                }
            }
        } catch (e: CancellationException) {
            println("Coroutine canceled (Kotlin): $e")
            throw e
        }
    }
    childStarted.await()

    // Cancels the coroutine and interrupts the thread executing Thread.sleep()
    childJob.cancel()
}
//sampleEnd
}

{kotlin-runnable="true" id="interrupt-cancellation-example"}

Handle values safely when canceling coroutines Copy Chapter

When a suspended coroutine is canceled, it resumes with a CancellationException instead of returning any values, even if those values are already available.
This behavior is called prompt cancellation.
It prevents your code from continuing in a canceled coroutine's scope, such as updating a screen that's already closed.

Here's an example:

// Defines a coroutine scope that uses the UI thread
class ScreenWithButtons(private val scope: CoroutineScope) {
    fun loadAndUpdateButtons(filename: String) {
        scope.launch {
            // withContext() checks for cancellation before entering the block
            // and after the block returns
            val buttonNames = withContext(Dispatchers.IO) {
                // This is a blocking call that doesn't react to cancellation
                readLines(filename)
            }
            
            // It's safe to call updateUi()
            // because withContext() doesn't return if the coroutine is canceled,
            // and no code running on the UI thread can dispose of the buttons before this call
            updateUi(buttonNames)
        }
    }

    // Call this function only from the UI thread because it accesses the buttons
    // Throws an exception if called after the buttons are disposed
    private fun updateUi(buttonNames: List<String>) {
        // Placeholder code that updates the buttons with specified names
    }

    // Call this function only from the UI thread
    fun leaveScreen() {
        // Cancels the scope when leaving the screen
        // You can no longer update the UI
        scope.cancel()
    }
}

// UI controller code
setHandler(Event.ScreenClosed) {
    // Runs on the UI thread
    screenWithButtons.leaveScreen()
    buttons.dispose()
}

In this example, withContext(Dispatchers.IO) cooperates with cancellation and prevents updateUi() from running if the
leaveScreen() function cancels the coroutine before withContext(Dispatchers.IO) returns the button names.

While prompt cancellation prevents using values after they're no longer valid, it can also stop your code while an important value is still in use, which might lead to losing that value.
This can happen when a coroutine receives a value, such as an AutoCloseable resource, but is canceled before it can reach the part of the code that closes it.
To prevent this, keep cleanup logic in a place that's guaranteed to run even when the coroutine receiving the value is canceled.

Here's an example:

/* Detailed source-code truncated for AI context efficiency. */

In this example, storing the BufferedReader in a variable and closing it in the finally block ensures the resource is released even if the coroutine is canceled.

Run non-cancelable blocks Copy Section

You can prevent cancellation from affecting certain parts of a coroutine.
To do so, pass NonCancellable as an argument to the withContext() coroutine builder function.

Avoid using NonCancellable with other coroutine builders like .launch() or .async(). Doing so disrupts structured concurrency by breaking the parent-child relationship.

{style="warning"}

NonCancellable is useful when you need to ensure that certain operations, such as closing resources with a suspending close() function,
complete even if the coroutine is canceled before they finish.

Here's an example:

import kotlinx.coroutines.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
val serviceStarted = CompletableDeferred<Unit>()

fun startService() {
    println("Starting the service...")
    serviceStarted.complete(Unit)
}

suspend fun shutdownServiceAndWait() {
    println("Shutting down...")
    delay(100.milliseconds)
    println("Successfully shut down!")
}

suspend fun main() {
    withContext(Dispatchers.Default) {
        val childJob = launch {
            startService()
            try {
                awaitCancellation()
            } finally {
                withContext(NonCancellable) {
                    // Without withContext(NonCancellable),
                    // this function doesn't complete because the coroutine is canceled
                    shutdownServiceAndWait()
                }
            }
        }
        serviceStarted.await()
        childJob.cancel()
    }
    println("Exiting the program")
}
//sampleEnd

{kotlin-runnable="true" id="noncancellable-blocks-example"}

Timeout Copy Chapter

A timeout allows you to automatically cancel a coroutine after a specified duration.
You can use it to stop operations that take too long.

For example, if a request to download a picture from a server times out, you can retry or fallback to the local cache.

To specify a timeout, use the withTimeoutOrNull() function with a Duration:

import kotlinx.coroutines.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun slowOperation(): String {
    try {
        delay(300.milliseconds)
        return "A"
    } catch (e: CancellationException) {
        println("The slow operation has been canceled: $e")
        throw e
    }
}

suspend fun fastOperation(): String {
    try {
        delay(15.milliseconds)
        return "B"
    } catch (e: CancellationException) {
        println("The fast operation has been canceled: $e")
        throw e
    }
}

suspend fun main() {
    withContext(Dispatchers.Default) {
        val slow = withTimeoutOrNull(100.milliseconds) {
            slowOperation()
        }
        println("The slow operation finished with $slow")
        val fast = withTimeoutOrNull(100.milliseconds) {
            fastOperation()
        }
        println("The fast operation finished with $fast")
    }
}
//sampleEnd

{kotlin-runnable="true" id="timeout-example"}

If the timeout exceeds the specified Duration, withTimeoutOrNull() returns null.

Topics Coroutines Debugging (`docs/topics/coroutines-debugging.md`) Copy Section

https://github.com/Kotlin/kotlinx.coroutines/edit/master/docs/topics/

Debugging applications that use coroutines can be challenging because multiple coroutines can run concurrently, suspend on one thread, and resume on another.
Their execution order and the threads they use can also change between runs, making it difficult to follow the execution of a particular coroutine.

On the JVM, you can use the following features to make debugging coroutines easier:

• Debug mode adds a unique name to each coroutine so you can identify it in a debugger and diagnostic output.
• Stack trace recovery adds information about where a coroutine receives an exception instead of an expected result.
• The debug agent tracks active coroutines, reports their state, and more.

Debug mode and stack trace recovery are available in the kotlinx-coroutines-core module.
The debug agent is available in the kotlinx-coroutines-debug module.

The debug agent isn't supported on Android.

{style="note"}

Enable debug mode Copy Chapter

Debug mode assigns a unique name to every launched coroutine.
You can see the coroutine names in a Java debugger, in the coroutine's string representation, and in a thread's name while it runs the coroutine.
Debug mode has negligible runtime overhead, so you can keep it enabled to simplify logging and diagnostics.

When you run your code with Java assertions enabled, the kotlinx.coroutines library automatically enables debug mode.
Unit tests run with assertions enabled by default, so you don't need to enable debug mode explicitly for them.

To enable debug mode explicitly, configure your build tool, such as Gradle or Maven, or IDE run configuration to pass the -Dkotlinx.coroutines.debug argument to the JVM that runs your application.

In IntelliJ IDEA, follow these steps to enable debug mode:

1. In the Run widget, select the run/debug configuration you want to update, then select More Actions | Edit:

   {width="600"}

   If you don't have a run/debug configuration, select Current File in the Run widget, then select More Actions | Run with Parameters to open the run configuration settings.

   {width="600"}

   {style="note"}

2. In the Run/Debug Configurations dialog, enter -Dkotlinx.coroutines.debug in the VM options field, and click OK:

   {width="600"}

Stack trace recovery Copy Chapter

When a coroutine receives an exception from another coroutine through a suspending function such as Deferred.await(),
the exception's stack trace doesn't contain the stack frames from the receiving coroutine.
Without these stack frames, the stack trace doesn't show where Deferred.await() is called or which functions lead to that call, which can make debugging difficult.

The kotlinx.coroutines library adds this information using stack trace recovery, which creates a copy of the exception with additional stack frames.

When the receiving coroutine resumes, it throws the copy instead of the original exception.
The original exception becomes the cause of the copy.
If the original exception has suppressed exceptions, they remain attached to it instead of being copied.
Keeping them attached to the original exception can prevent cycles in the exception chain and crashes in some frameworks.

Debug mode enables stack trace recovery by default.
To disable stack trace recovery in debug mode, pass the -Dkotlinx.coroutines.stacktrace.recovery=false VM option.

Here's an example to demonstrate the difference between stack traces with and without stack trace recovery:

import kotlinx.coroutines.*

object UserProfileService :
    CoroutineScope by CoroutineScope(CoroutineName("UserProfileService")) {

    private fun parseUserProfile(): String {
        error("Invalid user profile")
    }

    private fun loadUserProfile(): String {
        return parseUserProfile()
    }

    // Runs in the coroutine that calls this function
    suspend fun awaitUserProfile() {
        // Starts a new coroutine
        val userProfile = async(Dispatchers.Default) {
            // The new coroutine throws the exception
            loadUserProfile()
        }

        // The coroutine running awaitUserProfile()
        // receives the exception through the await() function
        userProfile.await()
    }
}

suspend fun main() {
    UserProfileService.awaitUserProfile()
}

In this example, the parseUserProfile() function throws the exception in the coroutine started by the .async() builder function.
The coroutine that calls awaitUserProfile() receives the exception through the Deferred.await() function.

With stack trace recovery disabled, the stack trace shows where parseUserProfile() throws the exception in the coroutine created by the .async() function,
but it doesn't include the Deferred.await() call in the awaitUserProfile() function:

{width="600"}

With stack trace recovery enabled, the stack trace also includes the Deferred.await() call in the awaitUserProfile() function:

{width="600"}

Stack trace recovery for custom exceptions Copy Section

Stack trace recovery can copy an exception automatically when its class has a public constructor that accepts a message, a cause, both, or no arguments at all.

If you want the kotlinx.coroutines library to recover the stack trace of an exception that requires additional constructor arguments,
such as a line number or an error code, implement the StackTraceRecoverable interface.

The StackTraceRecoverable interface is part of the Kotlin standard library, so you can implement it without adding a dependency on the kotlinx.coroutines library.

The StackTraceRecoverable interface is available on all targets, but the kotlinx.coroutines library uses it for stack trace recovery only on the JVM.

{style="note"}

To implement the interface, override the copyForStackTraceRecovery() function.
In the override, return a new exception instance for stack trace recovery, or null if you don't want the kotlinx.coroutines library to copy the exception.

These APIs are Experimental and require opt-in with the
@OptIn(ExperimentalStdlibCoroutineSupportApi::class) annotation.

Here's an example of a custom exception that preserves a line property when it creates a new instance for stack trace
recovery:

import kotlinx.coroutines.*
import kotlin.coroutines.ExperimentalStdlibCoroutineSupportApi
import kotlin.coroutines.debug.StackTraceRecoverable

@OptIn(ExperimentalStdlibCoroutineSupportApi::class)
class FileEditException
// The implementation requires a private constructor
// to pass the cause to the IllegalStateException constructor
private constructor(
    val line: Int,
    private val detail: String,
    cause: Throwable?,
) : IllegalStateException("When editing line $line: $detail", cause),
    // Implements StackTraceRecoverable for stack trace recovery
    StackTraceRecoverable<FileEditException> {

    constructor(line: Int, detail: String) : this(line, detail, null)

    // Copies the line number and message details
    override fun copyForStackTraceRecovery(): FileEditException =
        FileEditException(line, detail, this)
}

private fun editFile() {
    throw FileEditException(15, "Unexpected token")
}

suspend fun main() {
    supervisorScope {
        // Starts a new coroutine
        val fileEdit = async(Dispatchers.Default) {
            // Throws the original exception
            editFile()
        }
        
        // Stack trace recovery creates a copy of the exception,
        // adds the calling coroutine's stack frames, and throws the copy
        fileEdit.await()
    }
}

With debug mode enabled, the output contains the recovered copy followed by the original exception as its cause.

Exception in thread "main" com.example.FileEditException: When editing line 15: Unexpected token
	at com.example.RecoveryExampleKt.editFile(RecoveryExample.kt:54)
	at com.example.RecoveryExampleKt.access$editFile(RecoveryExample.kt:1)
	at com.example.RecoveryExampleKt$main$2$fileEdit$1.invokeSuspend(RecoveryExample.kt:62)
	at _COROUTINE._BOUNDARY._(CoroutineDebugging.kt:42)
	at com.example.RecoveryExampleKt$main$2.invokeSuspend(RecoveryExample.kt:67)
Caused by: com.example.FileEditException: When editing line 15: Unexpected token
	at com.example.RecoveryExampleKt.editFile(RecoveryExample.kt:54)
	at com.example.RecoveryExampleKt.access$editFile(RecoveryExample.kt:1)
	at com.example.RecoveryExampleKt$main$2$fileEdit$1.invokeSuspend(RecoveryExample.kt:62)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:34)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:100)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:586)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:807)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:717)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:704)

{collapsible="true" collapsed-title="StackTraceRecoverable example output"}

The debug agent Copy Chapter

The kotlinx-coroutines-debug module provides a debug agent for JVM applications.
The agent tracks coroutines as they are created, suspended, and resumed.

The DebugProbes API is the main entry point for the debug agent.
You can use it to print active coroutines and their current state.
The output includes stack traces that show where each coroutine was created and where it is suspended.
You can also use it to print a coroutine dump for the hierarchy of a specific Job or CoroutineScope.

If you enable DebugProbes in a production environment, it can significantly reduce your application's performance when it creates a stack trace for each new coroutine.
To avoid this overhead, set DebugProbes.enableCreationStackTraces to false.

The kotlinx-coroutines-debug module provides automatic BlockHound integration.
You can use it to detect blocking operations in coroutine contexts where they aren't allowed.

For setup instructions, see the BlockHound quick start guide.

{style="note"}

Add the debug agent dependency Copy Section

To use the debug agent in your project, add the kotlinx-coroutines-debug dependency:

// build.gradle.kts
dependencies {
    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-debug:%coroutinesVersion%")
}

<dependency>
    <groupId>org.jetbrains.kotlinx</groupId>
    <artifactId>kotlinx-coroutines-debug</artifactId>
    <version>%coroutinesVersion%</version>
    <scope>test</scope>
</dependency>

Track coroutines with the debug agent Copy Section

To start tracking coroutines with the debug agent, you can either:

• Add -javaagent:/path/to/kotlinx-coroutines-debug-%coroutinesVersion%.jar to your VM options to load the debug agent when the application starts.
• Call the DebugProbes.install() function before starting the coroutines you want to track.

Starting with JDK 21, dynamically loading the debug agent with the DebugProbes.install() function can produce a warning.
To avoid this warning, load the agent with the -javaagent VM option.

{style="note"}

With the debug agent active, you can use the following APIs:

• DebugProbes.dumpCoroutines() prints all active coroutines.
• DebugProbes.dumpCoroutinesInfo() returns information about active coroutines.
• DebugProbes.printJob() prints a coroutine dump for the hierarchy of a Job.
• DebugProbes.printScope() prints a coroutine dump for the hierarchy of a CoroutineScope.

Here's an example that uses the debug agent to print active coroutines and the coroutine hierarchy for a specific Job:

import kotlinx.coroutines.*
import kotlinx.coroutines.debug.*
import kotlin.time.Duration.Companion.seconds

private suspend fun loadAccount() {
    delay(5.seconds)
}

private suspend fun loadPreferences() {
    delay(5.seconds)
}

private suspend fun loadUserProfile() = coroutineScope {
    launch { loadAccount() }
    launch { loadPreferences() }
}

@OptIn(ExperimentalCoroutinesApi::class)
fun main() {
    // Installs the debug agent
    // This is only required if you don't use the -javaagent VM option
    DebugProbes.install()

    runBlocking {
        // Starts a coroutine with two child coroutines
        val loadingJob = launch {
            loadUserProfile()
        }

        // Gives the child coroutines time to suspend
        delay(1.seconds)

        // Prints all active coroutines
        DebugProbes.dumpCoroutines()

        println("============")

        // Prints the loading job and its child coroutines
        DebugProbes.printJob(loadingJob)
    }
}

With debug mode enabled, running the example produces the following output:

/* Detailed source-code truncated for AI context efficiency. */

{collapsible="true" collapsed-title="Debug mode example output"}

Print active coroutines when JUnit tests time out Copy Section

You can set a timeout for JUnit tests with the corresponding CoroutinesTimeout API, depending on the JUnit version.
The API installs debug probes automatically.
If a test doesn't complete before the timeout, it prints all active coroutines and their stack traces and fails the test.

JUnit 4

To set a timeout for JUnit 4 tests and print all active coroutines and their stack traces if they exceed it, use the CoroutinesTimeout rule:

import kotlinx.coroutines.*
import kotlinx.coroutines.debug.junit4.CoroutinesTimeout
import org.junit.Rule
import org.junit.Test
import kotlin.time.Duration

@OptIn(ExperimentalCoroutinesApi::class)
class UserProfileTest {
    @get:Rule
    val timeout = CoroutinesTimeout.seconds(1)

    private suspend fun loadUserProfile() {
        withContext(Dispatchers.IO) {
            // Simulates an operation that doesn't complete
            delay(Duration.INFINITE)
        }
    }

    @Test
    fun loadsUserProfile() = runBlocking {
        val loadingJob = launch {
            loadUserProfile()
        }

        // Waits for the coroutine, so the test doesn't complete
        loadingJob.join()
    }
}

After one second, the rule reports that the test timed out and prints all active coroutines and their stack traces.
The test then fails with a TestTimedOutException.

/* Detailed source-code truncated for AI context efficiency. */

{collapsible="true" collapsed-title="CoroutinesTimeout JUnit4 example output"}

JUnit 5

To apply a timeout to all test functions in a class, add the @CoroutinesTimeout annotation to the class:

import kotlinx.coroutines.*
import kotlinx.coroutines.debug.junit5.CoroutinesTimeout
import org.junit.jupiter.api.Test
import kotlin.time.Duration

@OptIn(ExperimentalCoroutinesApi::class)
// Sets a one-second timeout for all test functions in the class
@CoroutinesTimeout(testTimeoutMs = 1_000)
class UserProfileTest {
    private suspend fun loadUserProfile() {
        withContext(Dispatchers.IO) {
            // Simulates an operation that doesn't complete
            delay(Duration.INFINITE)
        }
    }

    @Test
    fun loadsUserProfile() = runBlocking {
        val loadingJob = launch {
            loadUserProfile()
        }

        // Waits for the coroutine, so the test doesn't complete
        loadingJob.join()
    }
}

{validate="false"}

After one second, the CoroutinesTimeout API reports the timeout and prints all active coroutines and their stack traces.
The test then fails with a CoroutinesTimeoutException.

/* Detailed source-code truncated for AI context efficiency. */

{collapsible="true" collapsed-title="CoroutinesTimeout JUnit5 example output"}

Resolve `kotlinx-coroutines-debug` resource conflicts on Android Copy Section

The debug agent isn't supported on Android.

The kotlinx-coroutines-debug module has transitive dependencies on JNA, JNA Platform, Byte Buddy, and Byte Buddy Agent.
Some of these dependencies contain resources with the same paths.
When Android merges dependency resources, the duplicate paths can cause a DuplicateRelativeFileException resulting in a build failure.

To resolve the build failure while keeping the kotlinx-coroutines-debug dependency, exclude the conflicting resources with the following packaging configuration in your build.gradle.kts file:

// build.gradle.kts
android {
    packaging {
        resources {
            // Excludes license files from JNA and JNA Platform
            excludes += setOf(
                "META-INF/AL2.0",
                "META-INF/LGPL2.1",
            )

            // Excludes the ASM license file from Byte Buddy
            excludes += "META-INF/licenses/ASM"

            // Retains one copy of each Byte Buddy Agent file
            pickFirsts += setOf(
                "win32-x86-64/attach_hotspot_windows.dll",
                "win32-x86/attach_hotspot_windows.dll",
            )
        }
    }
}

What's next Copy Chapter

Learn how to debug coroutines in IntelliJ IDEA in Debug coroutines using IntelliJ IDEA and Debug Kotlin Flow using IntelliJ IDEA.

Topics Coroutines Flow Operators (`docs/topics/coroutines-flow-operators.md`) Copy Section

https://github.com/Kotlin/kotlinx.coroutines/edit/master/docs/topics/

Flow operators let you transform and process values in a flow pipeline.
Kotlin provides two main kinds of flow operators:

• Intermediate operators return a new downstream flow that consumes values from upstream flows and applies an operation to them.
• Terminal operators trigger the execution of the flow pipeline by collecting an upstream flow. They may also return a result.

While the kotlinx.coroutines library offers a wide range of flow operators,
you can also define custom ones when you need behavior that the built-in operators don't provide.

The following sections include examples for custom implementations alongside the corresponding built-in operators.

{style="tip"}

Intermediate operators Copy Chapter

Intermediate operators return a new downstream flow that consumes values from upstream flows.
You can chain several intermediate operators to build a flow pipeline before collecting the final result.

Intermediate operators can be classified by purpose into the following categories:

• Transforming operators transform values before emitting them downstream.
• Filtering and size-limiting operators control which upstream values continue downstream.
• Concurrent processing operators let emissions run separately from collection.
• Combining operators collect values from multiple upstream flows and emit them into one downstream flow.
• Lifecycle operators run actions in response to specific events during flow collection, such as when collection starts or when the upstream flow completes.

Transforming operators Copy Section

Transforming operators transform the values emitted by an upstream flow.
You can use them to convert values to another type, skip values, or emit additional values downstream.

Transforming operators accept suspending lambdas, so their lambdas can call suspending functions while processing each emitted value.
They still process values sequentially unless the flow pipeline uses an operator that introduces concurrency.

{style="note"}

The .transform() operator is a general transforming operator that you can use as the basis for more specific transforming operators, such as .map() and .filter().

Here's an example that uses the .transform() operator to emit each upstream value as many times as its value:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// A simplified custom implementation of the default .transform() operator
inline fun <T, R> Flow<T>.myTransform(
    // Accepts a suspending lambda that can emit values downstream
    crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = flow {
    // Collects values from the upstream flow
    [email protected] { value ->
        // Applies the transformation and emits values to the downstream flow
        [email protected](value)
    }
}

// Uses the default .transform() operator
suspend fun main() = withContext(Dispatchers.Default) {
    val flow = (0..4).asFlow().transform { value ->
        // Emits each value as many times as its value
        repeat(value) {
            emit(value)
        }
    }
    println(flow.toList())
    // [1, 2, 2, 3, 3, 3, 4, 4, 4, 4]
}

{kotlin-runnable="true"}

You can use the .map() operator to transform each upstream value into one downstream value.

Here's an example that uses .map() to multiply each value by four:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// A simplified custom implementation of the default .map() operator
inline fun <T, R> Flow<T>.myMap(
    crossinline transform: suspend (value: T) -> R
): Flow<R> = transform { value ->
    emit(transform(value))
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Multiplies each upstream value by four
    val flow = (0..4).asFlow().map { it * 4 }
    println(flow.toList())
    // [0, 4, 8, 12, 16]
}
//sampleEnd

{kotlin-runnable="true"}

To emit only upstream values that match a condition, use the .filter() operator.

Here's an example that emits values where dividing by 3 leaves a remainder of 1:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// A simplified custom implementation of the default .filter() operator
inline fun <T> Flow<T>.myFilter(
    crossinline predicate: suspend (value: T) -> Boolean
): Flow<T> = transform { value ->
    // Emits only values that match the condition
    if (predicate(value))
        emit(value)
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Emits only values where dividing by 3 leaves a remainder of 1
    val flow = (0..10).asFlow().filter { it % 3 == 1 }
    println(flow.toList())
    // [1, 4, 7, 10]
}
//sampleEnd

{kotlin-runnable="true"}

Some operators can combine the behavior of other transforming operators, such as .map() and .filter(), by transforming values and emitting only the results that match a condition.

For example, use .mapNotNull() to transform each upstream value and emit only non-null results:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// A simplified custom implementation of the default .mapNotNull() operator
inline fun <T, R: Any> Flow<T>.myMapNotNull(
    crossinline transform: suspend (value: T) -> R?
): Flow<R> = transform { value ->
    transform(value)?.let { transformed ->
        emit(transformed)
    }
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Converts each string to Double and skips values that can't be converted
    val flow = flowOf("1.2", "10", "11", "error", "0.000")
        .mapNotNull { it.toDoubleOrNull() }
    
    println(flow.toList())
    // [1.2, 10.0, 11.0, 0.0]
}
//sampleEnd

{kotlin-runnable="true"}

Filtering and size-limiting operators Copy Section

Filtering and size-limiting operators control which values continue downstream from a flow.
You can use them to remove repeated consecutive values, skip values from the beginning of a flow, or cancel collection after a specified number of values.

To ignore repeated consecutive values, use the .distinctUntilChanged() operator.
It emits a value only when it differs from the previously emitted value:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// A simplified custom version of the default .distinctUntilChanged() operator
fun <T> Flow<T>.myDistinctUntilChanged(): Flow<T> = flow {
    var lastEmitted: Any? = Any() // A value that's equal only to itself
    [email protected] { value ->
        if (lastEmitted != value) {
            [email protected](value)
            lastEmitted = value
        }
    }
}

suspend fun main() = withContext(Dispatchers.Default) {
    // Removes repeated consecutive values from the upstream flow
    val flow = flowOf(1, 2, 3, 3, 3, 4, 5, 5, 1).distinctUntilChanged()
    println(flow.toList())
    // [1, 2, 3, 4, 5, 1]
}

{kotlin-runnable="true"}

You can use the .drop() operator to skip the first values emitted by an upstream flow.
For example, .drop(2) skips the first two values and emits the remaining values downstream:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// A simplified custom version of the default .drop() operator
fun <T> Flow<T>.myDrop(count: Int): Flow<T> = flow {
    require(count >= 0)
    var elementsAlreadyDropped = 0
    [email protected] { value ->
        if (elementsAlreadyDropped == count) {
            [email protected](value)
        } else {
            ++elementsAlreadyDropped
        }
    }
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Skips the first two values from the upstream flow    
    val flow = flowOf(1, 2, 3, 4, 5).drop(2)
    println(flow.toList())
    // [3, 4, 5]
}
//sampleEnd

{kotlin-runnable="true"}

To cancel collection after a fixed number of values, use the .take() operator.
Here's an example that uses the .take() operator to collect only the first three values:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.*
import java.io.IOException
import kotlin.time.Duration.Companion.milliseconds

// A simplified custom version of the default .take() operator
fun <T> Flow<T>.myTake(count: Int): Flow<T> = flow {
    require(count > 0)
    val cancellationException = CancellationException()
    var elementsRemaining = count
    try {
        [email protected] {
            emit(it)
            --elementsRemaining
            if (elementsRemaining == 0) {
                // Cancels the upstream flow after the requested number of values
                throw cancellationException
            }
        }
    } catch (e: Throwable) {
        if (e === cancellationException) {
            // Handles the CancellationException used to cancel the upstream flow
            // Completes the flow after the set number of values in .myTake()
        } else {
            // Rethrows unexpected exceptions
            throw e
        }
    }
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Collects only the first three values from the upstream flow
    val flow = (0..1000).asFlow().take(3)

    println(flow.toList())
    // [0, 1, 2]
}
//sampleEnd

{kotlin-runnable="true"}

Concurrent processing operators Copy Section

By default, a flow pipeline processes values sequentially.
The upstream flow emits a value, and collectors process it before the next value is emitted.

To run the upstream flow concurrently with downstream collection, use concurrent processing operators to introduce a buffer.
A buffer stores values that the upstream flow has emitted, but the collector hasn't processed yet.

One operator that introduces this buffer is the .buffer() operator.
It lets you configure the buffer capacity and what happens when the buffer is full, for example:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    flow {
        repeat(10) {
            emit(it)
            println("Emitted $it!")
        }
    }
        // Lets the upstream flow emit up to four values ahead of the collector
        .buffer(4)
        .collect {
            println("Processed $it!")
            delay(20.milliseconds)
        }
}
//sampleEnd

{kotlin-runnable="true"}

When the collector is slower than the upstream flow, the pipeline needs a way to handle values that the collector hasn't processed yet.

By default, the collector applies backpressure to the upstream flow.
With this strategy, the upstream flow suspends when the buffer is full and resumes when the collector frees up space.

To drop values instead of suspending the upstream flow, set the onBufferOverflow parameter:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    flow {
        repeat(10) {
            emit(it)
            println("Emitted $it!")
        }
    }
        // Stores up to four values before applying the overflow behavior
        // Drops the oldest buffered value when the buffer is full
        .buffer(4, onBufferOverflow = BufferOverflow.DROP_OLDEST)
        .collect { value ->
            println("Processed $value!")
            delay(20.milliseconds)
        }
}
//sampleEnd

{kotlin-runnable="true"}

You can also use the .conflate() operator, which is a shorthand for buffer(1, onBufferOverflow = BufferOverflow.DROP_OLDEST).
Use it when you only want to process the latest values, skipping the values emitted while the previous value is being collected:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    flow {
        repeat(10) {
            emit(it)
            println("Emitted $it!")
        }
    }.conflate().collect {
        println("Processed $it!")
        delay(20.milliseconds)
    }
}
//sampleEnd

{kotlin-runnable="true"}

The .conflate() operator only affects which buffered values the collector processes.
It doesn't cancel processing that has already started.
To do that, use collectLatest() instead.

In the previous examples, the .buffer() and .conflate() operators run the upstream flow concurrently in a separate coroutine without changing its coroutine context.

To run the upstream flow in a different coroutine context, use the .flowOn() operator.
If this changes the dispatcher, .flowOn() collects the upstream flow in a separate coroutine and uses a buffer between upstream emission and downstream collection.

Here's a simplified example that uses .flowOn() to run the upstream flow in Dispatchers.IO:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    flow {
        repeat(10) {
            emit(it)
        }
        println("Finished emitting!")
    }.flowOn(Dispatchers.IO).collect {
        println("Received $it!")
        delay(10.milliseconds)
    }
}
//sampleEnd

{kotlin-runnable="true"}

In this example, the .flowOn() operator can introduce concurrent upstream processing, but the buffer behavior isn't configured explicitly.

To configure both the coroutine context of the upstream flow and the buffer behavior, combine .flowOn() with .buffer() or .conflate().
When you use these operators together, the operators perform operator fusion and share a single buffer.

Here's an example that uses .flowOn(Dispatchers.IO) to run the upstream flow in Dispatchers.IO and .conflate() to keep the newest buffered value:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.Random
import kotlin.time.Duration.Companion.milliseconds
import kotlin.math.round

//sampleStart
fun awaitSensorSignal(): SensorSignal {
    Thread.sleep(10)
    val reading =
        round(Random.nextDouble(25.0, 100.0) * 100.0)/100.0
    println("Measured $reading as the temperature")
    return SensorSignal(temperatureCelsius = reading)
}

data class SensorSignal(
    val temperatureCelsius: Double
)

suspend fun sendLatestTemperature(temperatureCelsius: Double) {
    println("Starting to send $temperatureCelsius...")
    delay(50.milliseconds)
    println("Sent $temperatureCelsius.")
}

suspend fun main() = withContext(Dispatchers.Default) {
    val smartHomeTemperatureFlow = flow {
        while (true) {
            val signal = awaitSensorSignal()
            emit(signal.temperatureCelsius)
            println("Emitted $signal")
        }
    }
        // Runs the upstream flow in Dispatchers.IO
        .flowOn(Dispatchers.IO)
        // Keeps the newest buffered value and drops older ones
        .conflate()
        // Collects the first two values from the upstream flow
        .take(2)
        .collect { temperature ->
            println("Received $temperature!")
            sendLatestTemperature(temperature)
        }
}
//sampleEnd

{kotlin-runnable="true"}

Combining operators Copy Section

Combining operators consume values from multiple upstream flows and return a single downstream flow.
Use them when collectors need values from more than one flow.

To pair values from two upstream flows, use the .zip() operator.
It combines the first value from each flow, then the second value from each flow, and so on.
The resulting flow completes as soon as one of the upstream flows completes.

Here's an example:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.Random
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Emits a ticker value every 100 milliseconds
    val tickerFlow = flow {
        while (true) {
            emit(Unit)
            delay(100.milliseconds)
        }
    }

    val start = TimeSource.Monotonic.markNow()
    tickerFlow
        // Combines each ticker emission with the next number
        .zip(flowOf(1, 2, 3)) { _, value ->
            value
        }.collect {
            println("${start.elapsedNow()}: received $it")
        }
}
//sampleEnd

{kotlin-runnable="true"}

To combine the latest value from multiple flows, use the .combine() operator.
It emits a new value when any upstream flow emits a value, using the latest value from each upstream flow:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

//sampleStart
enum class Theme {
    Dark,
    Light,
}

data class UiState(
    val messages: List<String>,
    val theme: Theme,
)

val messagesFlow = MutableStateFlow(
    listOf(
        "Hello!",
        "Is anyone here?",
    )
)

val themeFlow = MutableStateFlow(
    Theme.Light
)

// Combines the latest values from both upstream flows
val uiStateFlow = combine(messagesFlow, themeFlow) { messages, theme ->
    UiState(messages, theme)
}

suspend fun main() {
    withContext(Dispatchers.Default) {
        // Uses UNDISPATCHED to subscribe before the first update happens
        val uiUpdateJob = launch(start = CoroutineStart.UNDISPATCHED) {
            uiStateFlow.collect {
                // Draws the UI
                println(it)
            }
        }
        messagesFlow.update { messages -> messages + "I'll be back!" }
        delay(100.milliseconds)
        
        themeFlow.value = Theme.Dark
        delay(100.milliseconds)
        
        uiUpdateJob.cancel()
    }
}
//sampleEnd

{kotlin-runnable="true"}

In this example, combine() creates uiStateFlow from the latest values of messagesFlow and themeFlow.
Updating either upstream flow emits a new UiState with the latest messages and theme.

If you want to collect values from multiple flows concurrently and emit their values into one downstream flow, use the .merge() operator:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

//sampleStart
interface UiEvent

class ClickEvent: UiEvent

class RightClickEvent: UiEvent

suspend fun main() {
    withContext(Dispatchers.Default) {

        val clickFlow = MutableSharedFlow<ClickEvent>()
        val rightClickFlow = MutableSharedFlow<RightClickEvent>()

        coroutineScope {
            // Uses UNDISPATCHED to subscribe before the first update happens
            val collectJob = launch(start = CoroutineStart.UNDISPATCHED) {

                // Collects both upstream flows concurrently and emits their values downstream
                merge(clickFlow, rightClickFlow).collect {
                    println("Observed an event: $it")
                }
            }
            clickFlow.emit(ClickEvent())
            delay(100.milliseconds)
            
            clickFlow.emit(ClickEvent())
            delay(100.milliseconds)
            
            rightClickFlow.emit(RightClickEvent())
            delay(100.milliseconds)
            
            collectJob.cancel()
        }
    }
}
//sampleEnd

{kotlin-runnable="true"}

Lifecycle operators Copy Section

Lifecycle operators accept a suspending lambda that runs at a specific point during flow collection.
You can use them to place logic before the flow is collected, before each value is emitted,
after collection completes, or when the flow completes without emitting values.

The .onStart() operator runs its lambda before the upstream flow is collected.
For code that needs to run before each emitted value, use .onEach().

Similarly to .onStart(), you can use .onSubscription() for hot flows to run code after a subscriber starts collecting the flow, but before it collects any emitted values.

{style="note"}

Here's an example that uses these operators to print a message before collection starts and before each value is emitted downstream:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

// A simplified custom version of the default .onStart() operator
fun <T> Flow<T>.myOnStart(
    action: suspend FlowCollector<T>.() -> Unit
): Flow<T> = flow {
    [email protected]()
    [email protected](this@flow)
}

suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf("Page 1", "Page 2", "Page 3").onStart {
            println("Processing pages!")
        }.onEach {
            println("Emitted $it")
        }.collect {
            println("Collected $it")
        }
    }
}

{kotlin-runnable="true"}

To run code after collection completes, use the .onCompletion() operator.
Its lambda can emit values downstream when the upstream flow completes successfully, for example:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

// A simplified custom version of the default .onCompletion() operator
fun <T> Flow<T>.myOnCompletion(
    action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit
): Flow<T> = flow {
    var exception: Throwable? = null
    try {
        [email protected](this@flow)
    } catch (e: Throwable) {
        // Run `action`, but if `action` calls `emit`, throw `e` from it
        FlowCollector<T> { throw e }.action(e)
        throw e
    }
    [email protected](null)
}

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf("Page 1", "Page 2", "Page 3").onCompletion {
            println("Almost done...")
            // Emits an additional value after the upstream flow completes
            emit("Last Page!")
        }.collect {
            println("Collected $it")
        }
    }
}
//sampleEnd

{kotlin-runnable="true"}

To run code when the upstream flow completes without emitting any values, use the .onEmpty() operator:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

// A simplified custom version of the default .onEmpty() operator
fun <T> Flow<T>.myOnEmpty(
    action: suspend FlowCollector<T>.() -> Unit
): Flow<T> = flow {
    var emittedSomething = false
    [email protected] { value ->
        emittedSomething = true
        [email protected](value)
    }
    if (!emittedSomething) {
        action()
    }
}

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf("Page 1", "Page 2", "Page 3").onEmpty {
            // Doesn't print anything, because the upstream flow emits values
            println("No pages to load!")
        }.collect()
        flowOf<Int>().onEmpty {
            println("No pages to load!")
            // No pages to load!
        }.collect()
    }
}
//sampleEnd

{kotlin-runnable="true"}

Terminal operators Copy Chapter

Terminal operators collect a flow.
You can use them to consume emitted values, return a result based on the collected values, or collect a flow in a specific CoroutineScope.

To collect a flow, use the collect() operator.
If you pass a lambda to collect(), it receives each emitted value:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf(1, 2, 3).collect {
            println("Collected $it!")
        }
    }
}
//sampleEnd

{kotlin-runnable="true"}

You can also call collect() without a lambda:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf(1, 2, 3).onEach {
            println("Collected $it!")
        }.collect()
    }
}
//sampleEnd

{kotlin-runnable="true"}

If you want to collect a flow but cancel unfinished work when a new value is emitted, use the collectLatest() operator:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flow {
            println("Emitting Page 1")
            emit("Page 1")
            delay(50.milliseconds)
            println("Emitting Page 2 in quick succession")
            emit("Page 2")
            delay(200.milliseconds)
            println("Emitting Page 3")
            emit("Page 3")
        }.flowOn(Dispatchers.IO).collectLatest {
            println("Starting to process $it!")
            try {
                delay(100.milliseconds)
            } catch (e: CancellationException) {
                println("Canceled processing $it.")
                throw e
            }
            println("Done processing!")
        }
    }
}
//sampleEnd

{kotlin-runnable="true"}

Some terminal operators collect a flow and return a result based on the collected values.
For example, you can use the .first() operator to return the first emitted value and then cancel collection:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource


suspend fun main() { 
    withContext(Dispatchers.Default) {
        val firstValue = flowOf(1, 2, 3).first()

        println(firstValue)
        // 1
    }
}

{kotlin-runnable="true"}

You can collect emitted values into a collection with the .toList() or .toSet() operators:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// A simplified custom implementation of the default .toList() operator
suspend fun <T> Flow<T>.myToList(): List<T> = buildList {
    [email protected] { value ->
        // Adds each emitted value to the resulting list
        add(value)
    }
}

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        val list = flowOf(1, 2, 3).toList()
        println(list)
        // [1, 2, 3]

        val set = flowOf(1, 2, 2, 3).toSet()
        println(set)
        // [1, 2, 3]
    }
}
//sampleEnd

{kotlin-runnable="true"}

To combine emitted values into a single result, use the .reduce() or .fold() operators.
The .fold() operator uses the value you provide as its starting value, and the .reduce() operator uses the first emitted value instead.

Here's an example:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        // Uses the first emitted value as the starting value
        val reduced = flowOf(1, 2, 3).reduce { accumulator, value ->
            accumulator + value
        }

        // Starts with the provided starting value
        val folded = flowOf(1, 2, 3).fold(2) { accumulator, value ->
            accumulator + value
        }

        println(reduced)
        // 6

        println(folded)
        // 8
    }
}
//sampleEnd

{kotlin-runnable="true"}

Collect a flow in a specific `CoroutineScope` Copy Section

When a screen or another long-lived object needs values from a flow, start the collector in that object's CoroutineScope.
This ensures that canceling the object's CoroutineScope when the object is destroyed also cancels collection.

To collect a flow in a specific CoroutineScope, use the .launchIn() terminal operator.
This operator returns the Job of the collecting coroutine.

Here's an example where a screen collects values from a StateFlow and stops the collecting coroutine when the screen closes:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

// A simplified custom version of the default .launchIn() operator
fun <T> Flow<T>.myLaunchIn(scope: CoroutineScope): Job = scope.launch {
    [email protected]()
}

//sampleStart
data class Coordinate(val x: Int, val y: Int)

class MyScreen(val scope: CoroutineScope) {
    private val _mousePosition =
        MutableStateFlow<Coordinate>(Coordinate(0, 0))
    val mousePosition get() = _mousePosition.asStateFlow()

    init {
        // Starts collecting the StateFlow in the screen's CoroutineScope
        mousePosition.onEach {
            updateStatusBar()
        }.launchIn(scope)
    }

    fun moveMouse(newCoordinate: Coordinate) {
        _mousePosition.value = newCoordinate
    }

    private fun updateStatusBar() {
        println("Mouse is at ${_mousePosition.value}")
    }
}

suspend fun main() {
    withContext(Dispatchers.Default) {
        val childScope = CoroutineScope(
            currentCoroutineContext() + Job(currentCoroutineContext()[Job])
        )
        val screen = MyScreen(childScope)
        delay(100.milliseconds)
        
        screen.moveMouse(Coordinate(10, 15))
        delay(100.milliseconds)
        
        screen.moveMouse(Coordinate(1, 3))
        delay(100.milliseconds)
        
        childScope.cancel()
    }
}
//sampleEnd

{kotlin-runnable="true"}

Topics Coroutines Flow (`docs/topics/coroutines-flow.md`) Copy Section

https://github.com/Kotlin/kotlinx.coroutines/edit/master/docs/topics/

A flow represents a sequential stream of values that can be produced asynchronously.
Unlike a suspending function, which returns one value, you can use flows to work with multiple sequential values over time.

You can use flows to create flow pipelines that load data progressively, react to event streams, and model subscription-style APIs.

A flow pipeline is a sequence of operations that involves the following roles:

• Emitter: produces values.
• Intermediate operators (optional): consume values from a flow, apply an operation to them, and return another flow.
• Collector: consumes values from a flow.

Here's a simple example that shows how these pipeline roles work together:

import kotlinx.coroutines.flow.*

//sampleStart
suspend fun main() {
    // The emitter produces values
    flowOf(0x4B, 0x6F, 0x74, 0x6C, 0x69, 0x6E)
        // The intermediate operator consumes values,
        // applies an operation, and returns another flow
        .map { value -> value.toChar() }
        // The collector consumes the transformed values
        .collect { updatedValue ->
            println("Say '$updatedValue'!")
        }
}
//sampleEnd

{kotlin-runnable="true"}

In a flow, values move from an emitter toward a collector, from upstream to downstream.
Intermediate operators collect an upstream flow, apply an operation to its values, and return a new downstream flow.
That downstream flow can become the upstream flow for the next collector.

{width=700}

Kotlin provides the following flow types:

• Cold flows start producing values when collected.
  Each collector triggers a new, independent execution of the flow.
• Hot flows emit values independently of collectors and share the same stream of values with all collectors.

You can use the Turbine library to test Kotlin flows.
It simplifies collecting and asserting flow emissions in unit tests, including completion and failure cases.

{style="tip"}

Cold flows Copy Chapter

Like sequences, cold flows are lazy.

The code block of a cold flow builder doesn't run until a collector collects it.
Each new collector starts a new execution of the flow.

Create a cold flow Copy Section

To create a cold flow, use the flow() builder function.
Inside its block, use the emit() function to emit values to collectors:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

//sampleStart
fun main() {
    // Creates a flow
    val pageFlow = flow {
        for (page in 1..3) {
            println("Loading page $page...")

            // Emits each page as it is loaded
            emit("Page $page")
        }
    }
    println("Creating a cold flow doesn't run it!")
}
//sampleEnd

{kotlin-runnable="true"}

In this example, the flow() builder function returns a Flow<T>, but it doesn't start executing its block.
A cold flow is like a recipe: it defines how to produce values, but it only starts producing values when you collect it.

You can also create cold flows with the following functions:

• flowOf(): creates a flow from provided values.
• .asFlow(): converts an existing iterable, such as a range, into a flow.

Here's an example:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() {
    // Creates a flow from provided values
    val predefinedPageFlow = flowOf("Page 1", "Page 2", "Page 3")
    // Creates a flow from a range
    val generatedPageFlow = (1..3).asFlow()
}

Collect a cold flow Copy Section

To collect a cold flow, use the collect() function, which triggers the emissions from the upstream flow.
If you pass a lambda to collect(), it receives each emitted value:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        val pageFlow = flow {
            for (page in 1..3) {
                println("Loading page $page...")
                emit("Page $page")
            }
        }
        // Collects the flow with a lambda that receives each emitted page
        pageFlow.collect { page ->
            println("Processing $page...")
            delay(100.milliseconds)
            println("Done processing $page.")
        }
    }
}
//sampleEnd

{kotlin-runnable="true"}

Each call to collect() runs the entire cold flow from the beginning.
If multiple collectors collect the same cold flow, each collector triggers its own collection:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() {
    val pageFlow = flow {
        // Reads the name of the current coroutine
        val coroutineName = currentCoroutineContext()[CoroutineName]?.name

        println("Starting emissions in $coroutineName")
        for (page in 1..3) {
            println("Loading page $page in $coroutineName")
            emit("Page $page")
        }
        println("Done emitting in $coroutineName")
    }

    withContext(Dispatchers.Default) {
        // Launches a collector that processes each page slowly
        launch(CoroutineName("a slow coroutine")) {
            pageFlow.collect {
                println("Processing $it slowly")
                delay(100.milliseconds)
                println("Done processing $it slowly")
            }
        }

        // Launches a collector that processes each page quickly
        launch(CoroutineName("a fast coroutine")) {
            pageFlow.collect {
                println("Processing $it quickly")
                delay(10.milliseconds)
                println("Done processing $it quickly")
            }
        }
    }
}
//sampleEnd

{kotlin-runnable="true"}

In this example, CoroutineName adds a name to each coroutine.
You can use CoroutineName for debugging. Here, it helps show which collector runs each collection.

Intermediate flow operators Copy Section

Intermediate operators apply operations to an upstream flow and return a new downstream flow.
They are cold, so the returned flow doesn't start processing values until it's collected, even when the upstream flow is hot.

The kotlinx.coroutines library offers a wide range of intermediate flow operators for transforming and processing flows.
You can also define custom operators yourself when you need behavior that the built-in operators don't provide.

Here's an example of a simplified custom .map() operator that applies a transformation to each emitted value:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

//sampleStart
// A simplified custom implementation of the default .map() operator
fun <T, R> Flow<T>.myMap(transform: suspend (value: T) -> R): Flow<R> = flow {
    // Collects values from the upstream flow
    [email protected] { value ->
        // Transforms each collected value and emits the result
        emit(transform(value))
    }
}

suspend fun main() {
    // Creates a flow, applies the custom map operator, and collects the transformed values
    flowOf(1, 2, 3).myMap { 2 * it }.collect {
        println("Collecting $it")
    }
}
//sampleEnd

{kotlin-runnable="true"}

Call suspending functions inside a flow builder

Unlike in sequences, you can call suspending functions inside a flow() builder function:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

//sampleStart
suspend fun loadPage(): Int {
    delay(100)
    return 3
}

suspend fun main() {
    flow {
        emit(loadPage())
    }.collect {
        println(it)
        // 3
    }
}
//sampleEnd

{kotlin-runnable="true"}

However, a flow() builder function must emit values from the same coroutine context where it runs.
You can't start another coroutine that calls emit() in its block, and you can't change the coroutine context with withContext():

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

//sampleStart
suspend fun main() {
    // This fails with an exception!
    flow {
        // Changes the coroutine context with withContext()
        withContext(Dispatchers.IO) {
            emit('a')
        }
    }.collect { 
        println("This never prints")
    }
}
//sampleEnd

{kotlin-runnable="true" validate="false"}

This restriction applies to the flow() builder function.

If you need the upstream flow to run in a different coroutine context, you can change it with the .flowOn() operator.

Alternatively, you can use channelFlow() to emit values from multiple coroutines.

Change the coroutine context of a cold flow with `.flowOn()`

By default, a cold flow runs in the same coroutine context as the collector.

If you want the flow to run in a different coroutine context, use the .flowOn() operator.
This operator is context-preserving.
It changes only the coroutine context of the upstream flow while keeping the downstream flow in the caller's context.

Here's an example of a cold flow that emits values in one coroutine context and collects them in another:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default + CoroutineName("downstream")) {
        flow {
            val coroutineName = currentCoroutineContext()[CoroutineName]?.name

            // Emits in the coroutine context applied with .flowOn()
            println("Emitting '1' in $coroutineName")
            // Emitting '1' in upstream
            emit(1)

        // Changes the coroutine context of the upstream flow
        }.flowOn(Dispatchers.IO + CoroutineName("upstream"))
            .collect {
            val coroutineName = currentCoroutineContext()[CoroutineName]?.name

            // Collects in the caller's coroutine context
            println("Collecting '$it' in $coroutineName")
            // Collecting '1' in downstream
        }
    }
}
//sampleEnd

{kotlin-runnable="true"}

Handle exceptions in flows Copy Section

Both emitters and collectors can throw exceptions.

If you don't handle an exception during flow collection, it propagates upstream from the collector and is thrown to the caller of the collect() function.

You can handle such exceptions by wrapping the collect() function in a try-catch block, for example:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

class MyFlowException(message: String) : Exception(message)

//sampleStart
suspend fun main() {
    val myFlow = flow {
        try {
            // The emit() function calls the lambda passed to collect()
            emit('a')
        } catch (e: MyFlowException) {
            println("Collector threw $e")

            // Rethrows the downstream exception
            throw e
        }
    }
    // Wraps flow collection in try-catch
    try {
        myFlow.collect {
            // Throws an exception from the collect() lambda
            throw MyFlowException("Can't process '$it'!")
        }
    } catch (e: MyFlowException) {
        println("Flow collection failed with $e")
        // Rethrows the exception to the caller
        throw e
    }
}
//sampleEnd

{kotlin-runnable="true" validate="false"}

In this example, the collector throws an exception when it receives a value from the emit() function.
The flow() builder function catches this downstream exception.

When you catch an exception thrown by the collector inside a flow builder function, rethrow it.
This preserves exception transparency and lets the caller of collect() handle the exception.

Use the `.catch()` operator to handle upstream exceptions

To handle exceptions before they reach the collector, use the .catch() operator.

You can use the .catch() operator to handle exceptions from the upstream flow, for example by using the emit() function to emit a fallback value downstream:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

//sampleStart
suspend fun main() {
    flow {
        emit("a")
        emit("b")

        // Throws an exception from the upstream flow
        throw UnsupportedOperationException(
            "I am tired of listing letters"
        )
    }.catch { upstreamException ->
        println("Upstream completed with $upstreamException!")

        // Emits a fallback value downstream
        emit("Upstream terminated with an exception!")
    }.collect {
        println("Got '$it'")
    }
}
//sampleEnd

{kotlin-runnable="true"}

In this example, the upstream flow emits values before throwing an exception.
The .catch() operator handles the exception and emits "Upstream terminated with an exception!" as a fallback value.

When a flow is expected to throw some exceptions during normal operation, handle the recoverable exceptions in .catch() and rethrow any unexpected exceptions.

Here's an example where a flow loads data and reports its progress:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.*
import java.io.IOException
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
sealed interface LoadingState {
    sealed interface Terminal: LoadingState
    object Started: LoadingState
    data class Percentage(val percents: Int): LoadingState
    object Failed: Terminal
    object Done: Terminal
}

fun loadBlob(url: String) = flow {
    emit(LoadingState.Started)

    val failureChancePerStep = 1 - java.lang.Math.pow(0.99, 10.0)

    repeat(10) { step ->
        if (Random.nextDouble() < failureChancePerStep)
            throw IOException("Failed to load!")
        emit(LoadingState.Percentage((step + 1) * 10))
        delay(10.milliseconds)
    }
    emit(LoadingState.Done)
}.catch { e ->
    println("Loading data failed with $e")
    if (e is IOException) {
        // Handles an expected exception
        emit(LoadingState.Failed)
    } else {
        // Rethrows unexpected exceptions, so the collect() fails with them
        throw e
    }
}

suspend fun main() {
    loadBlob("https://example.com/").collect {
        println("Got '$it'")
    }
}
//sampleEnd

{kotlin-runnable="true" validate="false"}

In this example, if loading fails with an expected exception, the .catch() operator uses the emit() function to emit a fallback state.
For unexpected exceptions, rethrow them in the .catch() operator.
This lets the caller of the collect() function receive exceptions that the flow doesn't handle.

The .catch() operator doesn't handle exceptions thrown by the collector.
If the lambda passed to collect() throws an exception, handle it with a try-catch block around the collect() function:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

suspend fun main() {
    val myFlow = flow {
        for (char in listOf('a', 'o', '5', 'c')) {
            try {
                emit(char)
            } catch (e: IllegalArgumentException) {
                println("Collector doesn't support character '$char': $e")

                // Rethrows the downstream exception
                throw e
            }
        }
    }.catch { e ->
        // Doesn't run because the exception happens downstream
        println("Upstream threw an exception: $e")
    }

    try {
        myFlow.collect {
            require(!it.isDigit()) { "Digits are not allowed!" }
        }
    } catch (e: IllegalArgumentException) {
        // Handles the exception from the collect() lambda
        println("Flow collection failed with $e")
    }
}

{kotlin-runnable="true"}

Since the collect() lambda runs after .catch(), you can't use .catch() to handle exceptions thrown from it.
To handle exceptions from code that runs for each emitted value with .catch(), place that code in .onEach() before .catch().

The .onEach() operator runs its lambda before each value is emitted downstream.
If .catch() handles an exception from .onEach(), the flow completes and doesn't emit the next value:

Here's an example:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.*
import java.io.IOException
import kotlin.time.Duration.Companion.milliseconds

suspend fun main() {
    flowOf('a', 'o', '5', 'c')
        // Runs before each value is emitted downstream
        .onEach {
            require(!it.isDigit()) { "Digits are not allowed!" }
            println("Got '$it'")
        }
        .catch { e ->
            println("Caught an exception: $e")
        }
        .collect()
}

{kotlin-runnable="true"}

In this example, the .onEach() operator runs upstream from .catch(), so the .catch() operator handles the exception when the require() check fails for '5'.

Restart the upstream flow after an exception

Some operations may temporarily fail, such as a network request that loses its connection.
For these cases, you can use the .retry() operator to restart the upstream flow after an exception.

The .retry() operator receives an exception and restarts collection when its lambda returns true, up to the specified number of retries.
For example, .retry(3) retries the upstream flow up to three times after the first failed attempt.

If the lambda returns false, .retry() stops retrying and rethrows the exception.

For more control over the retry logic, use the .retryWhen() operator.
Like .retry(), it receives the exception, but it also receives the current attempt number and can emit values before retrying.

{style="note"}

Here's an example that retries loading up to three times after an IOException:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.*
import java.io.IOException
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds

//sampleStart
sealed interface LoadingState {
    sealed interface Terminal: LoadingState
    object Started: LoadingState
    data class Percentage(val percents: Int): LoadingState
    object Failed: Terminal
    object Done: Terminal
}

fun loadBlob(url: String) = flow {
    emit(LoadingState.Started)

    val failureChancePerStep = 1 - java.lang.Math.pow(0.99, 10.0)

    repeat(10) { step ->
        if (Random.nextDouble() < failureChancePerStep)
            throw IOException("Failed to load!")
        emit(LoadingState.Percentage((step + 1) * 10))
        delay(10.milliseconds)
    }
    emit(LoadingState.Done)
}.retry(3) { e ->
    if (e is IOException) {
        // This is an expected error
        // Waits for one second before retrying
        delay(1.seconds)
        true
    } else {
        // Stops retrying and rethrows unexpected exceptions
        false
    }
}

suspend fun main() {
    loadBlob("https://example.org/").collect {
        println("Got $it")
    }
}
//sampleEnd

{kotlin-runnable="true" validate="false"}

Flow cancellation Copy Section

Flow cancellation stops collection when the result is no longer needed, such as when a request times out.

Flow collection is tied to the coroutine that calls the collect() function.
When that coroutine is canceled, the collection stops, and the upstream flow is canceled too.

To cancel flow collection, call the cancel() function on the Job of the collecting coroutine:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.*
import java.io.IOException
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
val myFlow = flow {
    var i = 0
    try {
        while (true) {
            println("Emitting $i")
            emit(i)
            println("Emitted $i")
            ++i
            delay(10.milliseconds)
        }
    } catch (e: Throwable) {
        println("Upstream finished with $e")
        throw e
    }
}

suspend fun main() {
    coroutineScope {
        val job = launch {
            try {
                myFlow.collect {
                    println("Processing $it")
                    delay(5.milliseconds)
                }
            } catch (e: Throwable) {
                println("Collection finished with $e")
                throw e
            }
        }
        delay(100.milliseconds)

        // Cancels the coroutine that collects the flow
        job.cancel()
    }
}
//sampleEnd

{kotlin-runnable="true"}

The collector can also cancel the upstream flow while the collecting coroutine stays active.
To do this, throw a CancellationException from the collector.

The .take() operator uses this behavior to stop collecting after a fixed number of values.
For example, .take(3) collects only the first three values from the upstream flow, then cancels it.

Here's an example that uses a simplified version of the .take() operator:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.*
import java.io.IOException
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
// Defines a simplified version of the default .take() operator
fun <T> Flow<T>.myTake(count: Int): Flow<T> = flow {
    require(count > 0)
    val cancellationException = CancellationException()
    var elementsRemaining = count
    try {
        [email protected] {
            emit(it)
            --elementsRemaining
            if (elementsRemaining == 0) {
                // Cancels the upstream flow after the requested number of values
                throw cancellationException
            }
        }
    } catch (e: Throwable) {
        if (e === cancellationException) {
            // Handles the CancellationException used to cancel the upstream flow
            // Completes the flow after the set number of values in .myTake()
        } else {
            // Rethrows unexpected exceptions
            throw e
        }
    }
}

suspend fun main() {
    (0..1000).asFlow().myTake(3).collect {
        println("Got $it")
    }
}
//sampleEnd

{kotlin-runnable="true"}

In this example, the .myTake() function emits values from the upstream flow until all requested values are emitted.
Then it throws a CancellationException to cancel the upstream flow.

Emit values concurrently with `channelFlow()` Copy Section

The flow() builder function is simple and efficient for flows that emit values from one coroutine.
If you want to emit values from multiple coroutines concurrently into the same flow, use the channelFlow() builder function.
You can use it for concurrent work that reports results progressively, such as loading data from multiple sources.

The channelFlow() builder function creates a cold flow that uses a channel to send values from multiple coroutines.
Inside the builder, use the send() function instead of the emit() function to produce values.

Here's an example that uses channelFlow() to collect two flows concurrently and re-emit their values with a simplified version of the .merge() operator:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.*
import java.io.IOException
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
// Defines a simplified version of the default .merge() operator
fun <T> Flow<T>.myMerge(other: Flow<T>): Flow<T> = channelFlow {
    // CoroutineScope and SendChannel are available as receivers here
    // Launches a coroutine that collects the receiver flow
    launch {
        // Collects the receiver flow
        [email protected] {
            send(it)
        }
    }
    launch {
        // Launches a coroutine that collects the other flow
        other.collect {
            // Calls SendChannel.send
            send(it)
        }
    }
}

suspend fun main() {
    val flow1 = (0..3).asFlow().onEach { delay(20.milliseconds) }
    val flow2 = (6..9).asFlow().onEach { delay(50.milliseconds) }
    flow1.myMerge(flow2).collect { println(it) }
}
//sampleEnd

{kotlin-runnable="true"}

The channelFlow() builder function uses a buffered channel, allowing producers to send values ahead of the collector until the buffer is full.
By default, the buffer can hold up to 64 values.
When the buffer is full, producers suspend until the buffer has free space.

You can change the buffer capacity with the .buffer() operator.
For example, .buffer(12) lets producers send up to 12 values ahead of the collector, while .buffer(0) removes the buffer, so each value is sent only when the collector can receive it.

Here's an example:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.*
import java.io.IOException
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() {
    val oneHundredNumbers = channelFlow {
        repeat(100) {
            println("Sending $it")
            send(it)
        }
    }

    // Uses the default buffer capacity
    oneHundredNumbers.collect {
        println("Processing $it")
        delay(10.milliseconds)
    }
  
    // Removes the buffer so sending and processing interleave from the start
    oneHundredNumbers.buffer(0).collect {
        println("Processing $it")
        delay(10.milliseconds)
    }
}
//sampleEnd

{kotlin-runnable="true"}

In this example, oneHundredNumbers flow uses the default buffer capacity, while oneHundredNumbers.buffer(0) flow has no buffer.

With the default buffer capacity, the producer sends values quickly until the buffer becomes full.
After that, send() suspends until the buffer has free space, so Sending and Processing messages start to interleave.

With .buffer(0), each send() call waits until the collector can receive the value, so Sending and Processing interleave from the start.

Hot flows Copy Chapter

Hot flows are shared streams that emit values independently of collectors.
They keep emitting values even when no collector is active, and multiple collectors can collect the same emissions from an already active stream instead of starting a new execution.

A collector of a hot flow is called a subscriber.

You can use hot flows when multiple parts of an application need to react to the same stream of updates, such as incoming chat messages, user actions, or UI state changes.

Kotlin provides two hot flow types:

• SharedFlow broadcasts values to multiple subscribers. Use it when you need to broadcast events that happen over time, such as messages or notifications.
• StateFlow is a specialized SharedFlow that always holds the latest state value. Use it when you need to represent state that changes over time, such as UI state.

Create a `SharedFlow` Copy Section

SharedFlow is a hot flow that broadcasts emitted values that happen over time to its subscribers.

You can create a SharedFlow with the MutableSharedFlow() function.

A MutableSharedFlow exposes functions for emitting values.
If you expose it directly, code outside the class can emit values into the flow.

To prevent this, store the mutable flow in a private backing property and expose a read-only SharedFlow with the .asSharedFlow() function.
To emit values to subscribers, use the emit() function on the MutableSharedFlow:

data class Message(
    val senderId: Int,
    val time: Instant,
    val text: String,
)

class Chatroom {
    // Stores the SharedFlow in a private backing property
    private val _messages = MutableSharedFlow<Message>()

    // Exposes a read-only SharedFlow to subscribers
    val messages: SharedFlow<Message>
        get() = _messages.asSharedFlow()

    suspend fun sendMessageToEveryone(message: Message) {
        // Emits the message to subscribers
        _messages.emit(message)
    }
}

Just like with cold flows, you can use the collect() function to collect values from a SharedFlow.

You can also configure a SharedFlow to immediately replay already emitted values to new subscribers.
The replay cache works like a small history buffer, storing a fixed number of previous emissions.

To set how many previous emissions a new subscriber receives, use the replay parameter in MutableSharedFlow():

// Sets the number of already emitted messages new subscribers receive on subscription
const val MESSAGES_TO_REMEMBER = 10

class Chatroom {
    private val _messages = MutableSharedFlow<Message>(

        // Replays the set amount of last emitted messages to new subscribers
        replay = MESSAGES_TO_REMEMBER
    )

    val messages: SharedFlow<Message>
        get() = _messages.asSharedFlow()

    suspend fun sendMessageToEveryone(message: Message) {
        // Emits the message to subscribers of the messages flow 
        _messages.emit(message)
    }
}

Collecting a hot flow doesn't complete by itself, so you must cancel the collecting coroutines when you no longer need them.

A hot flow doesn't have a close or cancel operation.
Canceling collection only stops the corresponding subscriber from collecting.
To stop new emissions, cancel the coroutine or scope that produces values for the hot flow.

{style="note"}

Let's look at an example that uses a SharedFlow to model a chat room, which sends each new message to active subscribers and replays recent messages to subscribers that join later:

/* Detailed source-code truncated for AI context efficiency. */

{kotlin-runnable="true"}

In this example, CoroutineStart.UNDISPATCHED starts each collecting coroutine immediately.

This ensures that each coroutine reaches collect(), subscribes to messages, and suspends before sendMessageToEveryone() emits messages.
Without it, a collecting coroutine might start later and miss earlier emissions if the replay cache is too small.

Use explicit backing fields to expose hot flows

You can use explicit backing fields to expose a read-only SharedFlow while keeping a mutable backing field inside the class.

Explicit backing fields define the implementation type in the field declaration.
Inside the class, the compiler smart casts the property to the backing field type, so you can call the emit() function without a separate private backing property.

Explicit backing fields don't create the read-only wrapper that .asSharedFlow() provides.
Use this pattern only when downcasting the exposed flow isn't a concern.

{style="warning"}

Here's an example:

/* Detailed source-code truncated for AI context efficiency. */

{kotlin-runnable="true"}

Create a `StateFlow` Copy Section

StateFlow is a hot flow that stores a single state value and emits updates when that value is replaced with a new one.
New subscribers receive the current value as soon as they start collecting, and then receive new values each time the state updates.

You can use StateFlow to represent state that changes over time, such as loading progress, UI state, or the state of an object.

To create a StateFlow, use the MutableStateFlow() function with an initial value:

// Creates a MutableStateFlow with LoadingState.Started as the initial value
val result = MutableStateFlow<LoadingState>(LoadingState.Started)

To set the current state, use the value property:

fun loadBlob(url: String): StateFlow<LoadingState> {
    val result = MutableStateFlow<LoadingState>(LoadingState.Started)

    DownloadManager.startLoading(
        url,
        onPercentageLoaded = { percentage ->
            // Replaces the current state with the latest progress
            result.value = LoadingState.Percentage(percentage)
        },
        onCompletion = {
            // Replaces the current state with the completion state
            result.value = LoadingState.Done
        },
        onFailure = {
            // Replaces the current state with the failure state
            result.value = LoadingState.Failed
        }
    )
}

Setting value is thread-safe and replaces the current state, but updating value based on its previous value isn't atomic.
When a new state depends on the previous state, use .update() instead.

{style="note"}

Similar to MutableSharedFlow, MutableStateFlow exposes APIs for emitting updates.
If you expose it directly, any code that receives it can update the state by downcasting it to MutableStateFlow.

To prevent this, expose the mutable flow as a read-only StateFlow with the .asStateFlow() function:

fun loadBlob(url: String): StateFlow<LoadingState> {
    val result = MutableStateFlow<LoadingState>(LoadingState.Started)

    DownloadManager.startLoading(
        url,
        onPercentageLoaded = { percentage ->
            result.value = LoadingState.Percentage(percentage)
        },
        onCompletion = {
            result.value = LoadingState.Done
        },
        onFailure = {
            result.value = LoadingState.Failed
        }
    )

    // Exposes the loading state as a read-only StateFlow
    return result.asStateFlow()
}

Here's an example that uses StateFlow to report loading progress from a callback-based API:

/* Detailed source-code truncated for AI context efficiency. */

{kotlin-runnable="true"}

This example uses GlobalScope only to keep the callback-based API short.
In your own applications, pass a CoroutineScope to the function that starts the work, such as startLoading() in this example, and launch the coroutine in that scope so callers can cancel the work when they no longer need it.

{style="note"}

Since StateFlow is a hot flow, collection doesn’t complete on its own.
In this example, the .takeWhile() operator stops collection when loading reaches a terminal state.

A StateFlow emits an update only when the new value differs from the current value.

Avoid storing mutable objects in a StateFlow.
Mutating the object itself doesn't replace the current value, so collectors don't receive an update.

{style="warning"}

You can also update a StateFlow by calculating the new state from the current state.
Use the .update() function for these updates.
The .update() function updates the value atomically, which helps when multiple coroutines update the same MutableStateFlow.

If you only need to update a shared value and don't need to observe state changes over time, use the Kotlin Atomics APIs, such as AtomicInt or AtomicReference.

{style="note"}

Here's an example where likes are stored in a StateFlow, and each new state is calculated from the previous state:

/* Detailed source-code truncated for AI context efficiency. */

{kotlin-runnable="true"}

In this example, the .update() function increments the number of likes atomically.
This prevents lost updates when multiple coroutines call the like() function at the same time.

Store accumulated state in a `StateFlow`

Sometimes you might want subscribers to receive the result of all previous emissions, not only the latest emitted value.

For example, a chat room can keep its message history as a single state value.
When a new user joins the chat room, they receive the current message history first.
Then they continue receiving updates when new messages arrive.

You can model this behavior with a StateFlow.

To do so, store the full message history as the current value with StateFlow<List<Message>>, instead of broadcasting each chat message as a separate event with SharedFlow<Message>:

/* Detailed source-code truncated for AI context efficiency. */

{kotlin-runnable="true"}

In this example, messageHistory stores the full list of previous messages as the current state.
When a new message is sent, the .update() function creates a new list from the previous history and adds the new message atomically.

Updating immutable collections by creating a new collection can take more time as the collection grows.
You can create persistent collections with the Experimental kotlinx.collections.immutable library to make immutable collection updates more efficient.

{style="tip"}

Since messageHistory is a StateFlow, subscribers receive the current message history when they start collecting.
After that, they receive a new list each time a message is sent, which changes the chat history.

Convert cold flows to hot flows Copy Section

A cold flow runs its upstream operations separately for each collector.
When multiple subscribers need emissions from the same upstream collection, you can convert the cold flow to a hot flow that shares that collection with its subscribers.

The following simplified version of .shareIn() demonstrates this idea by collecting a cold flow once, emitting its values into a MutableSharedFlow, and exposing it as a read-only SharedFlow:

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.*

fun <T> Flow<T>.simpleShareIn(scope: CoroutineScope): SharedFlow<T> {
    val sharedFlow = MutableSharedFlow<T>()
    scope.launch {
        [email protected] {
            sharedFlow.emit(it)
        }
    }
    return sharedFlow.asSharedFlow()
}

suspend fun main() { 
    
}

In this example, simpleShareIn() starts a new coroutine in the provided scope.
To stop collecting from the upstream flow, cancel the scope that runs the collecting coroutine.

If the upstream flow throws an exception, this collecting coroutine fails.
Use operators such as .catch() or .retry() before sharing the flow to handle upstream exceptions before the collecting coroutine fails.

The built-in .shareIn() function provides this pattern without requiring you to create a MutableSharedFlow yourself.
It also adds options for controlling when upstream collection starts and stops, and how many previous emissions new subscribers receive.

To use the built-in .shareIn() function, provide the following arguments:

• The coroutine scope where the upstream flow is collected.
• The SharingStarted strategy that controls when upstream collection starts and stops.
  For example, SharingStarted.Eagerly starts upstream collection immediately in the provided scope, before any subscriber starts collecting.
• The optional replay value that controls how many previous emissions new subscribers receive.

The .shareIn() function collects the upstream flow in the provided coroutine scope and broadcasts its emissions to subscribers.

Here's an example where .shareIn() converts a cold flow to a hot flow that shares serialized chat messages with multiple subscribers:

/* Detailed source-code truncated for AI context efficiency. */

{kotlin-runnable="true"}

In this example, the .map() operator creates a cold flow that serializes each message.
Without the .shareIn() function, each collector runs that serialization separately.
The .shareIn() function shares one upstream collection, so each message is serialized once and then shared with all subscribers.

Because SharingStarted.Eagerly starts upstream collection immediately, the derived hot flow starts collecting chatroom.messages as soon as .shareIn() is called.

Similarly, to convert a cold flow to a StateFlow, use the .stateIn() function.

Unlike .shareIn(), .stateIn() requires an initial value because a StateFlow must always have a current value.

For example:

val lastUpdateFlow: StateFlow<Instant?> =
    chatroom
        .messageHistory
        .map { currentHistory -> currentHistory.lastOrNull()?.time }
        .stateIn(
            // Starts the sharing coroutine in this scope
            // The upstream flow, including .map(), runs in that coroutine
            derivedFlowsScope,

            // Starts collecting when the first subscriber appears
            // and stops when the last subscriber disappears
            SharingStarted.WhileSubscribed(),
            // Sets the initial state before the first upstream emission
            null,
        )

Cancel hot flows Copy Section

Hot flows don't stop when a subscriber is canceled.

When you cancel the coroutine that collects a hot flow, you only cancel that subscriber.
The hot flow can still emit values to other subscribers, and the coroutine that produces those values can keep running.

Hot flows themselves don't have a cancellation operation.
To cancel a hot flow, cancel the coroutine or scope that produces values for it.

Hot flows created with the .shareIn() or .stateIn() extension functions continue to collect the upstream flow until the sharing coroutine is canceled.
To stop collection from the upstream flow, cancel the scope that runs the sharing coroutine.

You can also stop upstream collection automatically when there are no subscribers with SharingStarted.WhileSubscribed().