Ich versuche, RxJs dazu zu bringen, eine Observable in meinem Stream zu durchlaufen, bis sie sich in einem bestimmten Zustand befindet, und dann den Stream weiterlaufen zu lassen. Insbesondere konvertiere ich eine synchrone do / while-Schleife in RxJs, aber ich gehe davon aus, dass die gleiche Antwort auch für eine for- oder while-Schleife verwendet werden kann.

Ich dachte, ich könnte doWhile () dafür verwenden, aber es scheint, dass die Bedingungsfunktion keinen Zugriff auf das Element im Stream hat, was mir den Zweck zu nehmen scheint.

Ich bin mir nicht ganz sicher, wie die richtige reaktive Terminologie für das ist, was ich will, aber hier ist ein Beispiel dafür, was ich vorhabe:

var source = new Rx.Observable.of({val: 0, counter: 3});

source.map(o => {
  o.counter--;
  console.log('Counter: ' + o.counter);

  if (!o.counter) {
    o.val = "YESS!";
  }
  return o;
})
.doWhile(o => { 
  return o.counter > 0; 
})
.subscribe(
    function (x) {
        console.log('Next: ' + x.val);
    },
    function (err) {
        console.log('Error: ' + err);   
    },
    function () {
        console.log('Completed');   
    });

Die erwartete Ausgabe wäre:

Counter: 3
Counter: 2
Counter: 1
Counter: 0
Next: YESS!
Completed

ngenommen, dies ist ein lösbares Problem. Ich bin nicht sicher, wie Sie den "Anfang" markieren, an den Sie zurückkehren möchten, wenn Sie eine Schleife ausführe