Rxjs: Unexpected behaviour of first operator together with switchMap

[artall64]

RxJS version: 5.2.0

Code to reproduce:
var x = Rx.Observable.interval(1000)
.do( x=> console.log("One"))
.first()
.switchMap(x => Rx.Observable.interval(1000))
.do( x=> console.log("Two"))
.subscribe((x) => {})

Expected behaviour:
// In console you will see:
// One
// Two
// Two
// Two
// Two
// etc...

Actual behaviour:
// In console you will see:
// One
// One
// Two
// One
// Two
// One
// etc...

Additional information:
I've expected that first() in this case should behave exactly the same as a take(1) but behaviour is quite different

[trxcllnt]

This sounds like a bug in first

[mpodlasin]

@trxcllnt In a way it is firsts fault and in a way it is not. I implemented forcing unsubscription logic when first value arrives to first - this is what take(1) does and it works, but for me it feels extremely hacky.

The problem is since every operator is implemented via Subscriber inheritance, it can override default unsubscription logic that happens when source completes or errors (defined in protected _complete and _error methods in Subscriber as seen here: https://github.com/ReactiveX/rxjs/blob/master/src/Subscriber.ts#L143).

This is exactly what switchMap does - if inner Observable (the one that is returned by function passed to switchMap) is still emitting values (as it is in the case of interval which will never stop), switchMap holds off with unsubscribing from its source (result of first in that case).

You can see it here: https://github.com/ReactiveX/rxjs/blob/master/src/operator/switchMap.ts#L113

My point is - should an operator have power to intercept unsubscription logic of source observable? When observable sends complete or error notification, shouldn't one central mechanism make sure that it is always unsubscribed, so that operators are not prone to such mistakes?

[mpodlasin]

As far as I understand, in tc39 spec, if I call complete on observer in my custom Observable, I do not have to trigger unsubscription logic by hand. I can trust that there exists a mechanism that will call it for me. Right now implementation of take (https://github.com/ReactiveX/rxjs/blob/master/src/operator/take.ts#L80) and my fix for first go against it, by not trusting that underlying mechanism will work properly and calling unsubscription by hand.

[mpodlasin]

In other words: shouldn't this: https://github.com/ReactiveX/rxjs/blob/master/spec/operators/switchMap-spec.ts#L57-L58 be absolutely illegal? We have source that completed, but we are still subscribing to it for some time.

[trxcllnt]

@Podlas29 ah, yes there is a deeper issue here. I'll comment on #2457 in more detail.

[benlesh]

This one is interesting. I'm actually curious if the subscription tests have a flaw or if it's the actual implementations of the operators (where they override various parts of Subscriber, perhaps they're not properly calling unsubscribe in a few places anymore)

[martinsik]

I think the same problem is with the takeUntil() operator. I saw this recently on StackOverflow:

const timeout$ = new Subject();

Observable
  .of(1)
  .takeUntil(timeout$)
  .delay(1000)
  .subscribe(x => console.log(x));

timeout$.next();
timeout$.complete();

I'd expect this example not to print anything because I triggered the notification Observable in takeUntil() but it does print 1.

See demo: https://jsbin.com/wimebel/3/edit?js,console

This can be easily fixed by switching the order of takeUntil and delay but the current behavior seems unexpected.

[lock[bot]]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.