How to combine two data-flows in Android app architecture?

Question

I was building a chat application in Android in accordance with the app architecture guide. I have stored the user's data in Room Database in sync with Firebase Cloud Firestore, and user's active-status in Firebase Realtime Database.

In Datasource layer, users' data is exposed in the form of PagingDataFlow and a single user's active-status is exposed in form of Flow. From Activity/Fragment layer, both of these combined together form a UI state. Now, either in Repository or in ViewModel, I have to combine these two and form the UserUI unit and pass on that flow to View layer. I guess I have to do this in Repository layer. But I have no clue how to do it.

       -----------------                             ------------------
       |  UserLocalDS  |                             |  UserRemoteDS  |
       -----------------                             ------------------
           ↑       |                                     ↑        |
   (no_arg)|       |Flow<PagingData<User>>        user_id|        |Flow<UserStatus>
           |       ↓                                     |        ↓
       ----------------------------------------------------------------
       |                       UserRepository                         |
       ----------------------------------------------------------------
                                      |
                                      |Flow<UserUI>
                                      ↓
                               --------------
                               | UserListVM |
                               --------------
                                      |
                                      |Flow<UserUI>
                                      ↓
                               --------------
                               |UserListView|
                               --------------

Each emission from UserLocalDS is a PagingData<User> which contains a paged list of User. For each such User, I have to pass user.user_id to UserRemoteDS and obtain a Flow<UserStatus>, which emits UserStatus, which has to be merged with User to result in a PagingData<UserUI>.

This is pseudocode for what is expected. But this is not proper.

    pagedUsersFlow = userLocalDS.getPagedUsersFlow()
    pagedUsersFlow.foreach { user ->
        userStatusFlows[user.user_id] = userRemoteDS.getUserStatusFlow(user.user_id)
    }
    pagedUsersFlow.transform { user ->
        user + userStatusFlows[user.user_id].collect()
    }

Answer 1

How about this:

val f1: Flow<List<Int>> = flow {
    emit(listOf(1, 2, 3))
    emit(listOf(4, 5, 6))
}

fun fs(i: Int): Flow<String> = flow {
    repeat((Math.random() * 10).toInt()) {
        delay((Math.random() * 1000).toLong())
        emit("_$i")
    }
}

val f2: Flow<Pair<Int, String>> = f1.flatMapMerge { l ->
    l.asFlow().flatMapMerge { n ->
        fs(n).map { s ->
            n to s
        }
    }
}
f2.onEach {
    println(it)
}.collect()

Sample output:

(5, _5)
(3, _3)
(4, _4)
(2, _2)
(6, _6)
(5, _5)
(3, _3)

This will produce a flow of the user updates of every user. Each emission will contain both the user and the status.

To simplify things, take Pair<Int, String> as the combination of User and UserStatus, i.e. UserUI, Flow<List<Int>> as Flow<PagingData<User>>, and Flow<String> as Flow<UserStatus>.

You probably shouldn't use the Flow.*latest operators because they will drop emissions when there's back-pressure. And there could be back-pressure if calling getUserStatusFlow for an entire PagingData<User> is ever slower than Flow<PagingData<User>> is emitting PagingData<User>>.

You'll need a way to convert a PagingData<User> to a Flow<User>. (I do this with l.asFlow(), where asFlow is provided by kotlinx.coroutines) That should be pretty straightforward if PagingData is a type of collection that allows iteration. If you don't know how to do this, ping me in the comments and I'll add it.

Answer 2

I've come up with the following implementation:

pagedUsersFlow.flatMapLatest {
     flow {
         it.map { user ->
             userRemoteDS.getUserStatusFlow(user.id).onEach { status ->
                 emit(UserUI(user, status))
             }.launchIn(coroutineScope)
         }
     }
}.collect { userUI ->
    // use userUI
}

---

Update:

pagedUsersFlow.map {
    it.map { user ->
        val status = userRemoteDS.getUserStatusFlow(user.id).firstOrNull() ?: UserStatus("offline")
        UserUI(user, status)
    }
}.collect { pagingUI: PagingData<UserUI> ->
    // ...
}

The drawaback of this solution is that it will not update user status if it is changed on Firestore. Actually I don't think there is a good solution to what you want to achieve using PagingData. I would recommend to reconsider your data structures and use List instead of PagingData.