Upsample eine Rx beobachtbar basierend auf einer anderen

Question

Ich bin neu in reaktiven Erweiterungen, also bitte entschuldigen Sie meine Ignoranz. Ich habe zwei beobachtbare Objekte obj1 und obj2, die Elemente zurückgeben. In einem bestimmten Zeitfenster T (Puffer nehme ich an) möchte ich die seltenste Itemsequenz upsampeln. Siehe, ein schematisches Beispiel.

  Time of returned objects 
     obj1 t1 t2 t3 t4 t5 t6 ... 
     obj2 t1   t2  t3 ... 
    os_obj2 t1 t1 t2 t2 t2 t3 ... 
time_window <-----T-----><-----T-----> 

Im aufwärtsgesampelten Verfahren legen wir obj2 Artikel im selben Ort (Zeit) kehrten die Produkte von obj1 existieren.

Die Auswahl von dem, was obj2 Artikel wir auf einem bestimmten obj1 Artikel Zeit ist abhängig von dem obj2 Artikel ist näher an der spezifischen obj1 Artikel Zeit replizieren soll.

Könnten Sie bitte einen Weg vorschlagen (mögliche Funktionen oder eine Pipeline von Funktionen), um dies zu erreichen?

Danke.

Answer 1

BEARBEITEN Mir ist gerade aufgefallen, dass Sie Objekte aussenden wollen, die zuvor von anderen Observablen nicht emittiert wurden. Dies kann nicht erreicht werden, da es Voraussagen für die Zukunft erfordert.

Wenn obj2 vorhersagbare Elemente zu regelmäßigen Zeiten aussendet, können Sie etwas tun. Make obj2 emittieren diese Elemente eine halbe Periode früher, so können Sie nur die neuesten Artikel von obj2 verwenden, und müssen keine zukünftigen Artikel voraussagen.

Das folgende Beispiel gibt die neuesten Artikel aus obj2 aus, zu Zeiten von obj1. Am einfachsten ist es, Observable.withLatestFrom zu verwenden.

var obj1 = Rx.Observable.interval(500).map(x => "obj1." + x).take(10); 
 
var obj2 = Rx.Observable.interval(300).map(x => "obj2." + x).take(10); 
 

 
obj2.withLatestFrom(obj1).pluck(1).subscribe(x => console.log(x));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.0-beta.11/Rx.umd.js"></script>

PS: Zuerst dachte ich, sample verwendet werden kann, aber es hat für mich nicht für Überabtastungs zu arbeiten.

Answer 2

Jeder Wert in einem beobachtbaren Stream hat zwei Elemente: Die Zeit, in der er angezeigt wird, und der Wert. Also lassen Sie uns Ihren Schaltplan ändern leicht:

Timeline 0--1--2--3--4--5--6--7--8... 
XObserve x0-x1-x2-x3-x4-x5-x6-x7-x8... 
YObserve y0-------y1-------y2------... 

So haben wir zwei Observablen, gibt ein Wert jede Sekunde (der Einfachheit halber sagen wir mal die Timeline eine zweite Skala ist), die andere sendet einen Wert alle drei Sekunden. Wenn Sie all dies wünschen, möchten Sie, dass ein dritter Stream ZObserve eine Upsampling-Technik verwendet, um Werte für den Y-Stream mit der Geschwindigkeit des X-Streams zu erzeugen, so?

Timeline 0--1--2--3--4--5--6--7--8-... 
XObserve x0-x1-x2-x3-x4-x5-x6-x7-x8... 
YObserve y0-------y1-------y2------... 
ZObserve y0-y0-y1-y1-y1-y2-y2-y2-??... 

Wie Sie den Wert für ZObserve bei T8 sehen kann, ist ??, weil ich keine Ahnung, was es ist. Ich sehe auch nicht, wie ich bei T2 und T5 hätte vorhersagen können, was der nächste Wert aus dem Y-Stream sein wird. Wenn es zuverlässige Regeln für das gibt, was sie sein werden, können wir sie entsprechend codieren, einen neuen Stream erzeugen PredictedY und WithLatestFrom als @Tamas verwenden, um den Z-Stream zu erzeugen.

Answer 3

Ich schlage vor, CombineLatest Operator zu verwenden und vielleicht zur Probe danach