Merge branch 'master' into coroutines_course_empty_exercises

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/MainActivity.kt

@@ -50,5 +50,5 @@ class MainActivity : BaseActivity() {
         return itemDecorator
     }
 
-    override fun getToolbarTitle() = "Coroutine Usecases on Android"
+    override fun getToolbarTitle() = "Coroutines and Flows on Android"
 }

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/channels/3_drop_while_busy.kt

@@ -0,0 +1,66 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.channels
+
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.consumeEach
+import kotlinx.coroutines.channels.produce
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.launch
+
+/**
+ UseCase: We want to trigger some processing. While the downstream is currently
+ busy, we want to drop the current trigger emission.
+
+ With a SharedFlow, this is not possible, since we need to define a buffer size of
+ > 0 for buffer strategies like "DROP_LATEST". Channels however have a buffer sice of 0
+ by default.
+
+ Another option is to use the custom operator "dropIfBusy" (see below)
+
+ See also: https://stackoverflow.com/questions/64844821/how-to-drop-latest-with-coroutine-flowt/74560222#74560222
+**/
+
+fun <T> Flow<T>.dropIfBusy(): Flow<T> = flow {
+    coroutineScope {
+        val channel = produce {
+            collect { trySend(it) }
+        }
+        channel.consumeEach { emit(it) }
+    }
+}
+
+suspend fun main(): Unit = coroutineScope {
+
+    val channel = Channel<Int>()
+
+    launch {
+        channel
+            .consumeAsFlow()
+            .collect {
+                println("Process $it")
+                delay(1000)
+                println("$it processed")
+            }
+    }
+
+    launch {
+
+        delay(100)
+
+        // 1 should be processed
+        channel.trySend(1)
+        println("sharedFlow emits 1")
+
+        // 2 should not be processed since downstream is busy
+        channel.trySend(2)
+        println("sharedFlow emits 2")
+
+        // 3 should be processed again
+        delay(2000)
+        channel.trySend(3)
+        println("sharedFlow emits 3")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/10_buffer_in_stateflow.kt

@@ -0,0 +1,38 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.launch
+import kotlin.system.measureTimeMillis
+
+suspend fun main(): Unit = coroutineScope {
+
+    val flow = MutableStateFlow(0)
+
+    // Collector 1
+    launch {
+        flow.collect {
+            println("Collector 1 processes $it")
+        }
+    }
+
+    // Collector 2
+    launch {
+        flow.collect {
+            println("Collector 2 processes $it")
+            delay(100)
+        }
+    }
+
+    // Emitter
+    launch {
+        val timeToEmit = measureTimeMillis {
+            repeat(5) {
+                flow.emit(it)
+                delay(10)
+            }
+        }
+        println("Time to emit all values: $timeToEmit ms")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/1_buffer.kt

@@ -0,0 +1,24 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.flow
+
+suspend fun main() = coroutineScope {
+
+    val flow = flow {
+        repeat(5) {
+            println("Emitter:    Start Cooking Pancake $it")
+            delay(100)
+            println("Emitter:    Pancake $it ready!")
+            emit(it)
+        }
+    }.buffer()
+
+    flow.collect {
+        println("Collector:  Start eating pancake $it")
+        delay(300)
+        println("Collector:  Finished eating pancake $it")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/2_buffer_overflow_suspend.kt

@@ -0,0 +1,26 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.flow
+
+suspend fun main() = coroutineScope {
+
+    val flow = flow {
+        repeat(5) {
+            val pancakeIndex = it + 1
+            println("Emitter:    Start Cooking Pancake $pancakeIndex")
+            delay(100)
+            println("Emitter:    Pancake $pancakeIndex ready!")
+            emit(pancakeIndex)
+        }
+    }.buffer(capacity = 1, onBufferOverflow = BufferOverflow.SUSPEND)
+
+    flow.collect {
+        println("Collector:  Start eating pancake $it")
+        delay(300)
+        println("Collector:  Finished eating pancake $it")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/3_buffer_overflow_drop_oldest.kt

@@ -0,0 +1,26 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.flow
+
+suspend fun main() = coroutineScope {
+
+    val flow = flow {
+        repeat(5) {
+            val pancakeIndex = it + 1
+            println("Emitter:    Start Cooking Pancake $pancakeIndex")
+            delay(100)
+            println("Emitter:    Pancake $pancakeIndex ready!")
+            emit(pancakeIndex)
+        }
+    }.buffer(capacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
+
+    flow.collect {
+        println("Collector:  Start eating pancake $it")
+        delay(300)
+        println("Collector:  Finished eating pancake $it")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/4_buffer_overflow_drop_latest.kt

@@ -0,0 +1,26 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.flow
+
+suspend fun main() = coroutineScope {
+
+    val flow = flow {
+        repeat(5) {
+            val pancakeIndex = it + 1
+            println("Emitter:    Start Cooking Pancake $pancakeIndex")
+            delay(100)
+            println("Emitter:    Pancake $pancakeIndex ready!")
+            emit(pancakeIndex)
+        }
+    }.buffer(capacity = 1, onBufferOverflow = BufferOverflow.DROP_LATEST)
+
+    flow.collect {
+        println("Collector:  Start eating pancake $it")
+        delay(300)
+        println("Collector:  Finished eating pancake $it")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/5_buffer_unlimited_size.kt

@@ -0,0 +1,26 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.flow
+
+suspend fun main() = coroutineScope {
+
+    val flow = flow {
+        repeat(5) {
+            val pancakeIndex = it + 1
+            println("Emitter:    Start Cooking Pancake $pancakeIndex")
+            delay(100)
+            println("Emitter:    Pancake $pancakeIndex ready!")
+            emit(pancakeIndex)
+        }
+    }.buffer(capacity = UNLIMITED)
+
+    flow.collect {
+        println("Collector:  Start eating pancake $it")
+        delay(300)
+        println("Collector:  Finished eating pancake $it")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/6_collect_latest.kt

@@ -0,0 +1,25 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.collectLatest
+import kotlinx.coroutines.flow.flow
+
+suspend fun main() = coroutineScope {
+
+    val flow = flow {
+        repeat(5) {
+            val pancakeIndex = it + 1
+            println("Emitter:    Start Cooking Pancake $pancakeIndex")
+            delay(100)
+            println("Emitter:    Pancake $pancakeIndex ready!")
+            emit(pancakeIndex)
+        }
+    }
+
+    flow.collectLatest {
+        println("Collector:  Start eating pancake $it")
+        delay(300)
+        println("Collector:  Finished eating pancake $it")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/7_map_latest.kt

@@ -0,0 +1,29 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.mapLatest
+
+suspend fun main() = coroutineScope {
+
+    val flow = flow {
+        repeat(5) {
+            val pancakeIndex = it + 1
+            println("Emitter:    Start Cooking Pancake $pancakeIndex")
+            delay(100)
+            println("Emitter:    Pancake $pancakeIndex ready!")
+            emit(pancakeIndex)
+        }
+    }.mapLatest {
+        println("Add topping onto the pancake $it")
+        delay(200)
+        it
+    }
+
+    flow.collect {
+        println("Collector:  Start eating pancake $it")
+        delay(300)
+        println("Collector:  Finished eating pancake $it")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/8_conflate.kt

@@ -0,0 +1,24 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.conflate
+import kotlinx.coroutines.flow.flow
+
+suspend fun main() = coroutineScope {
+
+    val flow = flow {
+        repeat(5) {
+            println("Emitter:    Start Cooking Pancake $it")
+            delay(100)
+            println("Emitter:    Pancake $it ready!")
+            emit(it)
+        }
+    }.conflate()
+
+    flow.collect {
+        println("Collector:  Start eating pancake $it")
+        delay(300)
+        println("Collector:  Finished eating pancake $it")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/concurrency/9_buffer_in_sharedflow.kt

@@ -0,0 +1,38 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow.concurrency
+
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.launch
+import kotlin.system.measureTimeMillis
+
+suspend fun main(): Unit = coroutineScope {
+
+    val flow = MutableSharedFlow<Int>(extraBufferCapacity = 10)
+
+    // Collector 1
+    launch {
+        flow.collect {
+            println("Collector 1 processes $it")
+        }
+    }
+
+    // Collector 2
+    launch {
+        flow.collect {
+            println("Collector 2 processes $it")
+            delay(100)
+        }
+    }
+
+    // Emitter
+    launch {
+        val timeToEmit = measureTimeMillis {
+            repeat(5) {
+                flow.emit(it)
+                delay(10)
+            }
+        }
+        println("Time to emit all values: $timeToEmit ms")
+    }
+}

app/src/main/res/values/strings.xml

@@ -1,5 +1,5 @@
 <resources>
-    <string name="app_name">Coroutine Usecases on Android</string>
+    <string name="app_name">Coroutines and Flows on Android</string>
     <string name="perform_single_network_request_on_background_thread">Load recent Android Versions</string>
     <string name="perform_2_network_requests_sequentially">Load Features of most recent Android Version</string>
     <string name="load_features_of_android_versions_sequentially">Load Features of Android Versions sequentially</string>