RxJs 处理异常而不终止

Question

我不确定我在这方面有多正确，所以如果有专家能够纠正我，我也将不胜感激。我目前的理解是 observables 是惰性的，并且在订阅之前不会产生值。如果发生错误，observable 将不再发送任何值。在很多情况下，这不是我们想要的。

在下面的代码示例中，我获取了珀斯和伦敦的天气，如果发生错误，则返回一个指示发生错误的对象。这几乎意味着订阅者上的错误块不会被调用，但成功会被调用，我将不得不查看它是否失败并更改逻辑。这是最好的方法吗？？？？

zip 操作符是否还要等待所有 obs 产生一个值而不管顺序如何，并在所有的单个值产生时返回？

代码如下：

window.application = application || {};

window.application.stocks = (function(){


    function getWeather(city, country){

        return $.ajaxAsObservable({
            url: 'http://api.openweathermap.org/data/2.5/weather?q=' + city + ',' + country,
            dataType: 'jsonp'
        }).map(function(v){
            return {
                city : city,
                country : country,
                temperature : v.data.main.temp
            };
        });
    }

    var requests = Rx.Observable
                  .interval(500)
                  .timestamp()
                  .map(function(v){
                     var perth = getWeather('perth','au');
                     var london = getWeather('london','uk');

                     return Rx.Observable.zip(perth,london, function(a,b){
                         return {
                             a : a,
                             b : b,
                             timestamp : v.timestamp,
                             failure : false,
                         };
                     }).retry(3)
                       .catch(Rx.Observable.return({ failure: true }));

                   })
                  .concatAll();

    var selector = $('#weather');

    var subscriber = requests.forEach(
        function(data){

            if(data.failure){
                $('<li>Failure in receiving the temperature</li>').appendTo(selector);
                return;
            }

            $('<li> at: ' + data.timestamp + ' for: '  + data.a.city + ' - ' + data.a.temperature + '</li>').appendTo(selector);
            $('<li> at: ' + data.timestamp + ' for: '  + data.b.city + ' - '  + data.b.temperature + '</li>').appendTo(selector);
        },
        function(error){
            selector.text('');
            $('<li>Error: ' + error + '</li>').appendTo('#weather');
        },
        function(){

        }
    );

});

非常感谢任何建议和提示。

提前致谢。

布莱尔。

Answer 1

您可能应该将.forEach() 替换为.subscribe()，因为其中一个是另一个的别名，而.subscribe() 在这种情况下更常见且更明确。

subscribe() 可以接受三个函数作为参数onNext（第一个函数）、onError（第二个函数，可选）、onComplete（第三个函数，可选）。因此，您可能会通过在 subscribe() 中提供第二个函数来处理错误来实现您想要的。那么你就不需要.catch() 操作符来插入类似标志的事件了。

此外，对于您的特定用例，perth 和 london Observable 在概念上是独立的（它们彼此之间没有依赖关系），您希望使用 .combineLatest() 而不是 .zip()。两者的区别详见this answer。

Answer 2

有趣的是，这篇文章说您确实需要标记错误，以免蒸汽无法完成。

Idiomatic way to recover from stream onError

根据我目前对 RX 的理解和测试，如果在处理过程中抛出错误，订阅的 onError 处理程序将被调用，但它将完成流。

意味着订阅者将不会收到更多消息。