feat: complete streams when component is torn down

BREAKING CHANGE: streams are now completed when the component is torn
down.

src/index.js

@@ -8,27 +8,33 @@ import Rx from 'rx';
 
 export default function withObserve() {
     this.before('initialize', function () {
-        this.localSubscriptions = [];
+        this.localDisposables = [];
     });
 
     /**
-     * Observe a sequence which can be disposed of on teardown.
+     * Observe a sequence is disposed of on teardown.
      *
      * Takes the sequence to observe.
      * Returns the observable sequence.
      */
     this.observe = function (upstream) {
-        return Rx.Observable.create((o) => {
-            var subscription = upstream.subscribe(o);
-            // Store the subscription so that the mixin can dispose on teardown.
-            this.localSubscriptions.push(subscription);
-            return subscription;
+        return Rx.Observable.create(observer => {
+            var upstreamDisposable = upstream.subscribe(observer);
+            // Create our own disposable so we can track teardown. When that happens,
+            // notify the observer.
+            var disposable = Rx.Disposable.create(function () {
+                upstreamDisposable.dispose();
+                observer.onCompleted();
+            });
+            // Store the disposable so that the mixin can dispose on teardown.
+            this.localDisposables.push(disposable);
+            return disposable;
         }, upstream);
     };
 
     this.before('teardown', function () {
-        this.localSubscriptions.forEach(function (subscription) {
-            subscription.dispose();
+        this.localDisposables.forEach(function (disposable) {
+            disposable.dispose();
         });
     });
 }

src/with-observe.spec.js

@@ -13,6 +13,12 @@ describe('withObserve', function () {
         this.component = (new Component()).initialize(document.body);
     });
 
+    afterEach(function () {
+        try {
+            this.component.teardown();
+        } catch (e) {}
+    });
+
     it('should observe changed values', function (done) {
         this.component.observe(observable).subscribeOnNext(function (value) {
             if (value === 2) {
@@ -39,4 +45,16 @@ describe('withObserve', function () {
         subject.onNext('not subscribed');
         expect(called).toBe(2);
     });
+
+    it('should end the stream on teardown', function () {
+        var onNextSpy = jasmine.createSpy('onNextSpy');
+        var onErrorSpy = jasmine.createSpy('onErrorSpy');
+        var onCompletedSpy = jasmine.createSpy('onCompletedSpy');
+        this.component.observe(observable).subscribe(onNextSpy, onErrorSpy, onCompletedSpy);
+        this.component.teardown();
+        subject.onNext(10);
+        expect(onNextSpy).not.toHaveBeenCalledWith(10);
+        expect(onErrorSpy).not.toHaveBeenCalled();
+        expect(onCompletedSpy).toHaveBeenCalled();
+    });
 });