使用 rxjs 如何对 HTTP 请求进行排队，使其等待现有请求完成，然后仅发送最新请求？

Question

我有一个 observable 可以跟踪表单的变化。当表单更新时，它应该通过向服务器发送 HTTP PUT 请求来自动保存。我想要这种行为，以便在进行更改时发送第一个 HTTP 请求。对于任何后续更改，在前一个请求完成之前不应发送任何请求。当上一个请求完成后，我想只发送最新的更新。

我曾考虑过使用排气映射，例如formChanges$.pipe(exhaustMap(data => sendHttpUpdate(data))); 因为这会给我在请求进行时忽略所有更新的行为。但是，所有数据在执行过程中都会丢失，因此我需要重新触发 formChanges$ 可观察对象以发送最新更新，否则它将被忽略。

我也考虑过 concatMap 例如formChanges$.pipe(concatMap(data => sendHttpUpdate(data))); 因为这会等待上一个请求完成，但会按顺序发送所有更新，而不仅仅是最新的。

我想使用纯 rxjs 解决方案，即使用 Observables、Subjects 和运算符，而不使用变量设置状态。我是否缺少可以实现此目的的明显运算符或运算符组合？

这是我迄今为止最好的尝试：

    const $send = new Subject();

    formChanges$.pipe(
        sample($send),
        switchMap(this.fakeHttpRequest),
      )
      .subscribe(handleResponse);



    formChanges$.pipe(first())
      .subscribe(() => $send.next());

其中 fakeHttpRequest 是一个函数，它接受值 formChanges$ 并返回 HTTP 请求的 Observable。

这并不令人满意，因为我必须在第一次更改表单后手动触发第一次发送，我觉得这并不能很好地传达代码的意图。

Answer 1

我不敢相信我花了 9 个月的时间才弄清楚这一点，但万一其他人为此大发雷霆：

我意识到没有一个映射运算符（concatMap、exhaustMap）完全符合要求，因此您需要处理映射，然后自己订阅哪些值。解决方案是将每个表单值映射到可观察的 HTTP 请求（尚未订阅，因此没有请求）。然后将此 observable 作为参数传递给throttle。 Throttle 接受一个 observable，它订阅并忽略来自原始源的值，直到完成。 Throttle 还接受一个配置参数，允许您定义前导值和尾随值的行为 - 使其非常灵活。对于这种情况，我们需要两者，因为我们想知道第一个请求的结果和最后一个缓冲请求的结果。

从这里我们现在有了一个包含我们想要的所有源值的可观察对象，即请求数据。但是，我们想要响应数据！简单地再次订阅最终会导致重复的 HTTP 调用（可能有不同的结果）。不要害怕，因为这是shareReplay 来救援的地方。如果我们通过 shareReplay 将原始请求 observable 管道化，那么后续订阅 observable 现在只会得到原始响应，而不是进行新的调用。由于我们有一个 observable 的 observables，我们将要使用 mergeAll 运算符将其展平。我们现在可以订阅它以获取我们的 HTTP 请求的结果。

把这一切放在一起：

formChanges$.pipe(
  map(formValues => this.fakeHttpRequest(formValues).pipe(shareReplay(1)))
  throttle(x => x, {leading: true, trailing: true}),
  mergeAll(),
).subscribe((result) => console.log(result));

根据内幕消息，我假设您关心结果。如果这不是真的，那么这会更简单。我们可以简单地让限制发送请求，我们不必担心重复请求：

formChanges$.pipe(
  map(formValues => this.fakeHttpRequest(formValues)))
  throttle(x => x, {leading: true, trailing: true}),
).subscribe();

Answer 2

嗯，如果我理解的话，这样的事情可能会奏效：

changeStarted: Subject<boolean>; // Keep track if save started;
changeComplete: Subject<boolean>; // Keep track if save finished;

constructor() {
  this.changeStarted =  Subject<boolean>(); 
  this.changeComplete =  Subject<boolean>(); 
  
  formChanges$.pipe(
    bufferToggle(this.changeStarted, () => this.firstCall ? interval(1) : this.changeComplete) // once change has started wait till change is completed.
  ).subscribe(x => {
    this.changeStarted.next(true) //signal to start buffering.
    // Make the http request
    this.fakeHttpRequest.subscribe(() => this.changeComplete.next(true));

}