http server as Rx observable

httprx.js

'use strict';

var http = require('http');
var Rx = require('rx');
var qs = require('qs');


var server, source, subscription;

server = http.createServer();
server.setTimeout(30000);
server.listen(8000);

source = Rx.Observable.fromEventPattern(
    server.addListener.bind(server, 'request'),
    server.removeListener.bind(server, 'request'),
    function select(args) {
        return { req: args[0], res: args[1] };
    }
).takeUntil(Rx.Observable.fromEvent(server, 'close'));

subscription = source.subscribe(
    function onNext(e) {
        var req, res, body;

        req = e.req;
        res = e.res;
        body = '';

        req.setEncoding('utf8');
        res.setHeader('content-type', 'text/html');

        Rx.Node.fromStream(req).subscribe(
            function (x) {
                body += x;
            },
            function (e) {
                res.statusCode = 500;
                res.end();
            },
            function () {
                console.log(req.body = qs.parse(body));
                res.statusCode = 200;
                res.end('<form method="post"><input type="hidden" name="foo" value="test"><input type="submit"></form>');
            }
        );
    },
    function onError(e) {
        console.error(e);
    },
    function onComplete() {
        console.log('complete');
    }
);

process.on('SIGINT', server.close.bind(server));