RxJs Dealing with Exceptions without termination

Question

I am not sure how correct I am on this so if any experts are able to correct me on this it would also be appreciated. My current understanding is that observables are lazy and do not produce values until subscribed to. If an error occurs the observable sends no more values. In a lot of cases this is not what is wanted.

In the code sample below I am getting the weather for perth and london and if an error occurs returning an object indicating the error has occured. This pretty much means that the error block on the subscriber will not get called but the success will and I will have to look at whether it failed and change logic. Is this the best way to do this????

Also does the zip operator wait for all obs to produce a value regardless of ordering and return when a single value from all has been produced?

Code as below:

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(){

        }
    );

});

Any suggestions and tips is greatly appreciated.

Thanks in advance.

Blair.

Answer 1

Interesting as this post says that you do need to flag errors so that the steam does not complete.

Idiomatic way to recover from stream onError

From my current understanding and testing of RX if an error is thrown during process the onError handler of the subscription will get called but it will complete the stream.

Meaning that the subscriber will receive no further messages.

Answer 2

You probably should replace .forEach() with .subscribe() since one is an alias of the other, and .subscribe() is more common and more explicit in this case.

subscribe() can accept three functions as arguments onNext (first function), onError (second function, optional), onComplete (third function, optional). Hence, you probably will achieve what you want by just providing a second function in the subscribe(), to handle errors. Then you don't need the .catch() operator to insert a flag-like event.

Also, for your specific use case, where perth and london Observables are conceptually independent (they have no dependencies with one another), you want to use .combineLatest() instead of .zip(). See this answer for details on the difference between the two.