Unidirectional Data Flow using Coroutines

Over the years the typical architecture of an Android application has changed a lot, from the first apps where all the logic was in the Activity a lot of different patterns have emerged. Right now MVP and MVVM are probably the most common architectural patterns for the UI (often together with something based on the clean architecture) but there are a lot of alternatives. For example, many developers are using something inspired by Redux, the state container usually used in React-based applications (web or mobile).

In the last months, many frameworks have been presented by various companies:

• mobius developed by Spotify
• MVICore developed by Badoo
• MvRx developed by Airbnb after they stopped using React Native
• RxRedux developed by Freeletics

All these frameworks are heavily based on RxJava and allow to implement the concept called Unidirectional Data Flow: everything starts with an observable of the UI events that are transformed in various ways to an observable of state updates that will be used to update the UI. This idea is really clean in theory, but can be a bit complicated to implement in practice.

In this post, we’ll see how to implement something based on the same concepts using the Kotlin coroutines. The final result is less functional and without some functionality (like the Time Travel Debugger available on MVICore) but really easy to implement and to understand.

The immutable state

All the unidirectional data flow frameworks are based on a concept: the view state is modelled using an immutable object. The view observes a flow of states, every time there is an update a new immutable object is emitted creating a copy of the previous state.

Kotlin doesn’t have a language construct to define immutable objects, the best way to achieve it is a data class with all the fields defined as val. Each field must be a primitive type or an instance of an immutable class.

For example, we can use the following data class to define the state of a View that shows a list of users, a loading state or an error message:

data class UserListViewState(
        val users: List<User> = emptyList(),
        val loading: Boolean = false,
        val error: Throwable? = null
)

A better way to model this state is using a sealed class that implements the LCE pattern (Loading / Content / Error), but here we use a data class to keep the example simple. More info about LCE pattern can be found here, a Kotlin example of how to implement it is here.

The business logic

The User class contains a boolean starred field and since the state is immutable it can’t be updated directly. Thus we need to create a new state to toggle it and update the UI. For example, we can create a method in a UseCase object to implement it (replaceAt is an extension method to create a new list replacing an item):

class MainUseCase {
    fun toggleUser(state: UserListViewState, pos: Int): 
            UserListViewState {
        val currentUser = state.users[pos]
        val newUser = currentUser.copy(starred = 
            !currentUser.starred)
        return state.copy(users = 
            state.users.replaceAt(pos, newUser))
    }
}

This method is similar to a Redux reducer, it creates a new state based on the old state and some parameters. It’s a pure function, the output depends only on the input and it doesn’t produce any side effects.

The Android stuff: ViewModel and Activity-Fragment

The MainUseCase object can be used in a ViewModel (here we are using the class defined in the Architecture Components) to easily manage the configuration changes:

class MainViewModel(
        private val useCase: MainUseCase
) : ViewModel() {

    val store = ViewStateStore(UserListViewState())
    //...

    fun toggleUser(position: Int) {
        val newState = useCase.toggleUser(
                store.state(), position)
        store.dispatchState(newState)
    }
}

The ViewStateStore is a simple class that wraps a LiveData, it forces the value to be not null thanks to a mandatory initial value and Kotlin nullability support:

class ViewStateStore<T : Any>(
        initialState: T
) {
    private val liveData = MutableLiveData<T>().apply {
        value = initialState
    }
    fun observe(owner: LifecycleOwner, observer: (T) -> Unit) =
            liveData.observe(owner, Observer { observer(it!!) })
    fun dispatchState(state: T) {
        liveData.value = state
    }
    fun state() = liveData.value!!
}

In an Activity or a Fragment we can register an observer that updates the UI on every new state:

viewModel.store.observe(this) {
    loading.isVisible = it.loading
    error.isVisible = it.error != null
    retry.isVisible = it.error != null
    recycler.isVisible = !it.loading && it.error == null
    adapter.list = it.users
}

Thanks to the LiveData we don’t need to remove the observer and to manage the Activity/Fragment lifecycle.

The architecture used in this example to manage the UI is based on three main component: an Activity/Fragment, a ViewModel and an UseCase. Every time there is a UI event (for example a click on a button) this flow is executed:

• a listener on the Activity/Fragment (or something similar managed using the Android data binding) invokes a method in the ViewModel;
• the ViewModel invokes an UseCase method passing the current state and some extra parameters. The current state is updated based on the value returned by the UseCase method;
• the UI is updated automatically because the Activity/Fragment observes the LiveData used to manage the current state.

What about async actions?

The previous example was really simple, a real application is usually more complicated and contains asynchronous code. In Kotlin we can manage async code using coroutines, so let’s try to change the example to execute a server call to toggle the starred field. We can define a suspend method in the UseCase that invokes a suspend method on a Repository (for example a Retrofit object) and updates the state accordingly:

suspend fun toggleUser(state: UserListViewState, position: Int): 
            UserListViewState {
    val currentUser = state.users[position]
    val newUser = repository.toggleUser(currentUser).await()
    return state.copy(users = 
            state.users.replaceAt(position, newUser))
}

Thanks to coroutines this async code is similar to the previous synchronous version.

In the ViewModel we need to use the launch method to execute a suspending method, at the end of the async execution the state is updated using the dispatchState method:

fun toggleUser(position: Int) {
    launch {
        val newState = useCase.toggleUser(store.state(), position)
        withContext(Main) {
            store.dispatchState(newState)
        }    
    }
}

Clicking on a user on the UI a server call is executed and then the UI is updated. This is just a demo, in a real project a try/catch is necessary to avoid an application crash in case of an exception.

Does this solution work? What happens if the server or the connection is slow and the user clicks on two items quickly? Two server calls are executed in parallel and, at the end of both calls, the state is updated. But there is something wrong, the state is updated based on the value at the beginning of the server call. So when the second server call returns the state is updated to a value that overwrites the modification done after the first call.

Let’s say there are two users in the UI, both not starred so the state is [false, false]. Then two server calls are executed, both executions of the toggleUser method in the UseCase have [false, false] as initial state parameter. When the first call returns the state is updated to [true, false]. When the second one returns the state is updated based on the initial state, so it’s updated to [false, true] ignoring the value already used to update the UI. We have a concurrency issue even using immutable objects! :(

To fix this issue we can create a new class that defines a method used to update the state, it’s just a wrapper to a function:

class Action<T>(private val f: T.() -> T) {
    operator fun invoke(t: T) = t.f()
}

Probably the same example could be implemented even without this class but we are using it because it simplifies the code and the error messages when something is wrong. Starting in Kotlin 1.3 this class can be defined as an inline class to avoid an extra object allocation.

Now we can modify the UseCase method to return an Action that will create the new state instead of the new state directly:

suspend fun toggleUser(state: UserListViewState, position: Int): 
            Action<UserListViewState> {
    val currentUser = state.users[pos]
    val newUser = repository.toggleUser(currentUser).await()
    return Action {
        copy(users = users.replaceAt(position, newUser))
    }
}

Thanks to the Kotlin syntax this code is simple and similar to the previous version.

Now we must modify the ViewModel to use this method, the important modification is that the Action is executed with the updated state as parameter (and not with the state available at the beginning of the method):

fun toggleUser(position: Int) {
    launch {
        val action = useCase.toggleUser(store.state(), position)
        withContext(Main) {
            dispatchState(action(store.state()))
        }
    }
}

The difference is important in case there are two parallel invocations, the new state is always created based on the last state available and on the same thread (the Android main thread in this example) to avoid concurrency problems.

The code can be simplified defining a dispatchAction in the ViewStateStore class (this class must implement CoroutineScope to define the default dispatcher to use and manage coroutines cancellation):

class ViewStateStore<T : Any>(
        initialState: T
) : CoroutineScope {
    //... 
    private val job = Job()

    override val coroutineContext: CoroutineContext = 
         job + Dispatchers.IO
    //...
    fun dispatchAction(f: suspend (T) -> Action<T>) {
        launch {
            val action = f()
            withContext(Main) {
                dispatchState(action(state()))
            }
        }
    }
}

And now the ViewModel code is just an invocation of the UseCase method inside the lambda used as dispatchAction parameter:

fun toggleUser(pos: Int) {
    store.dispatchAction { useCase.toggleUser(it, pos) }
}

Testing the use case

The business logic of the application is in the UseCase class, for this reason it’s really important to have an easy way to verify it using a JVM test. The UseCase of the previous example uses a Repository collaborator to execute the server call. In a test we can replace this object with a mock:

class MainUseCaseTest {

    val repository: Repository = mockk()

    val useCase = MainUseCase(repository)

    @Test
    fun toggleUserUsingRepository() {
        coEvery { repository.toggleUser(USER_1) } returns
                USER_1.copy(starred = true)

        val initialState = UserListViewState(listOf(USER_1, USER_2))
        val action = runBlocking {
            useCase.toggleUser(initialState, 0)
        }
        val finalState = action(initialState)
        
        assert(finalState.users[0].starred).isTrue()
    }
}

This test uses mockk to create mock objects and assertk to verify the field value of the resulting state. The UseCase doesn’t use any Android classes and doesn’t manage the threading so it’s not necessary to use any JUnit rules or something else to change the execution thread.

Multiple actions in a single method

Sometimes a single method in the UseCase must update the UI multiple times, for example to show a loading indicator at the beginning of an operation and the result (or an error message) at the end. In this case, the method cannot return an action, it’s necessary to return something similar to a RxJava Observable that can manage multiple actions. Using the coroutines we can use a Channel, it can be created using the produce method:

suspend fun getList(): ReceiveChannel<Action<UserListViewState>> = 
    produceActions {
        send(Action { copy(loading = true, error = null) })
        try {
            val users = repository.getList().await()
                    .map { it to repository.isStarred(it.id) }
                    .map { (user, deferred) ->
                        user.copy(starred = deferred.await())
                    }
            send(Action { copy(users = users, loading = false) })
        } catch (e: Exception) {
            send(Action { copy(error = e, loading = false) })
        }
    }

The send method is used to create and emit an Action that will create the new state. So in this method the channel returned will emit two actions: the first one will be used to create the state with the loading indicator visible and the second one to show the result or the error message.

The produceActions method is just a utility method used to simplify the code, it invokes the produce method to create a ReceiveChannel. Currently there are two possible implementation of this method, one uses the GlobalScope method and the other is a CoroutineScope extension function. I am not sure which one is the best solution and there is something strange using the produce method inside a suspending method, I have opened a bug on the Kotlin Coroutine GitHub repository to report it.

The users are loaded invoking two methods in the Repository object, in case this code becomes complex or must be reused in another ViewModel it can be easily moved to another class.

This method can be invoked in the ViewModel to obtain the channel to pass to a new method dispatchActions:

fun loadData() {
    store.dispatchActions(useCase.getList())
}

This new method is similar to the previous one but manages multiple actions instead of a single one. Using the coroutines the code is similar to a standard for loop even if the actions are produced asynchronously:

fun dispatchActions(channel: ReceiveChannel<Action<T>>) {
    launch {
        channel.consumeEach { action ->
            withContext(Main) {
                dispatchState(action(state()))
            }
        }
    }
}

Wrapping up

The complete example can be found in this GitHub repository, a more complex one with some extra features is available here. The architecture described in this post implements some of the Unidirectional Data Flow principles:

• the state is immutable, a new object is created based on the previous one every time there is an update;
• the actions (implemented as an UseCase method) are functions that create a new state. They can be tested easily using a JVM test and a library like Mockito or Mockk to replace the collaborators;
• the business logic doesn’t update the UI directly, it just updates the state.

The main components used in this example are three:

• Activity/Fragment: manages the user interface of the application, it invokes a ViewModel method on user interaction and registers an observer on the state LiveData to update the UI every time a new value is available;
• ViewModel: doesn’t contain any logic, it just manages a LiveData. Every time there is a user interaction it invokes an UseCase method and updates the LiveData using the new state created based on the Action/Channel returned;
• UseCase: similar to a Redux reducer, it creates a new state (or one or more actions to create a new state) based on the previous state and some parameters. In simple cases it can contain the business logic, in a real application it delegates the business logic to other objects.

However, comparing this example to the examples based on the libraries listed at the beginning of the post there are some differences:

• unidirectional data flow is used only in View/ViewModel communication: it’s not based on RxJava so there isn’t a unique flow from the UI events to the UI updates. The UI events are managed using standard methods that will return a new state. This can be a drawback but in my opinion is a big advantage because it allows keeping the code simple;
• using a Redux approach the actions should contain only the state updates and should be separated from the effects (like an asynchronous call to a server). Instead, the last example described in this post contains both actions and effects in a single method multiple. A method like this can become complicated but it’s pure Kotlin code so can be tested on the JVM and it’s easy to refactor it extracting some methods or delegate something to other objects.

I haven’t used yet this architecture in a real application, right now this an attempt to create something simple that uses the typical concepts of the Unidirectional Data Flow. For this reason, all the feedbacks are welcome, just write a comment here to tell me your opinion!