1 import { Subject } from './Subject';
2 import { queue } from './scheduler/queue';
3 import { Subscription } from './Subscription';
4 import { ObserveOnSubscriber } from './operators/observeOn';
5 import { ObjectUnsubscribedError } from './util/ObjectUnsubscribedError';
6 import { SubjectSubscription } from './SubjectSubscription';
7 export class ReplaySubject extends Subject {
8 constructor(bufferSize = Number.POSITIVE_INFINITY, windowTime = Number.POSITIVE_INFINITY, scheduler) {
10 this.scheduler = scheduler;
12 this._infiniteTimeWindow = false;
13 this._bufferSize = bufferSize < 1 ? 1 : bufferSize;
14 this._windowTime = windowTime < 1 ? 1 : windowTime;
15 if (windowTime === Number.POSITIVE_INFINITY) {
16 this._infiniteTimeWindow = true;
17 this.next = this.nextInfiniteTimeWindow;
20 this.next = this.nextTimeWindow;
23 nextInfiniteTimeWindow(value) {
24 if (!this.isStopped) {
25 const _events = this._events;
27 if (_events.length > this._bufferSize) {
33 nextTimeWindow(value) {
34 if (!this.isStopped) {
35 this._events.push(new ReplayEvent(this._getNow(), value));
36 this._trimBufferThenGetEvents();
40 _subscribe(subscriber) {
41 const _infiniteTimeWindow = this._infiniteTimeWindow;
42 const _events = _infiniteTimeWindow ? this._events : this._trimBufferThenGetEvents();
43 const scheduler = this.scheduler;
44 const len = _events.length;
47 throw new ObjectUnsubscribedError();
49 else if (this.isStopped || this.hasError) {
50 subscription = Subscription.EMPTY;
53 this.observers.push(subscriber);
54 subscription = new SubjectSubscription(this, subscriber);
57 subscriber.add(subscriber = new ObserveOnSubscriber(subscriber, scheduler));
59 if (_infiniteTimeWindow) {
60 for (let i = 0; i < len && !subscriber.closed; i++) {
61 subscriber.next(_events[i]);
65 for (let i = 0; i < len && !subscriber.closed; i++) {
66 subscriber.next(_events[i].value);
70 subscriber.error(this.thrownError);
72 else if (this.isStopped) {
73 subscriber.complete();
78 return (this.scheduler || queue).now();
80 _trimBufferThenGetEvents() {
81 const now = this._getNow();
82 const _bufferSize = this._bufferSize;
83 const _windowTime = this._windowTime;
84 const _events = this._events;
85 const eventsCount = _events.length;
87 while (spliceCount < eventsCount) {
88 if ((now - _events[spliceCount].time) < _windowTime) {
93 if (eventsCount > _bufferSize) {
94 spliceCount = Math.max(spliceCount, eventsCount - _bufferSize);
96 if (spliceCount > 0) {
97 _events.splice(0, spliceCount);
103 constructor(time, value) {
108 //# sourceMappingURL=ReplaySubject.js.map