c# - Observable.Create 不起作用并且 Observable.ToEnumerable 无限期阻塞

Question

为什么下面的代码不起作用？

var observable = Observable.Create<int>(o => new Action(() =>
{
    o.OnNext(0);
    o.OnCompleted();
}));

foreach (var item in observable.ToEnumerable()) // <-- Here blocks
{
    // Never get here
}

以下内容也不起作用：

Task.Run(() =>
{
    foreach (var item in observable.ToEnumerable()) // <-- Still blocks
    {
        // Never get here
    }
}).Wait();

有趣的是，Generate 方法按预期工作：

var observable = Observable.Generate(0, i => i < 1, i => i + 1, i => i);

foreach (var item in observable.ToEnumerable())
{
    // Is ok
}

Answer 1

这是不正确的：

var observable = Observable.Create<int>(o => new Action(() =>
{
    o.OnNext(0);
    o.OnCompleted();
}));

您的意思是提供一个调用OnNext订阅的函数。但是，您无意中调用了错误的重载，Observable.Create并为其提供了一个函数，该函数除了创建一个操作（将在处理订阅时调用）之外什么都不做。因此（错误地），您OnNext仅在处理订阅时才尝试调用。

这就是你的意思：

var observable = Observable.Create<int>(o =>
{
    o.OnNext(0);
    o.OnCompleted();

    return () => Console.WriteLine("Disposed!"); // this will be called when 
                                                  // subscription is disposed
});

或这个：

var observable = Observable.Create<int>(o =>
{
    o.OnNext(0);
    o.OnCompleted();

    return Disposable.Empty; // using System.Reactive.Disposables;
});