Блог

У меня есть некоторый код, который выполняет загрузку большого файла меньшими порциями. Вот что у меня есть:

 upload(project: Project): Observable<any> {

    var chunks = this.chunkFiles(project);

    var totalPercent = project.file
        .map(file => Math.ceil(file.size / this.chunkSize))
        .reduce((sum, curr) => sum   curr) * 100;

    var finishedPercent = 0;
    var currentPercent = 0;

    return from(chunks)
        .pipe(
            concatMap(request =>
            {
                return this.http.request(request)
                    .pipe(
                        retryWhen(error =>
                        {
                            return interval(1000)
                                .pipe(
                                    flatMap(count =>
                                    {
                                        if (count == 2)
                                        {
                                            return throwError(error);
                                        }
                                        return of(count);
                                    })
                                );
                        }));
            }),
            map(event =>
            {
                if (event.type === HttpEventType.UploadProgress)
                {
                    var progress = event as HttpProgressEvent;
                    currentPercent = Math.round(100 * progress.loaded / progress.total);
                    return {
                        type: event.type,
                        loaded: finishedPercent   currentPercent,
                        total: totalPercent
                    };
                }
                else if (event instanceof HttpResponse) 
                {
                    finishedPercent  = 100;
                    if (finishedPercent == totalPercent)
                    {
                        return event;
                    }
                }
            }),
            filter(response => response !== undefined)
        );
}
  

Предполагается, что он начнет загружать фрагменты файла, и как только одна из загрузок фрагмента завершится неудачей, она должна прекратиться, и ошибка должна распространиться на код, который вызывает мою функцию загрузки. Этот конкретный код выглядит следующим образом:

 onStartUpload = this.actions$
    .ofType(forRoot.START_UPLOAD).pipe(
    switchMap((action: forRoot.StartUpload) =>
        this.uploadService
            .upload(action.project).pipe(
            map((event) => {
                if (event.type === HttpEventType.UploadProgress) {
                    const progress = event as HttpProgressEvent;
                    const percentDone = Math.round(100 * progress.loaded / progress.total);
                    return new forRoot.UploadProgress(percentDone);
                } else if (event instanceof HttpResponse) {
                    return new forRoot.UploadSucceeded();
                } else {
                    console.log(event);
                }
            }),
            filter(a => a !== undefined),
            catchError(error => {
                if (error instanceof HttpErrorResponse amp;amp; error.status === 400) {
                    const message = this.getMessage(error);
                    return of(new forRoot.UploadFailed(message || "Bad request"));
                } else {
                    return of(new forRoot.UploadFailed("Network error"));
                }
            })))
        );
  

Проблема в том, что я получаю только сообщение «Ошибка сети», тогда как я ожидаю что-то со статусом ошибки 400 (поэтому я должен получать либо конкретное сообщение, либо «Неверный запрос».

Я предполагаю, что я неправильно обрабатываю ошибку при повторной попытке, но я попробовал несколько разных идей, и, похоже, ни одна из них не работает. Чего я не понимаю?

Отредактируйте, чтобы показать рабочий код без повторных попыток:

Вот код, который корректно обрабатывает сообщения об ошибках, однако в нем нет повторной попытки:

 return from(chunks)
    .pipe(
        concatMap(request =>
        {
            return this.http.request(request)
        }),
        map(event =>
        {
            if (event.type === HttpEventType.UploadProgress)
            {
                var progress = event as HttpProgressEvent;
                currentPercent = Math.round(100 * progress.loaded / progress.total);
                return {
                    type: event.type,
                    loaded: finishedPercent   currentPercent,
                    total: totalPercent
                };
            }
            else if (event instanceof HttpResponse) 
            {
                finishedPercent  = 100;
                if (finishedPercent == totalPercent)
                {
                    return event;
                }
            }
        }),
        filter(response => response !== undefined)
    );
  

Я предполагал, что есть что-то, что я могу добавить к вызову http.request, чтобы повторить указанное количество раз с задержкой между повторными попытками, а затем, если все они завершатся неудачей, выдать ошибку, аналогичную http.request . Похоже, что происходит то, что он переходит от кода concatMap к коду map. Я могу добавить вызовы console.log (событие) и увидеть сообщение об ошибке, возвращаемое с сервера. Я все еще довольно новичок в rxjs, поэтому, возможно, я не понимаю, но я ожидал, что после возникновения ошибки код карты не будет выполняться.

Проблема здесь:

 retryWhen(error => {     // << error is a stream
  return interval(1000)
    .pipe(
        flatMap(count => {
          if (count == 2){
              return throwError(error); // so here a stream is thrown
          }

          return of(count);
      })
    );
})
  

error аргумент — это не отдельная ошибка, а фактически наблюдаемая из всех возникающих ошибок.

Чтобы исправить это, вы должны вернуть observable, производный от error , например:

 retryWhen(errors$ =>
  errors$.pipe(
    delay(1000)
  )
)
  

Обновленный пример ограниченной обработки ошибок

 retryWhen(errors =>
  errors.pipe(
    // switchMap to add limiting logic
    switchMap((error, index) =>
      index == 3                      // if it is a 4th error
      ? throwError(error)             //  -- we rethrow it upstream
      : of(error).pipe( delay(1000) ) //  -- delay a retry otherwise
    )
  )
)
  

Вот пример повторной попытки с экспоненциальным возвратом.

И статья об обработке ошибок в rxjs (с примечанием о retryWhen досрочном завершении).

Надеюсь, это поможет