如何使用RxJS运算符避免多个嵌套订阅? [英] How can I avoid multiple nested subscriptions using RxJS operators?

问题描述

我正在使用Angular进行文件加密和上载类.这些操作很多都是异步的，因此我编写的方法返回了RxJS Observable.

I am working on a file encryption and upload class using Angular. Many of these operations are async and therefore the methods I wrote are returning RxJS Observables.

// 1.
private prepareUpload(file): Observable<T>;

// 2.
private encryptData(data, filekey): Observable<T>

// 3.
private uploadEncryptedData(formData, token, range): Observable<T>

// 4.
private completeUpload(updatedFilekey, token): Observable<T>

我想将此逻辑封装在公共的upload(file)方法中，最终我使用了嵌套订阅，并且可以使用，但是我知道这是错误的，并且出于多种原因，RxJS中使用了反模式.这是代码的简化版本:

I want to encapsulate this logic in a public upload(file) method and I ended up using nested subscriptions and it works but I know that it is wrong and an anti-pattern in RxJS for several reasons. Here is a simplified version of the code:

public upload(file) {
    const gen = this.indexGenerator(); // generator function

    this.prepareUpload(file).subscribe(values => {
    const [response, filekey, data] = values;

    this.encryptData(data, filekey).subscribe(encryptedDataContainer => {
      const formData = this.prepareEncDataUpload(encryptedDataContainer.data, file.name)
      const range = this.getRange(file.size, gen.next().value);

      this.uploadEncryptedData(formData, response.token, range).subscribe(() => {
        if (range.isFinalPart) {
            this.completeUpload(encryptedDataContainer.updatedFilekey, response.token).subscribe(console.log);
        }
      });

    });

  });

}

我无法使用多个RxJS运算符的组合来清除此代码.我的目标是避免嵌套订阅，而是在工作流完成时从公共upload()方法返回单个Observable.

I failed to clean this code using combinations of several RxJS operators. My goal is to avoid nested subscriptions and instead return a single Observable from the public upload() method when the workflow is completed.

谢谢！

推荐答案

您可以在RxJ中使用mergeMap和filter运算符并链接您的呼叫.您将需要创建一些函数级变量以在链接期间使用.

You can use mergeMap and filter operators from RxJs and chain your calls. You will need to create some function level variables to use during the chaining.

import { mergeMap, filter, catchError } from 'rxjs/operators`

public upload(file) {
    const gen = this.indexGenerator(); // generator function
    let range, token;
    this.prepareUpload(file)
      .pipe(
        mergeMap((values) => {
          const [response, filekey, data] = values;
          token = response.token;
          return this.encryptData(data, filekey);
        }),
        mergeMap(encryptedDataContainer => {
          const formData = this.prepareEncDataUpload(encryptedDataContainer.data, file.name)
          range = this.getRange(file.size, gen.next().value);

          return this.uploadEncryptedData(formData, token, range);
        }),
        filter(() => !!range.isFinalPart),
        mergeMap(() => {
          return this.completeUpload(encryptedDataContainer.updatedFilekey, token);
        })
        catchError((error) => {
          console.log(error);
          // handle the error accordingly.
        })
      )
      .subscribe(() => {
        console.log('success');
      });

  }

这篇关于如何使用RxJS运算符避免多个嵌套订阅?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！