Wie kann ich die beobachtbare Sequenz basierend auf der Bedingung von jedem Element ablaufen lassen?

Question

Ich habe eine Sequenz und ich muss es basierend auf der Bedingung von jedem Element wiederholen. Zum Beispiel muss ich ein Element erneut verarbeiten, wenn es mit einem Flag "fehlgeschlagen" markiert ist. Mein Problem ist, ich kann nicht finden, wie man den while-loop Betrieb tut.

TakeWhile ist fast das, was ich brauche, aber es wiederholt sich nicht.

Answer 1

Ich würde vorschlagen, die ursprüngliche Sequenz mit einer neuen Observablen, die Sie Objekte füttern, wenn sie ausfallen.

loopSequence 
    .Where(item => item.Failed) 
    .Subscribe(item => 
    { 
     try{ 
      //Do stuff 
      retries.OnNext(new Item { ..., Failed = false }); 
     } catch 
     { 
      retries.OnNext(new Item { ..., Failed = true }); 
     } 
    }); 

Sie sollen auch wirklich vorsichtig sein mit diesem Muster, da ein ständig: Statt

var retries = new ReplaySubject<Foo>(); 
var loopSequence = sequence.ToObservable().Merge(retries); 

loopSequence 
    .Where(item => item.Failed) 
    .Subscribe(item => 
    { 
     try{ 
      //Do stuff 
      item.Failed = false; 
     } catch 
     { 
      item.Failed = true; 
     } 
     retries.OnNext(item); 
    }); 

Es ist allgemein schlechte Praxis betrachtet Zustand der Objekte in den Observablen zu ändern, so dass Sie Transformationen vielleicht prüfen Schaffung Ein fehlendes Element wird Ihre Programmausführung in eine Art Endlosschleife bringen.

Answer 2

Sie können jeden Wert wickeln und die Operationen unabhängig wiederholen:

observableSequence 
    .SelectMany(item => 
    Observable.Return(item) 
     .Select(x => //Do Stuff) 
     //Optional argument, omitting retries infinitely until 
     //success 
     .Retry(3) 
) 

    .Subscribe(item => { 
    //Handle the results 
    }); 

Optional Return eine IScheduler als zweites Argument, das die Reihenfolge ändern können, dass Wiederholungen (rekursiv vs. Trampolin) verarbeitet bekommen.