retryWhen

signature: retryWhen(receives: (errors: Observable) => Observable, the: scheduler): Observable

Retry an observable sequence on error based on custom criteria.

retryWhen - 图1

Examples

Example 1: Trigger retry after specified duration

( jsBin |
jsFiddle )

  1. import { timer } from 'rxjs/observable/timer';
  2. import { interval } from 'rxjs/observable/interval';
  3. import { map, tap, retryWhen, delayWhen } from 'rxjs/operators';
  4. //emit value every 1s
  5. const source = interval(1000);
  6. const example = source.pipe(
  7. map(val => {
  8. if (val > 5) {
  9. //error will be picked up by retryWhen
  10. throw val;
  11. }
  12. return val;
  13. }),
  14. retryWhen(errors =>
  15. errors.pipe(
  16. //log error message
  17. tap(val => console.log(`Value ${val} was too high!`)),
  18. //restart in 5 seconds
  19. delayWhen(val => timer(val * 1000))
  20. )
  21. )
  22. );
  23. /*
  24. output:
  25. 0
  26. 1
  27. 2
  28. 3
  29. 4
  30. 5
  31. "Value 6 was too high!"
  32. --Wait 5 seconds then repeat
  33. */
  34. const subscribe = example.subscribe(val => console.log(val));
Example 2: Customizable retry with increased duration

(
StackBlitz
)

Credit to Maxim Koretskyi for the
optimization

  1. import { Observable } from 'rxjs/Observable';
  2. import { _throw } from 'rxjs/observable/throw';
  3. import { timer } from 'rxjs/observable/timer';
  4. import { mergeMap, finalize } from 'rxjs/operators';
  5. export const genericRetryStrategy = ({
  6. maxRetryAttempts = 3,
  7. scalingDuration = 1000,
  8. excludedStatusCodes = []
  9. }: {
  10. maxRetryAttempts?: number,
  11. scalingDuration?: number,
  12. excludedStatusCodes?: number[]
  13. } = {}) => (attempts: Observable<any>) => {
  14. return attempts.pipe(
  15. mergeMap((error, i) => {
  16. const retryAttempt = i + 1;
  17. // if maximum number of retries have been met
  18. // or response is a status code we don't wish to retry, throw error
  19. if (
  20. retryAttempt > maxRetryAttempts ||
  21. excludedStatusCodes.find(e => e === error.status)
  22. ) {
  23. return _throw(error);
  24. }
  25. console.log(
  26. `Attempt ${retryAttempt}: retrying in ${retryAttempt *
  27. scalingDuration}ms`
  28. );
  29. // retry after 1s, 2s, etc...
  30. return timer(retryAttempt * scalingDuration);
  31. }),
  32. finalize(() => console.log('We are done!'))
  33. );
  34. };
  1. import { Component, OnInit } from '@angular/core';
  2. import { catchError, retryWhen } from 'rxjs/operators';
  3. import { of } from 'rxjs/observable/of';
  4. import { genericRetryStrategy } from './rxjs-utils';
  5. import { AppService } from './app.service';
  6. @Component({
  7. selector: 'my-app',
  8. templateUrl: './app.component.html',
  9. styleUrls: [ './app.component.css' ]
  10. })
  11. export class AppComponent implements OnInit {
  12. constructor(private _appService: AppService) {}
  13. ngOnInit() {
  14. this._appService
  15. .getData(500)
  16. .pipe(
  17. retryWhen(genericRetryStrategy()),
  18. catchError(error => of(error))
  19. )
  20. .subscribe(console.log);
  21. // excluding status code, delay for logging clarity
  22. setTimeout(() => {
  23. this._appService
  24. .getData(500)
  25. .pipe(
  26. retryWhen(genericRetryStrategy({
  27. scalingDuration: 2000,
  28. excludedStatusCodes: [500]
  29. })),
  30. catchError(error => of(error))
  31. )
  32. .subscribe(e => console.log('Exluded code:', e.status));
  33. }, 8000);
  34. }
  35. }

Additional Resources


:file_folder: Source Code:
https://github.com/ReactiveX/rxjs/blob/master/src/internal/operators/retryWhen.ts