Rxjs 5: Wie eine Kette von Observable zu etablieren?

Question

Ich bin Anfänger zu RxJS, ich versuche, eine Kette von Observablen zu implementieren, die sich genau wie MS-Excel verhalten. Das Konzept: Nehmen wir an, das Excel hat 5 Spalten: Name, Alter, Geschlecht, Land, Postleitzahl. Wir können Filter für jede Spalte unabhängig anwenden, die sich auch auf die Datensätze auswirken, die in den anderen Spalten angezeigt werden.

Hier erhält die Datenquelle den Daten-Back-End-Dienst, die Datenquelle hat nur zwei Funktionen "addRecord" & "removeRecord".

Wie ich hier erreichen möchte, sagen wir, dass ich Observable erstellen und an die Datenquelle anhängen werde es als OBS-1 nennen, das Daten von der Datenquelle erhalten wird. Das OBS-1 kann eigene Filter haben. Sagen wir, ich werde eine andere Observable OBS-2 erstellen, die Daten OBS-1 erhalten wird (gefilterte Daten wenn irgendwelche Filter in OBS-1). Ein anderes Observable sagt OBS-3, die wieder Daten von OBS-2 erhalten (gefiltert, wenn überhaupt in OBS-2), und so weiter.

Wenn OBS-2 zerstört (nicht abonniert) wird das OBS-3 die Daten von OBS-1 erhalten.

Wie erreichen wir dies in den RxJs?

Answer 1

Ich denke, Sie haben ein paar Dinge über Rx missverstanden. Observables haben keine Filter und Sie können nicht "live" Filter hinzufügen oder entfernen. Observables leiten Daten auch nicht basierend darauf weiter, wer abonniert ist.

Stattdessen bauen Sie eine Anrufkette auf. Sie beginnen mit einer Quelle, die beobachtbar ist, wie eine aus dem addRecord und eine aus dem removeRecord Ereignis. Dann verketten Sie diese Observablen, um neue Observable durch verschiedene operators in Rx zu bilden, und schließlich abonnieren Sie die endgültige beobachtbare. Das Abonnieren aktiviert die gesamte Kette und wenn die Quellenereignisse ausgelöst werden, werden alle Operatoren ausgelöst und das Ereignis wird (falls nicht gefiltert) das subscribe erreichen.

Sie können tatsächlich das tun, was Sie in Rx beschreiben. Das Ändern eines Filters auf einer Observablen kann zum Beispiel relativ einfach mit switchMap durchgeführt werden, einem Operator, mit dem Sie eine Sequenz auf eine andere projizieren und jedes Mal auf die neue Sequenz wechseln können. Zum Beispiel filterSource.switchMap(filterFunction => Obs-1.filter(filterFunction)). Noch einfacher: Sie können das erste Abonnement einfach abbestellen und die Rx-Kette erneut einrichten. Die Verwendung der eingebauten Funktionen lässt jedoch viele Jonglierzustände aus der Gleichung heraus.

Allerdings vermute ich stark, dass Sie nicht wirklich Verhalten benötigen, das so kompliziert ist. Was Sie wollen, kann einfach so archiviert werden:

var Src-1 = fromEvent(dataSource, 'addRecord') // create the first source 
var Src-2 = fromEvent(dataSource, 'removeRecord') // and the other source 
var Obs-1 = Src-1.combineLatest(Src-2) // combine both sources 
    .filter(e => someCondition(e)) // filter the source 
var Obs-2 = Obs-1.mergeMap(e => someOtherCondition(e) ? Change(e) : Rx.Observable.of(e)) // on someOtherCondition, either transform the source with the `Change(e)` function. Or keep it unchanged with `of(e)` 
var Obs-3 = Obs-2.filter(e => anotherCondition(e)) // Filter again 
var sub = Obs-3.subscribe() // activate the sequence.