exhaustMap.js 2.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677
  1. import { OuterSubscriber } from '../OuterSubscriber';
  2. import { InnerSubscriber } from '../InnerSubscriber';
  3. import { subscribeToResult } from '../util/subscribeToResult';
  4. import { map } from './map';
  5. import { from } from '../observable/from';
  6. export function exhaustMap(project, resultSelector) {
  7. if (resultSelector) {
  8. return (source) => source.pipe(exhaustMap((a, i) => from(project(a, i)).pipe(map((b, ii) => resultSelector(a, b, i, ii)))));
  9. }
  10. return (source) => source.lift(new ExhaustMapOperator(project));
  11. }
  12. class ExhaustMapOperator {
  13. constructor(project) {
  14. this.project = project;
  15. }
  16. call(subscriber, source) {
  17. return source.subscribe(new ExhaustMapSubscriber(subscriber, this.project));
  18. }
  19. }
  20. class ExhaustMapSubscriber extends OuterSubscriber {
  21. constructor(destination, project) {
  22. super(destination);
  23. this.project = project;
  24. this.hasSubscription = false;
  25. this.hasCompleted = false;
  26. this.index = 0;
  27. }
  28. _next(value) {
  29. if (!this.hasSubscription) {
  30. this.tryNext(value);
  31. }
  32. }
  33. tryNext(value) {
  34. let result;
  35. const index = this.index++;
  36. try {
  37. result = this.project(value, index);
  38. }
  39. catch (err) {
  40. this.destination.error(err);
  41. return;
  42. }
  43. this.hasSubscription = true;
  44. this._innerSub(result, value, index);
  45. }
  46. _innerSub(result, value, index) {
  47. const innerSubscriber = new InnerSubscriber(this, value, index);
  48. const destination = this.destination;
  49. destination.add(innerSubscriber);
  50. const innerSubscription = subscribeToResult(this, result, undefined, undefined, innerSubscriber);
  51. if (innerSubscription !== innerSubscriber) {
  52. destination.add(innerSubscription);
  53. }
  54. }
  55. _complete() {
  56. this.hasCompleted = true;
  57. if (!this.hasSubscription) {
  58. this.destination.complete();
  59. }
  60. this.unsubscribe();
  61. }
  62. notifyNext(outerValue, innerValue, outerIndex, innerIndex, innerSub) {
  63. this.destination.next(innerValue);
  64. }
  65. notifyError(err) {
  66. this.destination.error(err);
  67. }
  68. notifyComplete(innerSub) {
  69. const destination = this.destination;
  70. destination.remove(innerSub);
  71. this.hasSubscription = false;
  72. if (this.hasCompleted) {
  73. this.destination.complete();
  74. }
  75. }
  76. }
  77. //# sourceMappingURL=exhaustMap.js.map