Rx 可观察对象能否优雅地处理运算符中的异常并继续

Question

即

通过传递错误条件而不停止整个可观察量？

My Observable 从用户提供的常见递送服务（FedEx、UPS、DHL 等）的包裹跟踪号列表开始，在线查找预计递送日期，然后根据从今天起的天数（即"3 天内"而不是"1 月 22 日"）返回这些日期。问题是，如果任何单个查找导致异常，则整个流将停止，并且不会查找其余代码。没有能力优雅地处理，比如说，UnknownTrackingCode Exception，所以可观察量不能保证它会查找用户提交的所有代码。

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
}

由于一个错误而停止可观察流是设计使然：

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

默认行为是可以克服的，但只有首先消除 Rx 的许多好处：

• 我可以包装LookupDeliveryDate，以便它返回日期代替异常（例如UnknownTrackingCode Exception的1899-12-31），但这可以防止"松散耦合的代码"，因为CalculateDaysFromToday需要处理这些特殊情况
• 我可以用 try/catch 和块包围每个匿名函数，但这基本上阻止了我使用 lambda
• 我可以使用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返回魔术日期而不是异常，然后添加一个.filter以在魔术日期到达CalculateDaysFromToay之前过滤掉它们。