Flow coroutine with Android

Tram Ho

3 years ago

The asynchronous suspend function returns 1 value, so how can it return multiple asynchronous values???

1. Define

Flow is built on Coroutine.
Multiple values can be provided and returned.
Is a data stream in that thread can compute asynchronously.
The emitted value must always be of the same type as the return value.

2. Main components in the data stream (stream data)

Producer: provides data to be added to the data stream. Flow can deliver data asynchronously. Understand simply when you emit() value into Flow.
Intermediaries: can modify the value emitted in the stream. Here it can be understood that the intermediate operators of Flow like .map(), .catch(), ….
Consumer (user): uses the value from the stream. Understand simply when you collect() Flow.

3. Restrictions

Flow runs sequentially.
For example, Producer is in a coroutine, while calling a suspend function, Producer suspends until the suspend function returns.

<span class="token keyword">class</span> <span class="token function">NewsRemoteDataSource</span><span class="token punctuation">(</span>
    <span class="token keyword">private</span> <span class="token keyword">val</span> newsApi<span class="token operator">:</span> NewsApi<span class="token punctuation">,</span>
    <span class="token keyword">private</span> <span class="token keyword">val</span> refreshIntervalMs<span class="token operator">:</span> Long <span class="token operator">=</span> <span class="token number">5000</span>
<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token keyword">val</span> latestNews<span class="token operator">:</span> Flow<span class="token operator">&lt;</span>List<span class="token operator">&lt;</span>ArticleHeadline<span class="token operator">&gt;</span><span class="token operator">&gt;</span> <span class="token operator">=</span> flow <span class="token punctuation">{</span>
        <span class="token keyword">while</span><span class="token punctuation">(</span><span class="token boolean">true</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
            <span class="token keyword">val</span> latestNews <span class="token operator">=</span> newsApi<span class="token punctuation">.</span><span class="token function">fetchLatestNews</span><span class="token punctuation">(</span><span class="token punctuation">)</span>
            <span class="token function">emit</span><span class="token punctuation">(</span>latestNews<span class="token punctuation">)</span> <span class="token comment">// Emits the result of the request to the flow</span>
            <span class="token function">delay</span><span class="token punctuation">(</span>refreshIntervalMs<span class="token punctuation">)</span> <span class="token comment">// Suspends the coroutine for some time</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>
<span class="token punctuation">}</span>

<span class="token comment">// Interface that provides a way to make network requests with suspend functions</span>
<span class="token keyword">interface</span> NewsApi <span class="token punctuation">{</span>
    <span class="token keyword">suspend</span> <span class="token keyword">fun</span> <span class="token function">fetchLatestNews</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token operator">:</span> List<span class="token operator">&lt;</span>ArticleHeadline<span class="token operator">&gt;</span>
<span class="token punctuation">}</span>

class NewsRemoteDataSource(

private val newsApi: NewsApi,

private val refreshIntervalMs: Long = 5000

) {

val latestNews: Flow<List<ArticleHeadline>> = flow {

while(true) {

val latestNews = newsApi.fetchLatestNews()

emit(latestNews) // 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 NewsApi {

suspend fun fetchLatestNews(): List<ArticleHeadline>

}

In the above example, Producer suspends until fetchLastedNew() make network request successfully. Only then will the result be returned in the stream.

Producer cannot emit a value from another Coroutine context.

Note: You should not call emit while in another CoroutineContext by creating a new Coroutine or using withContext {} in your code. You can use other Flow Builder (flow {}) like callbackFlow

4. Modify the stream

Intermediate operators can be used to modify the data stream without affecting the value returned to the Producer.
These operators are a function that, when applied to a data stream, sets up an operator that is not executed until the value is retrieved in the future.

Eg:
In the example below, the Repository uses an intermediate operator, .map() to transform the data displayed in the View.

<span class="token keyword">class</span> <span class="token function">NewsRepository</span><span class="token punctuation">(</span>
    <span class="token keyword">private</span> <span class="token keyword">val</span> newsRemoteDataSource<span class="token operator">:</span> NewsRemoteDataSource<span class="token punctuation">,</span>
    <span class="token keyword">private</span> <span class="token keyword">val</span> userData<span class="token operator">:</span> UserData
<span class="token punctuation">)</span> <span class="token punctuation">{</span>
    <span class="token comment">/**
     * Returns the favorite latest news applying transformations on the flow.
     * These operations are lazy and don't trigger the flow. They just transform
     * the current value emitted by the flow at that point in time.
     */</span>
    <span class="token keyword">val</span> favoriteLatestNews<span class="token operator">:</span> Flow<span class="token operator">&lt;</span>List<span class="token operator">&lt;</span>ArticleHeadline<span class="token operator">&gt;</span><span class="token operator">&gt;</span> <span class="token operator">=</span>
        newsRemoteDataSource<span class="token punctuation">.</span>latestNews
            <span class="token comment">// Intermediate operation to filter the list of favorite topics</span>
            <span class="token punctuation">.</span><span class="token function">map</span> <span class="token punctuation">{</span> news <span class="token operator">-&gt;</span> news<span class="token punctuation">.</span><span class="token function">filter</span> <span class="token punctuation">{</span> userData<span class="token punctuation">.</span><span class="token function">isFavoriteTopic</span><span class="token punctuation">(</span>it<span class="token punctuation">)</span> <span class="token punctuation">}</span> <span class="token punctuation">}</span>
            <span class="token comment">// Intermediate operation to save the latest news in the cache</span>
            <span class="token punctuation">.</span><span class="token function">onEach</span> <span class="token punctuation">{</span> news <span class="token operator">-&gt;</span> <span class="token function">saveInCache</span><span class="token punctuation">(</span>news<span class="token punctuation">)</span> <span class="token punctuation">}</span>
<span class="token punctuation">}</span>

class NewsRepository(

private val newsRemoteDataSource: NewsRemoteDataSource,

private val userData: UserData

) {

/**

* Returns the favorite latest news applying transformations on the flow.

* These operations are lazy and don't trigger the flow. They just transform

* the current value emitted by the flow at that point in time.

*/

val favoriteLatestNews: Flow<List<ArticleHeadline>> =

newsRemoteDataSource.latestNews

// Intermediate operation to filter the list of favorite topics

.map { news -> news.filter { userData.isFavoriteTopic(it) } }

// Intermediate operation to save the latest news in the cache

.onEach { news -> saveInCache(news) }

}

The intermediate operator can be applied after another intermediate operator, forming a sequence that is lazily executed when an item is emitted in the Flow.

5. Collect from 1 stream

Use a terminal operator to trigger the flow to start listening for the value. .collect() used to get all values in the stream when those values are emitted.
.collect() As a suspend function, they need to be executed inside a coroutine. They take a lambda as a parameter to call when a new value is returned.

<span class="token keyword">class</span> <span class="token function">LatestNewsViewModel</span><span class="token punctuation">(</span>
    <span class="token keyword">private</span> <span class="token keyword">val</span> newsRepository<span class="token operator">:</span> NewsRepository
<span class="token punctuation">)</span> <span class="token operator">:</span> <span class="token function">ViewModel</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>

    <span class="token keyword">init</span> <span class="token punctuation">{</span>
        viewModelScope<span class="token punctuation">.</span><span class="token function">launch</span> <span class="token punctuation">{</span>
            <span class="token comment">// Trigger the flow and consume its elements using collect</span>
            newsRepository<span class="token punctuation">.</span>favoriteLatestNews<span class="token punctuation">.</span><span class="token function">collect</span> <span class="token punctuation">{</span> favoriteNews <span class="token operator">-&gt;</span>
                <span class="token comment">// Update View with the latest favorite news</span>
            <span class="token punctuation">}</span>
        <span class="token punctuation">}</span>
    <span class="token punctuation">}</span>
<span class="token punctuation">}</span>

class LatestNewsViewModel(

private val newsRepository: NewsRepository

) : ViewModel() {

init {

viewModelScope.launch {

// Trigger the flow and consume its elements using collect

newsRepository.favoriteLatestNews.collect { favoriteNews ->

// Update View with the latest favorite news

}

Collect 1 Flow trigger to the Producer which will refresh the news and emit the results from the network request for a certain period of time. Producer will always work in while(true) loop, data stream will be closed when ViewModel clear or viewModelScope cancel.

Flow collection can be stopped for a number of reasons:
- Coroutines that use collect are destroyed, like the example above. This also means that it will stop all Producers below.
- The producer completes the emission of the item. In this case, the data stream is closed and the Coroutine that was used collect continues execution.
Flow is cold and lazy unless specified by other intermediate operators. This means that the Producer’s code is executed when the Terminal operator is called in the flow.

6. Execute in another CoroutineContext

By default, the Flow builder Producer executes in the CoroutineContext of the Coroutine it collects, and as noted above, the producer cannot emit values from another Coroutine. But there are still some unwanted usecases.
In the example above, the Repository layer should not perform the above operations Dispatcher.Main – used in viewModelScope.
To be able to change the CoroutineContext of a Flow use the intermediate operator .flowOn().
.flowOn(): Change the CoroutineContext of the upstream stream – i.e. Producer or whatever intermediate operator is applied before (or on) .flowOn(). Downstream flow – means the following intermediate operators .flowOn() along with Consumer) is unaffected and executes in the CoroutineContext used to collect from the flow. If there are multiple operators .flowOn(), it will change 1 Upstream from its position.

7. References

Share the news now