import { Subject } from './Subject'; import { queue } from './scheduler/queue'; import { Subscription } from './Subscription'; import { ObserveOnSubscriber } from './operators/observeOn'; import { ObjectUnsubscribedError } from './util/ObjectUnsubscribedError'; import { SubjectSubscription } from './SubjectSubscription'; export class ReplaySubject extends Subject { constructor(bufferSize = Number.POSITIVE_INFINITY, windowTime = Number.POSITIVE_INFINITY, scheduler) { super(); this.scheduler = scheduler; this._events = []; this._infiniteTimeWindow = false; 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; } } nextInfiniteTimeWindow(value) { const _events = this._events; _events.push(value); if (_events.length > this._bufferSize) { _events.shift(); } super.next(value); } nextTimeWindow(value) { this._events.push(new ReplayEvent(this._getNow(), value)); this._trimBufferThenGetEvents(); super.next(value); } _subscribe(subscriber) { const _infiniteTimeWindow = this._infiniteTimeWindow; const _events = _infiniteTimeWindow ? this._events : this._trimBufferThenGetEvents(); const scheduler = this.scheduler; const len = _events.length; let subscription; if (this.closed) { throw new ObjectUnsubscribedError(); } else if (this.isStopped || this.hasError) { subscription = Subscription.EMPTY; } else { this.observers.push(subscriber); subscription = new SubjectSubscription(this, subscriber); } if (scheduler) { subscriber.add(subscriber = new ObserveOnSubscriber(subscriber, scheduler)); } if (_infiniteTimeWindow) { for (let i = 0; i < len && !subscriber.closed; i++) { subscriber.next(_events[i]); } } else { for (let i = 0; i < len && !subscriber.closed; i++) { subscriber.next(_events[i].value); } } if (this.hasError) { subscriber.error(this.thrownError); } else if (this.isStopped) { subscriber.complete(); } return subscription; } _getNow() { return (this.scheduler || queue).now(); } _trimBufferThenGetEvents() { const now = this._getNow(); const _bufferSize = this._bufferSize; const _windowTime = this._windowTime; const _events = this._events; const eventsCount = _events.length; let spliceCount = 0; while (spliceCount < eventsCount) { if ((now - _events[spliceCount].time) < _windowTime) { break; } spliceCount++; } if (eventsCount > _bufferSize) { spliceCount = Math.max(spliceCount, eventsCount - _bufferSize); } if (spliceCount > 0) { _events.splice(0, spliceCount); } return _events; } } class ReplayEvent { constructor(time, value) { this.time = time; this.value = value; } } //# sourceMappingURL=ReplaySubject.js.map