Rx Observable 能否优雅地处理运算符中的异常并继续？

Question

即通过传递错误条件并且不停止整个Observable？

My Observable 以用户提供的常见递送服务（FedEx、UPS、DHL 等）包裹跟踪号码列表开始，在线查找预计递送日期，然后 returns 这些日期从今天算起的天数（即 "in 3 days" 而不是 "Jan 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
Handling Exceptions in Reactive Extensions without stopping sequence
https://groups.google.com/forum/#!topic/rxjava/trm2n6S4FSc

可以克服默认行为，但首先要消除 Rx 的许多优点：

我可以包装 LookupDeliveryDate 所以它 returns 日期代替例外（例如 1899-12-31 代表 UnknownTrackingCode Exception）但这会阻止 "loosely coupled code"，因为CalculateDaysFromToday 需要处理这些特殊情况
我可以用 try/catch 和块包围每个匿名函数，但这实际上阻止了我使用 lambdas
我可以使用 if/thens 来引导代码路径，但这可能需要维护一些状态并消除确定性评估
每个步骤的错误处理显然会阻止在 Subscriber
编写我自己的错误处理运算符是可能的，但文档很少

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

Answer 1

如果出现错误，您到底希望发生什么？您是只想丢弃该条目还是想让下游的某些东西对其进行处理？

如果你想让下游的东西采取一些行动，那么你实际上是在将错误转化为数据（有点像你的例子 returning 一个标记值 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);
        });

您的另一个选择是 "handle"/在错误发生时抑制它。根据您的需要，这可能就足够了。在这种情况下，只需 LookupDeliveryDate return 魔法日期而不是例外，然后添加 .filter 以在魔法日期到达 CalculateDaysFromToay.[=19= 之前过滤掉它们]

Rx Observable 能否优雅地处理运算符中的异常并继续？

Can an Rx Observable gracefully handle exceptions in an operator and continue?

system.reactive

reactive-programming

rx-java