fix(core): fix proper propagation of subscriptions in EventEmitter (#22016)

martinsik · alxhub · commit e81606c97aa4 · 2018-02-06T07:56:33.000-08:00
Closes #21999 PR Close #22016
diff --git a/packages/core/src/event_emitter.ts b/packages/core/src/event_emitter.ts
@@ -7,6 +7,7 @@
  */
 
 import {Subject} from 'rxjs/Subject';
+import {Subscription} from 'rxjs/Subscription';
 
 /**
  * Use by directives and components to emit custom Events.
@@ -111,6 +112,12 @@ export class EventEmitter<T> extends Subject<T> {
       }
     }
 
-    return super.subscribe(schedulerFn, errorFn, completeFn);
+    const sink = super.subscribe(schedulerFn, errorFn, completeFn);
+
+    if (generatorOrNext instanceof Subscription) {
+      generatorOrNext.add(sink);
+    }
+
+    return sink;
   }
 }
diff --git a/packages/core/test/event_emitter_spec.ts b/packages/core/test/event_emitter_spec.ts
@@ -7,6 +7,8 @@
  */
 
 import {AsyncTestCompleter, beforeEach, describe, expect, inject, it} from '@angular/core/testing/src/testing_internal';
+import {filter} from 'rxjs/operators';
+
 import {EventEmitter} from '../src/event_emitter';
 
 {
@@ -125,6 +127,59 @@ import {EventEmitter} from '../src/event_emitter';
       expect(e.observers.length > 0).toBe(true);
     });
 
+    it('remove a subscriber subscribed directly to EventEmitter', () => {
+      const sub = emitter.subscribe();
+      expect(emitter.observers.length).toBe(1);
+      sub.unsubscribe();
+      expect(emitter.observers.length).toBe(0);
+    });
+
+    it('remove a subscriber subscribed after applying operators with pipe()', () => {
+      const sub = emitter.pipe(filter(() => true)).subscribe();
+      expect(emitter.observers.length).toBe(1);
+      sub.unsubscribe();
+      expect(emitter.observers.length).toBe(0);
+    });
+
+    it('unsubscribing a subscriber invokes the dispose method', () => {
+      inject([AsyncTestCompleter], (async: AsyncTestCompleter) => {
+        const sub = emitter.subscribe();
+        sub.add(() => async.done());
+        sub.unsubscribe();
+      });
+    });
+
+    it('unsubscribing a subscriber after applying operators with pipe() invokes the dispose method',
+       () => {
+         inject([AsyncTestCompleter], (async: AsyncTestCompleter) => {
+           const sub = emitter.pipe(filter(() => true)).subscribe();
+           sub.add(() => async.done());
+           sub.unsubscribe();
+         });
+       });
+
+    it('error thrown inside an Rx chain propagates to the error handler and disposes the chain',
+       () => {
+         let errorPropagated = false;
+         emitter.pipe(filter(() => { throw new Error(); }), )
+             .subscribe(() => {}, err => errorPropagated = true, );
+
+         emitter.next(1);
+
+         expect(errorPropagated).toBe(true);
+         expect(emitter.observers.length).toBe(0);
+       });
+
+    it('error sent by EventEmitter should dispose the Rx chain and remove subscribers', () => {
+      let errorPropagated = false;
+      emitter.pipe(filter(() => true)).subscribe(() => {}, err => errorPropagated = true, );
+
+      emitter.error(1);
+
+      expect(errorPropagated).toBe(true);
+      expect(emitter.observers.length).toBe(0);
+    });
+
     // TODO: vsavkin: add tests cases
     // should call dispose on the subscription if generator returns {done:true}
     // should call dispose on the subscription on throw

Original file line number	Diff line number	Diff line change
`@@ -7,6 +7,7 @@`
`7`	`7`	`*/`
`8`	`8`
`9`	`9`	`import {Subject} from 'rxjs/Subject';`
	`10`	`+import {Subscription} from 'rxjs/Subscription';`
`10`	`11`
`11`	`12`	`/**`
`12`	`13`	`* Use by directives and components to emit custom Events.`
`@@ -111,6 +112,12 @@ export class EventEmitter<T> extends Subject<T> {`
`111`	`112`	`}`
`112`	`113`	`}`
`113`	`114`
`114`		`- return super.subscribe(schedulerFn, errorFn, completeFn);`
	`115`	`+ const sink = super.subscribe(schedulerFn, errorFn, completeFn);`
	`116`	`+`
	`117`	`+ if (generatorOrNext instanceof Subscription) {`
	`118`	`+ generatorOrNext.add(sink);`
	`119`	`+ }`
	`120`	`+`
	`121`	`+ return sink;`
`115`	`122`	`}`
`116`	`123`	`}`