bitinflow/server
1
0
Fork 0
mirror of https://github.com/bitinflow/server.git synced 2026-03-13 13:35:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

use bouncy to proxy requests

Browse Source 
- simplifies our codebase a lot!
- queued requests are currently broken
This commit is contained in:
Roman Shtylman
2013-11-03 16:56:41 -05:00
parent e7e112963c
commit 6afb75704a
4 changed files with 84 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
package.json
View File

@@ -11,7 +11,6 @@
"request": "2.11.4",
"book": "1.2.0",
"optimist": "0.3.4",
"http-raw": "1.1.0",
"debug": "0.7.2",
"bookrc": "0.0.1",
"book-git": "0.0.2",
@@ -24,7 +23,8 @@
"taters": "0.0.5",
"express": "3.2.6",
"makeup": "0.0.1",
"enchilada": "0.3.0"
"enchilada": "0.3.0",
"bouncy": "3.2.1"
},
"devDependencies": {
"mocha": "1.6.0",

128
proxy.js
View File

@@ -1,17 +1,9 @@
var http = require('http');
var net = require('net');
var EventEmitter = require('events').EventEmitter;
var log = require('bookrc');
var debug = require('debug')('localtunnel-server');
// here be dragons, understanding of node http internals will be required
var HTTPParser = process.binding('http_parser').HTTPParser;
// available parsers for requests
// this is borrowed from how node does things by preallocating parsers
var parsers = http.parsers;
var Proxy = function(opt, cb) {
if (!(this instanceof Proxy)) {
return new Proxy(opt, cb);
@@ -60,8 +52,7 @@ var Proxy = function(opt, cb) {
// clear waiting by ending responses, (requests?)
self.waiting.forEach(function(waiting) {
waiting[1].end();
waiting[3].end(); // write stream
waiting(null);
});
self.emit('end');
@@ -80,13 +71,6 @@ var Proxy = function(opt, cb) {
// a single connection is enough to keep client id slot open
clearTimeout(conn_timeout);
// allocate a response parser for the socket
// it only needs one since it will reuse it
socket.parser = parsers.alloc();
socket._orig_ondata = socket.ondata;
socket.ondata = upstream_response;
socket.once('close', function(had_error) {
debug('client %s closed socket (error: %s)', id, had_error);
@@ -100,7 +84,6 @@ var Proxy = function(opt, cb) {
}
// need to track total sockets, not just active available
debug('remaining client sockets: %s', self.sockets.length);
// no more sockets for this ident
@@ -118,10 +101,10 @@ var Proxy = function(opt, cb) {
self.sockets.push(socket);
var next = self.waiting.shift();
if (next) {
var wait_cb = self.waiting.shift();
if (wait_cb) {
debug('handling queued request');
self.proxy_request(next[0], next[1], next[2], next[3]);
self.next_socket(wait_cb);
}
});
@@ -132,104 +115,39 @@ var Proxy = function(opt, cb) {
Proxy.prototype.__proto__ = EventEmitter.prototype;
Proxy.prototype.proxy_request = function(req, res, rs, ws) {
Proxy.prototype.next_socket = function(cb) {
var self = this;
// socket is a tcp connection back to the user hosting the site
var sock = self.sockets.shift();
// TODO how to handle queue?
// queue request
if (!sock) {
debug('no more clients, queued: %s', req.url);
rs.pause();
self.waiting.push([req, res, rs, ws]);
return;
debug('no more client, queue callback');
return self.waiting.push(cb);
}
debug('handle req: %s', req.url);
// put the socket back
function done() {
if (!sock.destroyed) {
debug('retuning socket');
self.sockets.push(sock);
}
// pipe incoming request into tcp socket
// incoming request will close the socket when done
// lt client should establish a new socket once request is finished
// we do this instead of keeping socket open to make things easier
rs.pipe(sock);
sock.ws = ws;
sock.req = req;
// since tcp connection to upstream are kept open
// invoke parsing so we know when the response is complete
var parser = sock.parser;
parser.reinitialize(HTTPParser.RESPONSE);
parser.socket = sock;
// we have completed a response
// the tcp socket is free again
parser.onIncoming = function (res) {
parser.onMessageComplete = function() {
debug('ended response: %s', req.url);
// any request we had going on is now done
ws.end();
sock.end();
// no more forwarding
delete sock.ws;
delete sock.req;
delete parser.onIncoming;
};
var wait = self.waiting.shift();
debug('processing queued cb');
if (wait) {
return self.next_socket(cb);
}
};
rs.resume();
debug('processing request');
cb(sock, done);
};
Proxy.prototype.proxy_upgrade = function(req, socket, head) {
var sock = self.sockets.shift();
if (!sock) {
// no available sockets to upgrade to
// TODO queue?
return socket.end();
}
var stream = req.createRawStream();
sock.ws = ws;
sock.upgraded = true;
stream.once('end', function() {
delete sock.ws;
// when this ends, we just reset the socket to the lt client
// this is easier than trying to figure anything else out
sock.end();
});
stream.pipe(sock);
sock.once('end', socket.end.bind(ws));
Proxy.prototype._done = function() {
var self = this;
};
function upstream_response(d, start, end) {
var socket = this;
var ws = socket.ws;
if (!ws) {
return log.warn('no stream set for req:', socket.req.url);
}
ws.write(d.slice(start, end));
if (socket.upgraded) {
return;
}
var ret = socket.parser.execute(d, start, end - start);
if (ret instanceof Error) {
log.error(ret);
parsers.free(parser);
socket.destroy(ret);
}
}
module.exports = Proxy;

135
server.js
View File

@@ -1,12 +1,12 @@
var log = require('bookrc');
var express = require('express');
var bouncy = require('bouncy');
var taters = require('taters');
var enchilada = require('enchilada');
var makeup = require('makeup');
var engine = require('engine.io');
var browserkthx = require('browserkthx');
var debug = require('debug')('localtunnel-server');
var createRawServer = require('http-raw');
var Proxy = require('./proxy');
var rand_id = require('./lib/rand_id');
@@ -23,9 +23,7 @@ var stats = {
tunnels: 0,
};
// return true if request will be handled, false otherwise
function middleware(req, res) {
function maybe_bounce(req, res, bounce) {
// without a hostname, we won't know who the request is for
var hostname = req.headers.host;
if (!hostname) {
@@ -35,35 +33,9 @@ function middleware(req, res) {
var match = hostname.match(/^([a-z]{4})[.].*/);
// not for a specific client
// pass on to regular server
if (!match) {
var match = req.url.match(/\/([a-z]{4})$/);
var req_id;
if (req.url === '/?new') {
req_id = rand_id();
}
else if (match && match[1]) {
req_id = match[1];
}
// will not handle
if (!req_id) {
return false;
}
new_client(req_id, {}, function(err, info) {
if (err) {
res.statusCode = 500;
return res.end(err.message);
}
var url = 'http://' + req_id + '.' + req.headers.host;
info.url = url;
res.end(JSON.stringify(info));
});
return true;
return false;
}
var client_id = match[1];
@@ -83,38 +55,19 @@ function middleware(req, res) {
--stats.requests;
});
var rs = req.createRawStream();
var ws = res.createRawStream();
// get client port
client.next_socket(function(socket, done) {
var stream = bounce(socket); //, { headers: { connection: 'close' } });
// return the socket to the client pool
stream.once('end', function() {
done();
});
});
client.proxy_request(req, res, rs, ws);
return true;
}
var handle_upgrade = function(req, socket, head) {
var hostname = req.headers.host;
if (!hostname) {
return socket.end();
}
var match = hostname.match(/^([a-z]{4})[.].*/);
// not handled by us
if (!match) {
return false;
}
var client_id = match[1];
var client = clients[client_id];
// no such subdomain
if (!client) {
return socket.end();
}
client.handle_upgrade(req, socket, head);
return true;
};
function new_client(id, opt, cb) {
// can't ask for id already is use
@@ -150,22 +103,12 @@ function new_client(id, opt, cb) {
module.exports = function(opt) {
opt = opt || {};
var server = createRawServer();
var app = express();
app.set('view engine', 'html');
app.set('views', __dirname + '/views');
app.engine('html', require('hbs').__express);
app.use(function(req, res, next) {
if (middleware(req, res)) {
return;
}
next();
});
app.use(express.favicon());
app.use(browserkthx({ ie: '< 9' }));
@@ -181,6 +124,42 @@ module.exports = function(opt) {
app.use(express.static(__dirname + '/static'));
app.use(app.router);
app.get('/', function(req, res, next) {
if (!req.query.hasOwnProperty('new')) {
return next();
}
var req_id = rand_id();
debug('making new client with id %s', req_id);
new_client(req_id, opt, function(err, info) {
if (err) {
res.statusCode = 500;
return res.end(err.message);
}
var url = 'http://' + req_id + '.' + req.headers.host;
info.url = url;
res.end(JSON.stringify(info));
});
});
app.get('/:req_id', function(req, res, next) {
var req_id = req.param('req_id');
debug('making new client with id %s', req_id);
new_client(req_id, opt, function(err, info) {
if (err) {
res.statusCode = 500;
return res.end(err.message);
}
var url = 'http://' + req_id + '.' + req.headers.host;
info.url = url;
res.end(JSON.stringify(info));
});
});
app.get('/', function(req, res, next) {
return res.render('index');
});
@@ -214,17 +193,19 @@ module.exports = function(opt) {
eio_server.handleRequest(req, res);
});
server.on('request', app);
server.on('upgrade', handle_upgrade);
var app_port = 0;
var app_server = app.listen(app_port, function() {
app_port = app_server.address().port;
});
server.on('upgrade', function(req, socket, head) {
if (handle_upgrade(req, socket, head)) {
var server = bouncy(function(req, res, bounce) {
// if we should bounce this request, then don't send to our server
if (maybe_bounce(req, res, bounce)) {
return;
}
};
eio_server.handleUpgrade(req, socket, head);
bounce(app_port);
});
return server;
};

2
test/queue.js
View File

@@ -24,7 +24,7 @@ test('setup local http server', function(done) {
setTimeout(function() {
res.setHeader('x-count', req.headers['x-count']);
res.end('foo');
}, 100);
}, 500);
});
server.listen(function() {