RxJS + node.js Implementazione del server HTTP?

Question

Sono stato bene con node.js fino all'implementazione di RxJS.

Ecco il mio codice di prova studying-

reattivi-extensions/rxjs nodi https://github.com/Reactive-Extensions/rxjs-node

---

rx_http.js
(RxJS involucro del lib http di node.js)

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; 
}; 

---

server.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(" ")); 
}); 

---

Il codice termina con una tantum 'ciao mondo' uscita al browser, e il RxServer non reagisce ad un altro accesso (brwoser ricarica ecc).

Sono sulla strada per imparare la cosa RxJS, ma poca documentazione trovata sul web.

Dimmi cosa c'è che non va nel codice e se conosci implementazioni migliori, ti preghiamo di condividere. Grazie.

Answer 1

Chiamare completato sul soggetto quando arriva la prima richiesta termina la sequenza osservabile. Potresti rimuovere la linea e provare di nuovo.

Spero che aiuti.

Ahmet Ali Akkas

Answer 2

Uso Rx.Subject invece di Rx.AsyncSubject in rx_http.js.

AsyncSubject memorizza l'ultimo valore di onNext() e lo diffonde a tutti gli osservatori una volta completato. 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; 
};