使用RxJava定时器和超时嵌套订阅

使用RxJava2，我试图完成一些定时事件。使用RxJava定时器和超时嵌套订阅

我有一些异步事件可能会在不到一秒钟内完成，我想在继续使用异步操作的结果之前向用户显示至少一秒的消息（也许可以使用后面的结果）。

异步操作也可能超时，我想显示一条消息，而不是继续。

我可以通过使用zip()将定时器作为第一个参数，并将异步操作作为第二个操作符来完成此任务，但下一个“图层”应该如何处理？

这是我的代码到目前为止，这实际上确实工作，但我觉得很肮脏创建嵌套订阅（取代异步操作的使用just()，而忽略订阅线程）

mStrings只是一个BehaviorSubject<String>。

mStrings.onNext("Waiting 1 second. The result will fire at the Single.timer onComplete"); 
Single.zip(Single.timer(1, TimeUnit.SECONDS), Single.just("First"), (t1, t2) -> t2) 
    .timeout(15, TimeUnit.SECONDS, observer -> { 
     mStrings.onNext("First timeout fired"); 
    }) 
    .subscribe(s1 -> { 
     mStrings.onNext("First timer fired and returned " + s1); 
     Single.zip(Single.timer(1, TimeUnit.SECONDS), Single.just("Second"), (t1, t2) -> t2) 
     .timeout(15, TimeUnit.SECONDS, observer -> { 
      mStrings.onNext("Second timeout fired"); 
     }) 
     .subscribe(s2 -> { 
      mStrings.onNext("Second timer fired and returned " + s2 + ". Previous was " + s1); 
      Single.zip(Single.timer(1, TimeUnit.SECONDS), Single.just("Third"), (t1, t2) -> t2) 
       .timeout(15, TimeUnit.SECONDS, observer -> { 
        mStrings.onNext("Third timeout fired"); 
       }) 
       .subscribe(s3 -> { 
        mStrings.onNext("Third timer fired and returned " + s3 + ". Previous was " + s1 + " and " + s2); 
       }); 
     }); 
    });

其结果是：

17:53:53.219 Waiting 1 second. The result will fire at the Single.timer onComplete 
17:53:54.220 First timer fired and returned First 
17:53:55.224 Second timer fired and returned Second. Previous was First 
17:53:56.224 Third timer fired and returned Third. Previous was First and Second

我缺少一个操作符将是有意义的这种类型的流量？或者一些基本的方法？我知道我可以使用多个主题制定出一个替代解决方案，但似乎过分。

来源

2017-05-09 Tim K

因此，您想每隔1秒触发一次异步操作，并每隔1秒显示一次状态消息，以显示以前的所有异步操作结果？如果asycn结果在1秒后返回，该怎么办？ – yosriz

我想我解决了我自己的问题。 flatMap可用于将所有订阅压缩为一个，并且每个个人timer和timeout仍将按预期运行。

mStrings.onNext("Waiting 1 second. The result will fire at the Single.timer onComplete"); 
Single.zip(Single.timer(1, TimeUnit.SECONDS), Single.just("First"), (t1, t2) -> t2) 
    .timeout(2, TimeUnit.SECONDS, observer -> { 
     mStrings.onNext("First timeout fired"); 
    }) 
    .flatMap(s1 -> { 
     mStrings.onNext("First timer fired and returned " + s1); 
     return Single.zip(Single.timer(1, TimeUnit.SECONDS), Single.just("Second"), (t1, t2) -> t2) 
     .timeout(15, TimeUnit.SECONDS, observer -> { 
      mStrings.onNext("Second timeout fired"); 
     }) 
     .flatMap(s2 -> { 
      mStrings.onNext("Second timer fired and returned " + s2 + ". Previous was " + s1); 
      return Single.zip(Single.timer(1, TimeUnit.SECONDS), Single.just("Third"), (t1, t2) -> t2) 
       .timeout(15, TimeUnit.SECONDS, observer -> { 
        mStrings.onNext("Third timeout fired"); 
       }) 
       .flatMap(s3 -> { 
        mStrings.onNext("Third timer fired and returned " + s3 + ". Previous was " + s1 + " and " + s2); 
        return Single.just("Fourth"); 
       }); 
     }); 
    }) 
    .subscribe(s -> { 
     // ignored 
    }); 
}));

复杂的嵌套仍然可以通过在逻辑上更多地构成Observables来简化。

来源

2017-05-09 17:31:38

使用RxJava定时器和超时嵌套订阅

回答

相关问题