如何使用.Net IObservable::Retry with WhenAnyValue from ReactiveUI

Question

如果我有一个 INPC 支持类 Numbers 具有两个属性 A 和 B。我可以写这样的代码

Numbers numbers = new Numbers();
IObservable<double> o = numbers.WhenAnyValue(p=>p.A,p=>p.B,(a,b)=>a/b);

WhenAnyValue 是 ReactiveUI 库中的一种实用方法，用于从属性更改事件组合 observables。如果我再写。

o.Subscribe(v=>Console.WriteLine(v));

只要 A 或 B 发生变化，它就会打印 a/b。在我设置之前这一切都很好

numbers.B = 0;

现在 a/b 将抛出 DivideByZeroException 并且 observable 将终止。然而，这是一个 UI。我不希望 observable 终止。我只是希望忽略异常或记录它并继续。第一次尝试是查看 IObservable 包含一个名为 Retry 的扩展方法，该方法将在异常后重新连接到 observable。我们尝试

Numbers numbers = new Numbers();
IObservable<double> o = numbers
  .WhenAnyValue(p=>p.A,p=>p.B,(a,b)=>a/b)
  .Retry();

o.Subscribe(v=>Console.WriteLine(v));

但是，当我执行 numbers.B = 0 时，Retry 将忽略异常并重新连接，并且会立即一次又一次地失败，因为 WhenAnyValue 总是传递一个事件订阅。

所以看来我需要的是重试，它会在重新连接后忽略第一个输入 iff 它与导致断开第一个输入的错误的输入相同，除了我不认为RX 可以做到这一点。

有什么想法吗？

完整的测试用例

以下测试用例不会终止。

   public class Numbers : ReactiveObject
    {
        int _A;
        public int A 
        {
            get { return _A; }
            set { this.RaiseAndSetIfChanged(ref _A, value); }
        }

        int _B;
        public int B 
        {
            get { return _B; }
            set { this.RaiseAndSetIfChanged(ref _B, value); }
        }
    }

    [Fact]
    public void TestShouldTerminate()
    {

        var numbers = new Numbers();
        var o = numbers
            .WhenAnyValue(p => p.A, p => p.B, Tuple.Create)
            .Select(v=>v.Item1/v.Item2)
            .Select(v=>v+1)
            .Retry();

        double value = 0;
        o.Subscribe(v => value = v);

        numbers.A = 10;
        numbers.B = 20;

        value.Should().Be(1.5);
    }
 }



    }



    }

Answer 1

这在原版 RX 中无法处理。我创建了一个名为

的包装器

IObservableExceptional
IObserverExceptional

这改变了 RX 中错误处理的标准合约。错误现在不再终止 observable。它支持 LINQ，对于大多数用途来说应该是相当透明的。通过的测试用例是

    [Fact]
    public void ErrorsCanBePropogated()
    {
        var numbers = new Numbers();

        var list = new List<double>();
        var errors = new List<Exception>();

        numbers
            .WhenAnyValue(p => p.A, p => p.B, Tuple.Create)
            .ToObservableExceptional()
            .Select(v => v.Item1/v.Item2)
            .Subscribe(onNext: val=>list.Add(val), onError:err=>errors.Add(err));


        list.Count.Should().Be(0);
        errors.Count.Should().Be(1);

        numbers.A = 10;

        list.Count.Should().Be(0);
        errors.Count.Should().Be(2);

        numbers.B = 5;

        list.Count.Should().Be(1);
        list[0].Should().Be(2.0);
        errors.Count.Should().Be(2);
    }

三个新界面是

public interface IObservableExceptional<T>
{
    void Subscribe(IObserverExceptional<T> observer);

    IObservable<IExceptional<T>> Observable { get; }
}

public interface IObserverExceptional<T>
{
    void OnNext(IExceptional<T> t);
    void OnCompleted();
    IObserver<IExceptional<T>> Observer { get; }
}

public interface IExceptional<out T> : IEnumerable<T>
{
    bool HasException { get; }
    Exception Exception { get; }
    T Value { get; }
    string ToMessage();
    void ThrowIfHasException();
}

IObservableException 和 IExceptional 都支持 LINQ（即它们是 Monad）

在 IObservableExceptional 的 Select 或 SelectMany 组合子中抛出的任何异常都被包装为 IExceptional 对象并传递给订阅者。错误不会终止订阅。

存储库位于

https://github.com/Weingartner/Exceptional