var req = this._httpMessage;
var parser = this.parser;
+ assert(parser);
+
var ret = parser.execute(d);
if (ret instanceof Error) {
debug('parse error');
var res = parser.incoming;
req.res = res;
- socket.ondata = null;
- socket.onend = null;
+ socket.removeListener('data', socketOnData);
+ socket.removeListener('end', socketOnEnd);
parser.finish();
var bodyHead = d.slice(bytesParsed, d.length);
socket.removeListener('close', socketCloseListener);
socket.removeListener('error', socketErrorListener);
+ // TODO(isaacs): Need a way to reset a stream to fresh state
+ // IE, not flowing, and not explicitly paused.
+ socket._readableState.flowing = null;
+
req.emit(eventName, res, socket, bodyHead);
req.emit('close');
} else {
// send a final response after this client sends a request
// body. So, we must not free the parser.
parser.incoming.statusCode !== 100) {
+ socket.removeListener('data', socketOnData);
+ socket.removeListener('end', socketOnEnd);
freeParser(parser, req);
}
}
parser.maxHeaderPairs = 2000;
}
+ parser.onIncoming = parserOnIncomingClient;
socket.on('error', socketErrorListener);
- socket.ondata = socketOnData;
- socket.onend = socketOnEnd;
+ socket.on('data', socketOnData);
+ socket.on('end', socketOnEnd);
socket.on('close', socketCloseListener);
- parser.onIncoming = parserOnIncomingClient;
req.emit('socket', socket);
});
// info.url is not set for response parsers but that's not
// applicable here since all our parsers are request parsers.
function parserOnHeadersComplete(info) {
+ debug('parserOnHeadersComplete', info);
var parser = this;
var headers = info.headers;
var url = info.url;
if (parser) {
parser._headers = [];
parser.onIncoming = null;
- if (parser.socket) {
- parser.socket.onend = null;
- parser.socket.ondata = null;
+ if (parser.socket)
parser.socket.parser = null;
- }
parser.socket = null;
parser.incoming = null;
parsers.free(parser);
parser.maxHeaderPairs = 2000;
}
- socket.addListener('error', function(e) {
+ socket.addListener('error', socketOnError);
+ socket.addListener('close', serverSocketCloseListener);
+ parser.onIncoming = parserOnIncoming;
+ socket.on('end', socketOnEnd);
+ socket.on('data', socketOnData);
+
+ // TODO(isaacs): Move all these functions out of here
+ function socketOnError(e) {
self.emit('clientError', e, this);
- });
+ }
- socket.ondata = function(d) {
+ function socketOnData(d) {
+ debug('SERVER socketOnData %d', d.length);
var ret = parser.execute(d);
if (ret instanceof Error) {
debug('parse error');
// Upgrade or CONNECT
var bytesParsed = ret;
var req = parser.incoming;
+ debug('SERVER upgrade or connect', req.method);
- socket.ondata = null;
- socket.onend = null;
+ socket.removeListener('data', socketOnData);
+ socket.removeListener('end', socketOnEnd);
socket.removeListener('close', serverSocketCloseListener);
parser.finish();
freeParser(parser, req);
var eventName = req.method === 'CONNECT' ? 'connect' : 'upgrade';
if (EventEmitter.listenerCount(self, eventName) > 0) {
+ debug('SERVER have listener for %s', eventName);
var bodyHead = d.slice(bytesParsed, d.length);
- self.emit(eventName, req, req.socket, bodyHead);
+ // TODO(isaacs): Need a way to reset a stream to fresh state
+ // IE, not flowing, and not explicitly paused.
+ socket._readableState.flowing = null;
+ self.emit(eventName, req, socket, bodyHead);
} else {
// Got upgrade header or CONNECT method, but have no handler.
socket.destroy();
}
}
- };
+ }
- socket.onend = function() {
+ function socketOnEnd() {
+ var socket = this;
var ret = parser.finish();
if (ret instanceof Error) {
} else {
if (socket.writable) socket.end();
}
- };
+ }
- socket.addListener('close', serverSocketCloseListener);
// The following callback is issued after the headers have been read on a
// new message. In this callback we setup the response object and pass it
// to the user.
- parser.onIncoming = function(req, shouldKeepAlive) {
+
+ function parserOnIncoming(req, shouldKeepAlive) {
incoming.push(req);
var res = new ServerResponse(req);
// When we're finished writing the response, check if this is the last
// respose, if so destroy the socket.
- res.on('finish', function() {
+ res.on('finish', resOnFinish);
+ function resOnFinish() {
// Usually the first incoming element should be our request. it may
// be that in the case abortIncoming() was called that the incoming
// array will be empty.
m.assignSocket(socket);
}
}
- });
+ }
if (!util.isUndefined(req.headers.expect) &&
(req.httpVersionMajor == 1 && req.httpVersionMinor == 1) &&
self.emit('request', req, res);
}
return false; // Not a HEAD response. (Not even a response!)
- };
+ }
}
exports._connectionListener = connectionListener;
} else {
debug('encrypted.onend');
}
-
- if (this.onend) this.onend();
}
}
} else {
// Give them requested data
- if (this.ondata) {
- var self = this;
- this.ondata(pool, start, start + bytesRead);
-
- // Consume data automatically
- // simple/test-https-drain fails without it
- process.nextTick(function() {
- self.read(bytesRead);
- });
- }
this.push(pool.slice(start, start + bytesRead));
}
this.readable = this.writable = false;
}
- this.onend = null;
-
// shut down the socket when we're finished with it.
this.on('finish', onSocketFinish);
this.on('_socketEnd', onSocketEnd);
self.bytesRead += nread;
// Optimization: emit the original buffer with end points
- var ret = true;
- if (self.ondata) self.ondata(buffer);
- else ret = self.push(buffer);
+ var ret = self.push(buffer);
if (handle.reading && !ret) {
handle.reading = false;
maybeDestroy(self);
}
- if (self.onend) self.once('end', self.onend);
-
// push a null to signal the end of data.
self.push(null);
request_upgradeHead = upgradeHead;
- socket.ondata = function(d, start, end) {
- var data = d.toString('utf8', start, end);
+ socket.on('data', function(d) {
+ var data = d.toString('utf8');
if (data == 'kill') {
socket.end();
} else {
socket.write(data, 'utf8');
}
- };
+ });
});
}