Add AsyncSubject

[mattpodwysocki]

Add AsyncSubject to RxJS v5. Other subjects can be added in the future, but this one is tried and true and used by many community members.

[mattpodwysocki]

@Blesh @staltz correct me if I'm wrong, but wouldn't it be as simple as this?

import {Subject} from '../Subject';
import {Subscriber} from '../Subscriber';
import {Subscription} from '../Subscription';

export class AsyncSubject<T> extends Subject<T> {
  _value: T = void 0;
  _hasNext: boolean = false;
  _isScalar: boolean = true;

  constructor () {
    super();
  }

  _subscribe(subscriber: Subscriber<any>): Subscription<T> {
    const subscription = super._subscribe(subscriber);
    if (!subscription) {
      return;
    } else if (!subscription.isUnsubscribed && this._hasNext) {
      subscriber.next(this._value);
      subscriber.complete();
    }
    return subscription;
  }

  _next(value: T): void {
    this._value = value;
    this._hasNext = true;
  }

  _complete(): void {
    let index = -1;
    const observers = this.observers;
    const len = observers.length;

    // optimization -- block next, complete, and unsubscribe while dispatching
    this.observers = void 0; // optimization
    this.isUnsubscribed = true;

    if (this._hasNext) {
      while (++index < len) {
        observers[index].next(this._value);
        observers[index].complete();
      }
    } else {
      while (++index < len) {
        observers[index].complete();
      }
    }

    this.isUnsubscribed = false;
  }
}

[benlesh]

LGTM.

I think once this completes, you could set _isScalar to true on it and get some benefits from operators that fast-track scalar observables.

[benlesh]

minor thing would be to put observers[index] in a variable inside of that first loop, just to avoid the lookup twice. I'm unsure about the micro-optimization, but it would probably minify cleaner.

[benlesh]

... also, you'd want to default _isScalar to false until it completes, and hasValue, then set it to true. That way when used after that by operators that have optimizations for scalar observables, it should just pull the value out of value directly.

[antoineol]

👍 for AsyncSubject, thanks!

Just one question: what would be the difference with new ReplaySubject(1) (if any)? I just learnt about this alternative from angular2 gitter.

[staltz]

@architruc very different. AsyncSubject does not emit next values before it completes. It only emits on completion, and after completion for late subscribers.

[antoineol]

@staltz Okay, if I understand well ReplaySubject emits at each value (let's say each next() in the source), but AsyncSubject on complete() in the source, right?

Yes, it is different, seems I misunderstood, although AsyncSubject could still help in my usecase. Thanks!

[benlesh]

This can be closed now, I think.

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.