RxJava上的链式调用

Question

在有些情况下，我需要连锁RxJava调用。

最简单的一个。

ViewModel:

fun onResetPassword(email: String) {
    ...
    val subscription = mTokenRepository.resetPassword(email)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeOn(Schedulers.io())
        .subscribe(
            //UI update calls
        )
    ...
}

我的仓库

fun resetPassword(email: String): Single<ResetPassword> {
        return Single.create { emitter ->

            val subscription = mSomeApiInterface.resetPassword(email)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io())
                .subscribe({
                    emitter.onSuccess(...)
                }, { throwable ->
                    emitter.onError(throwable)
                })

            ...

        }
    }

我的问题

我是否需要为这两个调用添加:

.observeOn(AndroidSchedulers.mainThread())
.subscribeOn(Schedulers.io())

还是API调用的第二个调用就足够了？

Answer 1

不，你不需要添加

.observeOn(AndroidSchedulers.mainThread())
.subscribeOn(Schedulers.io())

和viewmodel的函数。

.observeOn 通常应该在处理ui渲染之前被调用。所以通常，你需要在更新ui或发出LiveData值之前，在ViewModel中调用它。

另外，你也不需要订阅到 "ViewModel"。mSomeApiInterface 在你的repo中，我认为最好是直接返回，因为它是从你的方法向上链的，就像这样。

fun resetPassword(email: String): Single<ResetPassword> {
        return mSomeApiInterface.resetPassword(email);
    }

如果你有任何需要的映射，你可以用正常的链子把它串起来

fun resetPassword(email: String): Single<ResetPassword> {
        return mSomeApiInterface.resetPassword(email)
                .map{it -> }
    }

这样你就可以把你的ViewModel代码写成如下的样子。

fun onResetPassword(email: String) {
    ...
    // note the switcing between subscribeOn and observeOn
    // the switching is in short: subscribeOn affects the upstream,
    // while observeOn affects the downstream.
    // So we want to do the work on IO thread, then deliver results
    // back to the mainThread.
    val subscription = mTokenRepository.resetPassword(email)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            //UI update calls
        )
    ...
}

这将在io线程上运行API请求，在mainThread上返回结果，这可能是你想要的。)

这个 艺术性 有一些关于 subscribeOn 和 observeOn 的很好的例子和解释，我强烈推荐大家去看看。

Answer 2

    Observable<RequestFriendModel> folderAllCall = service.getUserRequestslist(urls.toString());
        folderAllCall.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .map(result -> result.getRequested())
                .subscribe(this::handleResults, this::handleError);


   private void handleResults(List<Requested> folderList) {
        if (folderList != null && folderList.size() != 0) {

                            usersList.addAll(folderList);

            }
            adapter.notifyDataSetChanged();
        }
    }

    private void handleError(Throwable t) {
        Toast.makeText(getContext(),t.getMessage(),Toast.LENGTH_LONG).show();
    }

在接口中。

    @Headers({ "Content-Type: application/json;charset=UTF-8"})
@GET
Observable<RequestFriendModel> getUserRequestslist(@Url String url);

POJO模型。

public class RequestFriendModel {

@SerializedName("requested")
@Expose
private List<Requested> requested = null;

public List<Requested> getRequested() {
    return requested;
}

public void setRequested(List<Requested> requested) {
    this.requested = requested;
}
}