反应性可观察认购出售事项

Question

如果我可以访问我知道只会返回一个项目的IObservable,这是否有效并且它是最好的使用模式？

IDisposable disposable = null;
disposable = myObservable.Subscribe(x =>
  {
     DoThingWithItem(x);
     if (disposable != null)
     {
       disposable.Dispose();
     }
  });

Answer 1

返回Subscribe方法返回的一次性返回仅允许您在观察结果自然结束之前手动取消订阅观察者.

如果observable完成 - 使用OnCompleted或者OnError- 那么订阅已经为您处理.

试试这段代码:

var xs = Observable.Create<int>(o =>
{
    var d = Observable.Return(1).Subscribe(o);
    return Disposable.Create(() =>
    {
        Console.WriteLine("Disposed!");
        d.Dispose();
    });
});

var subscription = xs.Subscribe(x => Console.WriteLine(x));

如果你运行上面的程序,你会看到"Disposed!" 在observable完成时写入控制台,而无需调用.Dispose()订阅.

需要注意的一件重要事情是:垃圾收集器从不调用.Dispose()可观察的订阅,因此如果订阅在订阅超出范围之前没有(或可能没有)自然结束,则必须处置订阅.

以此为例,例如:

var wc = new WebClient();

var ds = Observable
    .FromEventPattern<
        DownloadStringCompletedEventHandler,
        DownloadStringCompletedEventArgs>(
            h => wc.DownloadStringCompleted += h,
            h => wc.DownloadStringCompleted -= h);

var subscription =
    ds.Subscribe(d =>
        Console.WriteLine(d.EventArgs.Result));

该ds观察到的将只重视事件处理程序时,它有一个订阅,当观察到完成或订阅配置只会脱离.因为它是一个事件处理程序,所以observable永远不会完成,因为它正在等待更多的事件,因此处理是从事件中分离的唯一方法(对于上面的例子).

如果你有一个FromEventPattern你知道只会返回一个值的observable,那么.Take(1)在订阅允许事件处理程序自动分离之前添加扩展方法是明智的,然后你不需要手动处理订阅.

像这样:

var ds = Observable
    .FromEventPattern<
        DownloadStringCompletedEventHandler,
        DownloadStringCompletedEventArgs>(
            h => wc.DownloadStringCompleted += h,
            h => wc.DownloadStringCompleted -= h)
    .Take(1);

我希望这有帮助.

Answer 2

免责声明:我还在学习Rx.所以我不是真正的专家,但我相信返回的一次性用品Subscribe只会取消订阅.此外,如果源完成,就像您的情况一样,取消订阅会自动完成.所以我认为Dispose这是多余的,可以安全地删除.

有关详细信息,请参阅此问题的答案.

Answer 3

该Take功能将完全符合您的要求。在这种情况下，Take(1)。

Answer 4

与一些评论相反，从内部处理订阅并不少见OnNext。

虽然这是真的，在处置OnCompleted和OnError由包装订阅该为你做了Subscribe扩展方法造成的，你可能要根据你观察到的值退订（像你的情况：一日一）。你并不总是有一个已知只产生一个值的 observable。

问题是您IDisposable只有在订阅后才能获得。一个 observable 可能会OnNext在它返回给你IDisposable取消订阅之前给你回电（取决于IScheduler它使用的东西）。

在System.Reactive.Disposables.SingleAssignmentDisposable这种情况下就派上用场了。它包裹了一个IDisposable您可以延迟分配，如果到时SingleAssignmentDisposable已经处理，将立即在分配时处理。它还带有一个属性IsDisposed，该属性最初是false并且设置为何true时Dispose()被调用。

所以：

IObservable<string> source = ...;

var subscription = new SingleAssignmentDisposable();
subscription.Disposable = source.Subscribe(x =>
{
    if (subscription.IsDisposed) // getting notified though I've told it to stop
        return;
    DoThingsWithItem(x);
    if (x == "the last item I'm interested in")
        subscription.Dispose();
});