1. The Muticasted Observable

DevUI is a team with both design and engineering perspectives, serving huawei DevCloud platform and huawei internal background systems, as well as designers and front-end engineers. Add devui Helper (Devui-Official) DevUIHelper plugin: Devuihelper-lsp (welcome Star)

1. Subject
2. AnonymousSubject
3. BehaviorSubject
4. ReplaySubject
5. AsyncSubject

Note: the source version of RxJS used in this article is 6.6.0

Subject

const subject = new Subject<number>(); // Call before subscribing will not print subject.next(1); // subscription data const subscription = subject.subscribe((value) => {console.log(' subscription data A: '+ value); }); // After subscribing, the call prints data. subject.next(2); // Print the result // subscribe data A: 2Copy the code

class Subject<T> extends Observable<T> { observers: Observer<T>[] = []; Next (value? : T) { if (! this.isStopped) { ... const { observers } = this; const len = observers.length; const copy = observers.slice(); for (let i = 0; i < len; i++) { copy[i].next(value); }}} // Next error(err: any) {... this.hasError = true; this.thrownError = err; this.isStopped = true; const { observers } = this; const len = observers.length; const copy = observers.slice(); for (let i = 0; i < len; i++) { copy[i].error(err); } this.observers.length = 0; } // complete similar to next complete() {... this.isStopped = true; const { observers } = this; const len = observers.length; const copy = observers.slice(); for (let i = 0; i < len; i++) { copy[i].complete(); } this.observers.length = 0; }}Copy the code

_subscribe(subscriber: Subscriber<T>): Subscription { if (this.hasError) { subscriber.error(this.thrownError); return Subscription.EMPTY; } else if (this.isStopped) { subscriber.complete(); return Subscription.EMPTY; } else {// If all is well, save observers here to the observers array. this.observers.push(subscriber); // Provide a subscription object that points to the current observer. return new SubjectSubscription(this, subscriber) } }Copy the code

asObservable(): Observable<T> {
  const observable = new Observable<T>();
  (<any>observable).source = this;
  return observable;
}
Copy the code

AnonymousSubject

export interface Observer<T> { closed? : boolean; next: (value: T) => void; error: (err: any) => void; complete: () => void; }Copy the code

export class AnonymousSubject<T> extends Subject<T> {
  constructor(protected destination?: Observer<T>, source?: Observable<T>) {
    super();
    this.source = source;
  }

  next(value: T) {
    const { destination } = this;
    if (destination && destination.next) {
      destination.next(value);
    }
  }

  error(err: any) {
    const { destination } = this;
    if (destination && destination.error) {
      this.destination.error(err);
    }
  }

  complete() {
    const { destination } = this;
    if (destination && destination.complete) {
      this.destination.complete();
    }
  }
}
Copy the code

_subscribe(subscriber: Subscriber<T>): Subscription { const { source } = this; if (source) { return this.source.subscribe(subscriber); } else { return Subscription.EMPTY; }}Copy the code

export class Subject<T> extends Observable<T> { lift<R>(operator: Operator<T, R>): Observable<R> { const subject = new AnonymousSubject(this, this); subject.operator = <any>operator; return <any>subject; }}Copy the code

BehaviorSubject

const subject = new BehaviorSubject<number>(0); // Subscribe to const subscriptionA = subject.subscribe((value) => {console.log(' subscribe data A: '+ value); }); // Call before subscribing will not print subject.next(1); Const subscriptionB = subject.subscribe((value) => {console.log(' subscribe data B: '+ value); }); // After subscribing, the call prints data. subject.next(2); // Prints the result // Subscription data A: 0 // Subscription data A: 1 // Subscription data B: 1 // Subscription data A: 2 //Copy the code

export class BehaviorSubject<T> extends Subject<T> { constructor(private _value: T) { super(); } get value(): T { return this.getValue(); } getValue(): T { if (this.hasError) { throw this.thrownError; } else if (this.closed) { throw new ObjectUnsubscribedError(); } else { return this._value; }}}Copy the code

next(value: T): void {
  super.next(this._value = value);
}
Copy the code

_subscribe(subscriber: Subscriber<T>): Subscription { const subscription = super._subscribe(subscriber); // As long as the subscriber is not closed, the currently stored data is sent to the subscriber. if (subscription && ! (<SubscriptionLike>subscription).closed) { subscriber.next(this._value); } return subscription; }Copy the code

AsyncSubject

const subject = new AsyncSubject<number>(); Const subscriptionA = subject.subscribe((value) => {console.log(' subscribe data A: '+ value); }); // Subscribing to subject.next(1); subject.next(2); subject.next(3); subject.next(4); Const subscriptionB = subject.subscribe((value) => {console.log(' subscribe data B: '+ value); }); // Again, there is no trigger to subscribe to subject.next(5); // But the completion method triggers a subject.plete () subscription; // Print the result // Subscribe data A: 5 // Subscribe data B: 5Copy the code

export class AsyncSubject<T> extends Subject<T> {
  private value: T = null;
  private hasNext: boolean = false;
  private hasCompleted: boolean = false;
}
Copy the code

next(value: T): void {
  if (!this.hasCompleted) {
    this.value = value;
    this.hasNext = true;
  }
}
Copy the code

complete(): void {
  this.hasCompleted = true;
  if (this.hasNext) {
    super.next(this.value);
  }
  super.complete();
}
Copy the code

ReplaySubject

const subject = new ReplaySubject<string>(3); Const subscriptionA = subject.subscribe((value) => {console.log(' subscribe data A: '+ value); }); subject.next(1); subject.next(2); subject.next(3); subject.next(4); Const subscriptionB = subject.subscribe((value) => {console.log(' subscribe data B: '+ value); }); // Print the result: // Subscribe data A: 1 // Subscribe data A: 2 // Subscribe data A: 3 // Subscribe data A: 4 // Subscribe data B: 2 // Subscribe data B: 3 // Subscribe data B: 4Copy the code

const subject = new ReplaySubject<string>(10, 1000); Const subscriptionA = subject.subscribe((value) => {console.log(' subscribe data A: '+ value); }); subject.next('number'); subject.next('string'); subject.next('object'); subject.next('boolean'); setTimeout(() => { subject.next('undefined'); Const subscriptionB = subject.subscribe((value) => {console.log(' subscribe data B: '+ value); }); }, 2000); // Print the result // Subscribe data A: number // Subscribe data A: string // Subscribe data A: object // Subscribe data A: Boolean // Subscribe data A: undefined // Subscribe data B: undefinedCopy the code

export class ReplaySubject<T> extends Subject<T> { private _events: (ReplayEvent<T> | T)[] = []; private _bufferSize: number; private _windowTime: number; private _infiniteTimeWindow: boolean = false; constructor(bufferSize: number = Number.POSITIVE_INFINITY, windowTime: number = Number.POSITIVE_INFINITY, private scheduler? : SchedulerLike) { super(); this._bufferSize = bufferSize < 1 ? 1 : bufferSize; this._windowTime = windowTime < 1 ? 1 : windowTime; if (windowTime === Number.POSITIVE_INFINITY) { this._infiniteTimeWindow = true; this.next = this.nextInfiniteTimeWindow; } else { this.next = this.nextTimeWindow; }}}Copy the code

• nextInfiniteTimeWindow
• nextTimeWindow

nextInfiniteTimeWindow

private nextInfiniteTimeWindow(value: T): void { const _events = this._events; _events.push(value); // Decide which data to keep in the buffer according to the data length and buffer size. if (_events.length > this._bufferSize) { _events.shift(); } super.next(value); }Copy the code

nextTimeWindow

class ReplayEvent<T> {
  constructor(
    readonly public time: number, 
    readonly public value: T
  ) {}
}
Copy the code

private nextTimeWindow(value: T): void {
  this._events.push(new ReplayEvent(this._getNow(), value));
  this._trimBufferThenGetEvents();

  super.next(value);
}
Copy the code

private _trimBufferThenGetEvents(): ReplayEvent<T>[] { const now = this._getNow(); const _bufferSize = this._bufferSize; const _windowTime = this._windowTime; const _events = <ReplayEvent<T>[]>this._events; const eventsCount = _events.length; let spliceCount = 0; // Since the buffer is FIFO, so the time order must be from small to large, so just need to find the partition // cut point, can determine the minimum buffer data length // degree. while (spliceCount < eventsCount) { if ((now - _events[spliceCount].time) < _windowTime) { break; } spliceCount++; } // Buffer length takes precedence over cutting, so if the buffer length is exceeded, the cut point is determined by the larger side. if (eventsCount > _bufferSize) { spliceCount = Math.max(spliceCount, eventsCount - _bufferSize); } if (spliceCount > 0) { _events.splice(0, spliceCount); } return _events; }Copy the code

Subscribe to the process

Subscribe (subscriber: subscriber <T>); Subscription { const _infiniteTimeWindow = this._infiniteTimeWindow; // Update buffer const _events = _infiniteTimeWindow? this._events : this._trimBufferThenGetEvents(); const len = _events.length; // create subscription let subscription: if (this.isStopped || this.hasError) { subscription = Subscription.EMPTY; } else { this.observers.push(subscriber); subscription = new SubjectSubscription(this, subscriber); If (_infiniteTimeWindow) {if (_infiniteTimeWindow) {for (let I = 0; i < len && ! subscriber.closed; i++) { subscriber.next(<T>_events[i]); }} else {// Window time is not infinite, buffer stores ReplayEvent for (let I = 0; i < len && ! subscriber.closed; i++) { subscriber.next((<ReplayEvent<T>>_events[i]).value); } } if (this.hasError) { subscriber.error(this.thrownError); } else if (this.isStopped) { subscriber.complete(); } return subscription; }Copy the code

The last

Join us

Previous articles are recommended

Observable & Subscription (RxJS)

Dark Mode and Thematic Development of Web Interfaces

“How to build a grayscale Publishing environment”