如何在Android App Architecture中组合两个数据流？

Question

我根据 app Architecture指南。我已经将用户的数据存储在 ROOM数据库与 firebase cloud Firestore 和用户的active-status同步，并在 firebase firebase实时数据库中。

在 dataSource 层中，用户的数据以 paggingdataflow 的形式公开，单个用户的活动状态以 flow flow 的形式公开。来自活动/ fragment 层，这两个组合在一起形成一个UI状态。现在，在存储库或 viewModel 中，我必须组合这两个并形成 userUi 单位，然后将该流程传递给> 查看层。我想我必须在存储库图层中执行此操作。但是我不知道该怎么做。

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

userLocalds 的每个发射都是 pagingdata＆lt; user＆gt; ，其中包含 user 的分页列表。对于每个这样的 user ，我必须将 user.user_id 传递给 userremoteds 并获得 flow＆lt＆userstatus＆gt; ，，发出 userstatus ，必须将其与 user 合并，以导致 pagingdata＆lt; userUi＆gt; 。

这是预期的伪代码。但这是不合适的。

    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

怎么样：

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()

样本输出：

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

这将产生每个用户的用户更新的流程。每种排放都将包含用户和状态。

要简化事物，请以 pair＆lt; int，字符串＆gt; 作为用户和 userstatus 的组合 flow＆lt; list＆lt; int＆gt;＆gt; 作为 flow＆lt; pagingData＆lt; user＆gt;＆gt;＆gt; ，以及 flow flow＆lt; string＆gt; as as as flow＆lt＆lt＆lt; lt; userstatus＆gt; 。

您可能不应该使用流。*最新的运算符，因为它们会在有后压时丢弃排放。如果调用 getUserStatusflow 对于整个 pagingdata＆lt; user＆gt; 都比 flow flow＆pagingdata＆lt;用户＆gt; gt; 正在排放 pagingData＆lt;用户＆gt; 。

您需要一种将 pagingData＆gt; 转换为 flow＆lt; user＆gt; 的方法。 （我使用 l.asflow（）进行此操作，其中 ASFLOW 由 kotlinx.coroutines ）提供，如果 PIGGDATA 是一种允许迭代的集合。如果您不知道该怎么做，请在评论中ping我，我会添加它。

Answer 2

我想出了以下实现：

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> ->
    // ...
}

该解决方案的绘图是，如果用户状态在 firestore上更改，则不会更新用户状态。实际上，我认为使用 pagingData 实现的目标没有一个很好的解决方案。我建议重新考虑您的数据结构并使用 list 而不是 pagingdata 。