send

Channel(0)#send

kotlinx.coroutines

1.3.1

channelFlow

callbackFlow

inline

get()

CoroutineScope.future{}

get()

get()

get()

CoroutineScope

future{}

supervisorScope

SupervisorJob()

SupervisorJob(coroutineScope.coroutineContext[Job])

cache.getIfPresentAsync("x").let { deferred ->
				assertNotNull(deferred)
				assertTrue(deferred.isCompleted)
				assertEquals(42, deferred.await())
			}

deferred!!.isCompleted

Flow<T>

ReceiveChannel<T>

@Suppress("DeprecatedCallableAddReplaceWith")
@Deprecated("You should be using Flows, not channels.")
fun <T> Flow<T>.collectAsReceiveChannel(scope: CoroutineScope): ReceiveChannel<T> {
    val flow = this

    val channel = Channel<T>()

    val flowJob = scope.launch {
        flow
            .onEach { item -> channel.send(item) }
            .collect()
    }

    channel.invokeOnClose {
        flowJob.cancel()
    }

    return channel
}

callbackFlow

close(exception)

cancel(CancellationException("Error occurred", exception))

callbackFlow

channelFlow

addListener

Observable.create { }

@Test
    fun listCanBeObserved() = runBlocking {
        dao.insert(thing)

        val data: List<Thing> = dao.observeAll().single()
        val expected = Thing()
        assertEquals(expected, data)
    }

fun listCanBeObserved() = runBlocking {
        dao.insert(thing)

        val data: List<Thing> = dao.observeAll().collect {
            val expected = Thing()
            assertEquals(expected, data)
        }
    }

GlobalScope.launch {
            Log.d("rafa", "thread:${Thread.currentThread()}") // this prints Thread[DefaultDispatcher-worker-1,5,main]
            someTextView.text = "new test"
        }

Iterable<Deferred<T>>

@ExperimentalCoroutinesApi // Since 1.2.1, tentatively till 1.3.0
public suspend inline fun <T> ObservableSource<T>.collect(action: (T) -> Unit) =
    openSubscription().consumeEach(action)

private val processor = scope.launch {
        val batch = ArrayList<String>(MAX_BATCH)
        
        suspend fun sendBatch() {
            val payload = batch.joinToString(separator = "\n")
            batch.clear()
            return sendPayload(payload)
        }
        
        try {
            while (isActive) {
                batch.add(hitChannel.receive())
                if (batch.size == MAX_BATCH) {
                    sendBatch()
                }
            }
        } finally {
            if (batch.isNotEmpty()) {
                sendBatch()
            }
        }
    }

sendBatch

isActive

withContext(NonCancellable)

BroadcastArrayChannel

start

class EncryptionStep{
	var key: SecretKey? = null
	val input: ByteWriteChannel get() = _input
	val output: ByteReadChannel get() = _output

	private val _input = ByteChannel()
	private val _output = ByteChannel()

	@UseExperimental(InternalAPI::class)
	fun CoroutineScope.start(encrypted: Pair<ByteReadChannel, ByteWriteChannel>) {

		launch {
			_input.read{
				runBlocking{
					when(val key = key){
						null -> encrypted.second.writeAvailable(it)
						else -> encrypted.second.writeAvailable(encryptData(key, it.moveToByteArray()))
					}
				}
			}
		}
	}
}


fun CoroutineScope.encryptionStep(encrypted: Pair<ByteReadChannel, ByteWriteChannel>): EncryptionStep {
	val step = EncryptionStep()
	
	with(step){
		start(encrypted)
	}

	return step
}

ConflatedBroadcastChannel

E/File$object: kotlinx.coroutines.channels.ClosedSendChannelException: Channel was closed
        at kotlinx.coroutines.channels.Closed.getSendException(AbstractChannel.kt:1048)
        at kotlinx.coroutines.channels.AbstractSendChannel.offer(AbstractChannel.kt:166)
        at kotlinx.coroutines.channels.ChannelCoroutine.offer(ChannelCoroutine.kt)

runBlocking

Dispatchers.Default

fun main() = runBlocking() {
    print(Thread.currentThread().name) // main
}

fun main() = runBlocking(Dispatchers.Default) {
    print(Thread.currentThread().name) // DefaultDispatcher-worker-1
}

fun main() {
  runBlocking {
      throw Exception("e")
  }
}

main.runBlocking

Dispatchers.Default

runBlocking(Dispatchers.Default)

BaseContinuationImpl.resumeWith

Dispatchers.Default

Module with the Main dispatcher is missing. Add dependency providing the Main dispatcher, e.g. 'kotlinx-coroutines-android'

fun main() = runBlocking {
    val job = Job()
    launch {
        delay(1000)
        job.completeExceptionally(IllegalStateException("Failed"))
    }
    try {
        job.join()
        println("foo")
    } catch (e: Exception) {
        println("bar")
        e.printStackTrace()
    }
}