fix(AsyncSubject): properly emits values during reentrant subscriptions (#6522)

Resolves #6520

Author: benlesh

spec/subjects/AsyncSubject-spec.ts

@@ -211,7 +211,7 @@ describe('AsyncSubject', () => {
     expect(calls).to.equal(1);
   });
 
-   it('should not be reentrant via next', () => {
+  it('should not be reentrant via next', () => {
     const subject = new AsyncSubject<number>();
     let calls = 0;
     subject.subscribe({
@@ -229,4 +229,25 @@ describe('AsyncSubject', () => {
 
     expect(calls).to.equal(1);
   });
+
+  it('should allow reentrant subscriptions', () => {
+    const subject = new AsyncSubject<number>()
+    let results: any[] = [];
+
+    subject.subscribe({
+      next: (value) => {
+        subject.subscribe({
+          next: value => results.push('inner: ' + (value + value)),
+          complete: () => results.push('inner: done')
+        });
+        results.push('outer: ' + value);
+      },
+      complete: () => results.push('outer: done')
+    });
+
+    subject.next(1);
+    expect(results).to.deep.equal([]);
+    subject.complete();
+    expect(results).to.deep.equal(['inner: 2', 'inner: done', 'outer: 1', 'outer: done']);
+  });
 });

src/internal/AsyncSubject.ts

@@ -14,10 +14,10 @@ export class AsyncSubject<T> extends Subject<T> {
 
   /** @internal */
   protected _checkFinalizedStatuses(subscriber: Subscriber<T>) {
-    const { hasError, _hasValue, _value, thrownError, isStopped } = this;
+    const { hasError, _hasValue, _value, thrownError, isStopped, _isComplete } = this;
     if (hasError) {
       subscriber.error(thrownError);
-    } else if (isStopped) {
+    } else if (isStopped || _isComplete) {
       _hasValue && subscriber.next(_value!);
       subscriber.complete();
     }