Add flow playground files

Author: LukasLechnerDev

.idea/compiler.xml

@@ -0,0 +1,30 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+
+// Sequence for synchronously computed values
+// Flow for asynchronously computed values
+
+private fun simple(): Flow<Int> = flow { // flow builder
+    for (i in 1..3) {
+        delay(100) // pretend we are doing something useful here
+        emit(i) // emit next value
+    }
+}
+
+fun main() = runBlocking<Unit> {
+    // Launch a concurrent coroutine to check if the main thread is blocked
+    launch {
+        for (k in 1..3) {
+            println("I'm not blocked $k")
+            delay(100)
+        }
+    }
+    // Collect the flow
+    simple().collect { value -> println(value) }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/1-Simple-Flows.kt

@@ -0,0 +1,22 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow
+
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.runBlocking
+
+private fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
+
+// Context preservation
+// Collection of a flow always happens in the context of the calling coroutine.
+
+private fun simple(): Flow<Int> = flow {
+    log("Started simple flow")
+    for (i in 1..3) {
+        emit(i)
+    }
+}
+
+fun main() = runBlocking<Unit> {
+    simple().collect { value -> log("Collected $value") }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/10-Flow-Context.kt

@@ -0,0 +1,23 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.runBlocking
+
+// Flow always has to emit in the context in which it was collected
+
+private fun simple(): Flow<Int> = flow {
+    // The WRONG way to change context for CPU-consuming code in flow builder
+    kotlinx.coroutines.withContext(Dispatchers.Default) {
+        for (i in 1..3) {
+            Thread.sleep(100) // pretend we are computing it in CPU-consuming way
+            emit(i) // emit next value
+        }
+    }
+}
+
+fun main() = runBlocking<Unit> {
+    simple().collect { value -> println(value) }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/11-Wrong-emission-withContext.kt

@@ -0,0 +1,26 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow
+
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.runBlocking
+
+private fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
+
+private fun simple(): Flow<Int> = flow {
+    for (i in 1..3) {
+        Thread.sleep(100) // pretend we are computing it in CPU-consuming way
+        log("Emitting $i")
+        emit(i) // emit next value
+    }
+}.flowOn(Dispatchers.Default) // RIGHT way to change context for CPU-consuming code in flow builder
+
+// flowOn creates another coroutine for an upstream flow when it has to change the CoroutineDispatcher in its context
+
+fun main() = runBlocking<Unit> {
+    simple().collect { value ->
+        log("Collected $value")
+    }
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/12-FlowOn-operator.kt

@@ -0,0 +1,30 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.runBlocking
+import kotlin.system.measureTimeMillis
+
+private fun simple(): Flow<Int> = flow {
+    for (i in 1..3) {
+        delay(100) // pretend we are asynchronously waiting 100 ms
+        emit(i) // emit next value
+    }
+}
+
+fun main() = runBlocking<Unit> {
+    val time = measureTimeMillis {
+        simple()
+            .buffer() // without buffer, code takes 1235ms to complete, otherwise 100
+            .collect { value ->
+                delay(300) // pretend we are processing it for 300 ms
+                println(value)
+            }
+    }
+    println("Collected in $time ms")
+}
+
+// we have effectively created a processing pipeline

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/13-Buffering.kt

@@ -0,0 +1,31 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.conflate
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.runBlocking
+import kotlin.system.measureTimeMillis
+
+// Speed up processing when both the emitter and collector are slow
+// it drops emitted values
+
+private fun simple(): Flow<Int> = flow {
+    for (i in 1..3) {
+        delay(100) // pretend we are asynchronously waiting 100 ms
+        emit(i) // emit next value
+    }
+}
+
+fun main() = runBlocking<Unit> {
+    val time = measureTimeMillis {
+        simple()
+            .conflate() // conflate emissions, don't process each one
+            .collect { value ->
+                delay(300) // pretend we are processing it for 300 ms
+                println(value)
+            }
+    }
+    println("Collected in $time ms")
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/14-Conflation.kt

@@ -0,0 +1,27 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.collectLatest
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.runBlocking
+import kotlin.system.measureTimeMillis
+
+private fun simple(): Flow<Int> = flow {
+    for (i in 1..3) {
+        delay(100) // pretend we are asynchronously waiting 100 ms
+        emit(i) // emit next value
+    }
+}
+
+fun main() = runBlocking<Unit> {
+    val time = measureTimeMillis {
+        simple()
+            .collectLatest { value -> // cancel & restart on the latest value
+                println("Collecting $value")
+                delay(300) // pretend we are processing it for 300 ms
+                println("Done $value")
+            }
+    }
+    println("Collected in $time ms")
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/15-Processing-the-latest-value.kt

@@ -0,0 +1,15 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow
+
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.zip
+import kotlinx.coroutines.runBlocking
+
+fun main() = runBlocking<Unit> {
+
+    val numbers = (1..3).asFlow()
+    val strings = flowOf("one", "two", "three")
+    numbers.zip(strings) { a, b -> "$a -> $b" } // compose a single string
+        .collect { println(it) } // collect and print
+}

app/src/main/java/com/lukaslechner/coroutineusecasesonandroid/playground/flow/16-Composing-multiple-flows-zip.kt

@@ -0,0 +1,20 @@
+package com.lukaslechner.coroutineusecasesonandroid.playground.flow
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.runBlocking
+
+fun main() = runBlocking<Unit> {
+
+    val numbers = (1..3).asFlow()
+        .onEach { delay(300) }
+
+    val strings = flowOf("one", "two", "three")
+        .onEach { delay(400) }
+
+    val startTime = System.currentTimeMillis() // remember the start time
+    numbers.combine(strings) { a, b -> "$a -> $b" } // compose a single string with "zip"
+        .collect { value -> // collect and print
+            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
+        }
+}