Channel.offer

Channel.send

flowB

val flowA: Flow<Boolean>

true

flowB

flowA

false

try {
   // some suspending operations to connect to web socket
} catch(e: Exception) {
   throw WebSocketConnectionFailedException("meaningful high-level msg", e)
}

withTimeout

TimeoutCancellationException

CancellationException

try {
   // so some operations to connect to web socket
} catch(e: CancellationException) {
   throw e // no wrapping here
} catch(e: Exception) {
   throw WebSocketConnectionFailedException("meaningful high-level msg", e)
}

fun scroll() = flow {
  try {
    // do stuff
  } finally {
    withContext(NonCancellable) { suspendingCleanUpFunction() }
  }
}

Job

Normal cancellation of a job is distinguished from its failure by the type of its cancellation exception cause. If the cause of cancellation is CancellationException, then the job is considered to be cancelled normally. This usually happens when cancel is invoked without additional parameters. If the cause of cancellation is a different exception, then the job is considered to have failed. This usually happens when the code of the job encounters some problem and throws an exception.

CancellationException

cause

cause

CancellationException

CancellationException

cause

CancellationException

CancellationException

CancellationException

cause

CancellationException

runBlocking

val (a, b) = runBlocking(Dispatchers.Default) {
        val aDeferred = async {getA()}
        val aDeferred = async {getB()}
        return@runBlocking aDeferred.await() to bDeferred.await()
}

Flow.cancellable()

flow

currentCoroutineContext

CoroutineScope

EXACTLY_ONCE

flow.chunked()

var caughtException = false
        val batches = batchResults(inputFlow, chunkSizePerMessage)
            .map { createResults(it) }
            .catch {
                caughtException=true
                emit(Results.TerminateWithError(it))
            }
        emitAll(batches)
        if(!caughtException) {
            emit(Results.SuccessfulFinish())
        }

var caughtException

function* generatorFunction(i) {
   yield i
   yield i +1
}

let gen = generator(5)
gen.next() //  value=5, done=false
gen.next() //  value=6, done=false
// WHAT I WANT IS THIS
gen.next(100) // <-- here I pass value and get back another value at next yield point

suspend fun <T> Iterable<T>.forEachParallel(
  f: suspend (T) -> Unit
) {
  coroutineScope {
    forEach { t ->
      launch {
        f(t)
      }
    }
  }
}

async

coroutineScope

coroutineScope

async

launch

/**
 * Classes and interfaces marked with this annotation are restricted when used as receivers for extension
 * `suspend` functions. These `suspend` extensions can only invoke other member or extension `suspend` functions on this particular
 * receiver and are restricted from calling arbitrary suspension functions.
 */
@SinceKotlin("1.3")
@Target(AnnotationTarget.CLASS)
@Retention(AnnotationRetention.BINARY)
public annotation class RestrictsSuspension

Foo

incoming: ReceiveChannel<String>

data class MyType(val str: String)

Foo

incomingMyType: ReceiveChannel<MyType>

MyType

String

Flow

val Foo.incomingMyType: ReceiveChannel<MyType>
    get() = produce {
        for (str in incoming) {
            send(MyType(str))
        }
    }

StateFlow<Unit?>

SharedFlow

@ExperimentalCoroutinesApi

CoroutineScope

sessionCoroutineContext

MessageRepository

suspend fun

MessageRepository

withContext(sessionCoroutineContext) { … }

viewModelScope

private suspend fun processAndSend() {
    documentService.getDocuments(reference) // returns a Sequence<Documents>
            .chunked(10)
            .map { chunkedDocuments ->
                val processedDocuments = httpClient.processDocuments(chunkedDocuments) // compile error, suspending call
                queueSender.send(processedDocuments) // compile error, suspending call
            }
}

// compile error = Suspension functions can be called only within coroutine body