来自 NodeJS Express 'post' 的 ReactiveX Observable

Question

是否可以从以下位置创建 ReactiveX Observable：

app = express();
app.post('/path', function() {...})

?

我的意思是，有一种方法可以创建一个可观察的fromEvent，我已经在其中使用了很多使用object.on('eventName', function(){}) 注册的事件，但来自 express 的帖子并不完全一样。

Answer 1

Rx-ify 一路快递？

尝试将 express 中路由的回调样式实现转换为 Rx 所增加的复杂性使我得出结论，这不是前进的方向。这是因为 express 不知道 Rx，因此没有订阅生命周期和多事件发射的概念。

martin 的答案是一个概念验证，你可以在其中清楚地看到上述问题。这意味着对于每条路线，您需要创建一个单独的Subject 或share() 一个主题，然后为每条路线制作大量过滤器以分离实现。这是冗余，因为 express 已经将您的请求路由到正确的处理程序。

将 Rx 世界与快速回调结合起来

如果您想获得 Expressjs 路由与 RxJs 结合的好处，我会将路由逻辑分开并将其保留在 express 回调中，实际实现将其用作 Rx 并结合 .toPromise() 来激活它：

app.post('/user/:id', (req, res) => {
  return getUserById(req.params.id)
    .toPromise()/* return the Rx stream as promise to express so it traces its lifecycle */
    .then(
      user => res.send(user),
      err => res.status(500).send(err.message)
    );
});    

function getUserById(id) {
  // stub implementation
  return Rx.Observable.of({ id, name: 'username' }) 
    .delay(100);
}

Answer 2

我认为最简单的方法是使用Subject 从express 重新发送事件。

var express = require('express');
var Rx = require('rxjs');
var app = express();

let subject = new Rx.Subject();

app.get('/', (req, res) => subject.next([req, res]));

subject
  .do(a => console.log('123345'))
  .subscribe(args => {
    let [req, res] = args;
    res.send('Hello World!');
  });

app.listen(3000, function () {
  console.log('Example app listening on port 3000!')
});

主要区别在于您需要将[req, res] 包装为一个数组，因为您希望将两者都传递给subject.next(...)。稍后可以使用 let [req, res] = args; 解压缩。

很遗憾，您不能使用 Observable.bindCallback，因为它只需要来自包装函数的单个响应，然后完成，这就是您现在在这种情况下想要的。