123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134 |
- import { Observable } from './Observable';
- import { Subscription, EMPTY_SUBSCRIPTION } from './Subscription';
- import { ObjectUnsubscribedError } from './util/ObjectUnsubscribedError';
- import { arrRemove } from './util/arrRemove';
- import { errorContext } from './util/errorContext';
- export class Subject extends Observable {
- constructor() {
- super();
- this.closed = false;
- this.currentObservers = null;
- this.observers = [];
- this.isStopped = false;
- this.hasError = false;
- this.thrownError = null;
- }
- lift(operator) {
- const subject = new AnonymousSubject(this, this);
- subject.operator = operator;
- return subject;
- }
- _throwIfClosed() {
- if (this.closed) {
- throw new ObjectUnsubscribedError();
- }
- }
- next(value) {
- errorContext(() => {
- this._throwIfClosed();
- if (!this.isStopped) {
- if (!this.currentObservers) {
- this.currentObservers = Array.from(this.observers);
- }
- for (const observer of this.currentObservers) {
- observer.next(value);
- }
- }
- });
- }
- error(err) {
- errorContext(() => {
- this._throwIfClosed();
- if (!this.isStopped) {
- this.hasError = this.isStopped = true;
- this.thrownError = err;
- const { observers } = this;
- while (observers.length) {
- observers.shift().error(err);
- }
- }
- });
- }
- complete() {
- errorContext(() => {
- this._throwIfClosed();
- if (!this.isStopped) {
- this.isStopped = true;
- const { observers } = this;
- while (observers.length) {
- observers.shift().complete();
- }
- }
- });
- }
- unsubscribe() {
- this.isStopped = this.closed = true;
- this.observers = this.currentObservers = null;
- }
- get observed() {
- var _a;
- return ((_a = this.observers) === null || _a === void 0 ? void 0 : _a.length) > 0;
- }
- _trySubscribe(subscriber) {
- this._throwIfClosed();
- return super._trySubscribe(subscriber);
- }
- _subscribe(subscriber) {
- this._throwIfClosed();
- this._checkFinalizedStatuses(subscriber);
- return this._innerSubscribe(subscriber);
- }
- _innerSubscribe(subscriber) {
- const { hasError, isStopped, observers } = this;
- if (hasError || isStopped) {
- return EMPTY_SUBSCRIPTION;
- }
- this.currentObservers = null;
- observers.push(subscriber);
- return new Subscription(() => {
- this.currentObservers = null;
- arrRemove(observers, subscriber);
- });
- }
- _checkFinalizedStatuses(subscriber) {
- const { hasError, thrownError, isStopped } = this;
- if (hasError) {
- subscriber.error(thrownError);
- }
- else if (isStopped) {
- subscriber.complete();
- }
- }
- asObservable() {
- const observable = new Observable();
- observable.source = this;
- return observable;
- }
- }
- Subject.create = (destination, source) => {
- return new AnonymousSubject(destination, source);
- };
- export class AnonymousSubject extends Subject {
- constructor(destination, source) {
- super();
- this.destination = destination;
- this.source = source;
- }
- next(value) {
- var _a, _b;
- (_b = (_a = this.destination) === null || _a === void 0 ? void 0 : _a.next) === null || _b === void 0 ? void 0 : _b.call(_a, value);
- }
- error(err) {
- var _a, _b;
- (_b = (_a = this.destination) === null || _a === void 0 ? void 0 : _a.error) === null || _b === void 0 ? void 0 : _b.call(_a, err);
- }
- complete() {
- var _a, _b;
- (_b = (_a = this.destination) === null || _a === void 0 ? void 0 : _a.complete) === null || _b === void 0 ? void 0 : _b.call(_a);
- }
- _subscribe(subscriber) {
- var _a, _b;
- return (_b = (_a = this.source) === null || _a === void 0 ? void 0 : _a.subscribe(subscriber)) !== null && _b !== void 0 ? _b : EMPTY_SUBSCRIPTION;
- }
- }
- //# sourceMappingURL=Subject.js.map
|