fix(operator): Fix take to complete when the source is re-entrant.

The TakeSubscriber compares the instance variable `count` to determine if it should complete or not.

If the source Observable is re-entrant, this check fails (because `count` is updated in each `next`), and take never completes.

spec/operators/take-spec.ts

@@ -2,6 +2,7 @@ import {expect} from 'chai';
 import * as Rx from '../../dist/cjs/Rx';
 declare const {hot, cold, asDiagram, expectObservable, expectSubscriptions};
 
+const Subject = Rx.Subject;
 const Observable = Rx.Observable;
 
 /** @test {take} */
@@ -135,4 +136,19 @@ describe('Observable.prototype.take', () => {
 
     source.subscribe();
   });
+
+  it('should complete when the source is reentrant', () => {
+    let completed = false;
+    const source = new Subject();
+    source.take(5).subscribe({
+      next() {
+        source.next();
+      },
+      complete() {
+        completed = true;
+      }
+    });
+    source.next();
+    expect(completed).to.be.true;
+  });
 });

src/operator/take.ts

@@ -76,9 +76,10 @@ class TakeSubscriber<T> extends Subscriber<T> {
 
   protected _next(value: T): void {
     const total = this.total;
-    if (++this.count <= total) {
+    const count = ++this.count;
+    if (count <= total) {
       this.destination.next(value);
-      if (this.count === total) {
+      if (count === total) {
         this.destination.complete();
         this.unsubscribe();
       }