blob: d6d8eb7dfd3bea416b98ada0c01b2dd279755a1f (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
import { Subject } from './Subject';
import { SchedulerLike } from './types';
import { queue } from './scheduler/queue';
import { Subscriber } from './Subscriber';
import { Subscription } from './Subscription';
import { ObserveOnSubscriber } from './operators/observeOn';
import { ObjectUnsubscribedError } from './util/ObjectUnsubscribedError';
import { SubjectSubscription } from './SubjectSubscription';
/**
* A variant of Subject that "replays" or emits old values to new subscribers.
* It buffers a set number of values and will emit those values immediately to
* any new subscribers in addition to emitting new values to existing subscribers.
*
* @class ReplaySubject<T>
*/
export class ReplaySubject<T> extends Subject<T> {
private _events: (ReplayEvent<T> | T)[] = [];
private _bufferSize: number;
private _windowTime: number;
private _infiniteTimeWindow: boolean = false;
constructor(bufferSize: number = Number.POSITIVE_INFINITY,
windowTime: number = Number.POSITIVE_INFINITY,
private scheduler?: SchedulerLike) {
super();
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;
}
}
private nextInfiniteTimeWindow(value: T): void {
const _events = this._events;
_events.push(value);
// Since this method is invoked in every next() call than the buffer
// can overgrow the max size only by one item
if (_events.length > this._bufferSize) {
_events.shift();
}
super.next(value);
}
private nextTimeWindow(value: T): void {
this._events.push(new ReplayEvent(this._getNow(), value));
this._trimBufferThenGetEvents();
super.next(value);
}
/** @deprecated This is an internal implementation detail, do not use. */
_subscribe(subscriber: Subscriber<T>): Subscription {
// When `_infiniteTimeWindow === true` then the buffer is already trimmed
const _infiniteTimeWindow = this._infiniteTimeWindow;
const _events = _infiniteTimeWindow ? this._events : this._trimBufferThenGetEvents();
const scheduler = this.scheduler;
const len = _events.length;
let subscription: 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<T>(subscriber, scheduler));
}
if (_infiniteTimeWindow) {
for (let i = 0; i < len && !subscriber.closed; i++) {
subscriber.next(<T>_events[i]);
}
} else {
for (let i = 0; i < len && !subscriber.closed; i++) {
subscriber.next((<ReplayEvent<T>>_events[i]).value);
}
}
if (this.hasError) {
subscriber.error(this.thrownError);
} else if (this.isStopped) {
subscriber.complete();
}
return subscription;
}
_getNow(): number {
return (this.scheduler || queue).now();
}
private _trimBufferThenGetEvents(): ReplayEvent<T>[] {
const now = this._getNow();
const _bufferSize = this._bufferSize;
const _windowTime = this._windowTime;
const _events = <ReplayEvent<T>[]>this._events;
const eventsCount = _events.length;
let spliceCount = 0;
// Trim events that fall out of the time window.
// Start at the front of the list. Break early once
// we encounter an event that falls within the window.
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<T> {
constructor(public time: number, public value: T) {
}
}
|
