fix(AsyncSubject): fix reentrancy issue in complete

From comment https://github.com/ReactiveX/rxjs/pull/5729/files/30d429cf1b791db15c04a61f6a683e189b53fb3e#r492314703

spec/subjects/AsyncSubject-spec.ts

@@ -191,4 +191,42 @@ describe('AsyncSubject', () => {
     subject.subscribe(observer);
     expect(observer.results).to.deep.equal([expected]);
   });
+
+  it('should not be reentrant via complete', () => {
+    const subject = new AsyncSubject<number>();
+    let calls = 0;
+    subject.subscribe({
+      next: value => {
+        calls++;
+        if (calls < 2) {
+          // if this is more than 1, we're reentrant, and that's bad.
+          subject.complete();
+        }
+      }
+    });
+
+    subject.next(1);
+    subject.complete();
+
+    expect(calls).to.equal(1);
+  });
+
+   it('should not be reentrant via next', () => {
+    const subject = new AsyncSubject<number>();
+    let calls = 0;
+    subject.subscribe({
+      next: value => {
+        calls++;
+        if (calls < 2) {
+          // if this is more than 1, we're reentrant, and that's bad.
+          subject.next(value + 1);
+        }
+      }
+    });
+
+    subject.next(1);
+    subject.complete();
+
+    expect(calls).to.equal(1);
+  });
 });

src/internal/AsyncSubject.ts

@@ -1,3 +1,4 @@
+/** @prettier */
 import { Subject } from './Subject';
 import { Subscriber } from './Subscriber';
 import { Subscription } from './Subscription';
@@ -11,6 +12,7 @@ import { Subscription } from './Subscription';
 export class AsyncSubject<T> extends Subject<T> {
   private value: T | null = null;
   private hasValue = false;
+  private isComplete = false;
 
   protected checkFinalizedStatuses(subscriber: Subscriber<T>) {
     const { hasError, hasValue, value, thrownError, isStopped } = this;
@@ -30,8 +32,11 @@ export class AsyncSubject<T> extends Subject<T> {
   }
 
   complete(): void {
-    const { hasValue, value } = this;
-    hasValue && super.next(value!);
-    super.complete();
+    const { hasValue, value, isComplete } = this;
+    if (!isComplete) {
+      this.isComplete = true;
+      hasValue && super.next(value!);
+      super.complete();
+    }
   }
 }