RxJava - Gegenteil von switchMap() Operator?

Question

Ich frage mich, ob es eine Möglichkeit gibt, bestehende Operatoren zu komponieren, um das Gegenteil von switchMap() durchzuführen.

Die switchMap() wird nach der letzten Emission jagen, die sie erhält, und alle Observable löschen, die sie zuvor ausgeführt hat. Nehmen wir an, ich habe es umgedreht, und ich möchte alle Emissionen ignorieren, die zu einem Operator kommen, während er mit der ersten empfangenen Emission beschäftigt ist. Es wird die Emissionen ignorieren, bis es den aktuellen Observable innerhalb davon ausstößt. Dann verarbeitet es die nächste Emission, die es empfängt.

Observable.interval(1, TimeUnit.SECONDS) 
     .doOnNext(i -> System.out.println("Source Emitted Value: " + i)) 
     .ignoreWhileBusyMap(i -> doIntensiveProcess(i).subcribeOn(Schedulers.computation())) 
     .subscribe(i -> System.out.println("Subscriber received Value: " + i)); 

Gibt es eine Möglichkeit, dies zu erreichen? In dem obigen Beispiel, wenn intensiveProcess() waren drei Sekunden dauern, würde die ignoreWhileBusyMap()0 verarbeiten aber wahrscheinlich Emissionen 1 und 2 von interval() kommen ignorieren würde .Es dann 3 verarbeitet aber wahrscheinlich ignorieren 4 und 5, und so weiter ...

Answer 1

sicher, Tor die Verarbeitung eines Wertes von einem boolean, die festgelegt wird, nachdem die Verarbeitung beendet:

AtomicBoolean gate = new AtomicBoolean(true); 

Observable.interval(200, TimeUnit.MILLISECONDS) 
.flatMap(v -> { 
    if (gate.get()) { 
     gate.set(false); 

     return Observable.just(v).delay(500, TimeUnit.MILLISECONDS) 
       .doAfterTerminate(() -> gate.set(true)); 
    } else { 
     return Observable.empty(); 
    } 
}) 
.take(10) 
.toBlocking() 
.subscribe(System.out::println, Throwable::printStackTrace); 

bearbeiten

Alternative:

Observable.interval(200, TimeUnit.MILLISECONDS) 
.onBackpressureDrop() 
.flatMap(v -> { 
    return Observable.just(v).delay(500, TimeUnit.MILLISECONDS); 
}, 1) 
.take(10) 
.toBlocking() 
.subscribe(System.out::println, Throwable::printStackTrace); 

Sie können onBackpressureDrop-onBackpressureLatest ändern sich mit dem letzten Wert weiterhin sofort.

Answer 2

Um Jeopardy Stil zu beantworten: Was ist concatMap?

concatMap wird zur ersten Observable abonnieren und werden nicht an nachfolgende Observable s, bis die vorherigen Observable Anrufe onComplete() abonnieren.

In dieser Hinsicht ist es das "Gegenteil" von switchMap, das eifrig von vorherigen Observable s abmeldet, wenn ein neues kommt.

concatMap will alles hören, dass jede Observable zu sagen hat, während switchMap ein sozialer Schmetterling ist, und bewegt sich, sobald eine andere beobachtbare verfügbar ist.