Festlegen eines Zeitlimits, bis die Daten von Observable

Question

Es gibt eine rx.Observable.timeout, aber es ist pro Element und nicht für alle Daten. Aber ich brauche eine Zeitüberschreitung, wenn Daten nicht innerhalb eines bestimmten Zeitraums ausgegeben werden.

final Subscriber<Integer> sub = new Subscriber<Integer>() { 
      @Override 
      public void onNext(Integer num) { 


       try { 
        Thread.sleep(100); 
       } catch (InterruptedException e) {} 
       System.out.println(num); 
      } 

      @Override 
      public void onCompleted() { 
       System.out.println("--------------- Done ---------------"); 
      } 

      @Override 
      public void onError(Throwable e) { 
       //System.out.println("Error!!! + e"); 
      } 
     }; 

     Observable.range(1, 10000) 
     .map(x -> x * 10) 
     //.last() 
     .timeout(10, TimeUnit.SECONDS) 
     .doOnError(e -> {System.out.println("Errrorrr " + e);}) 
     .subscribe(sub); 

Answer 1

Sie takeUntil(Observable) mit timer() in eine Ausnahme abgebildet verwenden:

source.takeUntil(
    Observable.timer(10, TimeUnit.SECONDS) 
    .flatMap(ignored -> Observable.error(new TimeoutException())) 
) 
.subscribe(/* ... */); 

Answer 2

Verwenden Sie zwei Quellen: eine, die Ihre Daten ausgibt, eine, die Ihre Zeitüberschreitungsausnahme nach einer Verzögerung auslöst. Zusammenführen der beiden:

Observable timeoutObservable = Observable.error(new TimeoutException()) 
    .delay(10, TimeUnit.SECONDS, true); 
return Observable.merge(dataObservable, timoutObservable);