Erhalten Sie Daten einmal von 2 Observablen

Question

Ich habe 2 Observables, und ich muss jedes beobachtbare Daten nur einmal erhalten.

Was ich getan habe, ist Abonnement in einem Abonnement, obwohl sie in der gleichen Zeit (parallel) ausgeführt werden können.

Der obige Code funktioniert gut, aber sie werden nicht in der gleichen Zeit ausgeführt, obwohl sie können.

Wie kann ich beide Observablen gleichzeitig ausführen und dann die von beiden empfangenen Daten verwenden?

Update: Mit forkJoin() nichts zu tun (kein Konsolenprotokoll aus dem folgenden Code)

var source = Observable.forkJoin(
    this.af.database.object('tables'), 
    this.af.database.object('sections') 
); 

var subscription = source.subscribe(
    function (x) { 
    console.log('Next: %s', x); 
    }, 
    function (err) { 
    console.log('Error: %s', err); 
    }, 
    function() { 
    console.log('Completed'); 
    }); 

Answer 1

eine Reihe von Möglichkeiten Es gibt (lesen, Operatoren), die Sie Quellen kombinieren helfen. Eine allgemeine Ressource kann here in dem Abschnitt Combining multiple observable sequences into a single sequence gefunden werden. Für Ihr spezielles Problem ist hier die kurze Liste, die den Schnitt zu machen scheint:

• combineLatest: sendet einen Wert, wenn 1. Beide Quellen mindestens einen Wert emittiert haben, 2. danach, jederzeit eine der Quellen gibt einen Wert aus.

Zum Beispiel: Rx.Observable.combineLatest(object('tables'), object('sections').take(1))

• withLatestFrom: gibt einen Wert an, wenn die erste Quelle einen Wert abgibt. Der emittierte Wert enthält den letzten emittierten Wert der zweiten Quelle.

Zum Beispiel: object('tables').withLatesFrom(object('sections').take(1))

• zip: aussendet ersten Wert, wenn die beiden Quellen emittiert einen Wert haben. gibt über denselben Prozess den N-ten Wert ab.

Zum Beispiel: Rx.Observable.zip(object('tables'), object('sections').take(1))

haben Jeder dieser Operatoren ähnlich, aber etwas andere Semantik. Ich würde annehmen, dass combineLatest ist, was Sie brauchen, aber überprüfen Sie das Dokument, richten Sie etwas Code ein und probieren Sie etwas aus und wenn das nicht funktioniert, kommen Sie zurück und posten Sie es hier.

Answer 2

Sie withLatestFrom Operator verwenden sollten.

sehen hier mehr auf withLatestFrom

/* Have staggering intervals */ 
var source1 = Rx.Observable.interval(140) 
    .map(function (i) { return 'First: ' + i; }); 

var source2 = Rx.Observable.interval(50) 
    .map(function (i) { return 'Second: ' + i; }); 

// When source1 emits a value, combine it with the latest emission from source2. 
var source = source1.withLatestFrom(
    source2, 
    function (s1, s2) { return s1 + ', ' + s2; } 
).take(4); 

var subscription = source.subscribe(
    function (x) { 
     console.log('Next: ' + x.toString()); 
    }, 
    function (err) { 
     console.log('Error: ' + err); 
    }, 
    function() { 
     console.log('Completed'); 
    }); 

// => Next: First: 0, Second: 1 
// => Next: First: 1, Second: 4 
// => Next: First: 2, Second: 7 
// => Next: First: 3, Second: 10 
// => Completed 

Answer 3

Die Dokumentation für forJoin erklärt es:

Läuft alle beobachtbaren Sequenzen parallel und ihre letzten Elemente sammeln.

Das letzte Element ich das letzte Element vor einer beobachtbaren abgeschlossen ist gefeuert übernehmen würde, so dass alle Observablen, die Sie forkJoin passieren muss abgeschlossen sein, bevor irgendetwas emittiert wird.

Da Sie nur das erste Element emittieren wollen versuchen, jede beobachtbare komplett nach seinem ersten emit zu machen:

var source = Observable.forkJoin(
    this.af.database.object('tables').take(1), 
    this.af.database.object('sections').take(1) 
); 

Answer 4

versuchen Sie dies:

var source = Observable.forkJoin(
     this.af.database.object('tables'), 
     this.af.database.object('sections') 
    ); 

var subscription = source.subscribe(
data => { 
     console.log(data[0]); 
     console.log(data[1]); 
     }, 
     err => console.error(err) 
);