非协调mergeMap行为

Question

我目前正在研究一种文件上传方法，它要求我限制并发请求的数量。

我已经开始写一个如何处理它的原型

const items = Array.from({ length: 50 }).map((_, n) => n);
from(items)
  .pipe(
    mergeMap(n => {
      return of(n).pipe(delay(2000));
    }, 5)
  )
  .subscribe(n => {
    console.log(n);
  });

但是，当我用实际调用换掉of时，它确实起了作用。它只处理一个块，所以假设20个文件中有5个

from(files)
  .pipe(mergeMap(handleFile, 5))
  .subscribe(console.log);

handleFile函数返回对我的自定义ajax实现的调用

import { Observable, Subscriber } from 'rxjs';
import axios from 'axios';

const { CancelToken } = axios;

class AjaxSubscriber extends Subscriber {
  constructor(destination, settings) {
    super(destination);
    this.send(settings);
  }

  send(settings) {
    const cancelToken = new CancelToken(cancel => {
      // An executor function receives a cancel function as a parameter
      this.cancel = cancel;
    });
    axios(Object.assign({ cancelToken }, settings))
      .then(resp => this.next([null, resp.data]))
      .catch(e => this.next([e, null]));
  }

  next(config) {
    this.done = true;
    const { destination } = this;
    destination.next(config);
  }

  unsubscribe() {
    if (this.cancel) {
      this.cancel();
    }
    super.unsubscribe();
  }
}

export class AjaxObservable extends Observable {
  static create(settings) {
    return new AjaxObservable(settings);
  }

  constructor(settings) {
    super();
    this.settings = settings;
  }

  _subscribe(subscriber) {
    return new AjaxSubscriber(subscriber, this.settings);
  }
}

所以看起来就像这样

function handleFile() {
  return AjaxObservable.create({
    url: "https://jsonplaceholder.typicode.com/todos/1"
  });
}

CodeSandbox

如果我从合并映射函数中删除并发参数，一切都可以正常工作，但它同时上传所有文件。有办法解决这个问题吗？

Answer 1

问题是我没有在complete()中调用AjaxSubscriber方法，所以我修改了代码如下：

pass(response) {
  this.next(response);
  this.complete();
}

并从axios调用：

axios(Object.assign({ cancelToken }, settings))
  .then(resp => this.pass([null, resp.data]))
  .catch(e => this.pass([e, null]));