当我在没有订阅的情况下从班级更改数据库时，Flow 的订阅没有反应

Question

我有一个带有 Room 和 Hilt 的应用程序（所有模块都是 SingletonComponent::class）。 我有 2 个接口：

1. SourceOfData - 用于订阅房间流并将值放入许多视图模型的 SharedFlow 中
2. SomeUseCase - 用于插入对数据库的更改

interface SomeUseCase {
    suspend fun addPerson(person: Person)
    suspend fun deletePersonById(personId: Int)
    suspend fun addItemByPersonId(personId: Int, item: Item)
    suspend fun deleteItemById(itemId: Int)
}

interface SourceOfData {
    fun personSharedFlowForMap(): SharedFlow<List<Person>>

    //Test actions
    suspend fun addPerson(person: Person)
    suspend fun deletePersonById(personId: Int)
    suspend fun addItemByPersonId(personId: Int, item: Item)
    suspend fun deleteItemById(itemId: Int)
}

DataImpl 来源

class SourceOfDataImpl @Inject constructor(
    private val personDao: PersonDao,
    private val itemDao: ItemDao
) : SourceOfData {
 
    private val _personFlowMap = MutableSharedFlow<List<Person>>()
    val personFlowMap: SharedFlow<List<Person>> = _personFlowMap.asSharedFlow()
 
    init {
        subscribeToPersonMapWithItemDao()
    }
 
    private fun subscribeToPersonMapWithItemDao() {
        CoroutineScope(Dispatchers.IO).launch {
            combine(
                personDao.getPersons(),
                itemDao.getItems()
            ) { personEntityList, itemEntityList ->
                personEntityList.map { personEntity ->
                    Person(
                        personEntity.id,
                        personEntity.name,
                        itemEntityList
                            .filter { it.personId == personEntity.id }
                            .map { it.toItem() }
                    )
                }
            }.collect {
                _personFlowMap.emit(it)
            }
        }
    }
...

基础视图模型

abstract class BaseViewModel(private val sourceOfData: SourceOfData) : ViewModel() {

    abstract fun updateUiStateMap(personUiMap: PersonUiMap)

    fun subscribeToDataMap() {
        viewModelScope.launch {
            sourceOfData.personSharedFlowForMap().collect {
                val personList = it
                updateUiStateMap(convertDataListToUiMap(personList))
            }
        }
    }
}

问题： 如果我尝试通过 SomeUseCase 更新数据库，SharedFlow 不会发生任何事情，但数据库会发生变化。如果我尝试使用 SourceOfData 界面，应用程序可以正常工作。

主视图模型

    //TODO("Example")
    suspend fun addPerson(personName: String) {
//        someUseCase.addPerson(Person(name = personName)) //TODO("Not OK")
//        sourceOfData.addPerson(Person(name = personName)) //TODO("OK")
    }

问题：我怎样才能正确解决这个问题？

当然，我可以将一个接口用于多个 Daos，并且没有任何问题。但我觉得这不是一个好办法。

Answer 1

像在

subscribeToPersonMapWithItemDao

中那样在数据源内部的某个地方凭空创建一个新的 CoroutineScope 违背了结构化并发的目的。该协程无法从任何地方控制。相反，您应该传递共享流应用作参数的范围。这也将使测试变得更加容易，因为您可以注入测试范围。使用依赖注入框架来提供适当的范围。

现在解决手头的问题。一般来说，您应该仅在 UI 中收集流量。对于您在

SourceOfDataImpl

中所做的事情来说，这当然是正确的。相反，您应该像这样转换流程：

val personFlowMap: SharedFlow<List<Person>> = combine(
    personDao.getPersons(),
    itemDao.getItems(),
) { personEntityList, itemEntityList ->
    personEntityList.map { personEntity ->
        Person(
            personEntity.id,
            personEntity.name,
            itemEntityList
                .filter { it.personId == personEntity.id }
                .map { it.toItem() },
        )
    }
}
    // only needed when the scope isn't already using that dispatcher, which it probably will when you inject it
    .flowOn(Dispatchers.IO)
    .shareIn(
        scope = scope, // the replacement for CoroutineScope() that should be provided as a parameter
        started = SharingStarted.WhileSubscribed(5_000),
        replay = 1,
    )

视图模型也是如此：它也不应该收集流量。它应该只是将流从数据源转换为 UI 状态对象。与数据源使用

combine

合并流的方式相同，您的视图模型应该使用

mapLatest

将原始数据转换为 UI 状态对象。最后，流量应转换为

StateFlow

。其工作原理与上面创建 SharedFlow 的代码非常相似；您只需使用

stateIn

而不是

shareIn

。您需要指定流将发出的初始值作为其第一个值，直到上游流发出其第一个值。现在您可以使用 UI 最终从视图模型中收集流量。

仅让您的 UI 收集流量应该可以解决您的问题。

最后一条评论：您可能应该将流转换从数据源移动到存储库类。这也是流应该转换为 SharedFlow 的地方。然后，您的视图模型将仅与存储库通信。