Observable.Throw实例

Question

我有点不熟悉观察，所以我只是在寻找一个例子，这将使我在正确的方向(也许教程？)所以这就是-我想要创建异步Observable并利用它创建一个Exception。以下是我的例子：

    protected IObservable<Tuple<DataPart1, DataPart2>> LoadAllDataFunc(string FileName)
    {
        return Observable.Start<Tuple<DataPart1, DataPart2>>(() =>
        {
            ConfigReaderWriter readerWriter = new ConfigReaderWriter();
            try
            {
                readerWriter.UnpackFile(fileName, out DataPart1, out DataPart2);
                return Tuple.Create(DataPart1, DataPart2);
            }
            catch (Exception exp_gen)
            {
                Observable.Throw<Exception>(exp_gen);
                return null;
            }
        });
    }

问题是，我认为我没有正确地抛出Exception。例如-任何订阅者：

    internal IObservable<DataPart2> GetProject()
    {
        if (this.GlobalDataPart2 != null)
            return Observable.Return(GlobalDataPart2);

        IObservable<Project> receivedData = null;

        var loadAll = LoadAllDataFunc(this.GlobalFileName).Subscribe(
                data => { receivedData = Observable.Return(data.Item1); },
                (ex) => { Observable.Throw<Exception>(ex); }
            );

        return receivedData;
    }

不会从Exception收到LoadAllDataFunc吗？即使是，如果出现异常，订户也会接收null。

那么-什么是正确的方式来抛出异常，从可观察的？

Answer 1

为了清楚地说明Observable.Throw的作用:它返回一个可观察的序列(指定元素类型的序列)，该序列的唯一作用是通过调用其OnError方法告诉所有观察者一个异常：

var err = Observable.Throw<int>(new Exception("Oops!"));
err.Subscribe(_ => {}, ex => { Console.WriteLine(ex.Message); }, () => {});
err.Subscribe(_ => {}, ex => { Console.WriteLine(ex.Message); }, () => {});

上面的代码会打印出Oops！两次。

正如Gideon所提到的，像Start这样的操作符将向OnError信道传播用户异常。但是，如果使用Observable.Create，则直接与观察者讨论错误情况：

var res = Observable.Create<int>(observer =>
{
    return scheduler.Schedule(() =>
    {
        var res = default(int);

        try
        {
            res = ComputationThatMayFail();
        }
        catch (Exception ex)
        {
            observer.OnError(ex);
            return;
        }

        observer.OnNext(res);
        observer.OnCompleted();
    });
});

实际上，上面显示的代码非常接近Start的功能。(唯一的区别是使用AsyncSubject缓存操作的结果。)

Answer 2

当使用Observable.Start时，最好的方法就是不捕获异常。Observable.Throw和Observable.Return只创建可观察的序列；它们不会像您所期望的那样，从返回IObservable的内部方法中执行任何特殊的操作。这就是为什么您没有从顶级代码中获得OnError调用的原因。对Observable.Throw的调用创建一个可观察的，对它不做任何操作，并从函数返回一个空值，然后将该值发送给观察者的OnNext。

您的第二段代码似乎存在相关问题。再一次，抛出的电话也无济于事。该订阅服务器的OnNext操作可能也不是您想要的。通常，您将从该函数返回null，因为LoadAllDataFunc的第一个结果可能不会在GetProject返回时出现(订阅通常不会阻塞)。我猜您想要做的是使用Select。

return LoadAllDataFunc(this.GlobalFileName)
       .Select(data => data.Item1); //should this be Item2?