fix(Subscriber): Can no longer subscribe to itself in a circular manner (#4106)

* test(mergeMap): add failing test
* fix(subscriber): don't unsubscribe self
When unsubscribing a subscriber's parent, make sure that the subscriber
itself is not unsubscribed.
Closes#4095
* refactor(mergeMap): simplify
* chore(typings): use union type for destination
* chore(test): remove only
* chore(test): use pipe
* test(Subscriber): fake add too
* test(internals): update for subscription changes
* refactor(zip): add to destination
* refactor(delay): add to destination
* refactor(delayWhen): add to destination
* refactor(exhaustMap): add to destination
* refactor(expand): add to destination
* refactor(mergeScan): add to destination
* refactor(observeOn): add to destination
* refactor(onErrorResumeNext): add to destination
* refactor(sequenceEqual): add to destination
* refactor(switchMap): add to destination
* chore(Subscriber): remove _addParentTeardownLogic
* chore(test): simplify mergeMap test
Remove the mapTo and the concat to reduce the number of subscribers to
make the test easier to reason with.

Author: cartant

spec/Subscriber-spec.ts

@@ -23,7 +23,8 @@ describe('Subscriber', () => {
 it('should accept subscribers as a destination if they meet the proper criteria',()=>{
 constfakeSubscriber={
 [rxSubscriber](this: any){returnthis;},
-_addParentTeardownLogic(){/* noop */}
+add(){/* noop */},
+syncErrorThrowable: false
 };
 
 constsubscriber=newSubscriber(fakeSubscriberasany);

spec/operators/mergeMap-spec.ts

@@ -1,6 +1,6 @@
 import{expect}from'chai';
 import{mergeMap,map}from'rxjs/operators';
-import{asapScheduler,defer,Observable,from,of}from'rxjs';
+import{asapScheduler,defer,Observable,from,of,timer}from'rxjs';
 import{hot,cold,expectObservable,expectSubscriptions}from'../helpers/marble-testing';
 
 declareconsttype: Function;
@@ -717,7 +717,7 @@ describe('mergeMap', () => {
 // Added as a failing test when investigating:
 // https://github.com/ReactiveX/rxjs/issues/4071
 
-constresults: any[]=[];
+constresults: (number|string)[]=[];
 
 of(1).pipe(
 mergeMap(()=>defer(()=>
@@ -744,7 +744,7 @@ describe('mergeMap', () => {
 // Added as a failing test when investigating:
 // https://github.com/ReactiveX/rxjs/issues/4071
 
-constresults: any[]=[];
+constresults: (number|string)[]=[];
 
 of(1).pipe(
 mergeMap(()=>
@@ -764,6 +764,30 @@ describe('mergeMap', () => {
 },0);
 });
 
+it('should support wrapped sources',(done: MochaDone)=>{
+
+// Added as a failing test when investigating:
+// https://github.com/ReactiveX/rxjs/issues/4095
+
+constresults: (number|string)[]=[];
+
+constwrapped=newObservable<number>(subscriber=>{
+constsubscription=timer(0,asapScheduler).subscribe(subscriber);
+return()=>subscription.unsubscribe();
+});
+wrapped.pipe(
+mergeMap(()=>timer(0,asapScheduler))
+).subscribe({
+next(value){results.push(value);},
+complete(){results.push('done');}
+});
+
+setTimeout(()=>{
+expect(results).to.deep.equal([0,'done']);
+done();
+},0);
+});
+
 type('should support type signatures',()=>{
 leto: Observable<number>;
 

spec/operators/observeOn-spec.ts

@@ -103,23 +103,21 @@ describe('observeOn operator', () => {
 .pipe(observeOn(asapScheduler))
 .subscribe(
 x=>{
-constobserveOnSubscriber=subscription._subscriptions[0];
-expect(observeOnSubscriber._subscriptions.length).to.equal(2);// one for the consumer, and one for the notification
-expect(observeOnSubscriber._subscriptions[1].state.notification.kind)
-.to.equal('N');
-expect(observeOnSubscriber._subscriptions[1].state.notification.value)
-.to.equal(x);
+// see #4106 - inner subscriptions are now added to destinations
+// so the subscription will contain an ObserveOnSubscriber and a subscription for the scheduled action
+expect(subscription._subscriptions.length).to.equal(2);
+constactionSubscription=subscription._subscriptions[1];
+expect(actionSubscription.state.notification.kind).to.equal('N');
+expect(actionSubscription.state.notification.value).to.equal(x);
 results.push(x);
 },
 err=>done(err),
 ()=>{
 // now that the last nexted value is done, there should only be a complete notification scheduled
 // the consumer will have been unsubscribed via Subscriber#_parentSubscription
-constobserveOnSubscriber=subscription._subscriptions[0];
-expect(observeOnSubscriber._subscriptions.length).to.equal(1);// one for the complete notification
-// only this completion notification should remain.
-expect(observeOnSubscriber._subscriptions[0].state.notification.kind)
-.to.equal('C');
+expect(subscription._subscriptions.length).to.equal(1);
+constactionSubscription=subscription._subscriptions[0];
+expect(actionSubscription.state.notification.kind).to.equal('C');
 // After completion, the entire _subscriptions list is nulled out anyhow, so we can't test much further than this.
 expect(results).to.deep.equal([1,2,3]);
 done();

spec/operators/switch-spec.ts

@@ -1,7 +1,7 @@
 import{expect}from'chai';
 import{hot,cold,expectObservable,expectSubscriptions}from'../helpers/marble-testing';
 import{Observable,of,NEVER,queueScheduler,Subject}from'rxjs';
-import{switchAll}from'rxjs/operators';
+import{map,switchAll}from'rxjs/operators';
 
 declarefunctionasDiagram(arg: string): Function;
 declareconsttype: Function;
@@ -222,9 +222,9 @@ describe('switchAll', () => {
 it('should not leak when child completes before each switch (prevent memory leaks #2355)',()=>{
 letiStream: Subject<number>;
 constoStreamControl=newSubject<number>();
-constoStream=oStreamControl.map(()=>{
-return(iStream=newSubject<number>());
-});
+constoStream=oStreamControl.pipe(
+map(()=>(iStream=newSubject<number>()))
+);
 constswitcher=oStream.pipe(switchAll());
 constresult: number[]=[];
 letsub=switcher.subscribe((x)=>result.push(x));
@@ -242,19 +242,24 @@ describe('switchAll', () => {
 
 it('should not leak if we switch before child completes (prevent memory leaks #2355)',()=>{
 constoStreamControl=newSubject<number>();
-constoStream=oStreamControl.map(()=>{
-return(newSubject<number>());
-});
+constoStream=oStreamControl.pipe(
+map(()=>newSubject<number>())
+);
 constswitcher=oStream.pipe(switchAll());
 constresult: number[]=[];
 letsub=switcher.subscribe((x)=>result.push(x));
 
 [0,1,2,3,4].forEach((n)=>{
 oStreamControl.next(n);// creates inner
 });
-// Expect two children of switch(): The oStream and the first inner
+// Expect one child of switch(): The oStream
 expect(
 (subasany)._subscriptions[0]._subscriptions.length
+).to.equal(1);
+// Expect two children of subscribe(): The destination and the first inner
+// See #4106 - inner subscriptions are now added to destinations
+expect(
+(subasany)._subscriptions.length
 ).to.equal(2);
 sub.unsubscribe();
 });

src/internal/Observable.ts

@@ -199,7 +199,7 @@ export class Observable<T> implements Subscribable<T> {
 if(operator){
 operator.call(sink,this.source);
 }else{
-sink._addParentTeardownLogic(
+sink.add(
 this.source||(config.useDeprecatedSynchronousErrorHandling&&!sink.syncErrorThrowable) ?
 this._subscribe(sink) :
 this._trySubscribe(sink)

src/internal/Subscriber.ts

@@ -45,7 +45,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
 /** @internal */syncErrorThrowable: boolean=false;
 
 protectedisStopped: boolean=false;
-protecteddestination: PartialObserver<any>;// this `any` is the escape hatch to erase extra type param (e.g. R)
+protecteddestination: PartialObserver<any>|Subscriber<any>;// this `any` is the escape hatch to erase extra type param (e.g. R)
 
 private_parentSubscription: Subscription|null=null;
 
@@ -78,7 +78,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
 consttrustedSubscriber=destinationOrNext[rxSubscriberSymbol]()asSubscriber<any>;
 this.syncErrorThrowable=trustedSubscriber.syncErrorThrowable;
 this.destination=trustedSubscriber;
-trustedSubscriber._addParentTeardownLogic(this);
+trustedSubscriber.add(this);
 }else{
 this.syncErrorThrowable=true;
 this.destination=newSafeSubscriber<T>(this,<PartialObserver<any>>destinationOrNext);
@@ -116,7 +116,6 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
 if(!this.isStopped){
 this.isStopped=true;
 this._error(err);
-this._unsubscribeParentSubscription();
 }
 }
 
@@ -130,7 +129,6 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
 if(!this.isStopped){
 this.isStopped=true;
 this._complete();
-this._unsubscribeParentSubscription();
 }
 }
 
@@ -156,20 +154,6 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
 this.unsubscribe();
 }
 
-/** @deprecated This is an internal implementation detail, do not use. */
-_addParentTeardownLogic(parentTeardownLogic: TeardownLogic){
-if(parentTeardownLogic!==this){
-this._parentSubscription=this.add(parentTeardownLogic);
-}
-}
-
-/** @deprecated This is an internal implementation detail, do not use. */
-_unsubscribeParentSubscription(){
-if(this._parentSubscription!==null){
-this._parentSubscription.unsubscribe();
-}
-}
-
 /** @deprecated This is an internal implementation detail, do not use. */
 _unsubscribeAndRecycle(): Subscriber<T>{
 const{ _parent, _parents }=this;
@@ -326,5 +310,5 @@ export class SafeSubscriber<T> extends Subscriber<T> {
 }
 
 exportfunctionisTrustedSubscriber(obj: any){
-returnobjinstanceofSubscriber||('_addParentTeardownLogic'inobj&&obj[rxSubscriberSymbol]);
+returnobjinstanceofSubscriber||('syncErrorThrowable'inobj&&obj[rxSubscriberSymbol]);
 }

src/internal/Subscription.ts

@@ -45,7 +45,6 @@ export class Subscription implements SubscriptionLike {
 constructor(unsubscribe?: ()=>void){
 if(unsubscribe){
 (<any>this)._unsubscribe=unsubscribe;
-
 }
 }
 

src/internal/observable/zip.ts

@@ -4,6 +4,7 @@ import { isArray } from '../util/isArray';
 import{Operator}from'../Operator';
 import{ObservableInput,PartialObserver}from'../types';
 import{Subscriber}from'../Subscriber';
+import{Subscription}from'../Subscription';
 import{OuterSubscriber}from'../OuterSubscriber';
 import{InnerSubscriber}from'../InnerSubscriber';
 import{subscribeToResult}from'../util/subscribeToResult';
@@ -126,6 +127,8 @@ export class ZipSubscriber<T, R> extends Subscriber<T> {
 constiterators=this.iterators;
 constlen=iterators.length;
 
+this.unsubscribe();
+
 if(len===0){
 this.destination.complete();
 return;
@@ -135,7 +138,8 @@ export class ZipSubscriber<T, R> extends Subscriber<T> {
 for(leti=0;i<len;i++){
 letiterator: ZipBufferIterator<any,any>=<any>iterators[i];
 if(iterator.stillUnsubscribed){
-this.add(iterator.subscribe(iterator,i));
+constdestination=this.destinationasSubscription;
+destination.add(iterator.subscribe(iterator,i));
 }else{
 this.active--;// not an observable
 }

src/internal/operators/delay.ts

@@ -2,6 +2,7 @@ import { async } from '../scheduler/async';
 import{isDate}from'../util/isDate';
 import{Operator}from'../Operator';
 import{Subscriber}from'../Subscriber';
+import{Subscription}from'../Subscription';
 import{Notification}from'../Notification';
 import{Observable}from'../Observable';
 import{MonoTypeOperatorFunction,PartialObserver,SchedulerAction,SchedulerLike,TeardownLogic}from'../types';
@@ -110,7 +111,8 @@ class DelaySubscriber<T> extends Subscriber<T> {
 
 private_schedule(scheduler: SchedulerLike): void{
 this.active=true;
-this.add(scheduler.schedule<DelayState<T>>(DelaySubscriber.dispatch,this.delay,{
+constdestination=this.destinationasSubscription;
+destination.add(scheduler.schedule<DelayState<T>>(DelaySubscriber.dispatch,this.delay,{
 source: this,destination: this.destination,scheduler: scheduler
 }));
 }
@@ -137,10 +139,12 @@ class DelaySubscriber<T> extends Subscriber<T> {
 this.errored=true;
 this.queue=[];
 this.destination.error(err);
+this.unsubscribe();
 }
 
 protected_complete(){
 this.scheduleNotification(Notification.createComplete());
+this.unsubscribe();
 }
 }
 

src/internal/operators/delayWhen.ts

@@ -132,6 +132,7 @@ class DelayWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
 protected_complete(): void{
 this.completed=true;
 this.tryComplete();
+this.unsubscribe();
 }
 
 privateremoveSubscription(subscription: InnerSubscriber<T,R>): T{
@@ -149,7 +150,8 @@ class DelayWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
 constnotifierSubscription=subscribeToResult(this,delayNotifier,value);
 
 if(notifierSubscription&&!notifierSubscription.closed){
-this.add(notifierSubscription);
+constdestination=this.destinationasSubscription;
+destination.add(notifierSubscription);
 this.delayNotifierSubscriptions.push(notifierSubscription);
 }
 }
@@ -199,6 +201,7 @@ class SubscriptionDelaySubscriber<T> extends Subscriber<T> {
 }
 
 protected_complete(){
+this.unsubscribe();
 this.subscribeToSource();
 }