Room Flow执行Insert/Update/Upsert时不发射更新的问题求助

问题描述 #

底层数据库表发生变更时，Flow不会发射更新。仓库调用网络接口获取数据并成功保存到Room数据库后，UI层无法收集到更新（需重启应用才能显示新数据），其余功能均正常。

相关代码实现 #

API Service #

interface AtemsService {

    @GET("$PATH/data/_apps.json")
    suspend fun getPackageData(): Map<String, AtemsPackageInfo>
}

Remote 实现 #

class AtemsRemoteImpl @Inject constructor(private val atemsService: AtemsService) : AtemsRemote {

    override suspend fun getAtemsPackageData(): Map<String, AtemsPackageInfo> =
        atemsService.getPackageData()
}

本地Dao #

@Dao
interface AtemsPackageDao {

    @Query("SELECT * FROM $TABLE_NAME")
    suspend fun getAll(): List<LocalAtemsPackage>

    @Upsert
    suspend fun upsertAll(localAtemsPackages: List<LocalAtemsPackage>)
}

仓库实现（原问题版本） #

class AtemsPackageRepositoryImpl @Inject constructor(
    private val atemsRemote: AtemsRemote,
    private val atemsPackageDao: AtemsPackageDao,
    private val atemsPackageMapper: AtemsPackageMapper,
) : AtemsPackageRepository {

    override fun observeAtemsPackages(): Flow<List<AtemsPackage>> {
        return flow {
            val localAtemsPackages = atemsPackageDao.getAll()
            emit(atemsPackageMapper.localToDomain(localAtemsPackages))
            if (localAtemsPackages.isEmpty()) {
                val fetchRemotePackages = atemsRemote.getAtemsPackageData()
                val convertToLocal = atemsPackageMapper.remoteToLocal(fetchRemotePackages)
                atemsPackageDao.upsertAll(convertToLocal)
            }
        }
    }
}

用例 #

class AppsUseCaseImpl @Inject constructor(
    atemsPackageRepository: AtemsPackageRepository,
) : AppsUseCase {

    override val observeApps: Flow<List<AtemsPackage>> = atemsPackageRepository.observeAtemsPackages()
}

ViewModel #

@HiltViewModel
class SidePanelAppListViewModel @Inject constructor(appsUseCase: AppsUseCase) : ViewModel() {

    val apps = appsUseCase.observeApps.stateIn(
        scope = viewModelScope,
        started = WhileUiSubscribed,
        initialValue = emptyList()
    )
}

视图 #

@Composable
fun SidePanelAppList(
    modifier: Modifier = Modifier,
    viewModel: SidePanelAppListViewModel = hiltViewModel(),
) {
    val apps by viewModel.apps.collectAsStateWithLifecycle()

    SidePanelAppList(
        modifier = Modifier.then(modifier),
        apps = apps
    )
}

问题根源 #

原仓库实现中，observeAtemsPackages()返回的是自定义Flow，仅在初始化时查询一次本地数据并发射。Room只有当@Query方法返回Flow类型时，才会自动监听数据库变更并发射新数据，而原Dao的getAll()是挂起函数返回普通列表，无法触发后续的更新通知。

修复后的代码实现 #

在Tenfour04的指导下，调整了代码逻辑（新增根据远程更新时间判断是否拉取数据的逻辑）：

首先需修改Dao，新增可观察的查询方法 #

@Dao
interface AtemsPackageDao {

    @Query("SELECT * FROM $TABLE_NAME")
    fun observeAll(): Flow<List<LocalAtemsPackage>> // 改为返回Flow

    @Query("SELECT * FROM $TABLE_NAME")
    suspend fun getAll(): List<LocalAtemsPackage>

    @Upsert
    suspend fun upsertAll(localAtemsPackages: List<LocalAtemsPackage>)
}

仓库实现（修复版本） #

@OptIn(ExperimentalCoroutinesApi::class)
override fun observeAtemsPackages(): Flow<List<AtemsPackage>> =
    atemsPackageDao.observeAll().transformLatest {
        val lastLocalUpdate = settingsDataStore.getAtemsRemoteLastUpdate.first()
        val lastRemoteUpdate = atemsRepoRemote.getAtemsLastUpdate().lastUpdated
        if (lastRemoteUpdate > lastLocalUpdate) {
            val fetchRemotePackages = atemsRepoRemote.getAtemsPackageData()
            val convertToLocal = atemsPackageMapper.remoteToLocal(fetchRemotePackages)
            atemsPackageDao.upsertAll(convertToLocal)
            settingsDataStore.setAtemsRemoteLastUpdate(lastRemoteUpdate)
        }
        emit(atemsPackageMapper.localToDomain(it))
    }

修复说明 #

1. Dao新增observeAll()方法返回Flow<List<LocalAtemsPackage>>，Room会自动监听该表的变更，每次数据变化时都会发射新的列表。
2. 仓库使用transformLatest操作符，在每次数据库数据更新时，先检查远程更新时间是否晚于本地记录的更新时间，若有则拉取远程数据并更新本地数据库。
3. 由于observeAll()的Flow会在数据库更新后自动发射新数据，UI层能实时收到更新并刷新界面。

内容的提问来源于stack exchange，提问作者rkr87