ConcurrentHashMap

ReadWriteLock

Flux<T>

Channel<T>

fun functionImpl(a: Int, b: String) {}

fun nonSuspending(function: (Int, String) -> Unit) {}
fun suspending(function: suspend (Int, String) -> Unit) {}

fun test() {
  nonSuspending(::functionImpl)
  suspending(::functionImpl)
}

In 'uncaughtExceptionHandler' ...

2019-10-09 14:49:52.581 4418-4418/my.app.id E/AndroidRuntime: FATAL EXCEPTION: main
    Process: my.app.id, PID: 4418
    kotlinx.coroutines.JobCancellationException: Job was cancelled; job=SupervisorJobImpl{Cancelled}@b43ae6d

val myEvent = Channel<Unit>(Channel.RENDEZVOUS)

viewModel.myEvent.consumeAsFlow()
	.onEach {
		Toast.makeText(requireContext(), R.string.error, Toast.LENGTH_LONG).show()
	}
	.launchIn(lifecycleScope)

Flow<Change<oldValue: String, newValue: String>>

debounce

Change

coroutineScope

withContext

coroutineScope

coroutineContext

withContext

coroutineContext

CoroutineScope.coroutineContext

onReceiveOrNull

ReceiveChannel

onReceiveOrNull

ReceiveChannel

ExperimentalCoroutinesApi

delay()

RxJavaPlugins.setErrorHandler { }
    val flow = rxSingle<Unit> { throw IOException("") }.toFlowable().asFlow()
    runBlocking {
        repeat(10000) {
            combine(flow, flow) { _, _ -> Unit }
                .catch {}
                .collect { }
        }
    }

private val job = Job()
    private val coroutineContext = job + Dispatchers.Main
    private val scope: CoroutineScope = CoroutineScope(coroutineContext)

   override fun onResume() {
        super.onResume()
        scope.launch { subscribe() }
    }

    private suspend fun subscribe() {
        dataStore.subscribe()
         val channel: ReceiveChannel = dataStore.receiveMessage()
        for (message in channel) {
            when (message.type) {
                "error" -> textView_scrollable.addTextNl("error: ${message.message}")
                "ticker" -> textView_scrollable.addTextNl("ticker: ${message.sequence}, ${message.productId}, ${message.price}")
            }
        }
    }

    override fun onPause() {
        dataStore.unsubscribe()
        job.cancel()
        super.onPause()
    }

consumeEach

private suspend fun subscribe() {
       dataStore.subscribe()
        val channel: ReceiveChannel = dataStore.receiveMessage()
       channel.cosumeEach { message ->
           when (message.type) { 
               "error" -> textView_scrollable.addTextNl("error: ${message.message}")
               "ticker" -> textView_scrollable.addTextNl("ticker: ${message.sequence}, ${message.productId}, ${message.price}")
           }
       }
   }

JobCancellationException

consumeEach

Performs the given action for each received element and cancels the channel after the execution of the block. If you need to iterate over the channel without consuming it, a regular for loop should be used instead.

try-finally

NonCancellable

SupervisorScope

private val termination = CountDownLatch(1)
private val scope = dispatcher + Job().apply {
    thread {
        // use non-dameon thread to keep application alive until all tasks completed.
        termination.await()
        while (children.any { it.isActive }) Thread.sleep(50)
    }
}

fun shutdown() {
    termination.countDown()
}

delay

delay

Dispatchers.Main.immediate

NoSuchMethodError

publish

actual suspend fun publish(topic: String, msg: MqttMessage, timeout: Long): MqttError? = try {
        client.publish(topic, PahoMqttMessage(msg.payload.toByteArray()))
        null
    } catch (mqttPersistenceEx: MqttPersistenceException) {
        MqttError(mqttPersistenceEx.message ?: "Message persistence failed.", MqttStatus.MSG_PERSISTENCE_FAILED)
    } catch (mqttEx: MqttException) {
        MqttError(mqttEx.message ?: "Message delivery failed.", MqttStatus.MSG_DELIVERY_FAILED)
    }

Exception in thread "main" java.lang.NoSuchMethodError: 'java.lang.Object org.digieng.kmqtt.client.MqttClient.publish$default(org.digieng.kmqtt.client.MqttClient, java.lang.String, org.digieng.kmqtt.client.MqttMessage, long, kotlin.coroutines.Continuation, int, java.lang.Object)'

mainScope.launch{
	
			onDateChanged // conflatedBroadcastChannel
                .asFlow()
                .distinctUntilChanged()
                .collect{ ... }

            onTimeChanged // conflatedBroadcastChannel
                .asFlow()
                .distinctUntilChanged()
                .collect{ ... }
}

readLine()

onStart

onCompletion

Flow

onStart

onCompletion

suspend

Basically, if you declare a function `suspend`, when this function gets called, its actual invocation is placed in a Queue of Jobs to be executed in some Dispatcher. 
If the Dispatcher Thread happen to be the same as the calling Thread, it may be possible that the invocation happens immediately in the same running loop.
After the *suspending function* executes free of exceptions, then code will resume at the line right after the function call site.

suspending function

The calling Dispatcher Thread returns at the calling site. When the *suspend function* is over the Continuation mechanics make sure it resumes in the same caller Dispatcher, not necessarily the same caller Thread but the same caller Dispatcher

Exception in thread "main" java.lang.NoSuchMethodError: org.digieng.kmqtt.client.MqttClient.connect$default(Lorg/digieng/kmqtt/client/MqttClient;Lorg/digieng/kmqtt/client/MqttConnectionOptions;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
	at org.example.mqtt_test.MainKt$main$1.invokeSuspend(main.kt:11)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(Dispatched.kt:241)
	at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:270)
	at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:79)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:54)
	at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:36)
	at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
	at org.example.mqtt_test.MainKt.main(main.kt:8)
	at org.example.mqtt_test.MainKt.main(main.kt)

private val _state = liveData<LoginViewState> {
        coroutineScope {
            launch {
                usernamePasswordChannel.consumeEach { (username, password) ->
                    validateLoginDataUseCase.validateLoginData(
                        username,
                        password
                    ).perform(latestValue!!).also { emit(it) }
                }
            }
            launch {
                isLoggingInChannel.consumeEach {
                    usernamePasswordChannel.value.let { (username, password) ->
                        performLoginUseCase.login(
                            username,
                            password
                        )
                    }.perform(latestValue!!).also { emit(it) }
                }
            }
        }
    }