Javascript 如何使用RxJS构建分组排队系统

我正在try 制作一个由多个队列组成的排队系统，以获取"标题".

import { Subject, timer } from 'rxjs';

enum Title {
  A = 'a',
  B = 'b',
  C = 'c',
}

type Queue = { [key: string]: User[] };
type TitleHolders = { [key: string]: User };
type Timers = { [key: string]: Subject<null> };

const HOLDING_TIME = 10_000;
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
const stamp = () => new Date().toISOString().split('T')[1].split('.')[0];
const log = (data: unknown) => console.log(stamp(), JSON.stringify(data));
const fetchTitle = (title: Title) =>
  sleep(10_000).then(() => ({ success: true }));

class User {
  title: Title | undefined;

  constructor(public name: string) {}

  toString() {
    return `${this.name}: '${this.title}'`;
  }
}

class TitleManager {
  #queue: Queue = {};
  #currentHolders: TitleHolders = {};
  #timers: Timers = {};

  #setTitle(t: Title) {
    if (this.#currentHolders[t]) {
      return;
    }
    const user = this.#queue[t].pop();
    const sub = new Subject<null>();
    sub.subscribe({
      complete: () => {
        const title = user.title;
        delete this.#currentHolders[title];
        delete this.#timers[user.name];
        user.title = undefined;
        log(`>>> '${t}' title REMOVED from ${user.name}`);
        if (this.#queue[title].length) this.#setTitle(title);
      },
    });
    this.#currentHolders[t] ??= user;
    this.#timers[user.name] = sub;
    user.title = t;
    log(`>>> '${t}' title ADDED to ${user.name}`);
    timer(HOLDING_TIME).subscribe(() => sub.complete());
  }

  async giveTitleToUser(t: Title, u: User) {
    this.#queue[t] ??= [];
    this.#queue[t].push(u);
    console.log('starting title request for', u.name);
    const { success } = await fetchTitle(t);
    // Return success status, only set title if if fetching was successful
    if (success) {
      this.#setTitle(t);
    }
  }

  removeUsersTitle(u: User) {
    this.#timers[u.name].complete();
  }
}

const titleManager = new TitleManager();

(async function () {
  const leela = new User('Leela');
  const fry = new User('Fry');

  await titleManager.giveTitleToUser(Title.A, leela);
  await titleManager.giveTitleToUser(Title.A, fry);
})();

(async function () {
  const bender = new User('Bender');

  await sleep(500);
  await titleManager.giveTitleToUser(Title.B, bender);
})();

Link to playground个

该队列一次只能处理1,fetchTitle个呼叫.只有当它完成时，它才应该运行下一个.我正在努力想办法实现这一目标.目前，您可以在日志(log)中看到，它启动了fetchTitle个请求中的2个，但没有完成1个请求.如果我可以将成功状态返回给呼叫方，也会很好，但我不知道这是否可能.