ReplaySubject.js 2.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950
  1. import { Subject } from './Subject';
  2. import { dateTimestampProvider } from './scheduler/dateTimestampProvider';
  3. export class ReplaySubject extends Subject {
  4. constructor(_bufferSize = Infinity, _windowTime = Infinity, _timestampProvider = dateTimestampProvider) {
  5. super();
  6. this._bufferSize = _bufferSize;
  7. this._windowTime = _windowTime;
  8. this._timestampProvider = _timestampProvider;
  9. this._buffer = [];
  10. this._infiniteTimeWindow = true;
  11. this._infiniteTimeWindow = _windowTime === Infinity;
  12. this._bufferSize = Math.max(1, _bufferSize);
  13. this._windowTime = Math.max(1, _windowTime);
  14. }
  15. next(value) {
  16. const { isStopped, _buffer, _infiniteTimeWindow, _timestampProvider, _windowTime } = this;
  17. if (!isStopped) {
  18. _buffer.push(value);
  19. !_infiniteTimeWindow && _buffer.push(_timestampProvider.now() + _windowTime);
  20. }
  21. this._trimBuffer();
  22. super.next(value);
  23. }
  24. _subscribe(subscriber) {
  25. this._throwIfClosed();
  26. this._trimBuffer();
  27. const subscription = this._innerSubscribe(subscriber);
  28. const { _infiniteTimeWindow, _buffer } = this;
  29. const copy = _buffer.slice();
  30. for (let i = 0; i < copy.length && !subscriber.closed; i += _infiniteTimeWindow ? 1 : 2) {
  31. subscriber.next(copy[i]);
  32. }
  33. this._checkFinalizedStatuses(subscriber);
  34. return subscription;
  35. }
  36. _trimBuffer() {
  37. const { _bufferSize, _timestampProvider, _buffer, _infiniteTimeWindow } = this;
  38. const adjustedBufferSize = (_infiniteTimeWindow ? 1 : 2) * _bufferSize;
  39. _bufferSize < Infinity && adjustedBufferSize < _buffer.length && _buffer.splice(0, _buffer.length - adjustedBufferSize);
  40. if (!_infiniteTimeWindow) {
  41. const now = _timestampProvider.now();
  42. let last = 0;
  43. for (let i = 1; i < _buffer.length && _buffer[i] <= now; i += 2) {
  44. last = i;
  45. }
  46. last && _buffer.splice(0, last + 1);
  47. }
  48. }
  49. }
  50. //# sourceMappingURL=ReplaySubject.js.map