doOnNext中不会发出可观察的startWith

Question

我有许多Observable用于我的应用程序中的网络请求.由于这么多是相同的,我对它们应用了一个Observable转换:

/**
 * Creates a transformer that applies the schedulers and error handling for all of the observables in this ViewModel.
 */
private fun applyTransformations(): Observable.Transformer<NetworkState, NetworkState> {
    return Observable.Transformer { observable ->
        observable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .onErrorReturn { NetworkState.Error(it) }
                .doOnNext { publishState(it) }
                .startWith(NetworkState.Loading())
    }
}

我试图通过以上目标实现的目标:

• 应用一致的调度程序
• 通过返回我的密封类的实例来处理任何错误.
• 通过发布observable返回的状态来处理任何onNext.
• 通过发送加载状态开始.

这很好用,但我注意到的是,当我调用startWith和加载状态时,它实际上从未实际处理过doOnNext().换句话说,publishState()永远不会调用我的加载状态.

在我设置observable的地方,我不打算添加订阅者,因为doOnNext()以上就是我需要的所有内容:

val subscription = repository.getInstagramPhotos(count)
        .map { mapIGPhotoResponse(it) }
        .compose(applyTransformations())
        .subscribe()

但是,如果我要提供上面的订户,它将处理加载状态.它还将处理两个onNext()调用 - 一个用于提供的订户,另一个用于doOnNext转换.

有没有办法修改此startWith调用以发出任何我指定的内容doOnNext？我正在使用RxJava 1.

编辑:只是为了澄清一些,如果我追踪发射的东西,我希望看到两件事.Loading -> Success.我实际看到的只是Success.如果我向观察者提供订阅者,我会看到Loading -> Success -> Success.

Answer 1

startWith应该在之前doOnNext.

Rxjava方法虽然看起来像是使用构建器模式,但实际上却没有.每次应用运算符时,它们都会返回一个新的observable.在你的情况下,你的doOnNext observable在你的observable开始之前就完成了,因此不会使用你在startWith中提供的东西来调用它的消费者.

理想情况下,您应该:

observable
                .startWith(NetworkState.Loading()) 
                .doOnNext { publishState(it) }
                .onErrorReturn { NetworkState.Error(it) }
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())

另外,onEror如果发生这种情况,请小心订阅没有消费者.由于没有任何东西可以使用onError,因此RxJava会使您的应用程序崩溃,因为它没有任何通知错误.考虑将doOnNext订阅中的成功消费者替换为成功消费者,如果要忽略它,请考虑替换错误消费者.

也doOnNext通常用于副作用,诸如记录和排序,他们更便利比真正的功能性操作符.