Ghi ngoại lệ có thể được ném từ Đăng ký OnNext Action

Question

Tôi có phần mới đối với Rx.NET. Có thể bắt một ngoại lệ có thể được ném bởi bất kỳ người đăng ký nào không? Hãy thực hiện theo các bước sau ...

handler.FooStream.Subscribe(
      _ => throw new Exception("Bar"), 
      _ => { }); 

Hiện tại tôi đang bắt trên cơ sở thuê bao với một ví dụ sau đây. Việc thực hiện mà chỉ sử dụng một ManualResetEvent để đánh thức một chủ đề chờ đợi.

public interface IExceptionCatcher 
{ 
    Action<T> Exec<T>(Action<T> action); 
} 

và sử dụng nó như vậy ...

handler.FooStream.Subscribe(
      _exceptionCatcher.Exec<Foo>(_ => throw new Exception("Bar")), //It's disappointing that this generic type can't be inferred 
      _ => { }); 

tôi cảm thấy như phải có một cách nào đó tốt hơn. Tất cả các khả năng xử lý lỗi trong Rx.NET có đặc biệt để xử lý các lỗi quan sát được không?

EDIT: Theo yêu cầu, triển khai của tôi là https://gist.github.com/1409829 (giao diện và triển khai được tách thành các hội đồng khác nhau trong mã sản phẩm). Phản hồi được hoan nghênh. Điều này có vẻ ngớ ngẩn, nhưng tôi đang sử dụng lâu đài windsor để quản lý nhiều người đăng ký Rx khác nhau. catcher ngoại lệ này được đăng ký với container như thế này

windsorContainer.Register(Component.For<IExceptionCatcher>().Instance(catcher)); 

Nó sau đó sẽ được sử dụng như thế này ở đâu observable là thể hiện của IObservable:

var exceptionCatcher = 
    new ExceptionCatcher(e => 
           { 
            Logger.FatalException(
             "Exception caught, shutting down.", e); 
            // Deal with unmanaged resources here 
           }, false); 


/* 
* Normally the code below exists in some class managed by an IoC container. 
* 'catcher' would be provided by the container. 
*/ 
observable /* do some filtering, selecting, grouping etc */ 
    .SubscribeWithExceptionCatching(processItems, catcher); 

Answer 1

Việc xây dựng trong điều hành Quan sát không làm những gì bạn đang yêu cầu theo mặc định (giống như các sự kiện), nhưng bạn có thể tạo một phương thức mở rộng để thực hiện việc này.

public static IObservable<T> IgnoreObserverExceptions<T, TException>(
           this IObservable<T> source 
           ) where TException : Exception 
{ 
    return Observable.CreateWithDisposable<T>(
     o => source.Subscribe(
      v => { try { o.OnNext(v); } 
        catch (TException) { } 
      }, 
      ex => o.OnError(ex), 
      () => o.OnCompleted() 
      )); 
} 

Sau đó, bất kỳ quan sát nào có thể được bao bọc bằng phương pháp này để có được hành vi bạn mô tả.