exhaustMap.js 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128
  1. import { OuterSubscriber } from '../OuterSubscriber';
  2. import { subscribeToResult } from '../util/subscribeToResult';
  3. /* tslint:enable:max-line-length */
  4. /**
  5. * Projects each source value to an Observable which is merged in the output
  6. * Observable only if the previous projected Observable has completed.
  7. *
  8. * <span class="informal">Maps each value to an Observable, then flattens all of
  9. * these inner Observables using {@link exhaust}.</span>
  10. *
  11. * <img src="./img/exhaustMap.png" width="100%">
  12. *
  13. * Returns an Observable that emits items based on applying a function that you
  14. * supply to each item emitted by the source Observable, where that function
  15. * returns an (so-called "inner") Observable. When it projects a source value to
  16. * an Observable, the output Observable begins emitting the items emitted by
  17. * that projected Observable. However, `exhaustMap` ignores every new projected
  18. * Observable if the previous projected Observable has not yet completed. Once
  19. * that one completes, it will accept and flatten the next projected Observable
  20. * and repeat this process.
  21. *
  22. * @example <caption>Run a finite timer for each click, only if there is no currently active timer</caption>
  23. * var clicks = Rx.Observable.fromEvent(document, 'click');
  24. * var result = clicks.exhaustMap((ev) => Rx.Observable.interval(1000).take(5));
  25. * result.subscribe(x => console.log(x));
  26. *
  27. * @see {@link concatMap}
  28. * @see {@link exhaust}
  29. * @see {@link mergeMap}
  30. * @see {@link switchMap}
  31. *
  32. * @param {function(value: T, ?index: number): ObservableInput} project A function
  33. * that, when applied to an item emitted by the source Observable, returns an
  34. * Observable.
  35. * @param {function(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number): any} [resultSelector]
  36. * A function to produce the value on the output Observable based on the values
  37. * and the indices of the source (outer) emission and the inner Observable
  38. * emission. The arguments passed to this function are:
  39. * - `outerValue`: the value that came from the source
  40. * - `innerValue`: the value that came from the projected Observable
  41. * - `outerIndex`: the "index" of the value that came from the source
  42. * - `innerIndex`: the "index" of the value from the projected Observable
  43. * @return {Observable} An Observable containing projected Observables
  44. * of each item of the source, ignoring projected Observables that start before
  45. * their preceding Observable has completed.
  46. * @method exhaustMap
  47. * @owner Observable
  48. */
  49. export function exhaustMap(project, resultSelector) {
  50. return (source) => source.lift(new SwitchFirstMapOperator(project, resultSelector));
  51. }
  52. class SwitchFirstMapOperator {
  53. constructor(project, resultSelector) {
  54. this.project = project;
  55. this.resultSelector = resultSelector;
  56. }
  57. call(subscriber, source) {
  58. return source.subscribe(new SwitchFirstMapSubscriber(subscriber, this.project, this.resultSelector));
  59. }
  60. }
  61. /**
  62. * We need this JSDoc comment for affecting ESDoc.
  63. * @ignore
  64. * @extends {Ignored}
  65. */
  66. class SwitchFirstMapSubscriber extends OuterSubscriber {
  67. constructor(destination, project, resultSelector) {
  68. super(destination);
  69. this.project = project;
  70. this.resultSelector = resultSelector;
  71. this.hasSubscription = false;
  72. this.hasCompleted = false;
  73. this.index = 0;
  74. }
  75. _next(value) {
  76. if (!this.hasSubscription) {
  77. this.tryNext(value);
  78. }
  79. }
  80. tryNext(value) {
  81. const index = this.index++;
  82. const destination = this.destination;
  83. try {
  84. const result = this.project(value, index);
  85. this.hasSubscription = true;
  86. this.add(subscribeToResult(this, result, value, index));
  87. }
  88. catch (err) {
  89. destination.error(err);
  90. }
  91. }
  92. _complete() {
  93. this.hasCompleted = true;
  94. if (!this.hasSubscription) {
  95. this.destination.complete();
  96. }
  97. }
  98. notifyNext(outerValue, innerValue, outerIndex, innerIndex, innerSub) {
  99. const { resultSelector, destination } = this;
  100. if (resultSelector) {
  101. this.trySelectResult(outerValue, innerValue, outerIndex, innerIndex);
  102. }
  103. else {
  104. destination.next(innerValue);
  105. }
  106. }
  107. trySelectResult(outerValue, innerValue, outerIndex, innerIndex) {
  108. const { resultSelector, destination } = this;
  109. try {
  110. const result = resultSelector(outerValue, innerValue, outerIndex, innerIndex);
  111. destination.next(result);
  112. }
  113. catch (err) {
  114. destination.error(err);
  115. }
  116. }
  117. notifyError(err) {
  118. this.destination.error(err);
  119. }
  120. notifyComplete(innerSub) {
  121. this.remove(innerSub);
  122. this.hasSubscription = false;
  123. if (this.hasCompleted) {
  124. this.destination.complete();
  125. }
  126. }
  127. }
  128. //# sourceMappingURL=exhaustMap.js.map