134 lines
No EOL
4.3 KiB
JavaScript
134 lines
No EOL
4.3 KiB
JavaScript
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
|