插入时多次触发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()
}

kotlin prettyprint-override">@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>>
}

基本上，如果我调用一次GetApplianceControl，就会得到所需的结果。然后我再次调用另一个序列号，它是空的，我得到了空数组。但是，我第三次调用，但是使用与第一次相同的序列号，在进行insert调用后，我得到了正确结果和空数组的混合。

像这样：

第一次呼叫，序列号“123”-

第二次呼叫，序列号“000”-

第三次呼叫，序列号“123”-

如果我从api响应中删除db插入，它会正常工作。调用插入后发生一切奇怪的事情。

编辑：从视图模型调用GetApplianceControl（）。

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

正如我在评论中提到的，您有两个订阅，但都没有在任何地方取消订阅，这可能会导致内存泄漏（当主题被释放时，它不会被释放），而且在这种实现中，您会忽略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()
}

我不确定它是否解决了最初的问题。但如果是这样-问题在平面地图观察ALSO将有助于查看ControlApi.getApplianceControlls（）实现。