.config/coc/extensions/node_modules/coc-prettier/node_modules/rxjs/src/internal/Subject.ts

   1 import { Operator } from './Operator';
   2 import { Observable } from './Observable';
   3 import { Subscriber } from './Subscriber';
   4 import { Subscription } from './Subscription';
   5 import { Observer, SubscriptionLike, TeardownLogic } from './types';
   6 import { ObjectUnsubscribedError } from './util/ObjectUnsubscribedError';
   7 import { SubjectSubscription } from './SubjectSubscription';
   8 import { rxSubscriber as rxSubscriberSymbol } from '../internal/symbol/rxSubscriber';
   9
  10 /**
  11  * @class SubjectSubscriber<T>
  12  */
  13 export class SubjectSubscriber<T> extends Subscriber<T> {
  14   constructor(protected destination: Subject<T>) {
  15     super(destination);
  16   }
  17 }
  18
  19 /**
  20  * A Subject is a special type of Observable that allows values to be
  21  * multicasted to many Observers. Subjects are like EventEmitters.
  22  *
  23  * Every Subject is an Observable and an Observer. You can subscribe to a
  24  * Subject, and you can call next to feed values as well as error and complete.
  25  *
  26  * @class Subject<T>
  27  */
  28 export class Subject<T> extends Observable<T> implements SubscriptionLike {
  29
  30   [rxSubscriberSymbol]() {
  31     return new SubjectSubscriber(this);
  32   }
  33
  34   observers: Observer<T>[] = [];
  35
  36   closed = false;
  37
  38   isStopped = false;
  39
  40   hasError = false;
  41
  42   thrownError: any = null;
  43
  44   constructor() {
  45     super();
  46   }
  47
  48   /**@nocollapse
  49    * @deprecated use new Subject() instead
  50   */
  51   static create: Function = <T>(destination: Observer<T>, source: Observable<T>): AnonymousSubject<T> => {
  52     return new AnonymousSubject<T>(destination, source);
  53   }
  54
  55   lift<R>(operator: Operator<T, R>): Observable<R> {
  56     const subject = new AnonymousSubject(this, this);
  57     subject.operator = <any>operator;
  58     return <any>subject;
  59   }
  60
  61   next(value?: T) {
  62     if (this.closed) {
  63       throw new ObjectUnsubscribedError();
  64     }
  65     if (!this.isStopped) {
  66       const { observers } = this;
  67       const len = observers.length;
  68       const copy = observers.slice();
  69       for (let i = 0; i < len; i++) {
  70         copy[i].next(value);
  71       }
  72     }
  73   }
  74
  75   error(err: any) {
  76     if (this.closed) {
  77       throw new ObjectUnsubscribedError();
  78     }
  79     this.hasError = true;
  80     this.thrownError = err;
  81     this.isStopped = true;
  82     const { observers } = this;
  83     const len = observers.length;
  84     const copy = observers.slice();
  85     for (let i = 0; i < len; i++) {
  86       copy[i].error(err);
  87     }
  88     this.observers.length = 0;
  89   }
  90
  91   complete() {
  92     if (this.closed) {
  93       throw new ObjectUnsubscribedError();
  94     }
  95     this.isStopped = true;
  96     const { observers } = this;
  97     const len = observers.length;
  98     const copy = observers.slice();
  99     for (let i = 0; i < len; i++) {
 100       copy[i].complete();
 101     }
 102     this.observers.length = 0;
 103   }
 104
 105   unsubscribe() {
 106     this.isStopped = true;
 107     this.closed = true;
 108     this.observers = null;
 109   }
 110
 111   /** @deprecated This is an internal implementation detail, do not use. */
 112   _trySubscribe(subscriber: Subscriber<T>): TeardownLogic {
 113     if (this.closed) {
 114       throw new ObjectUnsubscribedError();
 115     } else {
 116       return super._trySubscribe(subscriber);
 117     }
 118   }
 119
 120   /** @deprecated This is an internal implementation detail, do not use. */
 121   _subscribe(subscriber: Subscriber<T>): Subscription {
 122     if (this.closed) {
 123       throw new ObjectUnsubscribedError();
 124     } else if (this.hasError) {
 125       subscriber.error(this.thrownError);
 126       return Subscription.EMPTY;
 127     } else if (this.isStopped) {
 128       subscriber.complete();
 129       return Subscription.EMPTY;
 130     } else {
 131       this.observers.push(subscriber);
 132       return new SubjectSubscription(this, subscriber);
 133     }
 134   }
 135
 136   /**
 137    * Creates a new Observable with this Subject as the source. You can do this
 138    * to create customize Observer-side logic of the Subject and conceal it from
 139    * code that uses the Observable.
 140    * @return {Observable} Observable that the Subject casts to
 141    */
 142   asObservable(): Observable<T> {
 143     const observable = new Observable<T>();
 144     (<any>observable).source = this;
 145     return observable;
 146   }
 147 }
 148
 149 /**
 150  * @class AnonymousSubject<T>
 151  */
 152 export class AnonymousSubject<T> extends Subject<T> {
 153   constructor(protected destination?: Observer<T>, source?: Observable<T>) {
 154     super();
 155     this.source = source;
 156   }
 157
 158   next(value: T) {
 159     const { destination } = this;
 160     if (destination && destination.next) {
 161       destination.next(value);
 162     }
 163   }
 164
 165   error(err: any) {
 166     const { destination } = this;
 167     if (destination && destination.error) {
 168       this.destination.error(err);
 169     }
 170   }
 171
 172   complete() {
 173     const { destination } = this;
 174     if (destination && destination.complete) {
 175       this.destination.complete();
 176     }
 177   }
 178
 179   /** @deprecated This is an internal implementation detail, do not use. */
 180   _subscribe(subscriber: Subscriber<T>): Subscription {
 181     const { source } = this;
 182     if (source) {
 183       return this.source.subscribe(subscriber);
 184     } else {
 185       return Subscription.EMPTY;
 186     }
 187   }
 188 }