diff --git a/package.json b/package.json index 83bff31..489978f 100644 --- a/package.json +++ b/package.json @@ -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", diff --git a/proxy.js b/proxy.js index 84f4093..25aed6f 100644 --- a/proxy.js +++ b/proxy.js @@ -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; - diff --git a/server.js b/server.js index d8373b5..33df9c7 100644 --- a/server.js +++ b/server.js @@ -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; }; - diff --git a/test/queue.js b/test/queue.js index d7e292c..d75d920 100644 --- a/test/queue.js +++ b/test/queue.js @@ -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() {