/**
 * Returns the context of the current coroutine.
 */
@SinceKotlin("1.3")
@Suppress("WRONG_MODIFIER_TARGET")
@InlineOnly
public suspend inline val coroutineContext: CoroutineContext
    get() {
        throw NotImplementedError("Implemented as intrinsic")
    }

suspend inline val

suspending

Continuation<OriginalReturnType>

OriginalReturnType

Continuation

requestToken

createPost

suspend fun requestToken

requestToken(Continuation<Token>)

suspend fun createPost

createPost(Continution<Post>)

sm

fun getDefault(): Any {
    check(false) { "failed"}
    TODO()
}

fun method(param: Any = getDefault()) = flow<Any> {
    TODO()
}

fun method() = flow {
    method(getDefault())
        .collect {
            emit(it)
        }       
}

runBlocking { //code }

// You are not allowed to change this class.
class Foo : AutoCloseable {
    private val outstream = PipedOutputStream()
    private val instream = PipedInputStream(outstream)

    override fun close() {
        instream.close()
    }

    fun blockingCall() {
        instream.read() // blocks the thread until e.g. the stream is closed
    }
}

/**
 * Wraps the AutoCloseable usage so that the AutoCloseable is closed when the coroutine is cancelled.
 */
suspend fun <T: AutoCloseable> closeOnCancel(block: (T) -> Unit) {
    // How do I implement this function?
    // I have tried using suspendCancellableCoroutine, but I think I misunderstood that function.
}

fun cancelBlockingCall() {
    val foo = Foo()
    runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) {
        withTimeout(100) {
            foo.closeOnCancel { it.blockingCall() }
        }
        println("Yay, arriving here means problem solved")
    }
}

closeOnCancel

CoroutineScope

launch

CoroutineContext

Dispatchers.Main

ServiceLoader

Dispatchers.Main

// In ViewModel
    fun toggleScan() {
        when {
            isScanning.value -> stopScan()
            else -> {
                startScan()
                presenterScope.launch(Dispatchers.Default) {
                    delay(DISCOVERY_TIMEOUT)
                }
            }
        }
    }

LinuxArm32Hfp

java.lang.IllegalStateException: This job has not completed yet

@Test
fun testHotFlow() = coroutinesTestRule.testDispatcher.runBlockingTest {
    val eventChannel = Channel<String>()
    val events = eventChannel.receiveAsFlow()

    val event = events.first()
    assertEquals(event, null)
}

@Test
fun testHotFlow() = coroutinesTestRule.testDispatcher.runBlockingTest {
    val eventChannel = Channel<String>()
    val events = eventChannel.receiveAsFlow()

    launch {
        eventChannel.send("First element")
    }

    val event = events.first()
    assertNotNull(event)
}

suspendCancellableCoroutine {}

withTimeout {}

cont.resume()

suspendCancellableCoroutine {}

dispatcher.advanceUntilIdle()

getLongestCompletedStreak

private val taskStreaksMax =
        allTasksWithTaskStatisticsFlow.map { allTasksWithTaskStatistics ->
            allTasksWithTaskStatistics.map { taskWithTaskStatistics ->
                TaskStreak(
                    taskWithTaskStatistics.task.id,
                    taskWithTaskStatistics.taskStatistics.getLongestCompletedStreak()
                )
            }
        }.flowOn(defaultDispatcher)

val jvmTest by getting {
    dependencies {
        implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core-test:1.5.2")
    }
}

kotlinx.coroutines

Dispatchers.Default

onUndeliveredElement

Mutex

suspend fun List<TaskStatistic>.getCurrentCompletedStreak(): Int =
    withContext(Dispatchers.Default) {
        var currentStreak = 0
        for (statistic in this@getCurrentCompletedStreak.reversed()) {
            if (!statistic.taskCompleted) break
            currentStreak++
        }
        currentStreak
    }

await

enqueue

100_000

enqueue

suspendCoroutine

suspendCoroutine

Call<T>.await()

suspendCoroutine

Application

Application

GlobalScope.launch

Application

CoroutineScope

Application

application.launch

DelicateCoroutinesApi

private class JobMap<K> {
    private val jobs = atomic(mapOf<K, Job>())

    fun add(key: K, factory: () -> Job) {
        jobs.update {
            val job = factory()
            job.invokeOnCompletion { cause ->
                if (cause == null || cause !is CancellationException) jobs.update { it - key }
            }
            it + (key to job)
        }
    }

    fun remove(key: K) {
        jobs.getAndUpdate { it - key }[key]?.cancel()
    }
}

scope: CoroutineScope

val context = Dispatchers.Default + CoroutineActionId(id)
val newJob= scope.launch(context) { ... }

CoroutineActionId(id)

AbstractCoroutineContextElement

scope

val children = scope.coroutineContext[Job]?.children
val activeAction = children?.firstOrNull { it[CoroutineActionId]?.id == id }
if (activeAction != null) {
    activeAction.cancelChildren()
    activeAction.join()
}

Job

launch

val children = scope.coroutineContext[Job]?.children

CoroutineActionId

children?.forEach {
     println("#state: caid = "  + it[CoroutineActionId]?.id)
}

Channel

Queue

class Queue<T>(private val limit: Int = 5) {

  private val elements: MutableList<T> = mutableListOf()

  suspend fun enqueue(t: T): Unit = suspendCancellableCoroutine { cont ->
    var enqueued = false
    while (!enqueued) {
      if (elements.size < limit) {
        elements.add(t)
        enqueued = true
        cont.resume(Unit)
      }
    }
  }

  suspend fun dequeue(): T = suspendCancellableCoroutine { cont ->
    var dequeued = false
    while (!dequeued) {
      if (elements.isNotEmpty()) {
        val element = elements.first()
        elements.remove(element)
        dequeued = true
        cont.resume(element)
      }
    }
  }
}

KafkaProducer

send

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

val loggedIn: Flow<Boolean> = prefs.data
    .map { preferences ->
        preferences[authenticated] ?: false
    }

prefs

prefs.data.onStart{ emit (initialPrefs) }

val response = readPacketFlow().first { response -> response is Acknowledgement }
when (response) {
	Acknowledgement -> true
	else -> false
}