node.js - RxJS + node.js HTTP 服务器实现？

Question

在 RxJS 实现之前，我一直很好地使用 node.js。

这是我的试用代码学习-

响应式(Reactive)扩展/rxjs-node https://github.com/Reactive-Extensions/rxjs-node

---

rx_http.js
(node.js 的 http 库的 RxJS 包装器)

var Rx = require("./rx.min");
var http = require("http");
for(var k in http)
{
    exports[k] = http[k];
}
exports.createServer = function ()
{
    var subject = new Rx.AsyncSubject();
    var observable = subject.asObservable();
    observable.server = http.createServer(function (request, response)
    {
       subject.onNext({ request:request, response:response });
       subject.onCompleted();
    });
    return observable;
};

---

服务器.js

var http = require('./rx_http');

// rxServer
var serverObservable = http.createServer();
var port = 3000;
serverObservable.server.listen(port);
console.log("Server listening on port: "+port);

// HTTP request event loop function
serverObservable.subscribe(function (data)
{
    var req = data.request;
    console.log(req.headers);

    var res = data.response;
    res.writeHead(200, {'Content-Type':"text/html"});
    res.end("hello world");

    console.log("res content out");
});

// exceptiopn
process.on('uncaughtException', function (err)
{
    console.log(['Caught exception:', err.message].join(" "));
});

---

代码以一次性“hello world”输出到浏览器结束，RxServer 停止对另一个访问使用react(brwoser 重新加载等)。

我正在学习 RxJS 的东西，但是在网络上找到的文档很少。

告诉我代码有什么问题，如果您知道更好的实现，请分享。谢谢。

Answer 1

在 rx_http.js 中使用 Rx.Subject 而不是 Rx.AsyncSubject。

AsyncSubject 缓存 onNext() 的最后一个值，并在完成时将其传播给所有观察者。 AsyncSubject

exports.createServer = function ()
{
    var subject = new Rx.Subject();
    var observable = subject.asObservable();
    observable.server = http.createServer(function (request, response)
    {
       subject.onNext({ request:request, response:response });
    });
    return observable;
};