First Coroutines example

fun main() = runBlocking { // this: CoroutineScope
launch { // launch a new coroutine and continue
delay(1000L) // non-blocking delay for 1 second (default time unit is ms)
println(“World!”) // print after delay
}
println(“Hello”) // main coroutine continues while a previous one is delayed
}

It will print
Hello
World!

Let us discuss the keywords used here to create coroutines
1. launch()

• is a coroutine builder.
• is a coroutine builder.
• It launches a new coroutine concurrently with the rest of the code, which
• continues to work independently.
• That’s why Hello has been printed first.
• delay is a special suspending function.
• It suspends the coroutine for a specific time.
• Suspending a coroutine does not block the underlying thread, but allows other coroutines to run and use the underlying thread for their code.

2. runBlocking()
1. is also a coroutine builder that bridges the non-coroutine world of a regular fun main() and the code with coroutines inside of runBlocking { … } curly braces

2. This is highlighted in an IDE by this: CoroutineScope hint right after the runBlocking opening curly brace.
3. If you remove or forget runBlocking in this code, you’ll get an error on the launch call, since launch is declared only in the CoroutineScope:
Unresolved reference: launch.

COROUTINES

• Coroutine Concept
• Scope
• suspending functions
• Jobs
• Context
• Exception Handling

Use of Coroutines

• Threads are resource intensive — it need resource to start stop
• Coroutines are lightweight threads and uses thread pools
• Simplify async code, callbacks and synchronisation
• Simple syntax
• Can pause and resume at any time, On a number of thread

Scope Create and run coroutines, provides life cycle events
Context The scope provides a context in which the coroutine runs
Suspending function Functions that can be run in a coroutines
Jobs handle on coroutines
Deferred a future result of a coroutine
Dispatcher manages which thread(s) the coroutine runs on
Error handling

1. runBlocking
2. launch
3. GlobalScope
4. repeat

Scope
- It provides lifecycle methods for coroutines
- Allows us to start and stop coroutines
- GlobalScope.launch{} The scope of the coroutine is the lifecycle of entire application
- runBlocking Creates a scope and runs coroutines in a blocking way
- coroutineScope Creates a new scope does not complete until a children coroutines is complete

runBlocking{
 launch{
 delay(1000L)
 println(“Hello”)
 }
}

GlobalScope.launch{
 delay(1000L)
 println(“Hello”)
}

coroutineScope{
 delay(1000L)
 println(“Custom coroutinesScope”)
}

Context
- A context is a set of data relates to coroutines
- All coroutines have an associated context
- You can start and stop coroutines
- But context is the data associated with coroutines
- Important elements of context
- 1. Dispatcher — Which thread the coroutines is run on
- 2. Job — Handle on the coroutine lifecycle
- If you explore GlobalScope, you can see coroutineScope and its use inside it

Suspending Function
- A function that can be run in a coroutine
- Make callbacks seamless

suspend fun sayHello(){
 print(“”Hello)
}

GlobalScope.launch{
 sayHello()
}

Jobs
- A .launch() call returns a job
- Allows us to manipulate the coroutine lifecycle
- Live in the hierarchy of other jobs both as parents or children

job{
 job{}
job{
 job{} 
} 
}

val job1 = GlobalScope.launch{
 coroutineScope{
 val job2 = launch{
 // processing
 }
 }
 }

- Can access lifecycle variables and methods
- cancel()
- join()
- If a job is cancelled, all it’s parents and children will be cancelled too.

Dispatchers
- A dispatcher determines which thread or thread pool the coroutine runs on
- Different dispatchers are available depending on the task specificity

launch(Dispatchers.Default){
 // do some cpu intensive processing task here
}

Common Dispatchers

1. Main
- Main thread update in UI driven applicaiton(eg Android)
- Main dispatcher needs to be defined in grade
2. Default
- Useful for CPU intensive work
IO
- useful for network communication or reading or writing files
Unconfined
- Starts the coroutine in the inherited dispatcher that called it
newSingleThreadContext(“”MyThread)
- Forces creation of a new thread
- This is not used more often, because it more expensive and heavy

Async
- Another way to start a coroutines
- Just like launch, except it returns a result
- In the form of a deferred
- Deferred — a future promise of a returned value
- When we need the value, we call await() (blocking call)`
- If the value is available, it will return immediately
- If the value is not available, it will pause the thread until it is

suspend function getRandom() = Random.nextINt(1000)

val valueDeffered = GlobalScope.async{
 getRandom()
}
// DO ome processing here
val finalValue = valueDeffered(
```

withContext
- Allows us to easily change context
- Easily switch between dispatchers
- very Lightweight

launch(Dispatchers.Default){
 // default context
withCOntext(Dispatcher.IO){
 // IO context 
}
// back to default context
}

Exception Handling in Coroutines
-Exception behaviour depends on the coroutines builder
launch
- Propagates through the parent-child hierarchy
- The exception will be thrown immediately and jobs will fail
- use try-catch or an exception handler

async
- Exceptions are deferred until the result is consumed
- If the result is not consumed, the exception is never thrown
- try-catch in the coroutine or in the await() cal

val myHandler = CoroutineExceptionHandler { coroutineCOntext, throwable ->
 // handle exception
}

launch(myHandler){
 // do some task here
 throws IndexOutOfBoundException()
}
```
- If we want to pass Coroutine handler and also a Dispatcher then we can use like this
```
launch(Dispatchers.Default + myHandler){
 // do some task here
 throw IndexOutOfBoundException()
}

Sample Android Application for coroutines
- Objectives
- Get an image from a URL
- Apply processing
- Show it in an ImageView

• In this example we will create a CoroutineScope with using scope = main, because we have to update the image in main thread

private val coroutineScope = CoroutineScope(Dispatchers.Main)

coroutineScope.launch {
 val originalDeferred = coroutineScope.async(Dispatchers.IO) { getOriginalBitmap() } // This will download the image bitmap from the url

val originalBitmap = originalDeferred.await() // By calling this, it will wait until the result not getting

val filteredDeferred = coroutineScope.async(Dispatchers.Default) { applyFilter(originalBitmap) } // another background task to apply the filter

val filteredBitmap = filteredDeferred.await() // in this line it will execute

loadImage(filteredBitmap) // in this it will load the image to the Image view in the main thread
}

Click here for Basic Example here
Another Example to use Coroutines with Retrofit in MVVM pattern
- Get a list of country info from an endpoint
- Use coroutines with Retrofit and MVVM
- Show it in a RecyclerView
- Click here for Retrofit with Coroutines Example

Coroutines, Room, MVVM
- Objective of this project
- Create, Read and delete data in a Room Database.
- Use coroutines with Room and MVVM
- Navigate in an app based on that data
- Click here for Room databse with Coroutines Example
- IN this example we have used coroutines with Room database
- Initialize Coroutine scope

private val coroutineScope = CoroutineScope(Dispatchers.IO)
private val db by lazy { UserDatabase(getApplication()).userDao() }

- Now do the operation inside coroutine scope

coroutineScope.launch {
 val user = db.getUser(username)
 if(user != null) {
         withContext(Dispatchers.Main) {
             error.value = “User already exists”
          }
 } else {
         val user = User(username, password.hashCode(), info)
         val userId = db.insertUser(user)
         user.id = userId
         LoginState.login(user)

withContext(Dispatchers.Main) {
         signupComplete.value = true
      }
   }
}

Asynchronous Flow
- What is Asynchronous Flow
- Creating flows
- Properties
- Operators
- Buffering
- Composing flows
- Exception handling

Asynchronous Flow
- Flows emits value
- Some other coroutines to collect the emitted values
- A flow is a stream of values that are Asynchronously computed
- flow{…} — It’s a builder. It will transfer the value
- emit(value) — Transmit a value
- collect{…} — Receive the values. Until the collect method is not called it will not emmit the value

fun sendPrimes(): Flow<Int> = flow{
 val primeNumbers = listOf(2,3,5,7,11,13,17,19,23,29)
 primeNumbers.forEach {
 kotlinx.coroutines.delay(it*100L)
 emit(it)
 }
}

• We can call the flow method like this

sendPrimes().collect {
     println(“Received prime numnbers $it”)
}

Creating flow
- There different way to create a flow
- 1. Generate flow by emitting each value

fun sendNumbers() = flow {
    for(i in 1..10){
       emit(i)
    } 
}

using .asFlow()
- A collection can be converted directly into a flow using collection.asFlow() function

fun sendNumbers() = listOf(1,2,3,4,5).asFLow()

flowOf(T)
- We can create flow using **flowOf** function
- A flow can be generated from a number of parameters of any Type
```
fun sendNumbers() = flowOf(“One”, “Two”, “Three”)
```
Flow Properties
- Cold
 — Flows are cold
 — The code does not run until the **collect** function is called
- Cancelation
 — Flows cant be cancelled by itself
 — It will be cancelled when the encompassing coroutine is cancelled
 — Flow is transparent for cancellation

fun sendNumbersTOCheckCancellation() : Flow<Int> = flow{
 val list = listOf(1, 2, 3)
 list.forEach {
      kotlinx.coroutines.delay(100L)
      emit(it)
 }
}

runBlocking {
 val numbersFlow = sendNumbersTOCheckCancellation()
      println(“FLow has not started yet”)
      println(“Started Flow now”)
      withTimeoutOrNull(1000){
        numbersFlow.collect {
           println(it)
       }
     }
   }

- As per the above code it will print each number with 400 mlliseconds delay
- But it will cancel the flow after 1000 milliseconds defined withTimeoutOrNull(1000)
- We can cancelling the coroutine after 1000 milliseconds, so the flow will cancel
Operator
- It take an input flow, transform it and return an output flow
- Operators are cold
- The returning flow is synchronous
- **You can find all the operator by putting a dot after a flow

(1..10).asFlow()
- List of all the operators
- also
- apply
- let
- run
- runCatching
- takeIf
- take
- takeUnless
- reduce
- transform
- map
- broadcastIn
- buffer
- cancellable
- catch
- collectIndexed
- collectLatest
- combine
- combineTransform
- conflate
- count
- debounce
- filter
- filterNot
- first
- firstOrNull
- flatMapConcat
- flatMapMerge
- flatMapLatest
- fold
- launchIn
- mapLatest
- mapNotNull
- onCompletion
- onEach
- onEmpty
- onStart
- retry
- retryWhen
- runningReduce
- sample
- scan
- single
- singleOrNull
- takeWhile
- toCollection
- toList
- transformLatest
- transformWhile
- withIndex
- zip
- filterNotNull
1. Map

suspend fun mapOperator(){
 (1..10).asFlow()
 .map {
 delay(500L)
 “Mapping $it”
 }
 .collect {
 println(it)
 }
}

2. Filter

suspend fun filterOperator(){
 (1..10).asFlow()
 .filter { 
    it%2 == 0
 }
 .collect {
     println(it)
  }
}

3. transform
- General transformation operator
- Can emit any value at any point

suspend fun transformOperator(){
 (1..10).asFlow()
 .transform {
 emit(“Emitting string value $it”)
 emit(it)
 }
 .collect {
 println(it)
 }
}

4. Take
- Use only a number of values, disregard the rest

(1..10).asFLow(
.take(2)
.colect{
 println(it)
}
```
- Above code only first 2 values

Terminal flow
- There are few operators to convert the flow into a collection
- collect
- toList
- toSet
- reduce
reduce

suspend fun reduceOperator(){
 val size = 10
 val factorial = (1..size).asFlow()
 .reduce { accumulator, value ->
 accumulator * value
 }
 println(“Factorial of $size is $factorial”)
}

flowOn
- Change the context on which the flow is emitted

buffer
- In case of processing a flow takes a long time
- a buffer is useful to accumulate flow values that can be processed later
```
fun generate(): Flow<Int> = flow{
for(i in 1..3){
delay(100)
emit(i)
}
}

val time = measureTimeMillis {
generate()
 .reduce()
 .collect {
delay(300)
println(it)
}
}
println(“Collected in $time ms”)
```
Composing Flow
- If there are multiple flows and you want to combine those flows, then we can use the below operators
1. zip
- It will compose the value emitted by 2 flows
```
suspend fun zipExample(){
val english = flowOf(“One”, “Two”, “Three”)
val hindi = flowOf(“ek”, “do”, “teen”)
english.zip(hindi) { a,b -> “‘$a in Hindi is $b’”}
 .collect {
println(it)
}

}
```
Combine
- Combine the latest value of one flow with the latest value of other

## EXception Handling
- SImply use try catch block to handle exception
```
try{
(1..3).asFlow()
 .onEach{ check(it != 2) }
 .collect{ println(it) }
}
catch(e:Exception){
println(“Caught exception”)
}
```
### .catch() method in flow
```
(1..3).asFlow()
 .onEach{ check(it != 2) }
 .catch{ e -> println(“Caught Exception $e”) }
 .collect{ println(it) }
```
### onCompletion block
- Its same as final block
```
(1..3).asFlow()
 .onEach{ check(it != 2) }
 .onCOmpletion{ cause ->
if(cause != null)
println(“Flow Completed with $cause”)
else
println(“Flow Completed Successfully”)
}
 .catch{ e -> println(“Caught Exception $e”) }
 .collect{ println(it) }
```

Android application using flow

— Coroutines flow
- Retrofit
- MVVM
- Click here for Coroutines flow with Retrofit and MVVM
- Use coroutines flow to update an interface
- Use coroutines flow with Retrofit & MVVM
- Show news items periodically in a list

Channels

- Channel Producer
- Pipelines
- Fan-out
- Fan-in
- Buffered Channels
- Ticker channels

Channels
- A channel is a queue of data
- A coroutine can asynchronously put elements .send(data)
- Another can blockingly get elements .receive()
- A channel is closed when there are no more elements .close()
```
val channels = Channel<Int>()
launch {
for(x in 1..5)
channels.send(x * x)
}
for(i in 1..5)
println(channel.receive())
```
- Another Way
```
val channels = Channel<Int>()
launch {
for(x in 1..5)
channels.send(x * x)
channels.close()
}
for(i in channels)
println(i)
```
## Channel Produceer
- Allows a data source to create and return a channel
- CoroutineScope.produce{…}
- Need a coroutine scope to run the code
```
val channel = produce {
for(x in 1..5){
send(
}
}
```
## Pipelines
- A pipeline is a development pattern
- where one channel output is given an input to another channel
- One coroutine is producing a(potentially infinite) set of values, one or more coroutines are consuming and transforming those values.
```
fun CoroutineScope.produceNumbers() = Produce<Int>{
var x = 1
while(true)
send(x++)
}

fun CoroutineScope.square(numbers: ReceiveChannel<Int>){
for(x in numbers){
send(x*x)
}
}

val numbers = produceNumbers()
val swuares = square(numbers)
for(i in 1..5){
println(squares.receive())
}
println(“Done”)
coroutineContext.cancelChildren()
```
## Fan-out
- If multiple coroutines receive from the same channel values(work) are distributed among them
```
fun CoroutineScope.produceNumbers() = produce<Int>{
var x = 1
while(true){
send(x++)
delay(100L)
}
}

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
for(msg in channel){
println(“processor #$id received $msg”)
}
}

val processor = produceNumbers()
repeat(5) { launchProcessor(it, producer) }
delay(500)
producer.cancel()
```
## fan-in
- It’s opposite of fan out
- Multiple coroutines send values to one channel
```
suspend fun sendString(channel: SendChannel<String>, s: String, time: Long){
while(true){
delay(time)
channel.send(s)
}
}

val channel = Channel<String>()
launch{ sendString(channel, “message1”, 2000L) }
launch{ sendString(channel, “message1”, 2000L) }
repeat(6) { println(channel.receive()) }
coroutineCOntext.cancelChildern()
```
## Buffered Channel
- A Buffered channel has a limited capacity
- when the capacity is reached, the sender is paused
- when capacity becomes available new values can be sent
```

```
## Ticker Channels
- Periodically produces a unit after a given delay.
- Has an Optional initial delay
```
val tickerChannel = ticker(delaysMillis = 100, initialDelaysMillis = 0)
launch {
val startTime = System.currentTimeMillis()
tickerChannel.consumeEach{
val delta = System.currentTimeMillis() — startTime
println(“Received tick after $delta”)
}
}
delay(1000)
println(“Done”)
tickerChannel.cancel()
```
## Coroutine Concurrency and shared state