Kotlin Flow in Android

https://thoughtsoncloud.com/kotlin-flow-in-android/

Kotlin Flow in Android

A Kotlin Flow is a crucial constituent of this ecosystem. It resembles reactive streams and defines available methods and functions for producing, manipulating, and processing non-synchronous information streams.

Kotlin Flow in Android

by

Viktoriia Domukhovska

July 4, 2023

A Kotlin Flow is a crucial constituent of this ecosystem. It resembles reactive streams and defines available methods and functions for producing, manipulating, and processing non-synchronous information streams. Kotlin Flow enables you to build multiple data over a period, akin to an information thread, and proffers various operators to edit, filter and unite these values.

For instance:

fun simpleFlow(): Flow < String > = flow { // flow builder
    for (char in 'A'..
        'C') {
        delay(100) // pretend we are doing something useful here
        emit(char.toString()) // emit the next value
    }
}

This code generates an elementary Kotlin Flow that sequentially produces three letters. Every letter makes an inevitable delay, simulating some helpful work. For example, after executing the simpleFlow(), starting with the letter ‘A’ the program will emit every subsequent letter with a 100-millisecond interval. The outcome is a Flow that contains the letters ‘A’, ‘B’ and ‘C’.

Flow builder

A flow builder is an interface in Kotlin that offers multiple methods to construct flows. Below are several standard methods utilized to build flows:

•  flowOf: His function produces a flow from a predefined set of values. It allows you to define a fixed set of values emitted sequentially as flow elements.
•  asFlow: It converts other types of collections or succession into flows. For instance, we can transform a list or an array into a thread utilizing the asFlow. It enables working with existing collections or sequences as data streams.
• flow: This builder function allows us to specify a thread by employing a suspending lambda expression. We can emit values utilizing the emit function and execute asynchronous procedures in the lambda.
• callbackFlow: It transforms APIs based on event handlers into flows, allowing us to receive values asynchronously from callback functions and emit them as flow items.
• channelFlow: This procedure generates a flow utilizing a channel, a more versatile approach to managing communication between coroutines. Values are transmitted into the channel and received as flow elements.

Of course, the approach choice depends on the particular prerequisites and nature of information during operation.

There are the samples of the thread builder methods, along with their explanations:

flowOf

fun flowOfExample(): Flow < String > = flowOf("unit1", "unit2", "unit3", "unit4", "unit5")


suspend fun main() {
    flowOfExample().collect {
        println(it)
    }
}

This code showcases applying of the flowOf method to construct a Kotlin Flow that sequentially produces the values “unit1”, “unit2”, “unit3”, “unit4”, and “unit5”. The flowOf method accepts multiple values as arguments and constructs a stream that creates these data in a particular order. In this sample, the thread generates strings representing different units.

Next, we utilize the collect function to employ and execute every thread-generated data, printing them to the console.

asFlow

fun asFlowExample(): Flow < String > {
    val laptopBrands = listOf("Apple", "Dell", "HP")
    return laptopBrands.asFlow()
}

suspend fun main() {
    asFlowExample().collect {
        println(it)
    }
}

Program employing the asFlow function to edit a collection (an array of laptop brands) into a thread that generates these values sequentially. In this particular situation, the list of laptop brands “Apple”, “Dell” and “HP” is converted into a flow, which print employing the collect operator.

flow

fun flowExample(): Flow < String > = flow {
    for (i in 1. .5) {
        delay(1000)
        emit("unit$i")
    }
}


suspend fun main() {
    flowExample().collect {
        println(it)
    }
}

This sample illustrates applying the flow builder to construct a custom thread. In the beginning, the stream issues strings in the format “unit1”, “unit2”, and so on, up to “unit5”. In the following, the program separates every emission with a wait of 1 sec, simulated by the delay function. Afterward, by invoking the collect function in the flow, the generated values are employed and displayed to the display, showcasing the flow’s asynchronous nature, where each emission is processed independently with the specified delay in between.

callbackFlow

import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.callbackFlow
import kotlinx.coroutines.flow.collect


// Sample implementation of the SomeListener interface
interface SomeListener {
    fun onDataReceived(data: Int)
}


// Sample implementation of the registerListener function
fun registerListener(listener: SomeListener) {
    // Register the listener
}


// Sample implementation of the unregisterListener function
fun unregisterListener(listener: SomeListener) {
    // Unregister the listener
}


fun callbackFlowExample(): Flow < Int > = callbackFlow {
    val listener = object: SomeListener {
        override fun onDataReceived(data: Int) {
            offer(data)
        }
    }
    registerListener(listener)


    awaitClose {
        unregisterListener(listener)
    }
}


suspend fun main() {
    callbackFlowExample().collect {
        println(it)
    }
}

The callbackFlow function allows us to modify callback-oriented APIs into a thread. In this code, we develop a callbackFlow that observes events from SomeListener’s onDataReceived callback. Every obtained data furnishes to the thread utilizing the offer function. Upon terminating the stream, we unregister the listener by employing awaitClose.

channelFlow

fun channelFlowExample(): Flow < String > = channelFlow {
    send("Good")
    delay(1000)
    send("Day")
    close()
}


suspend fun main() {
    channelFlowExample().collect {
        value - >
            println("Have a $value!")
    }
}

The channelFlowExample function utilizes the channelFlow builder to develop a Flow. Within the channelFlow block, we employ the transmit function to release the value “Good” and “Day” into the thread. In the following, we introduce a one-second wait between the two releases by incorporating the delay function. In the end, we denote the conclusion of releases by invoking the close function to terminate the channel and indicating that the program will no longer produce any additional data.

In the primary function, we invoke the collect handler on the thread to get and operate the generated data. Afterward, we display the notice “Have a !” to the show for every value. This sample demonstrates how channelFlow can generate a custom flow built on a Channel and generates data with determined pauses.

This is a short video overview of Kotlin to understand the topic better.

Utilizing Kotlin thread in Android application programming

Kotlin Flow is employed in Android programming to administer non-synchronous procedures and administer information flow productively. Here are several methods for employing Kotlin flow:

1. Employing Flow in ViewModel: We can make a stream in ViewModel that releases information from a source, such as information storage or a network call. In the ViewModel, you can process this information utilizing various stream functions such as map, filter, transform, etc. Afterward, the information can be processed from your activity or fragment using collect, which subscribes to the thread and handles the obtained data.
2. Merging thread with LiveData: Kotlin flows integrates with the Android Architecture Component LiveData. You can convert a thread to LiveData using the asLiveData(), which allows thread utilization with other Android architecture elements such as Data Binding or Lifecycle.
3. Utilizing stream with Kotlin Coroutines: Kotlin thread is closely related to Coroutines. We can initiate a thread within a coroutine by utilizing the flow{…} handlers. Employing break procedures and various stream functions, we can proficiently process non-synchronous procedures and process results within the coroutine realization framework.
4. Error handling in thread: Kotlin thread can process errors and exceptions within the information stream. We can utilize the catch function to catch and process errors within the information flow, facilitating elegant management of situations when something goes wrong during the realization of asynchronous procedures.

In coroutines, a Kotlin flow is a unique construction that allows for the sequential emission of multiple values, unlike break procedures that yield one value. However, it shares similarities with an Iterator in that it generates a series of values but leverages break operators to generate and utilize data non-synchronously, allowing a thread to enable a flow to safely execute tasks such as forming network requests to develop the following value non-blocking the primary stream.

The information flow consists of 3 primary constituents:

• Producer: He develops and appends information to the thread. With the assistance of coroutines, streams can create information non-synchronously, enabling efficient management of asynchronous operations.
• Intermediaries (optional): They can modify every generated value within the stream or the flow. They allow for transformations, filtering, or other processes on the information while they process within the stream.
• Consumer: He is the entity that consumes or obtains the values the stream produces. It can manage, display, or manipulate information according to its needs.

Together, these entities generate a cohesive information stream where the producer develops information, intermediaries can change the information or thread, and the consumer retrieves the data from the stream for subsequent processing or display.

Creation Kotlin Flow for Android

Consider the example of a Kotlin Flow, where an information source periodically gets the last products. As break operators cannot give back several consistent data, the information source develops and handles a thread to do this. In this sample, the information source is the producer, generating and issuing values within the thread.

class ProductRemoteDataSource(
    private val productApi: ProductApi,
    private val refreshIntervalMs: Long = 5000
) {
    val latestProducts: Flow < List < Product >> = flow {
        while (true) {
            val latestProducts = productApi.fetchLatestProducts()
            emit(latestProducts) // Emits the result of the request to the flow
            delay(refreshIntervalMs) // Suspends the coroutine for some time
        }
    }
}


// Interface that provides a way to make network requests with suspend functions
interface ProductApi {
    suspend fun fetchLatestProducts(): List < Product >
}

In this sample, a ProductRemoteDataSource utilizes a ProductApi to fetch a list of products from a network information source. The fetchLatestProducts() method requests a network to retrieve the newest products. We use thread to construct a flow that produces the updated product list.

The flow builder is accomplished within a coroutine, allowing it to advantage of the equable non-synchronously APIs. Nevertheless, there are specific limitations to consider:

• Flows are consistent: When using the flow builder within a coroutine, the producer suspends execution until the downstream operators (such as break operators) return the result. This process ensures that the flow remains consistent and operates sequentially. For example, when obtaining the last products, the producer suspends until the fetchLatestProducts data retrieval completes before emitting the result to the flow.
• Emitting data in the same CoroutineContext: When utilizing the thread builder, the producer should not produce data from a distinct CoroutineContext. It’s essential to avoid calling handle within another CoroutineContext by building objects or appliances withContext parts of the program. In such cases, alternative flow builders like callbackFlow utilizing.

These considerations ensure the sequential nature of flows and maintain the integrity of their execution within coroutines. By adhering to these guidelines, we can create robust flows that operate non-synchronous procedures optimally and deliver information streamlined.

Edit the information thread

Intermediates can utilize auxiliary functions to alter the information thread without releasing the values. These handlers are functions that develop a succession of procedures when we apply them to an information thread. The program defers and implements these tasks when the data are at a later stage.

In the next program, the information warehouse layer shows the utilization of the transitional map function to edit the information before it demonstrates on the view: 

class ProductRepository(
    private val productRemoteDataSource: ProductRemoteDataSource,
    private val userData: UserData
) {
    /**
     * Returns the favorite latest products applying transformations on the flow.
     * These operations are lazy and don't trigger the flow. Instead, they transform
     * the current value emitted by the flow now.
     */
    val favoriteLatestProducts: Flow < List < Product >> =
        productRemoteDataSource.latestProducts
        // Intermediate operation to filter the list of favorite products
        .map {
            products - > products.filter {
                userData.isFavoriteProduct(it)
            }
        }
        // Intermediate operation to save the latest products in the cache
        .onEach {
            products - > saveInCache(products)
        }
}

A succession of transitional functions can use, one after another, to develop a sequence of processes that perform lazily when an object creates to the thread. However, it’s essential to note that merely implementing a transitional function to a flow does not initiate the stream collection.

Remember that these intermediate operators execute lazily, meaning their execution is triggered when a piece of information generates in the thread. Therefore, implementing mediating function enables us to modify and filter the information flow according to your requirements, enabling flexible information management and manipulation.

Gathering from a flow

We can employ a terminal function operator to initiate the thread and receive values: collect, toList, toSet, single, singleOrNull, first, firstOrNull, last, lastOrNull, count, reduce, fold, launchIn, collectLatest. One such operator is collect, which permits gathering every data from the thread as they handle.

The collect function in Kotlin Flow is a terminal operator employed to utilize and handle the output values of the flow. It takes a lambda function as a parameter, invoked for every subsequent value produced by the flow. Nevertheless, it is not a “break process” because it pauses the coroutine invoking collect up to the flow is a break.

Expanding on the last case, here’s a fundamental realization of a ViewModel that absorbs information from the warehouse layer: 

class ProductViewModel(
    private val productRepository: ProductRepository
): ViewModel() {


    init {
        viewModelScope.launch {
            // Trigger the flow and consume its elements using collect
            productRepository.favoriteLatestProducts.collect {
                favoriteProducts - >
                    // Update the view with the latest favorite products
            }
        }
    }
}

In this sample, the ProductViewModel class is accountable for ingesting the information from the favoriteLatestProducts thread in the ProductRepository. The collect operator listens for the most recent values the thread releases and accomplishes the necessary functions to refresh the view.

Accumulating the thread begins the process of ingesting information from the producer, who is accountable for refreshing the most recent products and operating the results fetching at a fixed interruption. 

The program can terminate flow collection for several causes:

• As illustrated in the last sample, when the coroutine responsible for gathering the flow is aborting, it cancels the fundamental producer. In other words, if the coroutine is terminated or interrupted, it causes the producer to stop generating new data for the thread.
• The information flow ends when the producer completes producing all the objects in the thread. The thread is on pause, and the coroutine invoked by the collect operator resumes its execution, which allows the coroutine to proceed with any subsequent code or tasks after the thread collection.
• In the event of an exception during the flow collection, it may cause the flow to terminate. Additionally, there are operators and functions available in Kotlin Flow that allow control of the termination of flow collection, like the take operator, which restricts the number of elements gathered from the flow.

Overall, the termination of flow collection can occur due to coroutine cancellation, completion of the flow’s data generation, exceptions, or explicit control mechanisms provided by flow operators.

Managing unforeseen exceptions

We can utilize the catch intermediate function to manage anomalies that may happen in the realization of the producer. He permits intercept and process exceptions within the flow. By applying the catch function, we can define a particular error-handling logic to manage any exclusions during the handling of the stream gracefully.

init {
    viewModelScope.launch {
        productRepository.favoriteLatestProducts
            // Intermediate catch operator. If an exception is thrown,
            // catch and update the UI
            .catch {
                exception - > notifyError(exception)
            }
            .collect {
                favoriteProducts - >
                    // Update View with the latest favorite products
            }
    }
}

In the last sample, if an exception happens, the collect lambda is not invoked because there is no new element.

Running in an alternative CoroutineContext

Primarily, the producer of a flow builder performs in the same CoroutineContext as the coroutine that gathers from it. That implies it cannot generate information from a dissimilar CoroutineContext, such as Dispatchers.Main utilized by viewModelScope. Nonetheless, this conduct may be undesirable in some samples, particularly involving the information warehouse layer.

To edit the CoroutineContext of a flow, we can apply the particular operator flowOn. By applying flowOn, we can edit the CoroutineContext of the upward thread, which involves the producer and any middle function preceding it. Nevertheless, it’s essential to note that the downward flow, consisting of intermediate processes after flowOn and the consumer, remains unaffected and continues producing on the CoroutineContext utilized for amassing from the stream. In scenarios containing multiple flowOn functions, every alters the upward thread from its actual position. 

class ProductRepository(
    private val productRemoteDataSource: ProductRemoteDataSource 
    private val userData: UserData
) {
    /**
     * Returns the favorite latest products applying transformations on the flow.
     * These operations are lazy and don't trigger the flow. Instead, they transform
     * the current value emitted by the flow now.
     */
    val favoriteLatestProducts: Flow < List < Product >>=
        productRemoteDataSource.latestProducts
        .map {
            products - > // Executes on the default dispatcher
                products.filter {
                    userData.isFavoriteProduct(it)
                }
        }
        .onEach {
            products - > // Executes on the default dispatcher
                saveInCache(products)
        }
        // flowOn affects the upstream flow ↑
        .flowOn(Dispatchers.Default)
        // the downstream flow ↓ is not affected
        .catch {
            exception - > // Executes in the consumer's context
                emit(lastCachedProducts())
        }
}

With the presented program, the onEach and map operators utilize the defaultDispatcher, while the catch operator and the consumer render on the Dispatchers.Main, which exploits the viewModelScope.

Flow in Jetpack libraries

Flow incorporating into various Jetpack libraries and high regard in Android third-party libraries. It is specifically suitable for live information modifications and continuous information flow.

When performing with Room, we can leverage stream to receive notifications about updates in a database. We can obtain real-time updates by returning a Flow type from your data access objects (DAO), which gives permissions to effortlessly integrate database changes into your application flow and act on them promptly.

@Dao
abstract class ProductDao {
    @Query("SELECT * FROM Product")
    abstract fun getProducts(): Flow < List < Product >>
}

Whenever there is a modification in the Product table, the program will release the newest list containing the updated items in the database.