Ich verwende den Operator withLatestFrom in RxJS wie gewohnt:
... um die neuesten Emissionen von source2 und source3 aktiv zu sammeln und alle drei Werte nur dann auszugeben, wenn source1 emittiert.
Ich kann jedoch nicht garantieren, dass source2 oder source3 Werte erzeugt haben, bevor source1 einen Wert erzeugt. Stattdessen muss ich warten, bis alle drei Quellen jeweils mindestens einen Wert erzeugen, bevor withLatestFrom ihr Ding machen darf.
Der Vertrag muss sein: wenn source1 ausgibt, dann wird combined immer , wenn die anderen Quellen schließlich produzieren. Wenn source1 während des Wartens auf die anderen Quellen mehrmals ausgibt, können wir den letzten Wert verwenden und die vorherigen Werte verwerfen. Bearbeiten: als Marmordiagramm:
Ich kann einen benutzerdefinierten Operator dafür erstellen, aber ich möchte sicherstellen, dass ich keine offensichtliche Möglichkeit verpasse, dies mit den Vanilla-Operatoren zu tun. Es fühlt sich fast so an, als ob ich combineLatest für die initiale Emission und dann für den Wechsel von auf withLatestFrom von da an möchte, aber ich konnte nicht herausfinden, wie das geht.
Bearbeiten: Vollständiges Codebeispiel aus der endgültigen Lösung:
%Vor% Ich denke, die Antwort ist mehr oder weniger wie Sie beschrieben haben, lassen Sie den ersten Wert ein combinedLatest , dann wechseln Sie zu withLatestFrom . Mein JS ist verschwommen, aber ich denke, es würde ungefähr so aussehen:
Sie sollten wahrscheinlich publish verwenden, um mehrere Abonnements zu vermeiden, so dass das wie folgt aussieht:
oder mit Pfeilfunktionen ...
%Vor%BEARBEITEN :
Ich sehe das Problem mit concat , das withLatestFrom bekommt die Werte nicht. Ich denke, das Folgende würde funktionieren:
... also nimm einen Wert mit combineLatest und dann den Rest mit withLatestFrom .
Ich war mit der angenommenen Antwort nicht ganz zufrieden, also fand ich eine andere Lösung. Viele Möglichkeiten, eine Katze zu häuten!
Mein Anwendungsfall umfasst nur zwei Streams - einen "Requests" -Stream und einen "Tokens" -Stream. Ich möchte Anfragen so schnell wie sie empfangen werden mit dem was auch immer der neueste Token ist. Wenn noch kein Token vorhanden ist, sollte gewartet werden, bis das erste Token angezeigt wird, und dann alle ausstehenden Anfragen auslösen.
Ich war mit der angenommenen Antwort nicht ganz zufrieden, also fand ich eine andere Lösung. Im Wesentlichen habe ich den Anfragestrom in zwei Teile aufgeteilt - vor und nach dem ersten Token. Ich puffern den ersten Teil und dann wieder alles auf einmal, sobald ich weiß, dass der Token-Stream nicht leer ist.
%Vor%Sehen Sie es hier live: Ссылка
Verwenden Sie combineLatest und filter , um Tupel zu entfernen, bevor der erste vollständige Satz gefunden wurde, und setzen Sie eine Variable, um die Filterung zu stoppen. Die Variable kann im Rahmen eines Wrapping defer sein, um Dinge richtig zu machen (Support Resubscription). Hier ist es in Java (aber die gleichen Operatoren existieren in RxJs):
Ich hatte ähnliche Anforderungen, aber für nur zwei Observable. Am Ende habe ich switchMap + verwendet:
%Vor%Also es:
In meinem Fall ist das zweite Observable ein ReplaySubject. Ich bin nicht sicher, ob es mit anderen beobachtbaren Typen funktioniert.
Ich denke das:
Ich war überrascht, dass withLatestFrom nicht auf das zweite Observable wartet.
Tags und Links rxjs