rx.Observable.timeout()方法的使用及代码示例

x33g5p2x  于2022-01-25 转载在 其他  
字(6.6k)|赞(0)|评价(0)|浏览(203)

本文整理了Java中rx.Observable.timeout()方法的一些代码示例，展示了Observable.timeout()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Observable.timeout()方法的具体详情如下：
包路径：rx.Observable
类名称：Observable
方法名：timeout

Observable.timeout介绍

[英]Returns an Observable that mirrors the source Observable but applies a timeout policy for each emitted item. If the next item isn't emitted within the specified timeout duration starting from its predecessor, the resulting Observable terminates and notifies observers of a TimeoutException.

Scheduler: This version of timeout operates by default on the computation Scheduler.
[中]返回一个Observable，该Observable镜像源Observable，但为每个发出的项应用超时策略。如果下一个项目没有在从其前一个项目开始的指定超时持续时间内发出，则生成的Observable将终止并通知观察者TimeoutException。
调度程序：此版本的超时默认在计算调度程序上运行。

代码示例

代码示例来源：origin: Netflix/servo

  1. final CountDownLatch completed = new CountDownLatch(1);
  2. final Subscription s = Observable.mergeDelayError(Observable.from(batches))
  3.   .timeout(timeoutMillis, TimeUnit.MILLISECONDS)
  4.   .subscribeOn(Schedulers.immediate())
  5.   .subscribe(updated::addAndGet, exc -> {

代码示例来源：origin: apache/incubator-gobblin

  1. if (callback == null) {
  2.  return new WriteResponseFuture<>(
  3.    observable.timeout(_operationTimeout, _operationTimeunit).toBlocking().toFuture(),
  4.    _defaultWriteResponseMapper);
  5. } else {
  6.  observable.timeout(_operationTimeout, _operationTimeunit)
  7.    .subscribe(new Subscriber<D>() {
  8.     @Override

代码示例来源：origin: THEONE10211024/RxJavaSamples

  1. private Observable<String> _getObservableTask_2sToComplete() {
  2.   return Observable.create(new Observable.OnSubscribe<String>() {
  3.     @Override
  4.     public void call(Subscriber<? super String> subscriber) {
  5.       _log(String.format("Starting a 2s task"));
  6.       subscriber.onNext("2 s");
  7.       try {
  8.         Thread.sleep(2000);
  9.       } catch (InterruptedException e) {
  10.         e.printStackTrace();
  11.       }
  12.       subscriber.onCompleted();
  13.     }
  14.   }).subscribeOn(Schedulers.computation()).timeout(3, TimeUnit.SECONDS);
  15. }

代码示例来源：origin: THEONE10211024/RxJavaSamples

  1. @OnClick(R.id.btn_demo_timeout_1_5s)
  2. public void onStart5sTask() {
  3.   _subscription = _getObservableFor5sTask()//
  4.      .timeout(2, TimeUnit.SECONDS, _getTimeoutObservable())
  5.      .subscribeOn(Schedulers.computation())
  6.      .observeOn(AndroidSchedulers.mainThread())
  7.      .subscribe(_getEventCompletionObserver());
  8. }

代码示例来源：origin: Q42/RxPromise

  1. /**
  2.  * Returns a promise that mirrors the source promise but applies a timeout policy.
  3.  * If the promise isn't completed within the specified timeout duration,
  4.  * the resulting promise is rejected with a {@code TimeoutException}.
  5.  */
  6. public Promise<T> timeout(long timeout, TimeUnit timeUnit) {
  7.   return new Promise<T>(observable.timeout(timeout, timeUnit));
  8. }

代码示例来源：origin: meltwater/rxrabbit

  1. @Override
  2. public Observable<Message> call(Observable<Message> input) {
  3.   final AtomicLong consumedCount = new AtomicLong(0);
  4.   return input
  5.       .doOnNext(message -> message.acknowledger.ack())
  6.       .timeout(timeout, timeUnit, just(STOP))
  7.       .takeUntil(message -> message == STOP || consumedEnough(consumedCount.incrementAndGet()) )
  8.       .filter(message -> message != STOP);
  9. }

代码示例来源：origin: MrFuFuFu/RxFace

  1. public Observable<FaceResponse> getDataPost(CustomMultipartTypedOutput listMultipartOutput) {
  2.   return mWebService.uploadImagePost(listMultipartOutput)
  3.       .timeout(Constants.TIME_OUT, TimeUnit.MILLISECONDS)
  4.       .concatMap(new Func1<FaceResponse, Observable<FaceResponse>>() {
  5.         @Override
  6.         public Observable<FaceResponse> call(FaceResponse faceResponse) {
  7.           return faceResponse.filterWebServiceErrors();
  8.         }
  9.       }).compose(SchedulersCompat.<FaceResponse>applyExecutorSchedulers());
  10. }

代码示例来源：origin: MrFuFuFu/RxFace

  1. public Observable<FaceResponse> getDataUrlGet(Map<String, String> options) {
  2.     return mWebService.uploadImageUrlGet(options)
  3.         .timeout(Constants.TIME_OUT, TimeUnit.MILLISECONDS)
  4.         .concatMap(new Func1<FaceResponse, Observable<FaceResponse>>() {
  5.           @Override
  6.           public Observable<FaceResponse> call(FaceResponse faceResponse) {
  7.             return faceResponse.filterWebServiceErrors();
  8.           }
  9.         }).compose(SchedulersCompat.<FaceResponse>applyExecutorSchedulers());//http://www.jianshu.com/p/e9e03194199e
  10.   }
  11. }

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Test
  2. public void sample_67() throws Exception {
  3.   rxLookupFlight("LOT 783")
  4.       .subscribeOn(Schedulers.io())
  5.       .timeout(100, TimeUnit.MILLISECONDS);
  6. }

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Override
  2.   public Observable<LocalDate> externalCall() {
  3.     return delegate
  4.         .externalCall()
  5.         .timeout(1, TimeUnit.SECONDS,
  6.            Observable.empty(),
  7.            scheduler);
  8.   }
  9. }

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Test
  2. public void sample_281() throws Exception {
  3.   risky()
  4.       .timeout(1, SECONDS)
  5.       .doOnError(th -> log.warn("Will retry", th))
  6.       .retry()
  7.       .subscribe(log::info);
  8. }

代码示例来源：origin: eventuate-clients/eventuate-client-java

  1. public void assertClientIsDisconnected() {
  2.   rx.Observable.interval(50, TimeUnit.MILLISECONDS)
  3.       .filter(ignore -> isClientDisconnected())
  4.       .take(1).timeout(300, TimeUnit.MILLISECONDS).toBlocking().first();
  5.  }
  6. }

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Test
  2. public void sample_249() throws Exception {
  3.   Observable
  4.       .interval(99, MILLISECONDS)
  5.       .debounce(100, MILLISECONDS)
  6.       .timeout(1, SECONDS);
  7. }

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Test
  2. public void sample_303() throws Exception {
  3.   risky()
  4.       .timeout(1, SECONDS)
  5.       .retry(10);
  6. }

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Test
  2. public void sample_72() throws Exception {
  3.   final Observable<Long> upstream = Observable.interval(99, MILLISECONDS);
  4.   upstream
  5.       .debounce(100, MILLISECONDS)
  6.       .timeout(1, SECONDS, upstream
  7.           .take(1)
  8.           .concatWith(
  9.               upstream
  10.                   .debounce(100, MILLISECONDS)
  11.                   .timeout(1, SECONDS, upstream)));
  12. }

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Test
  2. public void sample_74() throws Exception {
  3.   risky()
  4.       .timeout(1, SECONDS)
  5.       .retryWhen(failures -> failures
  6.           .zipWith(Observable.range(1, ATTEMPTS), (err, attempt) ->
  7.               attempt < ATTEMPTS ?
  8.                   Observable.timer(1, SECONDS) :
  9.                   Observable.error(err))
  10.           .flatMap(x -> x)
  11.       );
  12. }

代码示例来源：origin: eventuate-clients/eventuate-client-java

  1. public void assertSubscribed() {
  2.  frames.timeout(30, TimeUnit.SECONDS)
  3.      .filter(frame -> frame.frame().getCommand().equals(Frame.Command.SUBSCRIBE)).take(1).timeout(720, TimeUnit.SECONDS)
  4.      .toBlocking().first();
  5. }
  6. public void assertAcked(int count) {

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Test
  2. public void sample_89() throws Exception {
  3.   risky()
  4.       .timeout(1, SECONDS)
  5.       .retryWhen(failures -> failures
  6.           .zipWith(Observable.range(1, ATTEMPTS),
  7.               this::handleRetryAttempt)
  8.           .flatMap(x -> x)
  9.       );
  10. }

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Test
  2.   public void sample_66() throws Exception {
  3.     risky()
  4.         .timeout(1, SECONDS)
  5. //                .retryWhen(failures -> failures.take(10))
  6.         .retryWhen(failures -> failures.delay(1, SECONDS));
  7.   }

代码示例来源：origin: nurkiewicz/rxjava-book-examples

  1. @Test
  2. public void sample_48() throws Exception {
  3.   ConnectableObservable<Long> upstream = Observable
  4.       .interval(99, MILLISECONDS)
  5.       .publish();
  6.   upstream
  7.       .debounce(100, MILLISECONDS)
  8.       .timeout(1, SECONDS, upstream.take(1));
  9.   upstream.connect();
  10. }

内容来源于网络，如有侵权，请联系作者删除！

Observable

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多

    Observable类方法

    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com