fix(ReplaySubject): observer now subscribed prior to running subscrip…

…tion function (#2046)

Removed reliance on inherited `_subscribe` method from `Subject`. ReplaySubject has different behaviors during subscription for notifying new subscribers after a completion or error. After moving the `super._subscribe` call before the notification from saved `_events`, it was apparent that the inherited behavior from `Subject` was stopping the subscriber before it could be notified properly.

fixes #2044

spec/subjects/ReplaySubject-spec.ts

@@ -16,6 +16,21 @@ describe('ReplaySubject', () => {
  done();
  });
 
+ it('should add the observer before running subscription code', () => {
+ const subject = new ReplaySubject<number>();
+ subject.next(1);
+ const results = [];
+
+ subject.subscribe((value) => {
+ results.push(value);
+ if (value < 3) {
+ subject.next(value + 1);
+ }
+ });
+
+ expect(results).to.deep.equal([1, 2, 3]);
+ });
+
  it('should replay values upon subscription', (done: MochaDone) => {
  const subject = new ReplaySubject();
  const expects = [1, 2, 3];

src/ReplaySubject.ts

@@ -4,7 +4,8 @@ import { queue } from './scheduler/queue';
 import { Subscriber } from './Subscriber';
 import { Subscription } from './Subscription';
 import { ObserveOnSubscriber } from './operator/observeOn';
-
+import { ObjectUnsubscribedError } from './util/ObjectUnsubscribedError';
+import { SubjectSubscription } from './SubjectSubscription';
 /**
  * @class ReplaySubject<T>
  */
@@ -31,6 +32,18 @@ export class ReplaySubject<T> extends Subject<T> {
  protected _subscribe(subscriber: Subscriber<T>): Subscription {
  const _events = this._trimBufferThenGetEvents();
  const scheduler = this.scheduler;
+ let subscription: Subscription;
+
+ if (this.closed) {
+ throw new ObjectUnsubscribedError();
+ } else if (this.hasError) {
+ subscription = Subscription.EMPTY;
+ } else if (this.isStopped) {
+ subscription = Subscription.EMPTY;
+ } else {
+ this.observers.push(subscriber);
+ subscription = new SubjectSubscription(this, subscriber);
+ }
 
  if (scheduler) {
  subscriber.add(subscriber = new ObserveOnSubscriber<T>(subscriber, scheduler));
@@ -41,7 +54,13 @@ export class ReplaySubject<T> extends Subject<T> {
  subscriber.next(_events[i].value);
  }
 
- return super._subscribe(subscriber);
+ if (this.hasError) {
+ subscriber.error(this.thrownError);
+ } else if (this.isStopped) {
+ subscriber.complete();
+ }
+
+ return subscription;
  }
 
  _getNow(): number {