scan.js 3.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106
  1. import { Subscriber } from '../Subscriber';
  2. /* tslint:enable:max-line-length */
  3. /**
  4. * Applies an accumulator function over the source Observable, and returns each
  5. * intermediate result, with an optional seed value.
  6. *
  7. * <span class="informal">It's like {@link reduce}, but emits the current
  8. * accumulation whenever the source emits a value.</span>
  9. *
  10. * <img src="./img/scan.png" width="100%">
  11. *
  12. * Combines together all values emitted on the source, using an accumulator
  13. * function that knows how to join a new source value into the accumulation from
  14. * the past. Is similar to {@link reduce}, but emits the intermediate
  15. * accumulations.
  16. *
  17. * Returns an Observable that applies a specified `accumulator` function to each
  18. * item emitted by the source Observable. If a `seed` value is specified, then
  19. * that value will be used as the initial value for the accumulator. If no seed
  20. * value is specified, the first item of the source is used as the seed.
  21. *
  22. * @example <caption>Count the number of click events</caption>
  23. * var clicks = Rx.Observable.fromEvent(document, 'click');
  24. * var ones = clicks.mapTo(1);
  25. * var seed = 0;
  26. * var count = ones.scan((acc, one) => acc + one, seed);
  27. * count.subscribe(x => console.log(x));
  28. *
  29. * @see {@link expand}
  30. * @see {@link mergeScan}
  31. * @see {@link reduce}
  32. *
  33. * @param {function(acc: R, value: T, index: number): R} accumulator
  34. * The accumulator function called on each source value.
  35. * @param {T|R} [seed] The initial accumulation value.
  36. * @return {Observable<R>} An observable of the accumulated values.
  37. * @method scan
  38. * @owner Observable
  39. */
  40. export function scan(accumulator, seed) {
  41. let hasSeed = false;
  42. // providing a seed of `undefined` *should* be valid and trigger
  43. // hasSeed! so don't use `seed !== undefined` checks!
  44. // For this reason, we have to check it here at the original call site
  45. // otherwise inside Operator/Subscriber we won't know if `undefined`
  46. // means they didn't provide anything or if they literally provided `undefined`
  47. if (arguments.length >= 2) {
  48. hasSeed = true;
  49. }
  50. return function scanOperatorFunction(source) {
  51. return source.lift(new ScanOperator(accumulator, seed, hasSeed));
  52. };
  53. }
  54. class ScanOperator {
  55. constructor(accumulator, seed, hasSeed = false) {
  56. this.accumulator = accumulator;
  57. this.seed = seed;
  58. this.hasSeed = hasSeed;
  59. }
  60. call(subscriber, source) {
  61. return source.subscribe(new ScanSubscriber(subscriber, this.accumulator, this.seed, this.hasSeed));
  62. }
  63. }
  64. /**
  65. * We need this JSDoc comment for affecting ESDoc.
  66. * @ignore
  67. * @extends {Ignored}
  68. */
  69. class ScanSubscriber extends Subscriber {
  70. constructor(destination, accumulator, _seed, hasSeed) {
  71. super(destination);
  72. this.accumulator = accumulator;
  73. this._seed = _seed;
  74. this.hasSeed = hasSeed;
  75. this.index = 0;
  76. }
  77. get seed() {
  78. return this._seed;
  79. }
  80. set seed(value) {
  81. this.hasSeed = true;
  82. this._seed = value;
  83. }
  84. _next(value) {
  85. if (!this.hasSeed) {
  86. this.seed = value;
  87. this.destination.next(value);
  88. }
  89. else {
  90. return this._tryNext(value);
  91. }
  92. }
  93. _tryNext(value) {
  94. const index = this.index++;
  95. let result;
  96. try {
  97. result = this.accumulator(this.seed, value, index);
  98. }
  99. catch (err) {
  100. this.destination.error(err);
  101. }
  102. this.seed = result;
  103. this.destination.next(result);
  104. }
  105. }
  106. //# sourceMappingURL=scan.js.map