在不停止序列的情况下处理Reactive Extensions中的异常

Question

为什么RX有以下语法OnNext* (OnError|OnCompleted)？而不是(OnNext|OnError)* OnCompleted？这是从实现的角度来看很清楚(这也与通用语义IEnumerable和yield),但我想从现实生活中的情况不同.在现实生活中 - 生产者生成混合的数据流和异常(异常不会破坏生产者).

问题:如果我理解正确,唯一可行的解​​决方案是使可观察的返回复杂数据结构从初始数据和生成的异常(Observable.Timestamp()并.TimeInterval()具有类似的概念)结合起来,还是有其他选择？

---

目前我得出了以下解决方案:在可观察的生产者内部我手动处理exeptions并将它们传递给以下数据结构,这是我可观察的结果

public class ValueOrException<T>
{
    private readonly Exception ex;
    private readonly T value;

    public ValueOrException(T value, Exception ex)
    {
        this.value = value;
        this.ex = ex;
    }

    public ValueOrException(T value)
    {
        this.value = value;
    }

    public T Value
    {
        get { return this.value; }
    }

    public Exception Ex
    {
        get { return this.ex; }
    }
}

Answer 1

如果消费者是知道是否Exception预期的消费者,那么我说在这里抛出异常是不正确的.

你本质上是在试图传达状态或逻辑,这些异常并不意味着要做.例外意味着将系统(或至少是当前操作)暂停,因为已发生的事情代码本身并不知道如何从中恢复.

在这种情况下,它只是恰巧,一个Exception是部分业务逻辑/状态的,但是,这并不意味着你可以把他们.您需要将它们下游传递给您的消费者,然后让您的消费者处理它们.

A Tuple<T, Exception>会在这里起作用,但是对于类型(以及属性)缺乏特异性通常是混乱的,所以我建议创建一个专用类型来传达您的操作结果并通过IObservable<T>.