fix(Subscriber): don't leak destination

[cartant]

Description:

This PR adds a failing test that shows that subscriptions - i.e. subscribers - returned from subscribe leak their destinations - i.e. their observers or callbacks.

The PR changes Subscriber#unsubscribe to null the destination after super.unsubscribe is called.

This change broke three operators - bufferWhen was one of them - because if unsubscription was effected within onNext and if an error was thrown from within onNext (after the unsubscription) it'd be caught in the OperatorSubscriber and an attempt would be made to call this.destination.error, but this.destination would be null.

Rather than capture this.destination before the try/catch around onNext, onComplete, and onError, I just used the destination that was passed in to the OperatorSubscriber constructor - AFAICT, we can be sure that it's a Subscriber, because OperatorSubscriber is an internal, impementation detail.

Related issue (if exists): Nope, but there is a discussion: #6115

[benlesh]

I think this is okay... I'd just want to see if this works okay when you chain together a few operators. We altered OperatorSubscriber, but there's no new tests for operators related to that, so it has me a little concerned.

[cartant]

but there's no new tests for operators related to that

I'm not sure what additional tests you might be looking for. There are no explicit tests for OperatorSubscriber, AFAICT; it's tested via the operator tests themselves. And regarding operators being chained, pretty much every operator has a don't-break-subscription-chains test with something like this in it - i.e. several operators chained together:

rxjs/spec/operators/mergeMap-spec.ts

Lines 307 to 311 in 3ab3f6f

	const source = e1.pipe(
	map((x) => x),
	mergeMap(() => x),
	map((x) => x)
	);

Regarding the OperatorSubscriber change, these are the tests that fail without it. And they all fail because the subscribers to the selector-returned observables unsubscribe themselves and if a error is thrown from within the selector and is caught in OperatorSubscriber, this.destination will be null, etc.:

rxjs/spec/operators/bufferWhen-spec.ts

Lines 184 to 211 in 3ab3f6f

	it('should propagate error thrown from closingSelector', () => {
	testScheduler.run(({ hot, cold, expectObservable, expectSubscriptions }) => {
	const e1 = hot('--a--^---b---c---d---e---f---g---h------| ');
	const subs = ' ^--------------! ';
	const closings = [
	cold(' ---------------s--| '),
	cold(' ----------(s|) '),
	cold(' -------------(s|)')
	];
	const closeSubs0 = ' ^--------------! ';
	const expected = ' ---------------(x#) ';
	const values = { x: ['b', 'c', 'd'] };
	let i = 0;
	const result = e1.pipe(
	bufferWhen(() => {
	if (i === 1) {
	throw 'error';
	}
	return closings[i++];
	})
	);
	expectObservable(result).toBe(expected, values);
	expectSubscriptions(e1.subscriptions).toBe(subs);
	expectSubscriptions(closings[0].subscriptions).toBe(closeSubs0);
	});
	});

rxjs/spec/operators/zipAll-spec.ts

Lines 345 to 365 in 3ab3f6f

	it('should work with selector throws', () => {
	rxTestScheduler.run(({ hot, expectObservable, expectSubscriptions }) => {
	const a = hot('---1-^-2---4----| ');
	const asubs = ' ^-------! ';
	const b = hot('---1-^--3----5----|');
	const bsubs = ' ^-------! ';
	const expected = ' ---x----# ';
	const selector = function (x: string, y: string) {
	if (y === '5') {
	throw new Error('too bad');
	} else {
	return x + y;
	}
	};
	const observable = of(a, b).pipe(zipAll(selector));
	expectObservable(observable).toBe(expected, { x: '23' }, new Error('too bad'));
	expectSubscriptions(a.subscriptions).toBe(asubs);
	expectSubscriptions(b.subscriptions).toBe(bsubs);
	});
	});

rxjs/spec/operators/zipAll-spec.ts

Lines 222 to 240 in 3ab3f6f

	it('should work with non-empty observable and non-empty iterable selector that throws', () => {
	rxTestScheduler.run(({ hot, expectObservable, expectSubscriptions }) => {
	const a = hot('---^--1--2--3--|');
	const asubs = ' ^-----!';
	const b = [4, 5, 6];
	const expected = '---x--#';
	const selector = function (x: string, y: number) {
	if (y === 5) {
	throw new Error('too bad');
	} else {
	return x + y;
	}
	};
	expectObservable(of(a, b).pipe(zipAll(selector))).toBe(expected, { x: '14' }, new Error('too bad'));
	expectSubscriptions(a.subscriptions).toBe(asubs);
	});
	});
	});

So I'm not sure what else needs to be tested.

[benlesh]

That makes sense, thanks @cartant