Comment étendre correctement rxjs/Observable avec mon premier opérateur personnalisé ?

Question

J'essaie d'utiliser un modèle de réessai dans mes appels de service (en fait : @Effects dans ngrx/store) avec des intervalles de retard accrus. Puisque j'ai réussi à trouver un code fonctionnel pour un appel (même s'il n'est pas optimisé, je ne veux pas me concentrer sur ce point dans ma question), je voudrais maintenant l'extraire dans un opérateur Observable personnalisé et l'utiliser de manière répétée dans tous mes appels de service.

Je ne sais pas comment concevoir l'API/utilisation du nouvel opérateur et comment le faire reconnaître par TypeScript.

Le code ci-dessous ne fonctionne certainement pas, car il accumule probablement une multitude de problèmes.

Donc, maintenant j'ai un appel/effet comme suit :

  @Effect()
  loadData$: Observable<Action> = this.actions$
    .ofType(ActionTypes.LOAD_DATA)
    .pluck('payload')
    .switchMap(params => {
      return this.myService.getData(params)
        .map(res => new LoadDataCompleteAction(res))

        // ...and this part would have to be extracted: 
        .retryWhen(attempts => Observable
          .zip(attempts, Observable.range(1, 5))
          .flatMap((n, i) => {
            if (i < 4) {
              return Observable.timer(1000 * i);
            } else {
              throw(n);
            }
          })
        )
    })
    .catch(err => Observable.of(new LoadDataFailed()));

et ce que je recherche, c'est de pouvoir réutiliser la partie réessai dans d'autres effets, et d'avoir un modèle similaire à celui ci-dessous :

  @Effect()
  loadData$: Observable<Action> = this.actions$
    .ofType(ActionTypes.LOAD_DATA)
    .pluck('payload')
    .switchMap(params => {
      return this.myService.getData(params)
        .map(res => new LoadDataCompleteAction(res))
        .retryWhen(attempts => Observable.retryOrThrow(attempts, maxAttempts)

         // or maybe - that's my design question
         .retryOrThrow(attempts, maxAttempts)
    })
    .catch(err => Observable.of(new LoadDataFailed()));

Pour simplifier, nous pourrions supposer que le modèle de rappel de retard ( i * 1000 ) serait constant pour toute l'application.

Le code ci-dessous est ma tentative, mais il ne fonctionne évidemment pas.

declare module 'rxjs/Observable' {
  interface Observable<T> {
    retryOrThrow<T>(attempts: any, max: number): Observable<T>;
  }     
}

Observable.prototype.retryOrThrow = function(attempt, max) {
  console.log('retryOrThrow called');

  return Observable.create(subscriber => {
    const source = this;
    const subscription = source.subscribe(() => {
        // important: catch errors from user-provided callbacks
        try {
          subscriber
            .zip(attempt, Observable.range(1, max + 1))
            .flatMap((n, i) => {
              console.log(n, i);
              if (i < max) {
                return Observable.timer(1000 * i);
              } else {
                throw(n);
              }
            });
        } catch (err) {
          subscriber.error(err);
        }
      },
      // be sure to handle errors and completions as appropriate and send them along
      err => subscriber.error(err),
      () => subscriber.complete());

    // to return now
    return subscription;
  });
};

1. Je ne sais pas comment concevoir l'API pour le nouvel opérateur, quelle syntaxe conviendrait le mieux ici.
2. Je ne sais pas comment déclarer correctement le nouvel opérateur et l'espace de noms ou le module Observable, pour que TypeScript reconnaisse les nouveaux éléments.

Appel de service actualisé :

  getMocky(){
    const u = Math.random();

    const okUrl = 'http://www.mocky.io/v2/58ffadf71100009b17f60044';
    const erUrl = 'http://www.mocky.io/v2/58ffae7f110000ba17f60046';
    return u > 0.6 ? this.http.get(okUrl) : this.http.get(erUrl);
  }

Answer 1

Je ne peux pas répondre directement à votre question car je ne sais pas comment étendre rxjs avec un opérateur personnalisé.

Heureusement, dans votre cas, vous (et moi) n'avez pas besoin de savoir. Ce que vous demandez vraiment, c'est comment prédéfinir une chaîne d'opérateurs à réutiliser à plusieurs endroits.

Tout ce dont vous avez besoin est le let-Opérateur et il est assez simple à utiliser.

Commencez par extraire la logique que vous voulez réutiliser dans une fonction qui renvoie un observable :

function retryOrThrow(maxAttempts, timeout) {
  return (source) =>
    source
      .retryWhen(attempts => Observable
        .zip(attempts, Observable.range(1, maxAttempts + 1))
        .flatMap((n, i) => {
          if (i < maxAttempts) {
            return Observable.timer(timeout * i);
          } else {
            throw (n);
          }
        })
      );
}

Utilisez la fonction dans votre effet en utilisant let :

@Effect()
loadData$: Observable<Action> = this.actions$
  .ofType(ActionTypes.LOAD_DATA)
  .pluck('payload')
  .switchMap(params => {
    return this.myService.getData(params)
      .map(res => new LoadDataCompleteAction(res))
      // inject the pre-defined logic with the desired parameters
      .let(retryOrThrow(5, 1000))
    })
    .catch(err => Observable.of(new LoadDataFailed()));

Le moyen le plus simple de comprendre ce que let fait est de regarder son code source . C'est très simple, car il s'agit simplement d'appliquer une fonction donnée à l'observable-source.