Android flow get value

Lessons learnt using Coroutines Flow in the Android Dev Summit 2019 app

This article is about the best practices we found when using Flow in the Android Dev Summit (ADS) 2019 app; which has just been open sourced. Keep reading to find out how each layer of our app handles data streams.

The ADS app architecture follows the recommended app architecture guide, with the addition of a domain layer (of UseCases) which help separate concerns, keeping classes small, focused, reusable and testable:

Like many Android apps the ADS app lazily loads data from the network or a cache; we found this to be a perfect use case for Flow . For one shot operations, suspend functions were a better fit. There are two main commits that refactor the app to use Coroutines. The first commit migrates one-shot operations, and the second one migrates to data streams.

In this article, you can find the principles we followed to refactor the app from using LiveData in all the layers of the architecture to just use LiveData for communication between View and ViewModel, and Coroutines for the UseCase and lower layers of our architecture.

1. Prefer exposing streams as Flows (not Channels)

There are two ways you can deal with streams of data in coroutines: the Flow API and the Channel API. Channels are a synchronisation primitive whereas Flow is built to model streams of data: it’s a factory for subscriptions to streams of data. Channels can however be used to back a Flow , as we’ll see later.

Prefer exposing Flow since it gives you more flexibility, more explicit contracts and operators than Channel

Flows automatically close the stream of data due to the nature of the terminal operators which trigger the execution of the stream of data and complete successfully or exceptionally depending on all the flow operations in the producer side. Therefore, you can’t (nearly as easily) leak resources on the producer side. This is easier to do with Channels: the producer might not clean up heavy resources if the Channel is not closed properly.

The data layer of an app is responsible for providing data usually by reading from a database or fetching from the Internet. For example here’s a DataSource interface that exposes a stream of user event data:

2. How to use Flow in your Android app architecture

UseCase and Repository

The layers in-between View/ViewModel and the DataSource (i.e. UseCase and Repository in our case) often need to combine data from multiple queries or transform the data before it can be used by the ViewModel layer. Just like Kotlin sequences, Flow supports a large set of operators to transform your data. There are a wealth of operators already available, or you can create your own transformation (e.g. using the transform operator). However, Flow exposes suspend lambdas on many of the operators, there’s often no need to make a custom transform to accomplish complex tasks, just call suspend functions from inside your Flow .

In our ADS example, we want to combine the UserEventResult with session data in the Repository layer. We use the map operator to apply a suspend lambda to each value of the Flow retrieved from DataSource:

ViewModel

When performing UI ↔ ViewModel communication with LiveData , the ViewModel layer should consume the stream of data coming from the data layer using a terminal operator (e.g. collect , first or toList ).

If you’re converting a Flow to a LiveData , you can use the Flow.asLiveData() extension function from the androidX lifecycle LiveData ktx library. This is very convenient since it will share a single underlying subscription to the Flow and will manage the subscription based on the observers’ lifecycles. Moreover, LiveData also keeps the most recent value for late-coming observers and the subscription active across configuration changes. Check this simpler code that showcases how you can use the extension function:

Disclaimer: The code snippet above is not part of the app; it’s a simplified version of the code that showcases how you can use Flow.asLiveData() .

3. When to use a BroadcastChannel or Flow as an implementation detail

Back to the DataSource implementation, how can we implement the getObservableUserEvent function we exposed above? The team considered two alternatives implementations: the flow builder or the BroadcastChannel API. Each serve different use cases.

When to use Flow

Flow is a cold stream. A cold stream is a data source whose producer will execute for each listener that starts consuming events, resulting in a new stream of data being created on each subscription. Once the consumer stops listening or the producer block finishes, the stream of data will be closed automatically.

Flow is a great fit when the production of data needs to start/stop to match the observer

You can emit a limited or unlimited number of elements using the flow builder.

Flow tends to be used for expensive tasks as it provides automatic cleanup via coroutine cancellation. Notice that this cancellation is cooperative, a flow that never suspends can never be cancelled: in our example, since delay is a suspend function that checks for cancellation, when the subscriber stops listening, the Flow will stop and cleanup resources.

When to use BroadcastChannel

A Channel is a concurrency primitive for communicating between coroutines. A BroadcastChannel is an implementation of Channel with multicast capabilities.

There are some cases where you might want to use an implementation of BroadcastChannel in your DataSource layer:

Use BroadcastChannel when the producer(s) and consumer(s) have different lifetimes or operate completely independently of each other

The BroadcastChannel API is the perfect fit when you want the producer to follow a different lifecycle and broadcast the current result to anyone who’s listening. In this way, the producer doesn’t need to start every time a new listener starts consuming events.

You can still expose a Flow to the caller, they don’t need to know about how this is implemented. You can use the extension function BroadcastChannel.asFlow() to expose a BroadcastChannel as a Flow .

However, closing that Flow won’t cancel the subscription. When using BroadcastChannel , you have to take care of its lifecycle. They don’t know if there are listeners or not, and will keep resources alive until the BroadcastChannel is cancelled or closed. Make sure to close the BroadcastChannel when it’s no longer needed. Also, remember that a closed channel cannot be active again, you’d need to create a new instance.

An example of how to use the BroadcastChannel API can be found in the next section.

Disclaimer

Parts of the Flow and Channel APIs are still in experimental, they’re likely to change. There are some situations where you would currently use Channels but the recommendation in the future may change to use Flow . Specifically, the StateFlow and Flow’s share operator proposals may reduce the usage of Channel in the future.

4. Convert data streams callback-based APIs to Coroutines

Multiple libraries already support coroutines for data streams operations, including Room. For those that don’t, you can convert any callback-based API to Coroutines.

Flow implementation

If you want to convert a stream callback-based API to use Flow , you can use the channelFlow function (also callbackFlow , which shares the same implementation). channelFlow creates an instance of a Flow whose elements are sent to a Channel . This allows us to provide elements running in a different context or concurrently.

In the following sample, we want to emit the elements that we get from a callback into a Flow :

1. Create a flow with the channelFlow builder that registers a callback to a third party library.
2. Emit all items received from the callback to the Flow .
3. When the subscriber stops listening, we unregister the subscription to the API using the suspend funawaitClose .

BroadcastChannel implementation

For our stream of data that tracks user authentication with Firestore, we used the BroadcastChannel API as we want to register one Authentication listener that follows a different lifecycle and broadcasts the current result to anyone who’s listening.

To convert a callback API to BroadcastChannel you need a bit more code than with Flow . You can create a class where the instance of the BroadcastChannel can be kept in a variable. During initialisation, register the callback that sends elements to the BroadcastChannel as before:

5. Testing tips

To test Flow transformations (as we do in the UseCase and Repository layers), you can use the flow builder to return fake data. For example:

To test implementations of Flow successfully, a good idea is to use the take operator to get some items from the Flow and the toList operator as the terminal operator to get the results in a list. See an example of this in the following test:

The take operator is a great fit to close the Flow after you get the items. Not closing a started Flow (or BroadcastChannel ) after each test will leak memory and creates a flaky and inconsistent test suite.

Note: If the implementation of the DataSource is done with a BroadcastChannel , the code above is not enough. You have to manage its lifecycle by making sure you start the BroadcastChannel before the test and close it after the test finishes. If not, you’ll leak memory. You can see a test like this in this other Flow sample.

Testing Coroutines best practices also apply here. If you create a new coroutine in code under test, you might want to execute it in your test thread for a deterministic execution of your test. Check out more about this in the Testing Coroutines ADS 2019 talk.

Источник

Android flow get value

Flow is a new feature of Kotlin Coroutines. Whereas suspending functions in Kotlin can only return a single value; Flow can return multiple values.

You can think of Flow as a stream. Values flow downstream like water in a river. Values can be created from asynchronous actions such as network requests or database calls. If you are familiar with RxJava — Flow behaves similarly to Observable and Flowable . Flow is built on top of coroutines. We love coroutines because they bring structured concurrency to Kotlin.

In this article we will look into the basics of Flow and in which scenarios we can use it.

The first thing we see in the code above is the flow builder. Inside the flow builder you can do any asynchronous work and use emit to send some kind of value. A 100 millisecond delay is used to simulate a network request before emitting a value. delay is a handy suspend function which delays for a specified time before resuming where it left off. Two values are emitted before the flow completes. collect is also a suspend function and it is used to start collecting values from the flow. Flows are cold streams so the flow will not start emitting before collect is called. collect is similar to subscribe in RxJava . The flow builder is the producer and the collector is the consumer.

Operators

Flow can be used in a reactive programming style. It has a lot of different operators that can transform a flow with functional operators. Operators such as map , filter , reduce and many more. Flow supports suspending functions on most operators which means that you can execute sequential asynchronous tasks in operators like map .

Let’s get some data from our database and then from the network:

The code above will first get data from the database and emit it to the collector. It then fetches updated data from the network and emits it. For simplicity we skip the caching of network data to the database.

The view can listen to this stream of data and update the UI when new data is collected. We have used the map operator to transform which thread the code is running in. You can see that we introduced a new operator named flowOn which will switch the upstream context in the flow. If you want to execute an operation off the main thread you can set which kind of dispatcher you want to use ( Main , Default , IO , Unconfined ). This is similar to subscribeOn in RxJava .

Custom Operators

Creating your own custom operators is easily achieved. Here is a custom operator that transforms all strings to uppercase.

Data flow in Android

In Android we have to think about how technology works in the context of Fragment , Activity and ViewModel . There are multiple ways of doing this. Personally I use Flow in repositories and LiveData in ViewModel which I will come back to later. Here’s an example of how this might look.

This will print:

WeatherViewModel exposes MutableLiveData to WeatherFragment . In the init block we start to collect the weather flow. In the fragment we start observing the weather LiveData . If the orientation changes and the fragment is recreated the console will print description=Sunny 🙂 because LiveData caches the last item.

If we try and rewrite the same code without LiveData there are some consequences.

Instead of using LiveData to communicate with WeatherFragment we are using Flow . The code mostly works in the same way. If the orientation changes however, WeatherFragment will be recreated and attempt to collect the weather flow. This will cause the flow to emit all its values again. For this reason I like to use LiveData .

Jetbrains are currently working on something called DataFlow which will have similar behaviour as LiveData . You can find more info about DataFlow here

Testing

For testing flows you can use the take operator to get a fixed amount of items from the flow and toList as the terminal operator to get the result as a list. toList suspends until it receives the number of items set in the take operator or it waits until the flow ends. You can still use collect instead of toList but personally I prefer using toList because you end up with all elements so you have more control of the result.

The flow should emit three values in total. In the test above we wait for the flow to emit all values and then confirm the length and the values received. Be careful when using only the toList operator. If the flow emits an infinite amount of values the function would suspend forever. To prevent this from happening we have the take operator to set the number of values that we want to test.

Error handling

If you do not handle errors in your flow your app will crash like any other suspend function. You can use the traditional way of wrapping the flow with a try catch or you can use the catch operator to catch exceptions thrown upstream. One thing to note is that it doesn’t handle exceptions downstream. If the collect block throws an exception, your app will crash. To get around this we can use collect and onEach together to collect all upstream exceptions.

If an error is thrown the flow will still stop emitting new values. If it should continue after an exception you need to handle this case manually.

Conclusion

This was a short intro to Flow . The API is lightweight and makes it hard to leak subscriptions, which is a problem in RxJava . We get this for free with structured concurrency. No more silent resource leaks! Room and SqlDelight already have support for Flow making it easy to get started. If you want to read more about Flow you can find more resources below.

Источник