feat(operator): add finally

spec/operators/finally-spec.js

@@ -0,0 +1,47 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.finally()', function () {
+  it('should call finally after complete', function (done) {
+    var completed = false;
+    Observable.of(1, 2, 3)
+      .finally(function(x) {
+        expect(completed).toBe(true);
+        done();
+      })
+      .subscribe(null, null, function() {
+        completed = true;
+      });
+  });
+
+  it('should call finally after error', function (done) {
+    var thrown = false;
+    Observable.of(1, 2, 3)
+      .map(function(x) {
+        if(x === 3) {
+          throw x;
+        }
+        return x;
+      })
+      .finally(function(x) {
+        expect(thrown).toBe(true);
+        done();
+      })
+      .subscribe(null, function() {
+        thrown = true;
+      });
+  });
+
+  it('should call finally upon disposal', function (done) {
+    var disposed = false;
+    var subscription = Observable
+      .timer(100)
+      .finally(function(x) {
+        expect(disposed).toBe(true);
+        done();
+      }).subscribe();
+      disposed = true;
+      subscription.unsubscribe();
+  });
+});

src/Observable.ts

@@ -149,4 +149,7 @@ export default class Observable<T> {
 
   catch: (selector: (err: any, source: Observable<T>, caught: Observable<any>) => Observable<any>) => Observable<T>;
   retryWhen: (notifier: (errors: Observable<any>) => Observable<any>) => Observable<T>;
+
+  ensure: (ensure: () => void, thisArg?: any) => Observable<T>;
+  finally: (ensure: () => void, thisArg?: any) => Observable<T>;
 }

src/Rx.ts

@@ -143,6 +143,11 @@ import retryWhen from './operators/retryWhen';
 observableProto.catch = _catch;
 observableProto.retryWhen = retryWhen;
 
+import _finally from './operators/finally';
+
+observableProto.ensure = _finally;
+observableProto.finally = _finally;
+
 export default {
     Subject,
     Scheduler,

src/operators/finally.ts

@@ -0,0 +1,35 @@
+import Operator from '../Operator';
+import Observer from '../Observer';
+import Subscriber from '../Subscriber';
+import Subscription from '../Subscription';
+
+import tryCatch from '../util/tryCatch';
+import {errorObject} from '../util/errorObject';
+import bindCallback from '../util/bindCallback';
+
+export default function _finally<T>(finallySelector: () => void, thisArg?: any) {
+  return this.lift(new FinallyOperator(thisArg ?
+    <() => void> bindCallback(finallySelector, thisArg, 2) :
+    finallySelector));
+}
+
+export class FinallyOperator<T, R> extends Operator<T, R> {
+
+  finallySelector: () => void;
+
+  constructor(finallySelector: () => void) {
+    super();
+    this.finallySelector = finallySelector;
+  }
+
+  call(observer: Observer<T>): Observer<T> {
+    return new FinallySubscriber(observer, this.finallySelector);
+  }
+}
+
+export class FinallySubscriber<T> extends Subscriber<T> {
+  constructor(destination: Observer<T>, finallySelector: () => void) {
+    super(destination);
+    this.add(new Subscription(finallySelector));
+  }
+}