Start with the kotlin flow

Start with the kotlin flow Abhishesh Srivastava abhishesh_sri

• Threads • Coroutines

fun doLongRunningWork( callback: (x: Int) -> Unit) { thread {
val result = compute() callback(result) } }

suspend fun doLongRunningWork : Int() { val result = compute()
return result } suspend fun compute() : Int { ... }

fun doLongRunningWork( callback: (x: Int) -> Unit) { thread {
val result = compute() callback(result) } } suspend fun doLongRunningWork : Int() { val result = compute() return result } suspend fun compute() : Int { ... }

doLongRunningWork { doLongRunningWork { ... ... } } couroutineScope.launch {
val first = doLongRunningWork() val second = doLongRunningWork() ... ... }

doLongRunningWork { doLongRunningWork { doLongRunningWork { ... ... } }
} couroutineScope.launch { val first = doLongRunningWork() val second = doLongRunningWork() val third = doLongRunningWork() ... ... }

Repository updates UI suspend fun updates() : ??? { ...
}

• Asynchronous data stream • Built on top of coroutine
• Emits value • Completes normally OR with exception • Usually flows are cold

• Emits value • Completes normally OR with exception • Usually flows are cold flow<Int> { for (i in 1..10) { emit(i) } }

• Emits value • Completes normally OR with exception • Usually flows are cold flow<Int> { for (i in 1..10) { println(“emit: $i”) emit(i) } }

• Emits value • Completes normally OR with exception • Usually flows are cold flow<Int> { for (i in 1..10) { println(“emit: $i”) emit(i) } }.collect { println(“col: $it”) println(it) }

public interface Flow<out T> { public suspend fun collect(collector: FlowCollector<T>)
} public interface FlowCollector<in T> { public suspend fun emit(value: T) }

flow { emit(value) } flowOf(1, 2, 3) listOf<Int>(1, 2, 3,
4).asFlow() channelFlow<Int> { ... }

• map • filter • take • zip • buffer
• conflate • … etc

CoroutineScope(Dispatchers.Main).launch { flow { for (i in 1..3) { println("emit:
$i") emit(i) } }.map { println("map: $it") it * 2 }.collect { println("collect: $it") } }

$i") emit(i) } }.map { println("map: $it") it * 2 }.collect { println("collect: $it") } } emit: 1 map: 1 collect: 2 emit: 2 map: 2 collect: 4 emit: 3 map: 3 collect: 6

$i") emit(i) } }.filter { println("filter: $it") it % 2 == 0 }.collect { println("collect: $it") } }

$i") emit(i) } }.filter { println("filter: $it") it % 2 == 0 }.collect { println("collect: $it") } } emit: 1 filter: 1 emit: 2 filter: 2 collect: 2 emit: 3 filter: 3

CoroutineScope(Dispatchers.Main).launch { flow { for (i in 1..3) { delay(100)
emit(i) } }.collect { delay(200) println("collect: $it") } } Output 1 2 3 Time taken : 937ms

emit(i) } }.buffer(). collect { delay(200) println("collect: $it") } } Output 1 2 3 Time taken : 745ms

emit(i) } }.conflate() .collect { delay(300) println("collect: $it") } } Output 1 3

• collect • single • reduce • toList • launchIn
• … etc

• Context preservation • Exception transparency

CoroutineScope(Dispatchers.Main).launch { flow<Int> { withContext(Dispatchers.IO) { for (i in 1..10)
{ emit(i) } } }.collect { println(it) } }

CoroutineScope(Dispatchers.Main).launch { flow<Int> { withContext(Dispatchers.IO) { for (i in 1..10)
{ emit(i) } } }.collect { println(it) } } java.lang.IllegalStateException: Flow invariant is violated:

CoroutineScope(Dispatchers.Main).launch { flow<Int> { for (i in 1..10) { emit(i)
} }.flowOn(Dispatchers.IO) .collect { println(it) } } Changes the context of upstream flow

uiScope.launch { try { dataFlow().collect { data -> updateUI(data) }
} catch (e: Exception) { } }

uiScope.launch { dataFlow() .catch{ e -> println(“caught $e“} .collect {
value -> println(value) } } a downstream exception must always be propagated to the collector.

Retrofit supports suspend function Integrated into many jetpack libraries Other
reactive libraries can be easily migrated to Flow

Integrated with room @Dao abstract class Dao { @Query("SELECT *
FROM table") abstract fun getExamples(): Flow<List<Model>> }

Integrated with live data implementation "androidx.lifecycle:lifecycle-livedata-ktx:$lifecycle_version" fun <T> LiveData<T>.asFlow(): Flow<T>
{} fun <T> Flow<T>.asLiveData( context: CoroutineContext = EmptyCoroutineContext, timeout: Duration ): LiveData<T> { }

• Producer • consumer

class FakeFlowProducer : FlowProducer { fun data() = flow {
emit(ITEM_1) } fun allData() = listOf(ALL_VALUES).asFlow() } @Test fun fakeFlowProducerTest() = runBlocking { val producer = FakeFlowProducer() val firstItem = producer.data.first() assertThat(firstItem, isEqualTo(ITEM_1) }

class FakeFlowProducer : FlowProducer { fun data() = flow {
emit(ITEM_1) } fun allData() = listOf(ALL_VALUES).asFlow() } @Test fun fakeFlowProducerTest() = runBlocking { val producer = FakeFlowProducer() val firstItem = producer.data.first() assertThat(firstItem, isEqualTo(ITEM_1) } Creates a testcoroutinescope which have aa testcoroutinedispatcher & testcoroutineexceptionhandler. Also accepts and lambda test body.

@Test fun testMultipleValues() = runBlocking { val producer = FakeFlowProducer()
val multipleValues = producer.multipleValues().toList() assertThat(multipleValues, isEqualTo(ALL_VALUES)) }

producer.allData.drop(1).first() producer.allData.take(5).toList() producer.allData.take(5).toSet() producer.allData.takeWhile(predicate).take(2).toList() producer.allData.firstWhile(predicate) producer.allData.map(transformation).take(5) producer.allData.single() producer.allData.count() producer.allData.count(predicate)

Start with the kotlin flow

Start with the kotlin flow

More Decks by Abhishesh

Other Decks in Programming

Featured

Transcript