9

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

这是我正在学习的试用代码-

Reactive-Extensions / 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 停止对另一个访问（浏览器重新加载等）做出反应。

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

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

4

2 回答 2

5

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

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

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;
};
于 2015-01-16T22:35:19.777 回答
3

当第一个请求到达时，对主题调用 oncompleted 会结束可观察序列。能否请您删除该行再试一次。

我希望它有所帮助。

艾哈迈德·阿里·阿卡斯

于 2012-11-05T04:03:11.120 回答