Rxjs 多文件上传队列，具有重试和停止行为

Question

我正在尝试构建一段可重用的多文件上传代码。

我不想关心HTTP层的实现，我想纯粹关注流逻辑。

我构建了以下函数来模拟 HTTP 层:

let fakeUploadCounter = 0;

const fakeUpload = () => {
  const _fakeUploadCounter = ++fakeUploadCounter;
  return from(
    Array.from({ length: 100 })
      .fill(null)
      .map((_, i) => i)
  ).pipe(
    mergeMap(x =>
      of(x).pipe(
        delay(x * 100),
        switchMap(x =>
          _fakeUploadCounter % 3 === 0 && x === 25
            ? throwError("Error happened!")
            : of(x)
        )
      )
    )
  );
};

该函数模拟上传进度，每上传3个文件，进度到25%时就会失败。

抛开这个问题，让我们关注重要的一点:主流。

这是我想要实现的目标:

• 仅使用流，不使用命令式编程，也不使用tap将临时结果推送到主题中。我可以 build 这个。但我正在寻找一个优雅的解决方案
• 在上传某些文件时，我希望能够将更多文件添加到上传队列
• 由于浏览器只能同时处理 6 个 HTTP 调用，我不想占用太多的数量，我们应该只能同时上传 3 个文件。一旦一个文件完成、停止或抛出，另一个文件就应该启动
• 当文件上传抛出异常时，我们应该将该文件保留在文件列表中并仍然显示进度。它不会再增加，但至少用户可以看到它失败的地方。在这种情况下，我们应该在该行上看到一些文本，表明存在错误，并且重试按钮应该让我们再次上传，或者放弃按钮让我们完全删除它

这是一个直观的解释:

到目前为止，这是我得到的代码:

export class AppComponent {
  public file$$: Subject<File> = new Subject();
  public retryFile$$: Subject<File> = new Subject();
  public stopFile$$: Subject<File> = new Subject();

  public files$ = this.file$$.pipe(
    mergeMap(file =>
      this.retryFile$$.pipe(
        filter(retryFile => retryFile === file),
        startWith(null),
        map(() =>
          fakeUpload().pipe(
            map(progress => ({ progress })),
            takeUntil(
              this.stopFile$$.pipe(filter(stopFile => stopFile === file))
            ),
            catchError(() => of({ error: true })),
            scan(
              (acc, curr: { progress: number } | { error: true }) => ({
                ...acc,
                ...curr
              }),
              {
                file,
                progress: 0,
                error: false
              }
            )
          )
        )
      )
    ),
    mergeAll(3), // 3 upload in parallel maximum
    scan(
      (acc, curr) => ({
        ...acc,
        // todo we can't use the File reference directly here
        // but we shouldn't use the file name either
        // instead we should generate a unique ID for each upload
        [curr.file.name]: curr
      }),
      {}
    ),
    map(fileEntities => Object.values(fileEntities))
  );

  public addFile() {
    this.file$$.next(new File([], `test-file-${filesCount}`));
    filesCount++;
  }
}

这是 stackblitz 中您可以 fork 的代码:https://stackblitz.com/edit/rxjs-upload-multiple-files-v2?file=src/app/app.component.ts

我已经很接近了!如果您在右侧的 stackblitz 中打开实时演示，然后单击“添加文件”按钮，您将看到可以添加许多文件，并且它们都会被上传。第三个将会优雅地失败。

现在什么不按我想要的方式工作:

• 如果快速单击“添加文件”按钮超过 3 次，则队列中只会出现 3 个文件。我想要全部，但只能同时上传 3 个。然而，所有要上传的文件都应该显示在 View 中，等待开始
• 停止按钮应删除所有上传内容。是否上传或失败

感谢您的帮助

Answer 1

数字 1:

1. If you click quickly more than 3 times on the "add file" button, only 3 files will appear in the queue. I'd like to have all of them but only 3 should be uploading at the same time. Yet, all the files to be uploaded should be displayed in the view, just waiting to start

首先，这是一个很酷的问题，因为据我所知，您不能简单地组合现有的运算符(不会因为 partition 变得愚蠢)。您需要一个自定义运算符来拆分流。如果您不想订阅源两次，则应在拆分之前共享。

---

要按照您想要的方式实现您的解决方案，还有很多工作要做。 但是，就让您的流显示所有文件(无论它们当前是否正在加载)而言，实际上只缺少一个部分。

您想要分割您的直播。一个流应该发出默认值

{
  file,
  progress: 0,
  error: false
}`

立即文件，第二个流应该发出对这些文件的更新。第二个流将具有 mergeAll(3)，但第一个流不需要此限制，因为它不发出网络请求。您可以合并这两个流，并根据需要更新或添加新条目到输出中。

这是一个工作中的例子。我做了一个虚拟示例来稍微抽象一下实现细节。我从具有这种形状的对象数组开始，

{
  id: number,
  message: "HeyThere" + id,
  response: none
}

我做了一个假的 httpRequest 调用来丰富对象

{
  id: number,
  message: "HeyThere" + id,
  response: "Hello"
}

每次添加新对象或丰富对象时，流都会发出。但丰富流仅限一次最多 3 个 httpRequest 调用。

const httpRequest= () => {
  return timer(4000).pipe(
    map(_ => "Hello")
  );
}

const arrayO = [];
arrayO.length = 10;
from(arrayO).pipe(
  map((val, index) => ({
    id: index,
    message: "HeyThere" + index,
    response: "None"
  })),
  share(),
  s => merge(s, s.pipe(
    map(ob => httpRequest().pipe(
      map(val => ({...ob, response: val}))
    )),
    mergeAll(3)
  )),
  scan((acc, val: any) => {
    acc.set(val.id, val);
    return acc;
  }, new Map<number, any>()),
  debounceTime(250),
  map(mapO => Array.from(mapO.values()))
).subscribe(console.log);

我添加了去抖功能，因为我发现它使输出更容易理解。由于我同步添加了所有 10 个未丰富的对象，因此如果我不反跳，它只会向输出发送 10 个数组。另外，由于每个假 HttpRequest 正好需要 4 秒，因此每 4 秒我就会在输出中收到三个垃圾邮件数组。 Debounce 可阻止 UI 出现卡顿或控制台收到垃圾邮件。

数字 2

The stop button should remove any upload. Whether it's uploading or failed

这是一个蠕虫 jar 头，因为每个规范的解决方案都说你应该建立一个状态管理系统。这将是以一种统一的方式与“队列”、“正在加载”、“失败”和“已加载”中的文件进行交互的最简单方法。

使用 RxJS 实现轻量级 Redux 风格的状态管理系统非常容易(只需使用 scan 来管理状态，并使用表示事件的 JSON 对象来转换状态)。最困难的部分是管理当前的 httpRequests。您可能会创建一个自定义 mergeAll() 运算符来接收事件、删除排队的请求，甚至在必要时取消传输中的请求。

使用 stopFile$$ 可以取消中途请求，但如果人们想要停止尚未开始的文件加载(根据您的第一个要求，您希望那些也是可见的)。不管怎样，它有点脆弱，因为向一个主题发出声音永远不能保证有人在听。 redux 式管理是最佳选择的另一个原因。