如何手动完成 angular http observable

Question

我正在开发多个并行文件上传功能，能够 delete/cancel 正在进行的 http 调用。一旦所有电话都收到 completed/cancelled，我会通知消费者。

为此，我使用 forkJoin 组合各个 http observables。但是如果用户点击取消按钮，我不应该等待实际的 http 响应完成。

~~takeUntill 不会优雅地处理它，因为它只会在从底层源 http 流接收到下一个值后才起作用。~~

this.uploadFiles().subscribe(data => {
  console.warn("Upload completed now!!!!!", data);
});

uploadFiles() {
  return forkJoin(
    this.files.map(file => // returns array of observable
        this.uploadFile(file).pipe( catchError(() => of("Error re-emitted as success to prevent early exit"))))
  ).pipe(map(() => {
       // logic for some user friendly statistic
      return data;
    }));
}

Answer 1

将订阅分配给变量并使用按钮取消：

$http: Subscription;
this.$http = this.http.post(this.uploadUrl, formData, {
      reportProgress: false
      // observe: 'events',
});

在取消函数中：

cancelUpload(file: any) {
  file.uploadStatus = "cancelled"; 
  this.$http.unsubscribe();
}

Answer 2

使用 takeUntil 将 Subject 作为通知程序来完成 Observables。您可以将文件 ID 传递给 Subject，并在 takeUntil 中使用 filter 以仅取消具有给定 ID 的文件的文件上传。

使用 defaultIfEmpty 提供一个值，表示请求已取消。这也可以防止外部 forkJoin 在内部空请求完成时立即完成。

private cancelUpload$ = new Subject<number>();

uploadFiles() {
  let errorCount = 0, cancelledCount = 0, successCount = 0;
  return forkJoin(this.dummyFiles.map(file =>
    this.uploadFile(file).pipe(
      // react to CANCEL event
      map(response => response == 'CANCEL' ? ++cancelledCount : ++successCount),
      catchError(() => of(++errorCount))
    )
  )).pipe(map(() => ({ errorCount, successCount, cancelledCount })));
}

uploadFile(file: any) {
  http$.pipe(
    ...
    takeUntil(this.cancelUpload$.pipe(filter(id => id == file.id))), // cancel
    defaultIfEmpty('CANCEL'), // provide value when cancelled
  )
}

cancelUpload(file: any) {
  file.uploadStatus = "cancelled";
  this.cancelUpload$.next(file.id) // cancel action
}

https://stackblitz.com/edit/angular-zteeql-e1zacp

Answer 3

除了takeUntil我们还可以使用merge或race运算符来处理这种情况。然而，它并没有改变底层逻辑。

第 1 步：为个人上传创建 Subject

file.cancelUpload$ = new Subject();

第 2 步：将此主题与实际的 http 调用合并

如果任何可观察对象发出错误，

merge 将完成流。即当我们从 cancelUpload$ 主题发出错误时，http 请求将自动取消（查看网络选项卡）。

return merge(file.cancelUpload$, $http.pipe(...

第 3 步：实际取消代码

cancelUpload(file: any) {
  file.uploadStatus = "cancelled";
  file.cancelUpload$.error(file.uploadStatus);// implicitly subject gets completed
}

第 4 步：完成 cancelUpload$ 上传主题 error/success

这将确保 merge 操作完成，因为两个流现在都已完成。因此 forkJoin 将收到响应。

参考https://stackblitz.com/edit/angular-zteeql?file=src%2Fapp%2Fhttp-example.ts

  uploadFiles() {
    let errorCount = 0,cancelledCount = 0, successCount = 0;
    return forkJoin(
      this.dummyFiles
        .map(file =>
          this.uploadFile(file).pipe(
            catchError(() => of("Error re-emitted as success")) // value doesn't matter
          )
        )
    ).pipe(
      map(() => { // map would receive array of files in the order it was subscribed
        this.dummyFiles.forEach(file => {
          switch (file.uploadStatus) {
            case "success": successCount++; break;
            case "cancelled": cancelledCount++; break;
            case "error": errorCount++; break;
          }
        });
        return { errorCount, successCount, cancelledCount };
      })
    );
  }

  uploadFile(file: any) {
    const formData = new FormData();
    const binaryContent = new Blob([Array(1000).join("some random text")], {
      type: "text/plain"
    }); // dummy data to upload
    formData.append("file", binaryContent);
    const $http = this.http.post(this.uploadUrl, formData, {
      reportProgress: false
      // observe: 'events',
      // withCredentials: true
    });
    file.cancelUpload$ = new Subject();
    file.uploadStatus = "inProgress";
    return merge(
      file.cancelUpload$,
      $http.pipe(
        tap(data => {
          file.uploadStatus = "uploaded";
          file.cancelUpload$.complete();
        }),
        catchError(event => {
          file.uploadStatus = "error";
          file.cancelUpload$.complete();
          return throwError("error");
        })
      )
    );
  }

如何手动完成 angular http observable

How to manually complete the angular http observable

httpclient

rxjs

angular

angular8