我有一个序列,我需要根据每个元素的条件重复该序列.例如,如果元素标记有“失败”标志,则我需要对其进行重新处理.我的问题是我找不到如何进行while循环操作.

TakeWhile几乎是我所需要的,但是不会重复.

/*
* The following lines are just an example to comprehend the idea
*/
var observableSequence = sequence.ToObservable();
observableSequence
    //This ´DoWhile´ did not worked because does not accept each element as argument
    //and sequence at this point is not the same as `observableSequence`
    .DoWhile(() => sequence.Any(item => !item.Failed))
    .Where(item => item.Failed == true) //OK here i could put another condition for limited retries...
    .Subscribe(item => {
        try{
            //Do stuff...
            //. . .
            item.Failed = false;
        } catch
        {
            item.Failed = true;
        }
    });

解决方法:

我建议将原始序列与一个新的可观察对象合并,以便在对象出现故障时将其输入.

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);
    });

更改可观察对象的状态通常被认为是不好的做法,因此您可能需要考虑创建转换：

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

您还应该对这种模式非常小心,因为持续失败的项目会使您的程序执行陷入无限循环.