0

我正在慢慢地将我的应用程序从 RxJava 1 迁移到 RxJava 2。在更新了所有代码后一切正常，除了一个用例，我现在有点迷失了，我想我需要回到文件以正确获取它。

应用程序从网络加载一个集合，Asset并且此操作从 x 毫秒开始，它会显示加载动画。然后，当检索数据时，动画将停止/删除并显示数据。

这就是我使用 RxJava 1 所拥有的并且正在工作的东西：

getAssetsSubscription = new GetAssetsUseCase().execute()
                    .publish(new Func1<Observable<List<Asset>>, Observable<List<Asset>>>() {
                        @Override
                        public Observable<List<Asset>> call(Observable<List<Asset>> o) {
                            return o.timeout(LOADING_VIEW_THRESHOLD_MS, TimeUnit.MILLISECONDS,
                                    Observable.fromCallable(new Callable<List<Asset>>() {
                                         @Override
                                         public List<Asset> call() throws Exception {
                                             if (isAdded()) {
                                                 getActivity().runOnUiThread(new Runnable() {
                                                     @Override
                                                     public void run() {
                                                         setLoadingViewVisibility(true);
                                                     }
                                                 });
                                             }
                                             return null;
                                         }
                                     }
                                )
                            ).ignoreElements().mergeWith(o);
                        }
                    })
                    .subscribe(new Subscriber<List<Asset>>() {
                @Override
                public void onCompleted() {
                    // Do things...
                }

                @Override
                public void onError(Throwable e) {
                    // Do things...
                }

                @Override
                public void onNext(List<Asset> assets) {
                    // Do things...
                }
            });

这是我对 RxJava 2 的“翻译”：有了这个，数据永远不会显示，onComplete总是被调用，但onNext永远不会。未触发超时时也是如此。

disposables.add(new GetAssetsUseCase().execute().publish(new Function<Observable<List<Asset>>,
                    ObservableSource<List<Asset>>>() {
                @Override
                public ObservableSource<List<Asset>> apply(Observable<List<Asset>> listObservable) throws
                        Exception {
                    return listObservable.timeout(LOADING_VIEW_THRESHOLD_MS, TimeUnit.MILLISECONDS,
                            Observable.fromCallable(new Callable<List<Asset>>() {
                                @Override
                                public List<Asset> call() throws Exception {
                                    if (isAdded()) {
                                        getActivity().runOnUiThread(new Runnable() {
                                            @Override
                                            public void run() {
                                                setLoadingViewVisibility(true);
                                            }
                                        });
                                    }
                                    return null;
                                }
                            })
                    ).ignoreElements().mergeWith(Completable.fromObservable(listObservable)).toObservable();
                }
            })
            .subscribeWith(new DisposableObserver<List<Asset>>() {
                @Override
                public void onComplete() {
                    // Do things...
                }

                @Override
                public void onError(Throwable e) {
                    // Do things...
                }

                @Override
                public void onNext(List<Asset> assets) {
                    // Do things...
                }
            }));
4

2 回答 2

1

使用的原始代码Observable#mergeWith(Observable)。由于 RxJava2 在适当的情况下缩小了类型，因此在您修改后的代码中将其更改为Completable.mergeWith(Completable).

要获得与旧代码相同的行为，您需要更改操作顺序：

  • 从.ignoreElements().mergeWith(Completable.fromObservable(listObservable)).toObservable()
  • 到.ignoreElements().<List<Asset>>toObservable().mergeWith(listObservable)

因为Completable.fromObservable(...)基本上相当于Observable#ignoreElements().

此外，return null;这可能会导致 RxJava2 出现问题，因为合同规定null事件流中不能有任何值。考虑Observable.fromCallable(...)用Completable.fromRunnable(...).toObservable()

于 2016-12-14T14:33:51.420 回答
0

那是因为您的可调用返回null，这意味着它是一个终端事件。

@Override
public List<Asset> call() throws Exception {
    if(isAdded()) {
        getActivity().runOnUiThread(new Runnable() { 
            @Override
            public void run() {
                setLoadingViewVisibility(true);
            }
        });
    }
    return null;
}

应该

@Override
public List<Asset> call() throws Exception {
    if(isAdded()) {
        getActivity().runOnUiThread(new Runnable() { // TODO: move after `observeOn(AndroidSchedulers.mainThread())`
            @Override
            public void run() {
                setLoadingViewVisibility(true);
            }
        });
    }
    return Collections.emptyList();
}
于 2016-12-14T09:32:18.270 回答