Re-subscribe to an observable with rx-angular

Issue

I'd like to use angular-rx for a simple refresh button for results. If the user clicks the refresh button the results are reloaded. If the user clicks the the refresh button 100times in 1 second, only the latest results are loaded. If the results failed for some reason, that doesn't mean the refresh button should stop working.

To achieve the last point I'd like to keep a subscription (or resubscribe) even if it fails, but I can not work out how to do that?

This doesn't work, but here's a simple example where I try resubscribing on error:

var refreshObs = $scope.$createObservableFunction('refresh');

var doSubscribe = function () {
  refreshObs
  .select(function (x, idx, obs) {
      // get the results.
      // in here might throw an exception
  })
  .switch()
  .subscribe(
  function (x) { /*show the results*/ }, // on next
  function (err) { // on error
      doSubscribe(); // re-subscribe
  },
  function () { } // on complete
  );
};
doSubscribe();

I figure this is common enough there should be some standard practice to achieve this?

UPDATE

Using the suggested solution, this is what I've made to test:

// using angularjs and the rx.lite.js library
var testCount = 0;
var obsSubject = new rx.Subject(); // note. rx is injected but is really Rx
$scope.refreshButton = function () { // click runs this
  obsSubject.onNext();
};

obsSubject.map(function () {
  testCount++;
  if (testCount % 2 === 0) {
      throw new Error("something to catch");
  }
  return 1;
})
.catch(function (e) {
  return rx.Observable.return(1);
})
.subscribe(
    function (x) {
    // do something with results
});

And these are my test results:

1. Refresh button clicked
2. obsSubject.onNext() called
3. map function returns 1.
4. subscribe onNext is fired
5. Refresh button clicked
6. obsSubject.onNext() called
7. map function throw error
8. enters catch function
9. subscribe onNext is fired
10. Refresh button clicked
11. obsSubject.onNext() called
12. Nothing. I need to keep subscription

My understanding is that catch should keep the subscription, but my testing indicates it doesn't. Why?

Solution

Based on the context given in your comment, you want:

• Every refresh button to trigger a 'get results'
• Every error to be displayed to the user

You really do not need the resubscribing, it's an anti-pattern because code in Rx never depends on that, and the additional recursive call just confuses a reader. It also reminds us of callback hell.

In this case, you should:

• Remove the doSubscribe() calls, because you don't need them. With that code, you already have the behavior that every refresh click will trigger a new 'get results'.
• Replace select().switch() with .flatMap() (or .flatMapLatest()). When you do the select(), the result is a metastream (stream of streams), and you are using switch() to flatten the metastream into a stream. That's all what flatMap does, but in one operation only. You can also understand flatMap as .then() of JS Promises.
• Include the operator .catch() which will treat your error, as in a catch block. The reason you can't get more results after an error happens, is that an Observable is always terminated on an error or on a 'complete' event. With the catch() operator, we can replace errors with sane events on the Observable, so that it can continue.

To improve your code:

var refreshObs = $scope.$createObservableFunction('refresh');

refreshObs
  .flatMapLatest(function (x, idx, obs) {
    // get the results.
    // in here might throw an exception
    // should return an Observable of the results
  })
  .catch(function(e) {
      // do something with the error
      return Rx.Observable.empty(); // replace the error with nothing
  })
  .subscribe(function (x) { 
      // on next
  });

Notice also that I removed onError and onComplete handlers since there isn't anything to do inside them.

Also take a look at more operators. For instance retry() can be used to automatically 'get results' again every time an error happens. See https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/retry.md

Use retry() in combination with do() in order to handle the error (do), and allow the subscriber to automatically resubscribe to the source observable (retry).

refreshObs
  .flatMapLatest(function (x, idx, obs) {
    // get the results.
    // in here might throw an exception
    // should return an Observable of the results
  })
  .do(function(){}, // noop for onNext
  function(e) {
      // do something with the error
  })
  .retry()
  .subscribe(function (x) { 
      // on next
  });

See a working example here: http://jsfiddle.net/staltz/9wd13gp9/9/

Answered By - André Staltz