Flow in Kotlin and how to use it in Android Project?

Published in

ISOP Nepal

5 min readJun 20, 2020

With Kotlin Coroutine 1.2.0 alpha release Jetbrains came up with Flow API as part of it. With Flow in Kotlin now you can handle a stream of data that emits values sequentially.

In Kotlin, Coroutine is just the scheduler part of RxJava but now with Flow APIs, it can be alternative to RxJava in Android

What is Flow APIs in Kotlin Coroutines?

Flow API in Kotlin is another better way to handle the stream of data asynchronously that executes sequentially.

So, in RxJava, Observables type is an example of a structure that represents a stream of items. Its body does not get executed until it is subscribed to by a subscriber. and once it is subscribed, subscriber starts getting the data items emitted. Similarly, Flow works on the same condition where the code inside a flow builder does not run until the flow is collected.

Let us create an android project and then let’s start integrating the Kotlin Flow APIs.

Step 01.

Add the following in the app’s build.gradle,

implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.3"
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.3"

and in the project’s build.Gradle adds,

classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:1.3.61"

Step 02.

In FlowConceptActivity’s layout file let’s create a UI that will have a button.

<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout 
    xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".FlowConceptActivity">    <Button
        android:id="@+id/button"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        app:layout_constraintBottom_toBottomOf="parent"
        android:text="Learn Kotlin Flow"
        app:layout_constraintLeft_toLeftOf="parent"
        app:layout_constraintRight_toRightOf="parent"
        app:layout_constraintTop_toTopOf="parent" /></androidx.constraintlayout.widget.ConstraintLayout>

Step 03.

Now, let’s begin the implementation of Flow APIs in FlowConceptActivity. In onCreate() function of Activity lets add two function like,

override fun onCreate(savedInstanceState: Bundle?) {
    super.onCreate(savedInstanceState)
    setContentView(R.layout.activity_main)
    startFlow()
    buttonClicks()
}

Here, startFlow() is the function where we will define the flow and buttonClicks() is the function where we will click the button to display the data which is emitted from the flow.

We will declare a lateinit variable of Flow of Int type,

lateinit var flow: Flow<Int>

Step 04.

Now, in startFlow() it will emit value after 500milliseconds delay.

private fun startFlow() {
    flow = flow {
        Log.d(TAG, "flow Start")
        (0..10).forEach {
            // Emit items with 500 milliseconds delay
            delay(500)
            Log.d(TAG, "Emitting value $it")
            emit(it)

        }
    }.flowOn(Dispatchers.Default)
}

Here,

We will emit value from 0 to 10 at 500ms delay.
To emit the value we will use emit() which collects the value emitted. It is part of FlowCollector which can be used as a receiver.
and, at last, we use flowOn operator which means that shall be used to change the context of the flow emission. Here, we can use different Dispatchers like IO, Default, etc.
flowOn() is similar to subscribeOn() in RxJava

Step 05.

Now, we need to write buttonClicks() function where we need to print the values which we will emit from the flow.

private fun buttonClicks() {
    button.setOnClickListener {
        CoroutineScope(Dispatchers.Main).launch {
            flow.collect {
                Log.d(TAG, it.toString())
            }
        }
    }
}

When we click the button we will print the values one by one.

flow.collect now will start extracting/collecting the value from the flow on the Main thread as Dispatchers.Main is used in launch coroutine builder in CoroutineScope

And the output which will be printed in Logcat is,

D/FlowConceptActivity: flow Start
D/FlowConceptActivity: Emitting value 0
D/FlowConceptActivity: 0
D/FlowConceptActivity: Emitting value 1
D/FlowConceptActivity: 1
D/FlowConceptActivity: Emitting value 2
D/FlowConceptActivity: 2
D/FlowConceptActivity: Emitting value 3
D/FlowConceptActivity: 3
D/FlowConceptActivity: Emitting value 4
D/FlowConceptActivity: 4
D/FlowConceptActivity: Emitting value 5
D/FlowConceptActivity: 5
D/FlowConceptActivity: Emitting value 6
D/FlowConceptActivity: 6
D/FlowConceptActivity: Emitting value 7
D/FlowConceptActivity: 7
D/FlowConceptActivity: Emitting value 8
D/FlowConceptActivity: 8
D/FlowConceptActivity: Emitting value 9
D/FlowConceptActivity: 9
D/FlowConceptActivity: Emitting value 10
D/FlowConceptActivity: 10

As you can see Flow starts only when the button is clicked as it prints Start Flow and then starts emitting. This is what I meant by Flows are cold.

Let’s say we update the startFlow() function like,

private fun startFlow() {
    flow = flow {
        Log.d(TAG, "flow Start")
        (0..10).forEach {
            // Emit items with 500 milliseconds delay
            delay(500)
            Log.d(TAG, "Emitting value $it")
            emit(it)
        }
    }.map {
        it * it
    }.flowOn(Dispatchers.Default)
}

Here you can see we added map operator which will take each and every item will square itself and print the value.

Anything, written above flowOn will run in the background thread.

Builders in Flow

Flow builders are nothing but the way to build Flows. There are 4 types of flow builders,

flowOf() — It is used to create flow from a given set of values. For Eg.

flowOf(4, 2, 5, 1, 7).onEach { delay(400) }.flowOn(Dispatcher.Default)

Here, flowOf() takes fixed values and prints each of them after a delay of 400ms. When we attach a collector to the flow, we get the output

D/FlowConceptActivity: 4
D/FlowConceptActivity: 2
D/FlowConceptActivity: 5
D/FlowConceptActivity: 1
D/FlowConceptActivity: 7

asFlow() — It is an extension function that helps to convert type into flows. For Eg,

(1..5).asFlow().onEach{ delay(300)}.flowOn(Dispatchers.Default)

Here, we converted a range of values from 1 to 5 as flow and emitted each of them at a delay of 300ms. When we attach a collector to the flow, we get the output,

D/FlowConceptActivity: 1
D/FlowConceptActivity: 2
D/FlowConceptActivity: 3
D/FlowConceptActivity: 4
D/FlowConceptActivity: 5

The output is the number being printed from 1 to 5, whichever were present in the range.

flow{} — This example has been explained in the Android Example above. This is a builder function to construct arbitrary flows.
channelFlow{} — This builder creates cold-flow with the elements using send provided by the builder itself. For Example,

channelFlow {
    (0..10).forEach {
        send(it)
    }
}.flowOn(Dispatchers.Default)

This will print,

D/FlowConceptActivity: 0
D/FlowConceptActivity: 1
D/FlowConceptActivity: 2
D/FlowConceptActivity: 3
D/FlowConceptActivity: 4
D/FlowConceptActivity: 5
D/FlowConceptActivity: 6
D/FlowConceptActivity: 7
D/FlowConceptActivity: 8
D/FlowConceptActivity: 9
D/FlowConceptActivity: 10

flowOn() is similar to subscribeOn() in RxJava
Now, finally, let’s discuss how to use an operator of Flow in your project

Zip Operator

If you remember from above Android Example, we have two functions startFlow() and buttonClicks(). We will modify both functions in MainActivity.

First, we will declare two lateinit variables of Flow of String type,

lateinit var flowOne: Flow<String>
lateinit var flowTwo: Flow<String>

Now, the two functions startFlow() and buttonClicks() will be called in onCreate() of MainActivity.

override fun onCreate(savedInstanceState: Bundle?) {
    ...
    startFlow()
    buttonClicks()}

and in the startFlow() we will initialize two flows to the two variables,

private fun setupFlow() {
    flowOne = flowOf("Raju", "Manish", "Milan").flowOn(Dispatchers.Default)
    flowTwo = flowOf("Yadav", "Chaudhary", "Rai").flowOn(Dispatchers.Default)}

and in buttonClicks() we will zip both the flows using zip operator,

private fun buttonClicks() {
    button.setOnClickListener {
        CoroutineScope(Dispatchers.Main).launch {
            flowOne.zip(flowTwo)
            { firstName, lastName->
                "$firstName $lastName"
            }.collect {
                Log.d(TAG, it)
            }
        }
    }
}