是否可以在副作用运算符（doOnNext（））中使用方法（返回 Observable）？

Question

我刚刚开始将 RxJava 引入我现有的项目。

我有一个名为getUser() 的数据库操作方法（在数据层中），它返回一个Observable<User>。 我在 3 个不同的地方使用了这种方法。在这 3 个中，我需要在 2 个地方返回一个 Observable<User> 的方法。但是，问题出现在我需要在doOnNext()副作用运算符中访问此方法的地方。

AFAIK，doOnNext() 不会更改流并继续上一个线程。如果流已经在 io 线程上，我们可以继续新的后台操作，而无需创建/使用 Observable。

片段

getDataManager().getUser(id)
                        .doOnNext(user -> getDataManager().insertUser(user))
                        .subscribeOn(getSchedulerProvider().io())
                        .observeOn(getSchedulerProvider().ui())
                        .subscribe()

这个insertUser() 方法返回Observable<Long>，这就是我希望它用于其他部分的方式。但是对于这个副作用运算符，我需要它只返回 Long。

有什么方法可以删除 Observable 包装器吗？还是我需要创建两种不同的方法？

Answer 1

对于返回 Observable 的相关计算/IO，通常使用 flatMap 或 concatMap 运算符。

getDataManager()
.getUser(id)
.concatMap(user ->
    getDataManager()
    .insertUser(user)
)
.subscribeOn(getSchedulerProvider().io())
.observeOn(getSchedulerProvider().ui())
.subscribe()

如果您仍然需要原始的user，只需将其映射回来：

getDataManager()
.getUser(id)
.concatMap(user ->
    getDataManager()
    .insertUser(user)
    .ignoreElements()
    .andThen(Observable.just(user))
)
.subscribeOn(getSchedulerProvider().io())
.observeOn(getSchedulerProvider().ui())
.subscribe()