android – 在RxJava中使用“skipWhile”和“repeatWhen”来实现服务器轮询

概述我真的很喜欢Rx Java,它是一个很棒的工具,但有些时候很难理解它是如何工作的. 我们在 Android项目中使用Retrofit和RxJava,并且有以下用例： 我需要轮询服务器,重试之间有一些延迟,而服务器正在做一些工作.服务器完成后,我必须提供结果.所以我用RxJava成功完成了它,这里是代码片段： 我用“skipWhile”和“repeatWhen” Subscription check 我真的很喜欢Rx Java,它是一个很棒的工具,但有些时候很难理解它是如何工作的.
我们在 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是开源的,我只能阅读代码,但正如我所说 – 这真的很难理解.

谢谢.

解决方法 我以前没有和repeat一起工作,但是这个问题让我很好奇,所以我做了一些研究.

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 truesource    .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秒重复一次,直到源发出真实.我一直在采取,直到结果是真的.并且我过滤掉所有错误的通知,因此订阅者直到它为止才会获得它们.

总结

以上是内存溢出为你收集整理的android – 在RxJava中使用“skipWhile”和“repeatWhen”来实现服务器轮询全部内容，希望文章能够帮你解决android – 在RxJava中使用“skipWhile”和“repeatWhen”来实现服务器轮询所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。