Reactive Observable Subscription Disposal

Question

如果我有權訪問IObservable，但我知道只會返回一個項目，這是否會起作用並且是最佳使用模式？

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

Answer 1

聲明：我還在學習Rx。所以我不是專家，但我相信Subscribe返回的一次性將只退訂。此外，如果源完成，就像你的情況一樣，取消訂閱是自動完成的。所以我認爲Dispose有多餘，可以安全地刪除。

查看對此question的回答以獲取更多信息。

Answer 2

Take函數將完全符合您的要求。在這種情況下，Take(1)。

Answer 3

一次性通過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)); 

：

試試這個代碼「出售！」在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觀察到的，你知道會永遠只返回一個值，那麼它是明智訂閱允許事件處理程序自動分離之前添加.Take(1)擴展方法，然後你不需要手動處置訂閱。

像這樣：

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

我希望這有助於。

Answer 4

與一些評論相反，處理OnNext以內的訂閱並不罕見。

雖然OnCompleted和OnError的處理方式確實是通過Subscribe擴展方法創建的包裝訂閱完成的，但您可能希望根據您正在觀察的值取消訂閱（例如，您的情況：第一個） 。你並不總是有一個已知只能產生一個值的可觀測值。

問題是隻有訂閱後才能獲得IDisposable。即使在它返回以取消訂閱（取決於它使用的IScheduler之類的內容）之前，可觀察的客戶可能會回撥OnNext。

System.Reactive.Disposables.SingleAssignmentDisposable在這種情況下派上用場。它包裝了一個IDisposable，你可能會延遲分配，並且如果SingleAssignmentDisposable已經被處理掉了，它會立即處置它。此外，它還包含IsDisposed，最初爲false，並且在調用Dispose()時設置爲true。

所以：

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(); 
});