Функция Rxjs Retry with Delay

Я пытаюсь использовать retry с функцией delay, я ожидаю, что функция вызовет после задержки 1000 мс, но это не делает, что может быть ошибкой здесь? Посмотрите на консольный вывод, в то же время 16:22:48.

Я жду там 16:22:48, 16:22:59...

canCreate: boolean;
getSomeFunction(): Observable<boolean> {
        return new Observable<boolean>(
            observer => {
                const canCreate = null; // this is just null for now, will some value later
                if (canCreate == null) {
                    observer.error('error');
                } else {
                    observer.next(true);
                }
                observer.complete();
            }
        )
    }


this.getSomeFunction()
      .do((value) => {
        this.cCreate = value;
      }, (error) => {
         console.log(error + new Date().toTimeString());
      })
      .delay(1000)
      .retry(10)
      .subscribe(
        value => this.cCreate = value,
        error => {
          this.cCreate = false;
        },
        () => {}
      );
  }

и результат консоли:

Ответ 1

delay() используется для введения задержки между событиями, испускаемыми наблюдаемой. Но наблюдаемое никогда не испускает никакого события. Это просто ошибки сразу.

То, что вы ищете, это retryWhen(), который позволяет retryWhen(), через сколько времени повторить попытку:

RxJS 5:

  .retryWhen(errors => errors.delay(1000).take(10))

RxJS 6:

import { retryWhen, delay, take } from 'rxjs/operators'
someFunction().pipe(
  // ...
  retryWhen(errors => errors.pipe(delay(1000), take(10)))
)

Это завершит все наблюдаемое после 10 попыток. Если вы хотите сделать ошибку всей наблюдаемой после 10 попыток, наблюдаемая, возвращаемая обратным вызовом retryWhen, должна выдать:

RxJS 5:

  .retryWhen(errors => errors.delay(1000).take(10).concat(Observable.throw()))

RxJS 6:

import { retryWhen, delay, take, concatMap, throwError } from 'rxjs/operators'
someFunction().pipe(
  // ...
  retryWhen(errors => errors.pipe(delay(1000), take(10), concatMap(throwError)))
)

Ответ 2

Чтобы добавить к @JB Nizet ответ. Если вы пишете это в rxjs 5+ с lettable операторами, структурируйте его как

retryWhen(errors => errors.pipe(delay(1000), take(5)))

Ответ 3

Это может помочь вам

let values$ = Rx.Observable.interval(1000).take(5);
let errorFixed = false;

values$
.map((val) => {
   if(errorFixed) { return val; }
   else if( val > 0 && val % 2 === 0) {
      errorFixed = true;
      throw { error : 'error' };

   } else {
      return val;
   }
})
.retryWhen((err) => {
    console.log('retrying again');
    return err.delay(1000).take(3); // 3 times
})
.subscribe((val) => { console.log('value',val) });

Ответ 4

Работает на rxjs версии 6.3.3

https://stackblitz.com/edit/http-basics-8swzpy

Откройте консоль и увидите повторы

Образец кода

import { map, catchError, retryWhen, take, delay, concat } from 'rxjs/operators';
import { throwError } from 'rxjs';


export class ApiEXT {

    static get apiURL(): string { return 'http://localhost:57886/api'; };
    static httpCLIENT: HttpClient;

 static POST(postOBJ: any, retryCOUNT: number = 0, retryITNERVAL: number = 1000) {
        return this.httpCLIENT
            .post(this.apiURL, JSON.stringify(postOBJ))
            .pipe(
                map(this.handleSUCCESS),
                retryWhen(errors => errors.pipe(delay(retryITNERVAL), take(retryCOUNT), concat(throwError("Giving up Retry.!")))),
                catchError(this.handleERROR));
    }


  private static handleSUCCESS(json_response: string): any {
        //TODO: cast_and_return    
        return JSON.parse(json_response);

    }

 private static handleERROR(error: Response) {
        let errorMSG: string;
        switch (error.status) {
            case -1: errorMSG = "(" + error.status + "/" + error.statusText + ")" + " Server Not Reachable.!"; break;
            default: errorMSG = "(" + error.status + "/" + error.statusText + ")" + " Unknown Error while connecting with server.!"; break;
        }
        console.error(errorMSG);
        return throwError(errorMSG);
    }

}

Ответ 5

Я придумал следующее решение, использующее retryWhen и Observable.Interval, но в этом решении функция error подписки никогда не вызывает,

this.branchService.getCanCreate()
  .do((value) => {
    this.cCreate = value;
  }, (error) => {
    console.log('do', error + new Date().toTimeString());
  })
  .retryWhen(errors => {
    return Observable.interval(1000).take(3).concat(Observable.throw('error')));
  })
  .subscribe(
    value => {
      this.cCreate = !!value
      console.log('success', new Date().toTimeString());
    },
    error => {
      console.log('subscribe', error + new Date().toTimeString());
      this.cCreate = false;
    },
    () => {
      console.log('finally', new Date().toTimeString());
    }
  );

Ответ 6

Для ngrx5+ мы могли бы создать оператор:


function retryRequest(constructor: () => Observable, count: number, delayTime: number) {
  let index = 0;
  return of(1) // we need to repeat not the result of constructor(), but the call of constructor() itself
    .pipe(
      switchMap(constructor),
      retryWhen(errors => errors.pipe(
        delay(delayTime),
        mergeMap(error => {
          if (++index > count) {
            return throwError(error);
          }
          return of(error);
        })
      ))
    );
}

Ответ 7

RxJS предоставляет оператор повтора, который повторно подписывает Observable на заданное количество счетчиков при возникновении ошибки. Перед выдачей ошибки Observable повторно подписывается на заданное количество счетчиков оператором повтора, и, если ошибка все еще существует, выдается ошибка. Повторить полезно, чтобы попасть в URL много раз. Возможно, что из-за пропускной способности сети URL не возвращает успешные данные за один раз, а когда он удаляется, он может возвращать данные успешно. Если после повторного использования все еще есть ошибка в Observable, то catchError может быть использован для возврата Observable с данными, определенными пользователем по умолчанию.

getBook (id: number): Observable {вернуть this.http.get(this.bookUrl + "/" + id).pipe(retry (3), catchError (err => {console.log(err); возврат ( ноль); }) ); }

Ответ 8

Недавно я столкнулся с этой проблемой и обнаружил, что принятое решение можно улучшить.

Observable.pipe(
     retryWhen(errors => errors.pipe(
      delay(1000),
      take(10))),
    first(v => true),
    timeout(10000))

По сути, он повторяет попытку, как уже упоминалось, но это заканчивается сразу же, без добавления какого-либо (ошибочного) значения с помощью оператора "first".

Если не удается найти значение в течение тайм-фрейма, возникает ошибка.