throttle.js 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131
  1. import { OuterSubscriber } from '../OuterSubscriber';
  2. import { subscribeToResult } from '../util/subscribeToResult';
  3. export const defaultThrottleConfig = {
  4. leading: true,
  5. trailing: false
  6. };
  7. /**
  8. * Emits a value from the source Observable, then ignores subsequent source
  9. * values for a duration determined by another Observable, then repeats this
  10. * process.
  11. *
  12. * <span class="informal">It's like {@link throttleTime}, but the silencing
  13. * duration is determined by a second Observable.</span>
  14. *
  15. * <img src="./img/throttle.png" width="100%">
  16. *
  17. * `throttle` emits the source Observable values on the output Observable
  18. * when its internal timer is disabled, and ignores source values when the timer
  19. * is enabled. Initially, the timer is disabled. As soon as the first source
  20. * value arrives, it is forwarded to the output Observable, and then the timer
  21. * is enabled by calling the `durationSelector` function with the source value,
  22. * which returns the "duration" Observable. When the duration Observable emits a
  23. * value or completes, the timer is disabled, and this process repeats for the
  24. * next source value.
  25. *
  26. * @example <caption>Emit clicks at a rate of at most one click per second</caption>
  27. * var clicks = Rx.Observable.fromEvent(document, 'click');
  28. * var result = clicks.throttle(ev => Rx.Observable.interval(1000));
  29. * result.subscribe(x => console.log(x));
  30. *
  31. * @see {@link audit}
  32. * @see {@link debounce}
  33. * @see {@link delayWhen}
  34. * @see {@link sample}
  35. * @see {@link throttleTime}
  36. *
  37. * @param {function(value: T): SubscribableOrPromise} durationSelector A function
  38. * that receives a value from the source Observable, for computing the silencing
  39. * duration for each source value, returned as an Observable or a Promise.
  40. * @param {Object} config a configuration object to define `leading` and `trailing` behavior. Defaults
  41. * to `{ leading: true, trailing: false }`.
  42. * @return {Observable<T>} An Observable that performs the throttle operation to
  43. * limit the rate of emissions from the source.
  44. * @method throttle
  45. * @owner Observable
  46. */
  47. export function throttle(durationSelector, config = defaultThrottleConfig) {
  48. return (source) => source.lift(new ThrottleOperator(durationSelector, config.leading, config.trailing));
  49. }
  50. class ThrottleOperator {
  51. constructor(durationSelector, leading, trailing) {
  52. this.durationSelector = durationSelector;
  53. this.leading = leading;
  54. this.trailing = trailing;
  55. }
  56. call(subscriber, source) {
  57. return source.subscribe(new ThrottleSubscriber(subscriber, this.durationSelector, this.leading, this.trailing));
  58. }
  59. }
  60. /**
  61. * We need this JSDoc comment for affecting ESDoc
  62. * @ignore
  63. * @extends {Ignored}
  64. */
  65. class ThrottleSubscriber extends OuterSubscriber {
  66. constructor(destination, durationSelector, _leading, _trailing) {
  67. super(destination);
  68. this.destination = destination;
  69. this.durationSelector = durationSelector;
  70. this._leading = _leading;
  71. this._trailing = _trailing;
  72. this._hasTrailingValue = false;
  73. }
  74. _next(value) {
  75. if (this.throttled) {
  76. if (this._trailing) {
  77. this._hasTrailingValue = true;
  78. this._trailingValue = value;
  79. }
  80. }
  81. else {
  82. const duration = this.tryDurationSelector(value);
  83. if (duration) {
  84. this.add(this.throttled = subscribeToResult(this, duration));
  85. }
  86. if (this._leading) {
  87. this.destination.next(value);
  88. if (this._trailing) {
  89. this._hasTrailingValue = true;
  90. this._trailingValue = value;
  91. }
  92. }
  93. }
  94. }
  95. tryDurationSelector(value) {
  96. try {
  97. return this.durationSelector(value);
  98. }
  99. catch (err) {
  100. this.destination.error(err);
  101. return null;
  102. }
  103. }
  104. /** @deprecated internal use only */ _unsubscribe() {
  105. const { throttled, _trailingValue, _hasTrailingValue, _trailing } = this;
  106. this._trailingValue = null;
  107. this._hasTrailingValue = false;
  108. if (throttled) {
  109. this.remove(throttled);
  110. this.throttled = null;
  111. throttled.unsubscribe();
  112. }
  113. }
  114. _sendTrailing() {
  115. const { destination, throttled, _trailing, _trailingValue, _hasTrailingValue } = this;
  116. if (throttled && _trailing && _hasTrailingValue) {
  117. destination.next(_trailingValue);
  118. this._trailingValue = null;
  119. this._hasTrailingValue = false;
  120. }
  121. }
  122. notifyNext(outerValue, innerValue, outerIndex, innerIndex, innerSub) {
  123. this._sendTrailing();
  124. this._unsubscribe();
  125. }
  126. notifyComplete() {
  127. this._sendTrailing();
  128. this._unsubscribe();
  129. }
  130. }
  131. //# sourceMappingURL=throttle.js.map