RxJava + Retrofit 轮询

Question

RxJava + Retrofit 轮询

22

我的问题是我无法使用Retrofit获取无限流。在获取初始poll()请求的凭据后，我进行了初始poll()请求。如果没有更改，则每个poll()请求会在25秒后响应，或者如果有任何更改，则会更早地响应-返回changed_data[]。每个响应都包含下一个poll请求所需的timestamp数据-我应该在每个poll()响应后进行新的poll()请求。这是我的代码：

getServerApi().getLongPollServer() 
  .flatMap(longPollServer -> getLongPollServerApi(longPollServer.getServer()).poll("a_check", Config.LONG_POLLING_SERVER_TIMEOUT, 2, longPollServer.getKey(), longPollServer.getTs(), "") 
   .take(1) 
   .flatMap(longPollEnvelope -> getLongPollServerApi(longPollServer.getServer()).poll("a_check", Config.LONG_POLLING_SERVER_TIMEOUT, 2, longPollServer.getKey(), longPollEnvelope.getTs(), ""))) 
   .retry()
   .subscribe(longPollEnvelope1 -> {
   processUpdates(longPollEnvelope1.getUpdates());
});

我是RxJava的新手，可能有些地方还不太理解，但是我无法获取无限流。我只能获得3个调用，然后是onNext和onComplete。

另外，也许有更好的解决方案来实现Android上的长轮询吗？

- localhost

在你的情况下，我会考虑使用Observable.create()来实现自己的“Observable”。 - Dmitry Zaytsev

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- snodnipper · Accepted Answer

虽然不是理想的做法，但我认为你可以使用 RX 的副作用来实现所需的结果（'doOn' 操作）。

Observable<CredentialsWithTimestamp> credentialsProvider = Observable.just(new CredentialsWithTimestamp("credentials", 1434873025320L)); // replace with your implementation

Observable<ServerResponse> o = credentialsProvider.flatMap(credentialsWithTimestamp -> {
    // side effect variable
    AtomicLong timestamp = new AtomicLong(credentialsWithTimestamp.timestamp); // computational steering (inc. initial value)
    return Observable.just(credentialsWithTimestamp.credentials) // same credentials are reused for each request - if invalid / onError, the later retry() will be called for new credentials
            .flatMap(credentials -> api.query("request", credentials, timestamp.get()))  // this will use the value from previous doOnNext
            .doOnNext(serverResponse -> timestamp.set(serverResponse.getTimestamp()))
            .repeat();
})
        .retry()
        .share();

private static class CredentialsWithTimestamp {

    public final String credentials;
    public final long timestamp; // I assume this is necessary for you from the first request

    public CredentialsWithTimestamp(String credentials, long timestamp) {
        this.credentials = credentials;
        this.timestamp = timestamp;
    }
}

当订阅'o'时，内部可观察对象将重复。如果出现错误，则'o'将重试并从凭据流重新发出请求。

在您的示例中，通过更新时间戳变量来实现计算引导，这对于下一个请求是必要的。