test: add firehose tests

spec/operators/audit-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { TestScheduler } from 'rxjs/testing';
-import { of, interval, EMPTY } from 'rxjs';
+import { of, interval, EMPTY, Observable } from 'rxjs';
 import { audit, take, mergeMap } from 'rxjs/operators';
 import { observableMatcher } from '../helpers/observableMatcher';
 
@@ -436,4 +436,23 @@ describe('audit operator', () => {
       }
     );
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      audit(() => of(0)),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });

spec/operators/bufferCount-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
-import { Subject, of } from 'rxjs';
-import { bufferCount, mergeMap } from 'rxjs/operators';
+import { Subject, of, Observable } from 'rxjs';
+import { bufferCount, mergeMap, take } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { observableMatcher } from '../helpers/observableMatcher';
 
@@ -163,4 +163,23 @@ describe('bufferCount operator', () => {
       expectSubscriptions(e1.subscriptions).toBe(e1subs);
     });
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      bufferCount(1),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });

spec/operators/bufferWhen-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
-import { of, EMPTY } from 'rxjs';
-import { bufferWhen, mergeMap, takeWhile } from 'rxjs/operators';
+import { of, EMPTY, Observable } from 'rxjs';
+import { bufferWhen, mergeMap, takeWhile, take } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { observableMatcher } from '../helpers/observableMatcher';
 

spec/operators/catch-spec.ts → spec/operators/catchError-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { concat, defer, Observable, of, throwError, EMPTY, from } from 'rxjs';
-import { catchError, map, mergeMap, takeWhile, delay } from 'rxjs/operators';
+import { catchError, map, mergeMap, takeWhile, delay, take } from 'rxjs/operators';
 import * as sinon from 'sinon';
 import { createObservableInputs } from '../helpers/test-helper';
 import { TestScheduler } from 'rxjs/testing';
@@ -457,4 +457,24 @@ describe('catchError operator', () => {
     );
   });
 
+  // TODO: fix firehose unsubscription
+  it.skip('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      catchError(() => EMPTY),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
+
 });

spec/operators/concatAll-spec.ts

@@ -1,5 +1,5 @@
 import { expect } from 'chai';
-import { from, throwError, of } from 'rxjs';
+import { from, throwError, of, Observable } from 'rxjs';
 import { concatAll, take, mergeMap } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { observableMatcher } from '../helpers/observableMatcher';
@@ -520,4 +520,23 @@ describe('concatAll operator', () => {
       expectSubscriptions(e1.subscriptions).toBe(e1subs);
     });
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    of(synchronousObservable).pipe(
+      concatAll(),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });

spec/operators/concatMap-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { of, from, Observable } from 'rxjs';
-import { concatMap, mergeMap, map } from 'rxjs/operators';
+import { concatMap, mergeMap, map, take } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { observableMatcher } from '../helpers/observableMatcher';
 
@@ -808,4 +808,23 @@ describe('Observable.prototype.concatMap', () => {
       }
     );
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      concatMap(value => of(value)),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });

spec/operators/concatMapTo-spec.ts

@@ -1,7 +1,7 @@
 import { expect } from 'chai';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { of, from } from 'rxjs';
-import { concatMapTo, mergeMap } from 'rxjs/operators';
+import { of, from, Observable } from 'rxjs';
+import { concatMapTo, mergeMap, take } from 'rxjs/operators';
 
 /** @test {concatMapTo} */
 describe('Observable.prototype.concatMapTo', () => {
@@ -356,4 +356,23 @@ describe('Observable.prototype.concatMapTo', () => {
         done(new Error('Subscriber complete handler not supposed to be called.'));
       });
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      concatMapTo(of(0)),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });

spec/operators/concatWith-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { of, Observable } from 'rxjs';
-import { concatWith, mergeMap } from 'rxjs/operators';
+import { concatWith, mergeMap, take } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { assertDeepEquals, NO_SUBS } from '../helpers/test-helper';
 
@@ -338,4 +338,23 @@ describe('concat operator', () => {
       expectSubscriptions(e1.subscriptions).toBe(e1subs);
     });
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      concatWith(of(0)),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });

spec/operators/debounce-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
-import { NEVER, timer, of, EMPTY, concat, Subject } from 'rxjs';
-import { debounce, mergeMap, mapTo } from 'rxjs/operators';
+import { NEVER, timer, of, EMPTY, concat, Subject, Observable } from 'rxjs';
+import { debounce, mergeMap, mapTo, take } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
 
@@ -431,4 +431,23 @@ describe('debounce operator', () => {
 
     expect(results).to.deep.equal([1, 2]);
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      debounce(() => of(0)),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });

spec/operators/defaultIfEmpty-spec.ts

@@ -1,6 +1,7 @@
+import { expect } from 'chai';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { of } from 'rxjs';
-import { defaultIfEmpty, mergeMap } from 'rxjs/operators';
+import { of, Observable } from 'rxjs';
+import { defaultIfEmpty, mergeMap, take } from 'rxjs/operators';
 
 /** @test {defaultIfEmpty} */
 describe('Observable.prototype.defaultIfEmpty', () => {
@@ -84,4 +85,23 @@ describe('Observable.prototype.defaultIfEmpty', () => {
     expectObservable(e1.pipe(defaultIfEmpty('x'))).toBe(expected);
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      defaultIfEmpty(0),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });

spec/operators/dematerialize-spec.ts

@@ -1,5 +1,6 @@
-import { of, Notification, ObservableNotification } from 'rxjs';
-import { dematerialize, map, mergeMap, materialize } from 'rxjs/operators';
+import { expect } from 'chai';
+import { of, Notification, ObservableNotification, Observable } from 'rxjs';
+import { dematerialize, map, mergeMap, materialize, take } from 'rxjs/operators';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
 
 const NO_VALUES: { [key: string]: ObservableNotification<any> } = {};
@@ -174,4 +175,24 @@ describe('dematerialize operator', () => {
     );
     expectObservable(result).toBe(expected);
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable<number>(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      materialize(),
+      dematerialize(),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });

spec/operators/distinct-spec.ts

@@ -1,6 +1,7 @@
-import { distinct, mergeMap } from 'rxjs/operators';
+import { expect } from 'chai';
+import { distinct, mergeMap, take } from 'rxjs/operators';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { of } from 'rxjs';
+import { of, Observable } from 'rxjs';
 
 /** @test {distinct} */
 describe('distinct operator', () => {
@@ -212,4 +213,23 @@ describe('distinct operator', () => {
     expectSubscriptions(e1.subscriptions).toBe(e1subs);
     expectSubscriptions(e2.subscriptions).toBe(e2subs);
   });
+
+  it('should stop listening to a synchronous observable when unsubscribed', () => {
+    const sideEffects: number[] = [];
+    const synchronousObservable = new Observable<number>(subscriber => {
+      // This will check to see if the subscriber was closed on each loop
+      // when the unsubscribe hits (from the `take`), it should be closed
+      for (let i = 0; !subscriber.closed && i < 10; i++) {
+        sideEffects.push(i);
+        subscriber.next(i);
+      }
+    });
+
+    synchronousObservable.pipe(
+      distinct(),
+      take(3),
+    ).subscribe(() => { /* noop */ });
+
+    expect(sideEffects).to.deep.equal([0, 1, 2]);
+  });
 });