我正在慢慢地将我的应用程序从 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...
}
}));