zen-observable-ts - npm Package Compare versions

Comparing version 0.0.0 to 0.1.0

dist/src/zenObservable.d.ts

@@ -13,3 +13,3 @@ import { ZenObservable } from './types';
     constructor(subscriber: ZenObservable.Subscriber<T>);
-    subscribe(observerOrNext: ZenObservable.Observer<T>, error?: (error: any) => void, complete?: () => void): ZenObservable.Subscription;
+    subscribe(observerOrNext: ((value: T) => void) | ZenObservable.Observer<T>, error?: (error: any) => void, complete?: () => void): ZenObservable.Subscription;
     forEach(fn: (value: T) => void): Promise<void>;
@@ -16,0 +16,0 @@ map<R>(fn: (value: T) => R): Observable<R>;

dist/src/zenObservable.js

@@ -43,3 +43,4 @@ "use strict";
             if (cleanup != null) {
-                if (typeof cleanup.unsubscribe === 'function') {
+                if (typeof cleanup.unsubscribe ===
+                    'function') {
                     cleanup = cleanupFromSubscription(cleanup);
@@ -192,7 +193,7 @@ }
         if (typeof observerOrNext === 'function') {
-            observerOrNext = {
+            return new Subscription({
                 next: observerOrNext,
                 error: error,
                 complete: complete,
-            };
+            }, this._subscriber);
         }
@@ -209,5 +210,2 @@ return new Subscription(observerOrNext, this._subscriber);
                 start: function (subscription) {
-                    if (Object(subscription) !== subscription) {
-                        throw new TypeError(subscription + ' is not an object');
-                    }
                     this._subscription = subscription;
@@ -214,0 +212,0 @@ },

dist/tests/filter.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-var chai_1 = require("chai");
+var chai = require("chai");
 var zenObservable_1 = require("../src/zenObservable");
+var chaiAsPromised = require("chai-as-promised");
+chai.use(chaiAsPromised);
+var assert = chai.assert;
 describe('filter ', function () {
@@ -11,5 +14,41 @@ it('Basics', function () {
             .forEach(function (x) { return list.push(x); })
-            .then(function () { return chai_1.assert.deepEqual(list, [3, 4]); });
+            .then(function () { return assert.deepEqual(list, [3, 4]); });
     });
+    it('throws on not a function', function () {
+        var list = [];
+        return assert.throws(function () {
+            return zenObservable_1.default.from([1, 2, 3, 4]).filter(1).forEach(function (x) { return list.push(x); })
+                .then;
+        });
+    });
+    it('throws on error inside function', function (done) {
+        var error = new Error('thrown');
+        return assert.doesNotThrow(function () {
+            return zenObservable_1.default.from([1, 2, 3, 4])
+                .filter(function () { throw error; })
+                .subscribe({
+                error: function (err) {
+                    assert.equal(err, error);
+                    done();
+                },
+            });
+        });
+    });
+    it('does not throw on closed subscription', function () {
+        var list = [];
+        var obs = zenObservable_1.default.from([1, 2, 3, 4]);
+        obs.subscribe({}).unsubscribe();
+        return assert.doesNotThrow(function () { return obs.filter(function (x) { return x > 2; }).forEach(function (x) { return list.push(x); }).then; });
+    });
+    it('does not throw on internally closed subscription', function () {
+        var list = [];
+        var obs = new zenObservable_1.default(function (observer) {
+            observer.next(1);
+            observer.next(1);
+            observer.complete();
+            observer.next(1);
+        });
+        return assert.doesNotThrow(function () { return obs.filter(function (x) { return x > 2; }).forEach(function (x) { return list.push(x); }).then; });
+    });
 });
 //# sourceMappingURL=filter.js.map

dist/tests/flatMap.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-var chai_1 = require("chai");
+var chai = require("chai");
 var zenObservable_1 = require("../src/zenObservable");
+var chaiAsPromised = require("chai-as-promised");
+chai.use(chaiAsPromised);
+var assert = chai.assert;
 describe('flatMap', function () {
@@ -16,3 +19,3 @@ it('Observable.from', function () {
             .then(function () {
-            chai_1.assert.deepEqual(list, [1, 2, 3, 2, 4, 6, 3, 6, 9]);
+            assert.deepEqual(list, [1, 2, 3, 2, 4, 6, 3, 6, 9]);
         });
@@ -26,5 +29,64 @@ });
             .forEach(function () { return null; })
-            .then(function () { return chai_1.assert(false); }, function () { return chai_1.assert(true); });
+            .then(function () { return assert(false); }, function () { return assert(true); });
     });
+    it('throws on not a function', function () {
+        return assert.throws(function () {
+            return zenObservable_1.default.from([1, 2, 3, 4]).flatMap(1).forEach(function (x) { return void 0; }).then;
+        });
+    });
+    it('throws on error inside function', function (done) {
+        var error = new Error('thrown');
+        return assert.doesNotThrow(function () {
+            return zenObservable_1.default.from([1, 2, 3, 4])
+                .flatMap(function () { throw error; })
+                .subscribe({
+                error: function (err) {
+                    assert.equal(err, error);
+                    done();
+                },
+            });
+        });
+    });
+    it('calls inner unsubscribe', function (done) {
+        zenObservable_1.default.from(zenObservable_1.default.of(1))
+            .flatMap(function (x) {
+            return new zenObservable_1.default(function (observer) { return done; });
+        })
+            .subscribe({})
+            .unsubscribe();
+    });
+    it('does not throw on closed subscription', function () {
+        var list = [];
+        var obs = zenObservable_1.default.from([1, 2, 3, 4]);
+        obs.subscribe({}).unsubscribe();
+        return assert.doesNotThrow(function () {
+            return obs
+                .flatMap(function (x) {
+                return zenObservable_1.default.from([x * 1, x * 2, x * 3]);
+            })
+                .forEach(function (x) {
+                list.push(x);
+            }).then;
+        });
+    });
+    it('does not throw on internally closed subscription', function () {
+        var list = [];
+        var obs = new zenObservable_1.default(function (observer) {
+            observer.next(1);
+            observer.next(1);
+            observer.complete();
+            observer.next(1);
+        });
+        obs.subscribe({}).unsubscribe();
+        return assert.doesNotThrow(function () {
+            return obs
+                .flatMap(function (x) {
+                return zenObservable_1.default.from([x * 1, x * 2, x * 3]);
+            })
+                .forEach(function (x) {
+                list.push(x);
+            }).then;
+        });
+    });
 });
 //# sourceMappingURL=flatMap.js.map

dist/tests/map.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-var chai_1 = require("chai");
+var chai = require("chai");
 var zenObservable_1 = require("../src/zenObservable");
+var chaiAsPromised = require("chai-as-promised");
+chai.use(chaiAsPromised);
+var assert = chai.assert;
 describe('map', function () {
@@ -11,5 +14,35 @@ it('Basics', function () {
             .forEach(function (x) { return list.push(x); })
-            .then(function () { return chai_1.assert.deepEqual(list, [2, 4, 6]); });
+            .then(function () { return assert.deepEqual(list, [2, 4, 6]); });
     });
+    it('throws on not a function', function () {
+        return assert.throws(function () { return zenObservable_1.default.from([1, 2, 3, 4]).map(1).forEach(function (x) { return void 0; }).then; });
+    });
+    it('throws on error inside function', function (done) {
+        var error = new Error('thrown');
+        return assert.doesNotThrow(function () {
+            return zenObservable_1.default.from([1, 2, 3, 4])
+                .map(function () { throw error; })
+                .subscribe({
+                error: function (err) {
+                    assert.equal(err, error);
+                    done();
+                },
+            });
+        });
+    });
+    it('does not throw on closed subscription', function () {
+        var obs = zenObservable_1.default.from([1, 2, 3, 4]);
+        obs.subscribe({}).unsubscribe();
+        return assert.doesNotThrow(function () { return obs.map(function (x) { return x * 2; }).forEach(function (x) { return void 0; }).then; });
+    });
+    it('does not throw on internally closed subscription', function () {
+        var obs = new zenObservable_1.default(function (observer) {
+            observer.next(1);
+            observer.next(1);
+            observer.complete();
+            observer.next(1);
+        });
+        return assert.doesNotThrow(function () { return obs.map(function (x) { return x * 2; }).forEach(function (x) { return void 0; }).then; });
+    });
 });
 //# sourceMappingURL=map.js.map

dist/tests/reduce.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-var chai_1 = require("chai");
+var chai = require("chai");
 var zenObservable_1 = require("../src/zenObservable");
+var chaiAsPromised = require("chai-as-promised");
+chai.use(chaiAsPromised);
+var assert = chai.assert;
 describe('reduce ', function () {
@@ -12,3 +15,3 @@ it('No seed', function () {
             .forEach(function (x) {
-            chai_1.assert.equal(x, 21);
+            assert.equal(x, 21);
         });
@@ -22,3 +25,3 @@ });
             .forEach(function (x) {
-            chai_1.assert.equal(x, 1);
+            assert.equal(x, 1);
         });
@@ -32,3 +35,3 @@ });
             .forEach(function () { return null; })
-            .then(function () { return chai_1.assert(false); }, function () { return chai_1.assert(true); });
+            .then(function () { return assert(false); }, function () { return assert(true); });
     });
@@ -41,3 +44,3 @@ it('Seed', function () {
             .forEach(function (x) {
-            chai_1.assert.equal(x, 121);
+            assert.equal(x, 121);
         });
@@ -51,6 +54,54 @@ });
             .forEach(function (x) {
-            chai_1.assert.equal(x, 100);
+            assert.equal(x, 100);
         });
     });
+    it('throws on not a function', function () {
+        return assert.throws(function () {
+            return zenObservable_1.default.from([1, 2, 3, 4]).reduce(1).forEach(function (x) { return void 0; }).then;
+        });
+    });
+    it('throws on error inside function', function (done) {
+        var error = new Error('thrown');
+        return assert.doesNotThrow(function () {
+            return zenObservable_1.default.from([1, 2, 3, 4])
+                .reduce(function () { throw error; })
+                .subscribe({
+                error: function (err) {
+                    assert.equal(err, error);
+                    done();
+                },
+            });
+        });
+    });
+    it('does not throw on closed subscription', function () {
+        var obs = zenObservable_1.default.from([1, 2, 3, 4]);
+        obs.subscribe({}).unsubscribe();
+        return assert.doesNotThrow(function () {
+            return obs
+                .reduce(function (a, b) {
+                return a + b;
+            }, 100)
+                .forEach(function (x) {
+                assert.equal(x, 110);
+            }).then;
+        });
+    });
+    it('does not throw on internally closed subscription', function () {
+        var obs = new zenObservable_1.default(function (observer) {
+            observer.next(1);
+            observer.next(1);
+            observer.complete();
+            observer.next(1);
+        });
+        return assert.doesNotThrow(function () {
+            return obs
+                .reduce(function (a, b) {
+                return a + b;
+            }, 100)
+                .forEach(function (x) {
+                assert.equal(x, 102);
+            }).then;
+        });
+    });
 });
 //# sourceMappingURL=reduce.js.map

dist/tests/tests.d.ts

@@ -0,4 +1,6 @@
+import './filter.js';
 import './flatMap.js';
+import './forEach.js';
 import './reduce.js';
 import './map.js';
-import './filter.js';
+import './observer.js';

dist/tests/tests.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+require("./filter.js");
 require("./flatMap.js");
+require("./forEach.js");
 require("./reduce.js");
 require("./map.js");
-require("./filter.js");
+require("./observer.js");
 //# sourceMappingURL=tests.js.map

lib/zenObservable.d.ts

@@ -13,3 +13,3 @@ import { ZenObservable } from './types';
     constructor(subscriber: ZenObservable.Subscriber<T>);
-    subscribe(observerOrNext: ZenObservable.Observer<T>, error?: (error: any) => void, complete?: () => void): ZenObservable.Subscription;
+    subscribe(observerOrNext: ((value: T) => void) | ZenObservable.Observer<T>, error?: (error: any) => void, complete?: () => void): ZenObservable.Subscription;
     forEach(fn: (value: T) => void): Promise<void>;
@@ -16,0 +16,0 @@ map<R>(fn: (value: T) => R): Observable<R>;

lib/zenObservable.js

@@ -43,3 +43,4 @@ "use strict";
             if (cleanup != null) {
-                if (typeof cleanup.unsubscribe === 'function') {
+                if (typeof cleanup.unsubscribe ===
+                    'function') {
                     cleanup = cleanupFromSubscription(cleanup);
@@ -192,7 +193,7 @@ }
         if (typeof observerOrNext === 'function') {
-            observerOrNext = {
+            return new Subscription({
                 next: observerOrNext,
                 error: error,
                 complete: complete,
-            };
+            }, this._subscriber);
         }
@@ -209,5 +210,2 @@ return new Subscription(observerOrNext, this._subscriber);
                 start: function (subscription) {
-                    if (Object(subscription) !== subscription) {
-                        throw new TypeError(subscription + ' is not an object');
-                    }
                     this._subscription = subscription;
@@ -214,0 +212,0 @@ },

package.json

 {
   "name": "zen-observable-ts",
-  "version": "0.0.0",
+  "version": "0.1.0",
   "description": "An Implementation of ES Observables in Typescript",
@@ -24,3 +24,3 @@ "author": "Evans Hauser <[e-mail]>",
     "test-watch": "mocha --reporter spec --full-trace dist/tests/tests.js --watch",
-    "coverage": "istanbul cover ./node_modules/mocha/bin/_mocha -- --reporter dot --full-trace dist/tests/tests.js",
+    "coverage": "istanbul cover ./node_modules/.bin/_mocha -- --reporter dot --full-trace dist/tests/tests.js",
     "postcoverage": "remap-istanbul --input coverage/coverage.json --type lcovonly --output coverage/lcov.info",
@@ -43,2 +43,3 @@ "lint": "tslint --type-check -p tsconfig.json src/*.ts && tslint --type-check -p tsconfig.json tests/*.ts",
     "chai": "^4.0.2",
+    "chai-as-promised": "^7.1.1",
     "istanbul": "^0.4.4",
@@ -45,0 +46,0 @@ "mocha": "^3.2.0",

src/zenObservable.ts

@@ -83,5 +83,8 @@ import { ZenObservable } from './types';
         if (
-          typeof (<ZenObservable.Subscription>cleanup).unsubscribe === 'function'
+          typeof (<ZenObservable.Subscription>cleanup).unsubscribe ===
+          'function'
         ) {
-          cleanup = cleanupFromSubscription(cleanup as ZenObservable.Subscription);
+          cleanup = cleanupFromSubscription(
+            cleanup as ZenObservable.Subscription,
+          );
         } else if (typeof cleanup !== 'function') {
@@ -209,3 +212,2 @@ throw new TypeError(cleanup + ' is not a function');
   ): Observable<R> {
-
     if ((<ZenObservable.ObservableLike<R>>observable).subscribe) {
@@ -227,4 +229,4 @@ return new Observable(observer =>
         if (observer.complete) {
-        observer.complete();
-              }
+          observer.complete();
+        }
       });
@@ -261,3 +263,3 @@ }
   public subscribe(
-    observerOrNext: ZenObservable.Observer<T>,
+    observerOrNext: ((value: T) => void) | ZenObservable.Observer<T>,
     error?: (error: any) => void,
@@ -267,7 +269,10 @@ complete?: () => void,
     if (typeof observerOrNext === 'function') {
-      observerOrNext = {
-        next: observerOrNext,
-        error,
-        complete,
-      };
+      return new Subscription(
+        {
+          next: observerOrNext,
+          error,
+          complete,
+        },
+        this._subscriber,
+      );
     }
@@ -284,30 +289,28 @@
 
-      this.subscribe({
-        start(subscription) {
-          if (Object(subscription) !== subscription) {
-            throw new TypeError(subscription + ' is not an object');
-          }
+      this.subscribe(
+        <ZenObservable.Observer<T>>{
+          start(subscription: ZenObservable.Subscription) {
+            this._subscription = subscription;
+          },
 
-          this._subscription = subscription;
-        },
+          next(value: T) {
+            let subscription = this._subscription;
 
-        next(value) {
-          let subscription = this._subscription;
+            if (subscription.closed) {
+              return;
+            }
 
-          if (subscription.closed) {
-            return;
-          }
+            try {
+              fn(value);
+              return;
+            } catch (err) {
+              reject(err);
+              subscription.unsubscribe();
+            }
+          },
 
-          try {
-            fn(value);
-            return;
-          } catch (err) {
-            reject(err);
-            subscription.unsubscribe();
-          }
+          error: reject,
+          complete: resolve,
         },
-
-        error: reject,
-        complete: resolve,
-      });
+      );
     });
@@ -338,3 +341,3 @@ }
         },
-        error(e) {
+        error(e: any) {
           observer.error(e);
@@ -354,6 +357,5 @@ },
 
-
     return new Observable(observer => {
       this.subscribe({
-        next(value) {
+        next(value: T) {
           if (observer.closed) {
@@ -377,3 +379,3 @@ return;
 
-        error(e) {
+        error(e: any) {
           observer.error(e);
@@ -401,5 +403,5 @@ },
 
-    return new Observable(observer => {
+    return new Observable<R | T>(observer => {
       this.subscribe({
-        next(value) {
+        next(value: R | T) {
           if (observer.closed) {
@@ -414,3 +416,3 @@ return;
             try {
-              acc = fn(acc, value);
+              acc = fn(acc, <T>value);
             } catch (e) {
@@ -425,3 +427,3 @@ observer.error(e);
 
-        error(e) {
+        error(e: any) {
           observer.error(e);
@@ -456,3 +458,3 @@ },
       let outer = this.subscribe({
-        next(value) {
+        next(value: T) {
           let _value: ZenObservable.ObservableLike<R>;
@@ -470,6 +472,6 @@ if (fn) {
           Observable.from(_value).subscribe({
-            start(s) {
+            start(s: ZenObservable.Subscription) {
               subscriptions.push((this._subscription = s));
             },
-            next(data) {
+            next(data: R) {
               observer.next(data);
@@ -476,0 +478,0 @@ },

tests/filter.ts

@@ -1,7 +0,11 @@
-import { assert } from 'chai';
+import * as chai from 'chai';
 import Observable from '../src/zenObservable';
+import * as chaiAsPromised from 'chai-as-promised';
 
+chai.use(chaiAsPromised);
+const assert = chai.assert;
+
 describe('filter ', () => {
   it('Basics', () => {
-    let list: Array<number> = [];
+    const list: Array<number> = [];
 
@@ -13,2 +17,50 @@ return Observable.from([1, 2, 3, 4])
   });
+
+  it('throws on not a function', () => {
+    const list: Array<number> = [];
+    return assert.throws(
+      () =>
+        Observable.from([1, 2, 3, 4]).filter(<any>1).forEach(x => list.push(x))
+          .then,
+    );
+  });
+
+  it('throws on error inside function', done => {
+    const error = new Error('thrown');
+    return assert.doesNotThrow(() =>
+      Observable.from([1, 2, 3, 4])
+        .filter(() => {
+          throw error;
+        })
+        .subscribe({
+          error: err => {
+            assert.equal(err, error);
+            done();
+          },
+        }),
+    );
+  });
+
+  it('does not throw on closed subscription', () => {
+    const list: Array<number> = [];
+    const obs = Observable.from([1, 2, 3, 4]);
+    obs.subscribe({}).unsubscribe();
+    return assert.doesNotThrow(
+      () => obs.filter(x => x > 2).forEach(x => list.push(x)).then,
+    );
+  });
+
+  it('does not throw on internally closed subscription', () => {
+    const list: Array<number> = [];
+    const obs = new Observable<number>(observer => {
+      observer.next(1);
+      observer.next(1);
+      observer.complete();
+      observer.next(1);
+    });
+
+    return assert.doesNotThrow(
+      () => obs.filter(x => x > 2).forEach(x => list.push(x)).then,
+    );
+  });
 });

tests/flatMap.ts

@@ -1,4 +0,8 @@
-import { assert } from 'chai';
+import * as chai from 'chai';
 import Observable from '../src/zenObservable';
+import * as chaiAsPromised from 'chai-as-promised';
 
+chai.use(chaiAsPromised);
+const assert = chai.assert;
+
 describe('flatMap', () => {
@@ -28,2 +32,71 @@ it('Observable.from', () => {
   });
+
+  it('throws on not a function', () => {
+    return assert.throws(
+      () =>
+        Observable.from([1, 2, 3, 4]).flatMap(<any>1).forEach(x => void 0).then,
+    );
+  });
+
+  it('throws on error inside function', done => {
+    const error = new Error('thrown');
+    return assert.doesNotThrow(() =>
+      Observable.from([1, 2, 3, 4])
+        .flatMap(() => {
+          throw error;
+        })
+        .subscribe({
+          error: err => {
+            assert.equal(err, error);
+            done();
+          },
+        }),
+    );
+  });
+
+  it('calls inner unsubscribe', done => {
+    Observable.from(Observable.of(1))
+      .flatMap(x => {
+        return new Observable(observer => done);
+      })
+      .subscribe({})
+      .unsubscribe();
+  });
+
+  it('does not throw on closed subscription', () => {
+    const list: Array<number> = [];
+    const obs = Observable.from([1, 2, 3, 4]);
+    obs.subscribe({}).unsubscribe();
+    return assert.doesNotThrow(
+      () =>
+        obs
+          .flatMap(x => {
+            return Observable.from([x * 1, x * 2, x * 3]);
+          })
+          .forEach(x => {
+            list.push(x);
+          }).then,
+    );
+  });
+
+  it('does not throw on internally closed subscription', () => {
+    const list: Array<number> = [];
+    const obs = new Observable<number>(observer => {
+      observer.next(1);
+      observer.next(1);
+      observer.complete();
+      observer.next(1);
+    });
+    obs.subscribe({}).unsubscribe();
+    return assert.doesNotThrow(
+      () =>
+        obs
+          .flatMap(x => {
+            return Observable.from([x * 1, x * 2, x * 3]);
+          })
+          .forEach(x => {
+            list.push(x);
+          }).then,
+    );
+  });
 });

tests/map.ts

@@ -1,4 +0,8 @@
-import { assert } from 'chai';
+import * as chai from 'chai';
 import Observable from '../src/zenObservable';
+import * as chaiAsPromised from 'chai-as-promised';
 
+chai.use(chaiAsPromised);
+const assert = chai.assert;
+
 describe('map', () => {
@@ -13,2 +17,44 @@ it('Basics', () => {
   });
+
+  it('throws on not a function', () => {
+    return assert.throws(
+      () => Observable.from([1, 2, 3, 4]).map(<any>1).forEach(x => void 0).then,
+    );
+  });
+
+  it('throws on error inside function', done => {
+    const error = new Error('thrown');
+    return assert.doesNotThrow(() =>
+      Observable.from([1, 2, 3, 4])
+        .map(() => {
+          throw error;
+        })
+        .subscribe({
+          error: err => {
+            assert.equal(err, error);
+            done();
+          },
+        }),
+    );
+  });
+
+  it('does not throw on closed subscription', () => {
+    const obs = Observable.from([1, 2, 3, 4]);
+    obs.subscribe({}).unsubscribe();
+    return assert.doesNotThrow(
+      () => obs.map(x => x * 2).forEach(x => void 0).then,
+    );
+  });
+
+  it('does not throw on internally closed subscription', () => {
+    const obs = new Observable<number>(observer => {
+      observer.next(1);
+      observer.next(1);
+      observer.complete();
+      observer.next(1);
+    });
+    return assert.doesNotThrow(
+      () => obs.map(x => x * 2).forEach(x => void 0).then,
+    );
+  });
 });

tests/reduce.ts

@@ -1,4 +0,8 @@
-import { assert } from 'chai';
+import * as chai from 'chai';
 import Observable from '../src/zenObservable';
+import * as chaiAsPromised from 'chai-as-promised';
 
+chai.use(chaiAsPromised);
+const assert = chai.assert;
+
 describe('reduce ', () => {
@@ -53,2 +57,60 @@ it('No seed', () => {
   });
+
+  it('throws on not a function', () => {
+    return assert.throws(
+      () =>
+        Observable.from([1, 2, 3, 4]).reduce(<any>1).forEach(x => void 0).then,
+    );
+  });
+
+  it('throws on error inside function', done => {
+    const error = new Error('thrown');
+    return assert.doesNotThrow(() =>
+      Observable.from([1, 2, 3, 4])
+        .reduce(() => {
+          throw error;
+        })
+        .subscribe({
+          error: err => {
+            assert.equal(err, error);
+            done();
+          },
+        }),
+    );
+  });
+
+  it('does not throw on closed subscription', () => {
+    const obs = Observable.from([1, 2, 3, 4]);
+
+    obs.subscribe({}).unsubscribe();
+    return assert.doesNotThrow(
+      () =>
+        obs
+          .reduce((a, b) => {
+            return a + b;
+          }, 100)
+          .forEach(x => {
+            assert.equal(x, 110);
+          }).then,
+    );
+  });
+
+  it('does not throw on internally closed subscription', () => {
+    const obs = new Observable<number>(observer => {
+      observer.next(1);
+      observer.next(1);
+      observer.complete();
+      observer.next(1);
+    });
+    return assert.doesNotThrow(
+      () =>
+        obs
+          .reduce((a, b) => {
+            return a + b;
+          }, 100)
+          .forEach(x => {
+            assert.equal(x, 102);
+          }).then,
+    );
+  });
 });

tests/tests.ts

@@ -0,4 +1,6 @@
+import './filter.js';
 import './flatMap.js';
+import './forEach.js';
 import './reduce.js';
 import './map.js';
-import './filter.js';
+import './observer.js';

No alert changes

Improved metrics

Total package byte prevSize

231585

increased by178.61%

Number of package files

59

increased by43.9%

Lines of code

2785

increased by61.92%

Worsened metrics

Dev dependency count

14

increased by7.69%

No dependency changes