使用 RxJS 的排队功能

Question

我在后端使用带有 NodeJS 的 rxjs。 我有一个允许消费者运行远程纱线安装过程的 Rest API。 install 函数返回一个可观察的进程。因此，当模块安装成功时，它会在 observable 和 complete 中发出一个值。此时，Rest API 会返回一个响应给用户说安装成功。如果安装失败，进程将在流中抛出一个错误，Rest API 返回另一个带有错误信息的响应。

我的问题是：

API被消费者多次并行调用，所以后端会有并行安装。

我尝试过油门操作符来创建一个队列，但它使第一个流保持活动状态。因此，如果第一个进程“已完成”，则返回“true”但流未完成

export class MyService {
    // the function called by the REST API
    installGlobal(moduleName: string): Observable < boolean > {
        // I think, there are something to do here to make it queuing
        return this.run('yarn', ['global', 'add', moduleName]);
    }

    private run(cmd: string, args: string[]): Observable < boolean > {
        const cmd$ = fromPromise(spawn(cmd, args)).pipe(
            map(stdout => {
                this.logger.info(`Install Module Successfully`);
                this.logger.info(`stdout: ${stdout.toString()}`);
                return true;
            }),
            catchError(error => {
                const errorMessage: string = error.stderr.toString();
                return _throw(errorMessage.substr(errorMessage.indexOf(' ') + 1));
            })
        );
        return cmd$;
    }
} 

我的期望：

要么有多个请求，要么必须排队。所以第一个将被处理，所有并行的一次必须排队。处理第一个流时，它必须将响应返回给 API 使用者（如 200 完成）并从队列中恢复下一个流。

[UPDATE-01 July 2019]：添加示例

您可以在stackblitz获取代码演示

我已经重新实现了现有代码，并且通过多次订阅将调用队列的服务来模拟我的 API 调用

Answer 1

一个简单的 Rxjs 队列可以像下面这样完成

const queque=new Subject()
// sequential processing
queue.pipe(concatMap(item=>yourObservableFunction(item)).subscribe()
// add stuff to the queue 
queque.next(item)