Wie Sequenz von Ereignissen in einer Reihe mit RxJava durch seinen Wert filtern

Question

Ich produziere einen Strom von Ereignissen mit rxjava2 wie folgt.

Observable<Integer> observable = booleanPublisher.map(aBoolean -> { 
     if(aBoolean){ 
      return 1; 
     }else{ 
      return 0; 
     } 
    }).buffer(4).map(aLilst ->{ 
     int sum = 0; 
     for (Integer i: aLilst) { 
      sum +=i; 
     } 
     return sum; 
    }); 
    observable.subscribe(
      aInt ->{ 
       Log.v("Value",String.valueOf(aInt)); 
      } 
    ); 

Dies erzeugt einen Strom von int Ereignissen mit Werten von 0..4, Jetzt würde Ich mag diesen Strom filtern, ein Ereignis zu emittieren, wenn beispielsweise 2 ganzzahlige Ereignisse in einer Reihe sind kleiner als 4, und ein anderer Ereignis, wenn 10 Integer-Ereignisse in einer Zeile mit 4 Werten vorhanden sind. Ich versuche, in 2 verschiedenen Observablen und Verschmelzungen zu trennen, hatte aber kein Glück, da diese 2 Ereignisse sich gegenseitig zurücksetzen müssen.

Vielen Dank im Voraus.

Answer 1

Eine weitere Möglichkeit zur Vermeidung von zwei verschiedenen Abonnements zu verwenden, ist merge, auf diese Weise zu verwenden:

enum MatchResult{ 
    TWO_ITEM_SMALL, 
    FIVE_CONSECUTIVE 
} 


Observable<MatchResult> getObservable1(Observable<Integer> observable) { 
     return observable 
       .scan(0, (integer, integer2) -> { 
        if (integer2 < 4) { 
         return integer + 1; 
        } 
        return 0; 
       }).filter(integer -> integer >= 2) 
       .map(integer -> MatchResult.TWO_ITEM_SMALL); 
    } 

Observable<MatchResult> getObservable2(Observable<Integer> observable) { 
    return observable 
      .scan(0, (integer, integer2) -> { 
       if (integer2 == 4) { 
        return integer + 1; 
       } else { 
        return 0; 
       } 
      }).filter(integer -> integer >= 5) 
      .map(integer -> MatchResult.FIVE_CONSECUTIVE); 
    } 


Observable.merge(getObservable1(getSoruce()), getObservable2(getSoruce())) 
      .subscribe(matchResult -> { 
       if(matchResult == MatchResult.TWO_ITEM_SMALL) { 
        logd("two items in a row, smaller than 4."); 
       } else { 
        logd("five items in a row equals to 4"); 
       } 
      }); 

Answer 2

Eine Möglichkeit, dies zu erreichen, ist die Verwendung des Operators scan. Da die Dokumentation sagt:

Der Scan Operator wendet eine Funktion auf den ersten von der Quelle emittieren Artikel beobachtbar und gibt dann das Ergebnis dieser Funktion als eigene erste Emission.

Als Rückgabewert von der Funktion können Sie die Anzahl der Elemente in einer Zeile haben, die Ihrer Bedingung entspricht.

Ich denke, es gibt eine Möglichkeit, es in einem einzigen Abonnement zu tun, aber ich denke, es ist klarer, es zu teilen, mit einem ConnectableObservable durch den publish() Operator. Nehmen wir dieses Beispiel:

Observable<Integer> getSource() { 
    return Observable.just(1, 2, 3, 4, 4, 4, 4, 4, 2); 
} 

Jetzt erstellen wir den Beobachter, der für die zwei aufeinanderfolgenden kleinen als 4 Elemente überprüft.

Observable<Integer> source = getSource(); 

source.scan(0, (integer, integer2) -> { 
    if (integer2 < 4) { 
     return integer + 1; 
    } 
    return 0; 
    }).filter(integer -> integer >= 2) 
    .subscribe(integer -> logd("two items in a row, smaller than 4.")); 

und der Teilnehmer, die die Ereignisse für fünf (zur Vereinfachung) aufeinanderfolgende Werte gleich zu 4.

source.scan(0, (integer, integer2) -> { 
    if (integer2 == 4) { 
     return integer + 1; 
    } 
    return 0; 
    }).filter(integer -> integer >= 5) // for you it will be ten 
    .subscribe(integer -> logd("five items in a row equals to 4.")); 

Und jetzt Sie gerade anrufen erhält:

observable.connect(); 

HINWEIS: Die aktuelle Implementierung setzt den Zähler nicht zurück, wenn die Bedingung übereinstimmt. Dies bedeutet, dass, zum Beispiel, dass, wenn Sie für eine Folge von fünf aufeinanderfolgenden Werten suchen gleich zu 4 diese Sequenz:

4 4 4 4 4 4 

werden zwei Spiele erzeugen, weil die erste Übereinstimmung mit den ersten fünf Elementen der Sequenz ist , aber wenn Sie einen neuen Gegenstand mit dem Wert 4 erhalten, haben Sie immer noch eine Übereinstimmung.