Reactive Quan sát Subscription Xử

Question

Nếu tôi có quyền truy cập vào một IObservable mà tôi biết chỉ bao giờ sẽ trở lại một mặt hàng, sẽ làm việc này và nó là mô hình sử dụng tốt nhất?

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

Answer 1

Tuyên bố từ chối trách nhiệm: Tôi vẫn đang học Rx. Vì vậy, tôi không thực sự là một chuyên gia nhưng tôi tin rằng dùng một lần được trả về bởi Subscribe sẽ chỉ ngừng đăng ký thuê bao. Ngoài ra nếu nguồn hoàn thành, như trong trường hợp của bạn, việc hủy đăng ký được thực hiện tự động. Vì vậy, tôi nghĩ rằng Dispose có dư thừa và có thể được xóa an toàn.

Xem câu trả lời cho question này cho biết thêm.

Answer 2

Chức năng Take sẽ làm chính xác những gì bạn đang tìm kiếm. Trong trường hợp này, Take(1).

Answer 3

Phương tiện mở rộng được trả về bằng phương pháp mở rộng Subscribe chỉ được trả lại để cho phép bạn hủy đăng ký theo cách thủ công từ quan sát trước kết thúc tự nhiên có thể quan sát được.

Nếu có thể quan sát được - với OnCompleted hoặc OnError - thì đăng ký đã được xử lý cho bạn.

Hãy thử mã này:

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

Nếu bạn chạy ở trên, bạn sẽ thấy rằng "xử lý!" được ghi vào bàn điều khiển khi có thể quan sát được mà không cần bạn gọi số .Dispose() khi đăng ký.

Một điều quan trọng cần lưu ý: các bộ thu rác không bao giờ gọi .Dispose() trên đăng ký quan sát được, vì vậy bạn phải vút của đăng ký của bạn nếu họ không có (hoặc không có) đã kết thúc một cách tự nhiên trước khi thuê bao của bạn đi ra khỏi phạm vi.

Đi này, ví dụ:

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

Các ds thể quan sát được sẽ chỉ gắn với xử lý sự kiện khi nó có một thuê bao và sẽ chỉ tách khi chạy xong quan sát hoặc đăng ký sẽ được xử lý. Vì nó là một trình xử lý sự kiện quan sát được sẽ không bao giờ hoàn toàn bởi vì nó đang chờ đợi sự kiện hơn, và do đó xử lý là cách duy nhất để tách từ sự kiện này (đối với ví dụ trên).

Khi bạn có một FromEventPattern thể quan sát được mà bạn biết sẽ chỉ bao giờ trả lại một giá trị thì nó là khôn ngoan để thêm phương pháp .Take(1) phần mở rộng trước khi đăng ký để cho phép xử lý sự kiện để tự động tách và sau đó bạn không cần phải tự định đoạt đăng ký.

Giống như vậy:

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

Tôi hy vọng điều này sẽ giúp.

Answer 4

Ngược lại với một số ý kiến ​​nó không phải ở tất cả các bất thường để xử lý một thuê bao từ bên OnNext.

Trong khi đó là sự thật mà xử lý tại OnCompleted và OnError được thực hiện cho bạn bởi một thuê bao bọc rằng phương pháp Subscribe mở rộng tạo ra, bạn có thể muốn huỷ đăng ký dựa trên một giá trị mà bạn đang quan sát (như trong trường hợp của bạn: Ai là 1) . Bạn không phải luôn luôn có một quan sát được biết là chỉ tạo ra một giá trị.

Vấn đề là bạn chỉ nhận được IDisposable sau khi đã đăng ký. Một quan sát có thể gọi lại cho bạn trên OnNext ngay cả trước khi nó trả lại cho bạn số IDisposable để hủy đăng ký (tùy thuộc vào những thứ như IScheduler sử dụng).

System.Reactive.Disposables.SingleAssignmentDisposable có ích trong trường hợp này. Nó kết thúc tốt đẹp một số IDisposable mà bạn có thể chỉ định trễ và sẽ ngay lập tức bỏ nó khi chuyển nhượng nếu SingleAssignmentDisposable đã được xử lý trước đó. Ngoài ra, nó còn mang một thuộc tính IsDisposed, ban đầu là false và được đặt thành true khi gọi Dispose().

Vì vậy:

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