Room RxJava observable 在插入时触发多次

Question

我的存储库实现有一个奇怪的问题。每次我调用应该从数据库中获取数据并通过网络调用更新数据库的函数时，我都会从数据库观察者那里收到多个结果。

override fun getApplianceControls(
    serialNumber: SerialNumber
): Flowable<ApplianceControlState> {
    val subject = BehaviorProcessor.create<ApplianceControlState>()

    controlsDao.get(serialNumber.serial)
        .map { controls ->
            ApplianceControlState.Loaded(controls.toDomainModel())
        }
        .subscribe(subject)

    controlApi.getApplianceControls(serialNumber.serial)
        .flatMapObservable<ApplianceControlState> { response ->
            val entities = response.toEntity(serialNumber)
            // Store the fetched controls on the database.
            controlsDao.insert(entities).andThen(
                // Return an empty observable because the db will take care of emitting latest values.
                Observable.create { }
            )
        }
        .onErrorResumeNext { error: Throwable ->
            Observable.create { emitter -> emitter.onNext(ApplianceControlState.Error(error)) }
        }
        .subscribeOn(backgroundScheduler)
        .subscribe()


    return subject.distinctUntilChanged()
}

@Dao
interface ApplianceControlsDao {

    @Insert(onConflict = OnConflictStrategy.REPLACE)
    fun insert(controls: List<TemperatureControlEntity>): Completable

    @Query("SELECT * FROM control_temperature WHERE serial = :serial")
    fun get(serial: String): Flowable<List<TemperatureControlEntity>>
}

基本上，如果我调用getApplianceControls 一次，我会得到想要的结果。然后我再次调用，另一个序列号是空的，我得到了空数组。 然后我第三次调用，但序列号与第一次相同，在插入调用后我得到正确结果和空数组的混合。

像这样：

第一次调用，序列号“123” -> 加载（[control1，control2，control3]）

第二次调用，序列号“000” -> Loaded([])

第三次调用，序列号“123” -> Loaded([control1, control2, control3]), Loaded([]), Loaded([control1, control2, control3])

如果我从 api 响应中删除 db insert，它可以正常工作。在调用insert 之后，所有奇怪的事情都会发生。

编辑：getApplianceControls() 是从 ViewModel 调用的。

fun loadApplianceControls(serialNumber: SerialNumber) {
    Log.i("Loading appliance controls")

    applianceControlRepository.getApplianceControls(serialNumber)
        .subscribeOn(backgroundScheduler)
        .observeOn(mainScheduler)
        .subscribeBy(
            onError = { error ->
                Log.e("Error $error")
            },
            onNext = { controlState ->
                _controlsLiveData.value = controlState  
            }
        ).addTo(disposeBag)
}

Answer 1

正如我在评论中提到的，您有 2 个未在任何地方取消订阅的订阅，这可能会导致内存泄漏（在释放主题时它不会释放），而且通过这种实现，您会忽略 API 错误。 我会尝试将其更改为：

override fun getApplianceControls(serialNumber: SerialNumber): Flowable<ApplianceControlState> {

    val dbObservable = controlsDao.get(serialNumber.serial)
        .map { controls ->
            ApplianceControlState.Loaded(controls.toDomainModel())
        }

    val apiObservable = controlApi.getApplianceControls(serialNumber.serial)
        .map { response ->
            val entities = response.toEntity(serialNumber)
           // Store the fetched controls on the database.
           controlsDao.insert(entities).andThen( Unit )
        }
        .toObservable()
        .startWith(Unit)

    return Observables.combineLatest(dbObservable, apiObservable) { dbData, _ -> dbData }
        // apiObservable emits are ignored, but it will by subscribed with dbObservable and Errors are not ignored 
        .onErrorResumeNext { error: Throwable ->
            Observable.create { emitter -> emitter.onNext(ApplianceControlState.Error(error)) }
        }
        .subscribeOn(backgroundScheduler)
        //observeOn main Thread
        .distinctUntilChanged()
}

我不确定它是否能解决原来的问题。但如果是这样 - 问题出在flatMapObservable 查看controlApi.getApplianceControls() 的实现也很有用。