Probable serious bug in .take operator (and .first and probably others?)

[vekexasia]

Hello all, I spend the whole morning trying to figure out what was wrong in my code. Hope this will help.

RxJS version: 5.0.0-beta.12

Code to reproduce:

const source = new BehaviorSubject(42);
source
  .take(1)
  .subscribe(
    n => {
      console.log(n);
      source.next(1)
    },
    e => console.log(e),
    () => console.log('completed')
  );

Expected behavior:
Output:

42
completed

Actual behavior:

42

Additional information:
Since the underlying observable source emits something synchronously, the latter data is discarded by the take operator but this.destination.complete() within take implementation never gets called (cause count gets incremented by the source.next emission and when the if(this.count === total) gets evaluated this.count does not evaluate to 1 anymore but to 2.

Possible Solution:
saving current count in the _next closure will fix the problem.

Similar problem within .first operator
if .take(1) within the previous example is changed to .first() then an Call stack limit exceeded is triggered and all the 1s are getting through first. This is almost the same problem but the fix can be done by setting a _lock variable within the _emitFinal fn under the .first implementation.

Conclusion

While this might seem a stupid bug considering the sample. It happened to me in a situation where there was no direct correlation between the onNext fn and the source observable. Instead, in my code, I was calling a fn (not rx native) that changed the underlying data of the source observable causing it to re-emit the changes.

[vekexasia]

Additionally, adding .delay(1) after .take or .first will create a workaround on the issue

[kwonoj]

For me this seems same case to #1759 (comment), trying to synchronous reentrant causes it trapped can be resolved by introduce async boundaries as suggested in comment.

[vekexasia]

@kwonoj it's the same issue. but i don't agree with that comment i'll reply there. thank you.

[vekexasia]

Hello @trxcllnt I was pointed here thanks to my other #1993 .

The current .take implementation reads in ts:

protected _next(value: T): void {
    const total = this.total;
    const count = ++this.count;
    if (count <= total) {
      this.destination.next(value);
      if (count === total) {
        this.destination.complete();
        this.unsubscribe();
      }
    }
  }

but the js code in node version 5.0.0-beta.12 gets compiled as follows:

TakeSubscriber.prototype._next = function (value) {
        var total = this.total;
        if (++this.count <= total) {
            this.destination.next(value);
            if (this.count === total) {
                this.destination.complete();
                this.unsubscribe();
            }
        }
    };

as you can see, feeding an item will cause this.count to be incremented before this.destination.next(value) ends its execution.

I cloned rxjs and run build_all but it looks like I can't get the wrong js code in none of the generated .js

[kwonoj]

if you're referring this changes #1967, it's checked in but new package hasn't been published yet.

[benlesh]

but the js code in node version 5.0.0-beta.12 gets compiled as follows:

Whoa... That's not good. Seems like a problem with the TypeScript compiler.

[benlesh]

Okay, so for whatever reason, what's been deployed built with this bug, but I can't get any version of TSC to build the same code to have this error locally... very strange.

[benlesh]

I see... this was an actual bug in code that @trxcllnt fixed in #1967... I'm slow to catch on haha.

We can close this issue, as the next release will have the fix.

[lock]

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.