fix(Observable): Fix Observable.subscribe to add operator TeardownLogic to returned Subscription

[PSanetra]

The Observable.subscribe method does not add TeardownLogic of a lifted Operator to the returned Subscription. Therefore the TeardownLogic of such an Operator is never called on unsubscriptions. This PR fixes that issue.

[cartant]

What's the use case that requires this change?

[PSanetra]

Every Operator-interface implementation that returns a TeardownLogic could require this change. You could make an alternative change which removes the return type from the call method of the Operator-interface, because the return value is simply ignored at the moment.

E.g. the code coverage report of the shareReplay operator shows that its TeardownLogic is never executed in any test: https://coveralls.io/builds/20480419/source?filename=src/internal/operators/shareReplay.ts#L88

[cartant]

I'm curious as to whether your opening this PR had something to do with your looking at shareReplay. Did it?

That operator's current implementation has a bug and when this PR is merged the bug will be fixed and the implementation will no longer return a teardown function. I'm also wondering whether there are any other operator implementations that return teardown functions - I suspect there aren't.

[PSanetra]

@cartant yes, I remembered from that discussion that this issue exists, but fixing this issue does not change the behavior of shareReplay in a visible way as the unit tests have shown.

I still think that the existence of this issue is at least a (publicly visible) code smell which should be fixed, either by respecting the contract of the call method signature and therefore calling the TeardownLogic on unsubscriptions or by removing the return type from the call method signature. At the moment you could think by looking at the call signature that you could return a TeardownLogic which is called eventually, but in fact this never happens.

I'm also wondering whether there are any other operator implementations that return teardown functions - I suspect there aren't.

I suspect or hope this too, but we don't know since the Operator-interface is public API.

[cartant]

fixing this issue does not change the behavior of shareReplay

I was wondering whether you were looking at this change going some way to fixing the behaviour of shareReplay, that's all. It won't - as you've pointed out.

Although it won't be used anywhere in RxJS's internals - once the above-referenced PR is merged - I don't think the return type can be removed. It would be a breaking change as far as the TypeScript declarations are concerned. I'll have a closer look at what you've proposed, in a short while.

[cartant]

I don't see why this is necessary. How does this relate to the change you made in Observable.ts? If it's not directly related to that change - and it solves another problem - it should be in a separate PR.

[PSanetra]

This change prevents three unit tests from failing. Otherwise the change in Observable.ts would add Subscribers a second time to the _subscriptions array, because most built-in operators have their own Subscriber implementation which already add themself to the array:

rxjs/src/internal/Subscriber.ts

Line 76 in ef557bb

destinationOrNext.add(this);

Here are the failing unit tests:

3 failing

  1) observeOn operator
       should clean up subscriptions created by async scheduling (prevent memory leaks #2244):

      Uncaught AssertionError: expected 3 to equal 2
      + expected - actual

      -3
      +2
      
      at SafeSubscriber._next (spec/operators/observeOn-spec.ts:108:57)
      at SafeSubscriber.__tryOrUnsub (src/internal/Subscriber.ts:267:10)
      at SafeSubscriber.next (src/internal/Subscriber.ts:209:14)
      at Subscriber._next (src/internal/Subscriber.ts:139:22)
      at Subscriber.next (src/internal/Subscriber.ts:99:12)
      at Notification.observe (src/internal/Notification.ts:36:42)
      at AsapAction.ObserveOnSubscriber.dispatch (src/internal/operators/observeOn.ts:81:18)
      at AsapAction.AsyncAction._execute (src/internal/scheduler/AsyncAction.ts:121:12)
      at AsapAction.AsyncAction.execute (src/internal/scheduler/AsyncAction.ts:96:24)
      at AsapScheduler.flush (src/internal/scheduler/AsapScheduler.ts:17:26)
      at runIfPresent (src/internal/util/Immediate.ts:8:5)
      at src/internal/util/Immediate.ts:16:34
      at <anonymous>

  2) switchAll
       should not leak when child completes before each switch (prevent memory leaks #2355):

      AssertionError: expected 2 to equal 1
      + expected - actual

      -2
      +1
      
      at spec/operators/switch-spec.ts:239:10
      at Context.modified (spec/helpers/testScheduler-ui.ts:206:13)

  3) switchAll
       should not leak if we switch before child completes (prevent memory leaks #2355):

      AssertionError: expected 2 to equal 1
      + expected - actual

      -2
      +1
      
      at spec/operators/switch-spec.ts:258:10
      at Context.modified (spec/helpers/testScheduler-ui.ts:206:13)

[cartant]

I'm familiar with some of those tests. They are very low-level and fragile. It's likely that the change is unnecessary and that the low-level tests should be changed, instead.

I will have a close look at this later.

Thanks for the work that you've out into this PR.

[cartant]

@PSanetra Sorry for the delay in reviewing this. I'm working on this ATM. I'm going to revert the change made to Subscription and I'll update - and comment - the subscription-count-dependent, internal tests.

[cartant]

I've tried to make the changes mentioned in this review by pushing to your branch's PR, but I'm unable to do so. Could you please make the changes?

[cartant]

I've reviewed the failing tests and you are correct: your changes to Subscription are needed. However, I'd like them to be done a little differently.

I'd like the change you've made to the add method in Subscription.ts to be reverted and, instead, I'd like the check to be moved to _addParent - so that all of the parent checks are co-located. Like this:

  /** @internal */
  private _addParent(parent: Subscription): boolean {
    let { _parent, _parents } = this;
    if (_parent === parent) {
      // If the new parent is the same as the current parent, then do nothing.
      return false;
    } else if (!_parent) {
      // If we don't have a parent, then set this._parent to the new parent.
      this._parent = parent;
      return true;
    } else if (!_parents) {
      // If there's already one parent, but not multiple, allocate an Array to
      // store the rest of the parent Subscriptions.
      this._parents = [parent];
      return true;
    } else if (_parents.indexOf(parent) === -1) {
      // Only add the new parent to the _parents list if it's not already there.
      _parents.push(parent);
      return true;
    }
    return false;
  }

And, in the add method, I'd like the pushing of the subscription onto the array to be wrapped in a conditional that uses the return value of the above call:

    if (subscription._addParent(this)) {
      subscriptions.push(subscription);
    }

[PSanetra]

I have implemented your requested changes and also added you as collaborator to the repository fork.

[coveralls]

Pull Request Test Coverage Report for Build 7956

• 12 of 14 (85.71%) changed or added relevant lines in 2 files are covered.
• 1 unchanged line in 1 file lost coverage.
• Overall coverage increased (+0.04%) to 96.854%

---

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
src/internal/Subscription.ts	11	13	84.62%

Files with Coverage Reduction	New Missed Lines	%
src/internal/Subscription.ts	1	91.08%

Totals
Change from base Build 7954:	0.04%
Covered Lines:	5264
Relevant Lines:	5435

---

💛 - Coveralls

[cartant]

LGTM

[benlesh]

Thank you, @PSanetra!