fix(AsyncSubject) should not allow complete() after error (#2280)

* test(AsyncSubject): asyncsubject should not send complete event after error

* fix(AsynSubject): check if there is an error first

* test(AsyncSubject): asyncSubject should not send error event after complete

* fix(AsyncSubject): only pass error if asyncsubject is not completed

spec/subjects/AsyncSubject-spec.ts

@@ -133,6 +133,9 @@ describe('AsyncSubject', () => {
 
     subscription.unsubscribe();
     observer.results = [];
+
+    subject.error(new Error(''));
+
     subject.subscribe(observer);
     expect(observer.results).to.deep.equal(['done']);
   });
@@ -168,4 +171,25 @@ describe('AsyncSubject', () => {
     subject.subscribe(observer);
     expect(observer.results).to.deep.equal([expected]);
   });
+
+  it('should not allow send complete after error', () => {
+    const expected = new Error('bad');
+    const subject = new AsyncSubject();
+    const observer = new TestObserver();
+    const subscription = subject.subscribe(observer);
+
+    subject.next(1);
+    expect(observer.results).to.deep.equal([]);
+
+    subject.error(expected);
+    expect(observer.results).to.deep.equal([expected]);
+
+    subscription.unsubscribe();
+
+    observer.results = [];
+
+    subject.complete();
+    subject.subscribe(observer);
+    expect(observer.results).to.deep.equal([expected]);
+  });
 });

src/AsyncSubject.ts

@@ -11,15 +11,14 @@ export class AsyncSubject<T> extends Subject<T> {
   private hasCompleted: boolean = false;
 
   protected _subscribe(subscriber: Subscriber<any>): Subscription {
-    if (this.hasCompleted && this.hasNext) {
+    if (this.hasError) {
+      subscriber.error(this.thrownError);
+      return Subscription.EMPTY;
+    } else if (this.hasCompleted && this.hasNext) {
       subscriber.next(this.value);
       subscriber.complete();
       return Subscription.EMPTY;
-    } else if (this.hasError) {
-      subscriber.error(this.thrownError);
-      return Subscription.EMPTY;
     }
-
     return super._subscribe(subscriber);
   }
 
@@ -30,6 +29,12 @@ export class AsyncSubject<T> extends Subject<T> {
     }
   }
 
+  error(error: any): void {
+    if (!this.hasCompleted) {
+      super.error(error);
+    }
+  }
+
   complete(): void {
     this.hasCompleted = true;
     if (this.hasNext) {