javascript - 用 Observable 替换中间件模式(取消订阅？)

Question

在很多框架中都有使用中间件的通用方法。

在ExpressJS - 在这里我用 middleware-chain 处理请求:

app.use('/url', middleware1);
app.use('/url', middleware2);
app.use('/url', middleware3);

在Telegraf - 这里我用 middleware-chain 处理消息事件(作为机器人) :

bot.on('text', middleware1);
bot.on('text', middleware2);
bot.on('text', middleware3);

以上示例有效，我想实现以推迟某些请求/事件的实用程序。在这种情况下，如果我的 middleware-chain目前正忙于处理一些请求/事件 - 我可以将新传入的事件/请求存储在数据库中，稍后在我的 middleware-chain 时处理它将是免费的。这样做的主要原因是限制某些用户一次只能发送一个请求/事件并等待响应 middleware-chain很忙。

让我们假设我可以在数据库中存储/检索我需要的任何东西。但是，如果我从数据库中检索到一些请求/事件——我想通过我所有的中间件发送/传递这个请求/事件。在这种情况下，如果框架不支持，我需要重新实现事件冒泡。

但是我可以为此使用 Observable。

ExpressJS:

const sbj = new Subject();

app.use('/url', (...args) => {
  sbj.next(args)
});

observable.subscribe({
  next: middleware1
});
observable.subscribe({
  next: middleware2
});
observable.subscribe({
  next: middleware3
});

电报:

const sbj = new Subject();

bot.on('text', (...args) => {
  sbj.next(args)
});

observable.subscribe({
  next: middleware1
});
observable.subscribe({
  next: middleware2
});
observable.subscribe({
  next: middleware3
});

但我在 Observable-chain 中遗漏了一项重要功能我在middleware-chain => 在 middleware-chain我可以取消下一个中间件。例如。 middleware1 -> next -> middleware2 -> cancel .

如何使用 Observable 实现相同的行为？ (所以一个订阅可以取消另一个)？我认为应该有一个优雅的解决方案。

Answer 1

RxJS 有自己的“链接”机制，称为管道运算符。您可以使用各种类型的产品 as you can see here .如果你有可重用的中间件，你可以很好地将它们实现为你自己的操作符。例如，身份验证运算符可能如下所示:

function authenticator() = (obs$) => obs$.pipe(
  mergeMap(([req, res]) => req.isAuthenticated() ? of([req, res]) : throwError([res, "Must be authenticated!"]))
)

然后，您可以按如下方式使用它:

const sbj = new Subject();

app.use('/url', (...args) => {
  sbj.next(args)
});

sbj.pipe(
   authenticator(),
   // more operators
).subscribe(([req, res]) => {
   // send response
}, ([res, err]) => {
   // send error
})