RxJava - Single不运行

Question

我正在尝试转换此AsyncTask：

private static class UpdateFeedTask extends AsyncTask<Void, Void, Void> {
    private WeakReference<FeedFragment> weakFeedFragment;

    private UpdateFeedTask(FeedFragment feedFragment) {
        this.weakFeedFragment = new WeakReference<>(feedFragment);
    }

    @Override
    protected void onPreExecute() {
        if (!weakFeedFragment.get().swipeRefreshLayout.isRefreshing()) {
            weakFeedFragment.get().feedProgressBar.setVisibility(View.VISIBLE);
        }
    }

    @Override
    protected final Void doInBackground(Void... voids) {
        weakFeedFragment.get().submissionList = weakFeedFragment.get().paginator.next();

        return null;
    }

    @Override
    protected void onPostExecute(Void value) {
        weakFeedFragment.get().feedProgressBar.setVisibility(View.INVISIBLE);
        weakFeedFragment.get().swipeRefreshLayout.setRefreshing(false);
        weakFeedFragment.get().isLoadingNextSubmissions = false;

        FeedAdapter feedAdapter = (FeedAdapter) weakFeedFragment.get().recyclerView.getAdapter();
        feedAdapter.updateItems(weakFeedFragment.get().submissionList);
    }
}

进入单个Observable，但是当我调用它时，它根本不会启动：

private void UpdateFeed() {
    if (swipeRefreshLayout.isRefreshing()) {
        feedProgressBar.setVisibility(View.VISIBLE);
    }

    updateFeedObservable()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeWith(new SingleObserver<Listing<Submission>>() {

                @Override public void onSubscribe(Disposable d) {
                    d.dispose();
                }

                @Override public void onSuccess(Listing<Submission> submissions) {
                    Log.d(TAG, "onSuccess: WORKS"); //Nope
                    feedProgressBar.setVisibility(View.INVISIBLE);
                    swipeRefreshLayout.setRefreshing(false);
                    isLoadingNextSubmissions = false;

                    FeedAdapter feedAdapter = (FeedAdapter) recyclerView.getAdapter();
                    feedAdapter.updateItems(submissions);
                }

                @Override public void onError(Throwable e) {
                    Log.e(TAG, e.getMessage());
                }
            });

}

public Single<Listing<Submission>> updateFeedObservable() {
    return Single.defer(() -> Single.fromCallable(() -> {
        Log.d(TAG, "updateFeedObservable: WORKS"); //Nope
        return paginator.next();
    }));
}

我尝试更改调度程序，但据我所知，任何IO相关需要使用Schedulers.io我不知道问题究竟在哪里，任何帮助将不胜感激

Answer 1

在new SingleObserver<Listing<Submission>>()::onSubscribe()你立即打电话给d.dispose()，所以你永远不会得到onSuccess()电话，因为当你订阅你处理SingleObserver。

删除：d.dispose()

这个：

return Single.defer(() -> Single.fromCallable(() -> {
        Log.d(TAG, "updateFeedObservable: WORKS"); // Nope
        return paginator.next();
    }));

可以简化为：

return Single.fromCallable(Paginator::next); // method reference