Flow

Flow

interface DataSource {
    val flow: Flow<List<Foo>>
}

class Repository(
    private val source: DataSource
)
    val diff: Flow<Diff<Foo>> get() = TODO()
    suspend fun getLatest(): List<Foo> = TODO()
}

Repository

Diff

LiveData

val dispatcher = Dispatchers.Main.immediate
val scope = CoroutineScope(dispatcher)

fun main() {
    println("Started.")

    scope.launch { delay(30); println("Hi!") }

    println("Finished.")
}

yield()

1.3.2-1.3.60

ReceiveChannel<T>.onReceiveOrNull

onReceiveOrNull

onReceiveOrNull

onReceiveOrNull

import kotlinx.coroutines.channels.onReceiveOrNull as onReceiveOrNullExt

select {
    channel.onReceiveOrNullExt().invoke { foo ->
        [...]
    }

    anotherChannel.onReceive { bar ->
        [...]
    }
}

FileReader

continuation.invokeOnCancellation -> abort() -> onabort -> continuation.cancel()

suspend fun Blob.readAsDataUrl2(): String {
    val fileReader = FileReader()
    return suspendCancellableCoroutine { continuation ->
        fileReader.onload = {
            val dataUrl = fileReader.result as String
            continuation.resume(dataUrl)
        }
        fileReader.onerror = {
            continuation.resumeWithException(DomExceptionThrowable(fileReader.error as DOMException))
        }
        fileReader.onabort = {
            continuation.cancel()
        }
        continuation.invokeOnCancellation { fileReader.abort() }

        fileReader.readAsDataURL(this) // <-- Async work starts here
    }
}

suspend fun foo(): Foo {
    return runCatching {
        // do something with calling other suspend functions
        Foo()
    }.getOrElse { exception ->
        // do somethiing
        Foo()
    }
}

suspend fun foo(): Foo {
    return runCatching {
        // do something with calling other suspend functions
        Foo()
    }.getOrElse { exception ->
        if (exception is CancellationException) {
            throw exception
        }
        // do something
        Foo()
    }
}

class MainActivity : Activity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        CoroutineScope(SupervisorJob() + Dispatchers.Main).launch {
            while (true) {
                combine(flowOf(Unit), flowOf(Unit)) { _ -> Unit }.collect()
            }
        }
    }
}

Flags: 0x28c8bf46
    Package: com.example.weirdcrash v1 (1.0)
    Foreground: Yes
    Build: TCL/5009D/U5A_PLUS_3G:7.0/NRD90M/5009D_ALWE_V2.9_20180529:user/release-keys
    
    *** *** *** *** *** *** *** *** *** *** *** *** *** *** *** ***
    Build fingerprint: 'TCL/5009D/U5A_PLUS_3G:7.0/NRD90M/5009D_ALWE_V2.9_20180529:user/release-keys'
    Revision: '0'
    ABI: 'arm'
    pid: 29639, tid: 29639, name: mple.weirdcrash  >>> com.example.weirdcrash <<<
    signal 11 (SIGSEGV), code 1 (SEGV_MAPERR), fault addr 0x15608
        r0 00015000  r1 12d35700  r2 00000002  r3 00000000
        r4 af58a6d8  r5 12d31ee0  r6 12d37280  r7 00000000
        r8 12d35700  r9 ac684400  sl 12cb0a90  fp 12d33bf8
        ip bef2f84c  sp bef2f8b0  lr 9901f5ef  pc 9901f608  cpsr 000f0030
    
    backtrace:
        #00 pc 00000608  /dev/ashmem/dalvik-jit-code-cache (deleted)
        #01 pc 000005ed  /dev/ashmem/dalvik-jit-code-cache (deleted)
     29639

viewModelScope.launch {
                val userProfileResponse = apiManager.client!!.getProfile()
                if (userProfileResponse.isSuccessful) {
                    _viewState.value = _viewState.value?.copy(isLoading = false, profile = userProfileResponse.body())
                } else {
                    Log.e(TAG, "Not success")
                }
        }

BroadcastChannel

Coroutine

sealed class Tree {
    data class Parent(val children: List<Tree>): Tree()
    data class Leaf(val value: Int): Tree()
}

fun Tree.simpleSum(): Int {
    return when (this) {
        is Tree.Parent -> children.map { it.simpleSum() }.sum()
        is Tree.Leaf -> value
    }
}

fun Tree.concurrentSum(): Int {
    return when (this) {
        is Tree.Parent -> runBlocking {
            children.map { async { it.simpleSum() } }
                .awaitAll().sum()
        }
        is Tree.Leaf -> value
    }
}

suspend fun Tree.questionableSum(): Int {
    return when (this) {
        is Tree.Parent -> coroutineScope {
            children.map { async { it.questionableSum() } }
                .awaitAll().sum()
        }
        is Tree.Leaf -> value
    }
}

Job

concurrentSum

Job

questionableSum

Flow

1.3.2

@ExperimentalCoroutinesApi

launch { field.setAs { calculate() } }
...
private fun StringProperty.setAs(calc: () -> Any) = set(
    try {
        calc().toString()
    } catch (e: Exception) {
        e.message
    }
)

setAs

calc

flow {
  for (i in collection) {
    try {
      emit(i)
    } finally {
      doCleanupOn(i)
    }
  }
}

fun main() = runBlocking {
    val job = launch {
        println("Coroutine start")
        launch {
            println("Child coroutine start")        
            println("Child coroutine end")

            launch {

                println("Child-inner coroutine start")        
                println("Child-inner coroutine end")
            }
        }
        println("Coroutine end")
        println("Coroutine end!")
    }
    println("Join")
    println("Done")
}

suspend

Object

runBlocking {
  val channel = Channel<Int>(bufferSize)
  val producers = (0..producerCount).map { launch { produceValues(channel) } }

  while (producers.any { it.isActive }) {
    // time-of-check time-of-use race condition here. If there were any producers active at the time of check but had finished sending messages, we end up waiting on messages that we'll never receive
    val message = channel.receive()
    doSomethingWithMessage(message)
  }

  coroutineContext.cancelChildren()
}

CoroutineExceptionHandler

runBlocking

fun main(): Unit = runBlocking(CoroutineExceptionHandler { _, e ->
    println("Boom: $e")
}) {
    (0..10).map {
        launch {
            println("Throw in attempt #$it")
            throw RuntimeException("Attempt #$it")
        }
    }
    Unit
}