feat(operator): add retry

spec/operators/retry-spec.js

@@ -0,0 +1,69 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.retry()', function () {
+  it('should retry a number of times, without error, then complete', function (done) {
+    var errors = 0;
+    var retries = 2;
+    Observable.value(42)
+      .map(function(x){
+        if ((errors+=1) < retries){
+          throw 'bad';
+        }
+        errors = 0;
+        return x;
+      })
+      .retry(retries)
+      .subscribe(
+        function(x){
+          expect(x).toBe(42);
+        }, 
+        function(err){
+          expect('this was called').toBe(false);
+        }, done);
+  });
+  it('should retry a number of times, then call error handler', function (done) {
+    var errors = 0;
+    var retries = 2;
+    Observable.value(42)
+      .map(function(x){
+        if ((errors+=1) < retries){
+          throw 'bad';
+        }
+        return x;
+      })
+      .retry(retries-1)
+      .subscribe(
+        function(x){
+          expect(x).toBe(42);
+        }, 
+        function(err){
+          expect(errors).toBe(1); 
+          done();     
+        }, function(){
+          expect('this was called').toBe(false);
+        });
+  });
+  it('should retry until successful completion', function (done) {
+    var errors = 0;
+    var retries = 10;
+    Observable.value(42)
+      .map(function(x){
+        if ((errors+=1) < retries){
+          throw 'bad';
+        }
+        errors = 0;
+        return x;
+      })
+      .retry()
+      .take(retries)
+      .subscribe(
+        function(x){
+          expect(x).toBe(42);
+        }, 
+        function(err){
+          expect('this was called').toBe(false);
+        }, done);
+  });
+});

src/Observable.ts

@@ -157,6 +157,7 @@ export default class Observable<T> {
   multicast: (subjectFactory: () => Subject<T>) => ConnectableObservable<T>;
 
   catch: (selector: (err: any, source: Observable<T>, caught: Observable<any>) => Observable<any>) => Observable<T>;
+  retry: <T>(count: number) => Observable<T>;
   retryWhen: (notifier: (errors: Observable<any>) => Observable<any>) => Observable<T>;
   repeat: <T>(count: number) => Observable<T>;
 

src/Rx.ts

@@ -146,10 +146,12 @@ observableProto.defaultIfEmpty = defaultIfEmpty;
 observableProto.materialize = materialize;
 
 import _catch from './operators/catch';
+import retry from './operators/retry';
 import retryWhen from './operators/retryWhen';
 import repeat from './operators/repeat';
 
 observableProto.catch = _catch;
+observableProto.retry = retry;
 observableProto.retryWhen = retryWhen;
 observableProto.repeat = repeat;
 

src/operators/retry.ts

@@ -0,0 +1,38 @@
+import Operator from '../Operator';
+import Observer from '../Observer';
+import Subscriber from '../Subscriber';
+import Observable from '../Observable';
+
+export default function retry<T>(count: number = 0): Observable<T> {
+  return this.lift(new RetryOperator(count, this));
+}
+
+export class RetryOperator<T, R> extends Operator<T, R> {
+  constructor(private count: number, protected original:Observable<T>) {
+    super();
+  }
+
+  call(observer: Observer<T>): Observer<T> {
+    return new RetrySubscriber<T>(observer, this.count, this.original);
+  }
+}
+
+export class RetrySubscriber<T> extends Subscriber<T> {
+  private retries: number = 0;
+  constructor(destination: Observer<T>, private count: number, private original: Observable<T>) {
+    super(destination);
+  }
+
+  _error(err: any) {
+    const count = this.count;
+    if (count && count === (this.retries+=1)){
+      this.destination.error(err);
+    } else { 
+      this.resubscribe();
+    } 
+  }
+
+  resubscribe() {
+    this.original.subscribe(this);
+  }
+}