我们在 Android项目中使用Retrofit和RxJava,并且有以下用例：
我需要轮询服务器,重试之间有一些延迟,而服务器正在做一些工作.服务器完成后,我必须提供结果.所以我用RxJava成功完成了它,这里是代码片段：
我用“skipWhile”和“repeatWhen”
- Subscription checkJobSubscription = mDataManager.checkJob(prepareTweakJob)
- .skipWhile(new Func1<CheckJobResponse,Boolean>() {
- @Override
- public Boolean call(CheckJobResponse checkJobResponse) {
- boolean shouldSkip = false;
- if (SHOW_LOGS) Logger.v(TAG,"checkJob,skipWhile,jobStatus " + checkJobResponse.getJobStatus());
- switch (checkJobResponse.getJobStatus()){
- case CheckJobResponse.PROCESSING:
- shouldSkip = true;
- break;
- case CheckJobResponse.DONE:
- case CheckJobResponse.ERROR:
- shouldSkip = false;
- break;
- }
- if (SHOW_LOGS) Logger.v(TAG,shouldSkip " + shouldSkip);
- return shouldSkip;
- }
- })
- .repeatWhen(new Func1<Observable<? extends Void>,Observable<?>>() {
- @Override
- public Observable<?> call(Observable<? extends Void> observable) {
- if (SHOW_LOGS) Logger.v(TAG,repeatWhen " + observable);
- return observable.delay(1,TimeUnit.SECONDS);
- }
- }).subscribe(new Subscriber<CheckJobResponse>(){
- @Override
- public void onNext(CheckJobResponse response) {
- if (SHOW_LOGS) Logger.v(TAG,onSuccess,response " + response);
- }
- @Override
- public void onError(BaseError error) {
- if (SHOW_LOGS) Logger.v(TAG,onError,canEditTimeline,error " + error);
- Toast.makeText(ChoseEditOptionActivity.this,R.string.NETWORK__no_internet_message,Toast.LENGTH_LONG).show();
- }
- @Override
- public void onCompleted() {
- if (SHOW_LOGS) Logger.v(TAG,"onCompleted");
- }
- });
代码工作正常：
当服务器响应该作业正在处理时,我从“skipWhile”链返回“true”,原始Observable等待1秒并再次执行http请求.
重复此过程,直到我从“skipWhile”链返回“false”.
这是我不明白的一些事情：
>我在“skipWhile”的文档中看到它不会从原始Observable发出任何内容(onError,onNext,onComplete),直到我从其“call”方法返回“false”.所以如果它没有发出任何东西为什么“repeatWhen”Observable做它的工作呢？它等待一秒钟并再次运行请求.是谁推出的？
>第二个问题是：为什么“repeatWhen”中的Observable不能永远运行,我的意思是为什么当我从“skipWhile”返回“false”时它会停止重复？如果我返回“false”,我会在订阅者中成功登录.
>在“repeatWhile”的文档中,它说最终我在订阅者中调用了“onComplete”,但从未调用过“onComplete”.
>如果我改变链接“skipWhile”和“repeatWhen”的顺序没有区别.这是为什么 ？
我知道RxJava是开源的,我只能阅读代码,但正如我所说 – 这真的很难理解.
谢谢.
解决方法
skipWhile会发出onError和onCompleted,即使它在此之前永远不会返回true.因此,每次checkJob()发出onCompleted时都会调用repeatWhen.这回答了问题#1.
其余的问题都是基于错误的假设.您的订阅永远在运行,因为您的重复永远不会终止.那是因为repeatWhen是一个比你意识到的更复杂的野兽.其中的Observable在从源上完成onCompleted时会发出null.如果你接受并返回onCompleted然后它结束,否则如果你发出任何东西它重试.由于延迟只是发射并延迟它,它总是再次发射零.因此,它不断重新订阅.
那么#2的答案就是它永远在运行;您可能正在执行此代码之外的其他操作来取消订阅.对于#3,你永远不会完成onCompleted,因为它永远不会完成.对于#4,订单无关紧要,因为您无限期地重复.
现在的问题是,你如何得到正确的行为？它就像使用takeUntil而不是skipWhile一样简单.这样,你不断重复,直到得到你想要的结果,从而在你希望它结束时终止流.
这是一个代码示例：
- Observable<Boolean> source = ...; // Something that eventually emits true
- source
- .repeatWhen(completed -> completed.delay(1,TimeUnit.SECONDS))
- .takeUntil(result -> result)
- .filter(result -> result)
- .subscribe(
- res -> System.out.println("onNext(" + res + ")"),err -> System.out.println("onError()"),() -> System.out.println("onCompleted()")
- );
在这个例子中,源是发射布尔值.我每1秒重复一次,直到源发出真实.我一直在采取,直到结果是真的.并且我过滤掉所有错误的通知,因此订阅者直到它为止才会获得它们.
