onErrorResumeNext.js 4.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118
  1. import { FromObservable } from '../observable/FromObservable';
  2. import { isArray } from '../util/isArray';
  3. import { OuterSubscriber } from '../OuterSubscriber';
  4. import { subscribeToResult } from '../util/subscribeToResult';
  5. /* tslint:enable:max-line-length */
  6. /**
  7. * When any of the provided Observable emits an complete or error notification, it immediately subscribes to the next one
  8. * that was passed.
  9. *
  10. * <span class="informal">Execute series of Observables no matter what, even if it means swallowing errors.</span>
  11. *
  12. * <img src="./img/onErrorResumeNext.png" width="100%">
  13. *
  14. * `onErrorResumeNext` is an operator that accepts a series of Observables, provided either directly as
  15. * arguments or as an array. If no single Observable is provided, returned Observable will simply behave the same
  16. * as the source.
  17. *
  18. * `onErrorResumeNext` returns an Observable that starts by subscribing and re-emitting values from the source Observable.
  19. * When its stream of values ends - no matter if Observable completed or emitted an error - `onErrorResumeNext`
  20. * will subscribe to the first Observable that was passed as an argument to the method. It will start re-emitting
  21. * its values as well and - again - when that stream ends, `onErrorResumeNext` will proceed to subscribing yet another
  22. * Observable in provided series, no matter if previous Observable completed or ended with an error. This will
  23. * be happening until there is no more Observables left in the series, at which point returned Observable will
  24. * complete - even if the last subscribed stream ended with an error.
  25. *
  26. * `onErrorResumeNext` can be therefore thought of as version of {@link concat} operator, which is more permissive
  27. * when it comes to the errors emitted by its input Observables. While `concat` subscribes to the next Observable
  28. * in series only if previous one successfully completed, `onErrorResumeNext` subscribes even if it ended with
  29. * an error.
  30. *
  31. * Note that you do not get any access to errors emitted by the Observables. In particular do not
  32. * expect these errors to appear in error callback passed to {@link subscribe}. If you want to take
  33. * specific actions based on what error was emitted by an Observable, you should try out {@link catch} instead.
  34. *
  35. *
  36. * @example <caption>Subscribe to the next Observable after map fails</caption>
  37. * Rx.Observable.of(1, 2, 3, 0)
  38. * .map(x => {
  39. * if (x === 0) { throw Error(); }
  40. return 10 / x;
  41. * })
  42. * .onErrorResumeNext(Rx.Observable.of(1, 2, 3))
  43. * .subscribe(
  44. * val => console.log(val),
  45. * err => console.log(err), // Will never be called.
  46. * () => console.log('that\'s it!')
  47. * );
  48. *
  49. * // Logs:
  50. * // 10
  51. * // 5
  52. * // 3.3333333333333335
  53. * // 1
  54. * // 2
  55. * // 3
  56. * // "that's it!"
  57. *
  58. * @see {@link concat}
  59. * @see {@link catch}
  60. *
  61. * @param {...ObservableInput} observables Observables passed either directly or as an array.
  62. * @return {Observable} An Observable that emits values from source Observable, but - if it errors - subscribes
  63. * to the next passed Observable and so on, until it completes or runs out of Observables.
  64. * @method onErrorResumeNext
  65. * @owner Observable
  66. */
  67. export function onErrorResumeNext(...nextSources) {
  68. if (nextSources.length === 1 && isArray(nextSources[0])) {
  69. nextSources = nextSources[0];
  70. }
  71. return (source) => source.lift(new OnErrorResumeNextOperator(nextSources));
  72. }
  73. /* tslint:enable:max-line-length */
  74. export function onErrorResumeNextStatic(...nextSources) {
  75. let source = null;
  76. if (nextSources.length === 1 && isArray(nextSources[0])) {
  77. nextSources = nextSources[0];
  78. }
  79. source = nextSources.shift();
  80. return new FromObservable(source, null).lift(new OnErrorResumeNextOperator(nextSources));
  81. }
  82. class OnErrorResumeNextOperator {
  83. constructor(nextSources) {
  84. this.nextSources = nextSources;
  85. }
  86. call(subscriber, source) {
  87. return source.subscribe(new OnErrorResumeNextSubscriber(subscriber, this.nextSources));
  88. }
  89. }
  90. class OnErrorResumeNextSubscriber extends OuterSubscriber {
  91. constructor(destination, nextSources) {
  92. super(destination);
  93. this.destination = destination;
  94. this.nextSources = nextSources;
  95. }
  96. notifyError(error, innerSub) {
  97. this.subscribeToNextSource();
  98. }
  99. notifyComplete(innerSub) {
  100. this.subscribeToNextSource();
  101. }
  102. _error(err) {
  103. this.subscribeToNextSource();
  104. }
  105. _complete() {
  106. this.subscribeToNextSource();
  107. }
  108. subscribeToNextSource() {
  109. const next = this.nextSources.shift();
  110. if (next) {
  111. this.add(subscribeToResult(this, next));
  112. }
  113. else {
  114. this.destination.complete();
  115. }
  116. }
  117. }
  118. //# sourceMappingURL=onErrorResumeNext.js.map