捕捉异常可以从认购OnNext行动被抛出 [英] Catching exceptions which may be thrown from a Subscription OnNext Action

问题描述

我有点新的Rx.NET。是否有可能赶上它可以由任何用户被抛出的异常？看看下面的...

  handler.FooStream.Subscribe（
            _ =＆GT;抛出新的异常（酒吧），
            _ =＆GT; {}）;
 

目前我赶上在每个订阅的基础与下面的一个实例。它的实现只使用一个ManualResetEvent的唤醒等待线程。

 公共接口IExceptionCatcher
{
    动作＆LT; T＆GT; EXEC＆LT; T＆GT;（动作＆LT; T＆GT;动作）;
}
 

和使用它像这样...

  handler.FooStream.Subscribe（
            _exceptionCatcher.Exec＆LT;富＆GT;（_ =＆GT;抛出新的异常（酒吧）），//这是令人失望，这泛型类型无法推断
            _ =＆GT; {}）;
 

我觉得必须有一些更好的办法。是否所有的错误处理Rx.NET功能专为处理错误观测？

编辑：每请求，我实现 https://gist.github.com/1409829 （接口和实现分离成在督促code不同的组件）。反馈是受欢迎的。这可能看起来很傻，但我使用温莎城堡来管理许多不同的接收用户。此异常捕手，像这样的容器中注册

  windsorContainer.Register（Component.For＆LT; IExceptionCatcher＆GT;（）实例（捕手））;
 

然后，它将像这样使用，其中观察到是的IObservable实例：

  VAR exceptionCatcher =
    新ExceptionCatcher（E =＆GT;
                                {
                                    Logger.FatalException（
                                        捕获到异常，关停，E）;
                                    //这里处理非托管资源
                                }，FALSE）;
/ *
 *通常情况下，code以下存在于某些类通过一个IoC容器管理。
 *捕手'将由容器中提供。
 * /
可观察/ *做一些过滤，选择，分组等* /
    .SubscribeWithExceptionCatching（processItems，捕手）;
 

解决方案

内置的可观察运营商不这样做你所要求的在默认情况下（很像事件），但你可以做的扩展方法，将做到这一点

 公共静态的IObservable＆LT; T＆GT; IgnoreObserverExceptions＆LT; T，TException＆GT;（
                                这个的IObservable＆LT; T＆GT;资源
                               ）其中TException：异常
{
    返回Observable.CreateWithDisposable＆LT; T＆GT;（
        O =＆GT; source.Subscribe（
            V =＆GT; {{尝试o.OnNext（V）; }
                   赶上（TException）{}
            }，
            EX =＆GT; o.OnError（前）
            （）=＆GT; o.OnCompleted（）
            ））;
}
 

那么任何可观察到可以通过这种方法被包裹，让你描述的行为。

I'm somewhat new to Rx.NET. Is it possible to catch an exception which may be thrown by any of the subscribers? Take the following...

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

Currently I'm catching on a per subscription basis with an instance of the following. The implementation of which just uses a ManualResetEvent to wake up a waiting thread.

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

and using it like so...

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

I feel like there must be some better way. Are all of the error handling capabilities in Rx.NET specifically for dealing with errors observables?

EDIT: Per request, my implementation is https://gist.github.com/1409829 (interface and implementation separated into different assemblies in prod code). Feedback is welcome. This may seem silly, but I'm using castle windsor to manage many different Rx subscribers. This exception catcher is registered with the container like this

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

It would then be used like this where observable is instance of 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);

解决方案

The built-in Observable operators do not do what you are asking for by default (much like events), but you could make an extension method that would do this.

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

Then any observable could be wrapped by this method to get the behavior you described.

这篇关于捕捉异常可以从认购OnNext行动被抛出的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！