SwitchAll / SwitchLatest are broken

[jinroh]

switch operators do not seem to unsubscribe properly their inner subscriptions.

Here is a small repro code:

Observable.interval(1000)
  .switchLatest(function(i) {
    return Observable.interval(100).do(function(j) { console.log(i + "_" + j); });
  })
  .subscribe(function() {
  });

[benlesh]

Thank you for pointing this out. Looking into this and expanding tests uncovered a variety of anomalies. I've been heads down refactoring code in this area.

[benlesh]

@jinroh ... can you please checkout #317 and see if it solves your issues? I suspect it will. I was able to confirm your issue and write tests around it.

[jinroh]

I think it solves one of the problem.
Here the example code I use:

var Rx = require("./cjs/Rx");

var Observable = Rx.Observable;

Observable.interval(1000).switchLatest(function(i) {
  return Observable.interval(100).map(function(j) {
    console.log("inner", i + "_" + j);
    return i + "_" + j;
  });
}).subscribe(function(v) {
  console.log("outer", v);
});

Outer subscription behaves as expected now:

outer 0_0
outer 0_1
outer 0_2
outer 0_3
outer 0_4
outer 0_5
outer 0_6
outer 0_7
outer 0_8
outer 1_0
outer 1_1
outer 1_2
outer 1_3
...

However, the inner subscriptions (from the map operator in my example) do not seem to be cleaned up. The "inner interval" side effects are still present. It looks like the subscription graph is not entirely disposed as it should be, causing a leak:

inner 0_0
inner 0_1
inner 0_2
inner 0_3
inner 0_4
inner 0_5
inner 0_6
inner 0_7
inner 0_8
inner 0_9
inner 1_0
inner 0_10
inner 1_1
inner 0_11
inner 1_2
inner 0_12
inner 1_3
inner 0_13
...

Not sure if this problem is linked to the Switch.. operator or more general. In my tests, this is the only operator I've used that internally unsubscribe inner-subs.

[benlesh]

Returning subscriptions from the project function in the operator doesn't do anything special. They're not being cleaned up because you're creating new subscriptions and never unsubscribing therm. I'm actually amazed that worked at all. I'll check that out.

[jinroh]

Thanks !

They're not being cleaned up because you're creating new subscriptions and never unsubscribing therm.

I'm not sure I understand correctly your comment. switchLatest is similar to flatMapLatest right ? Shouldn't this operator unsubscribe inner-subscriptions on its own, with a concurrency of one ?

[trxcllnt]

@Blesh the project function in this example isn't returning a Subscription, the outer subscribe call is just indented a bit.

[jinroh]

@trxcllnt oh i see what @Blesh said now !
edited my comment to remove the ident.

[benlesh]

Sorry, I was looking at it from my phone before. Looked like it was all together on that little screen

[benlesh]

I see the problem... PR incoming

[jinroh]

(me thinking out loud)
I find this bug quite interesting because it shows that the marble tests, as I understand them, cannot validate the correction of these kind of issues (not that the previous tests forms could, but still...).

Could be interesting to find out how to simply test the unsubscription on inner side-effects...

[benlesh]

@jinroh this will require just a regular synchronous test.

[mattpodwysocki]

@jinroh you're absolutely right, marble diagrams will not fix this issue and I have said from the beginning that we need a hybrid approach with concrete numbers. For example, in RxJS the current version, we not only track the results, but also the subscriptions, which is not something the current implementation does. For example, with case, we have to track the subscription of all the inners

[benlesh]

Marble diagrams are amazing/perfect for verifying sequences of notifications.

What we need to add are the output of additional events that occur during the execution of the observable. Mostly just timings of inner subscriptions and unsubscriptions.