Rx Observable可以优雅地处理运算符中的异常并继续吗？

Question

即,通过传递错误条件而不是停止整个Observable？

My Observable以用户提供的常用送货服务(FedEx,UPS,DHL等)包裹追踪号码列表开始,在线查找预计送达日期,然后按照今天的天数返回这些日期(即"在3天"而不是"1月22日").问题是如果任何单个查找导致异常,整个流将停止,其余代码将不会被查找.例如,没有能力优雅地处理,UnknownTrackingCode Exception因此Observable不能保证它会查找用户提交的所有代码.

public void getDaysTillDelivery(List<String> tracking_code_list) {
        Observable o = Observable.from(tracking_code_list)
                   // LookupDeliveryDate performs network calls to UPS, FedEx, USPS web sites or APIs
                   // it might throw: UnknownTrackingCode Exception, NoResponse Exception, LostPackage Exception
                .map(tracking_code -> LookupDeliveryDate(tracking_code))
                .map(delivery_date -> CalculateDaysFromToday(delivery_date));
        o.subscribe(mySubscriber); // will handle onNext, onError, onComplete
}

由于一个错误而停止Observable流是设计的:

• http://reactivex.io/documentation/operators/catch.html
• 在不停止序列的情况下处理Reactive Extensions中的异常
• https://groups.google.com/forum/#!topic/rxjava/trm2n6S4FSc

可以克服默认行为,但只能首先消除Rx的许多好处:

• 我可以换行,LookupDeliveryDate所以它返回Dates代替Exceptions(例如1899-12-31for UnknownTrackingCode Exception)但这可以防止"松散耦合的代码",因为CalculateDaysFromToday需要处理这些特殊情况
• 我可以使用try/catch和阻止每个匿名函数,但这实际上阻止我使用lambdas
• 我可以if/thens用来指导代码路径,但这可能需要维护一些状态并消除确定性评估
• 显然,错误处理每个步骤会阻止整合所有错误处理 Subscriber
• 编写我自己的错误处理操作符是可能的,但记录很少

有没有更好的方法来处理这个？

Answer 1

如果出现错误，您究竟希望发生什么？您只是想扔掉该条目还是想要下游的某些东西对其进行处理？

如果您希望下游的某些东西采取某些操作，那么您实际上是将错误转换为数据（有点像返回代表1899-12-31错误的哨兵值的示例）。根据定义，这种策略意味着下游的所有内容都需要了解数据流可能包含错误而不是数据，并且必须修改它们来处理它。

Observable但您可以将流转变为值流，而不是产生神奇的值Either。该值要么是日期，要么是错误。下游的所有内容都会接收该Either对象，并可以询问它是否有值或错误。如果它有值，他们可以Either用计算结果生成一个新对象。如果它有错误并且他们无法对其执行任何操作，则他们自己可能会产生错误。

我不知道 Java 语法，但在 C# 中可能是这样的：

Observable.From(tracking_code_list)
    .Select(t =>
        {
            try { return Either.From(LookupDeliveryDate(t)); }
            catch (Exception e)
            {
                return Either.FromError<Date>(e);
            }
        })
     .Select(dateEither =>
        {
            return dateEither.HasValue ?
                Either.From(CalculateDaysFromToday(dateEither.Value)) :
                Either.FromError<int>(dateEither.Error);
        })
     .Subscribe(value =>
        {
            if (value.HasValue) mySubscriber.OnValue(value.Value);
            else mySubscribe.OnError(value.Error);
        });

您的另一个选择是在错误发生时“处理”/抑制错误。根据您的需要，这可能就足够了。在这种情况下，只需LookupDeliveryDate返回魔术日期而不是异常，然后添加 a.filter以在魔术日期到达 之前过滤掉魔术日期CalculateDaysFromToay。