RxJava: zwei Anrufe während der Zeitüberschreitung

Question

Meine Aufgabe ist es, den Teilnehmer anzurufen, wenn zwei oder mehr Ereignisse während der Zeitüberschreitung auftreten. Erstes Ereignis, das eine Zeitüberschreitung auslöst, wenn während der Zeitüberschreitung nichts passiert - es wird erneut mit der nächsten gestartet und so weiter. Ich verwende folgende Konstruktion:

boolean trigger = false; 
    Observable.interval(...) 
    .filter(...) 
    .timeout(10, TU.SECONDS, Observable.just(false)) 
    .subscribe(par -> 
     if (par){ 
     if (trigger){ 
      twoTimesInTimeout(); 
     } else { 
      trigger = true; 
     } 
     } else { 
     trigger = false; 
     } 
    ) 

Vielleicht gibt es eine Möglichkeit, ohne Abzug zu gehen?

Answer 1

Diese Methode sendet jede Emission erneut aus, wenn innerhalb einer bestimmten Anzahl von Millisekunden eine weitere Emission auftritt. Ich vergleiche die Zeit, in der jede nachfolgende Emission emittiert wird und emittiere, wenn das Zeitlimit erfüllt ist.

<T> Observable<T> emitIfWithin(Observable<T> obs, long millis){ 
    zip(obs.timestamp(), obs.skip(1).timestamp(), Pair::new) 
     .filter(pair -> pair.first.getTimestampMillis() - pair.first.getTimestampMillis() < millis) 
     .map(pair -> pair.first); 
} 

Und dann rufen Sie es mit Ihrem gewünschten Filter Wiederverwendung zu maximieren:

emitIfWithin(observable.ofType(A.class), 10000) 
    .subscribe(...);