node/lib/net.js

1267 строки
31 KiB
JavaScript
Исходник Обычный вид История

2011-11-02 21:00:57 +04:00
// Copyright Joyent, Inc. and other Node contributors.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the
// "Software"), to deal in the Software without restriction, including
// without limitation the rights to use, copy, modify, merge, publish,
// distribute, sublicense, and/or sell copies of the Software, and to permit
// persons to whom the Software is furnished to do so, subject to the
// following conditions:
//
// The above copyright notice and this permission notice shall be included
// in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
// USE OR OTHER DEALINGS IN THE SOFTWARE.
2011-06-16 23:11:05 +04:00
var events = require('events');
var stream = require('stream');
2011-06-16 23:11:05 +04:00
var timers = require('timers');
var util = require('util');
var assert = require('assert');
var cares = process.binding('cares_wrap');
var cluster;
2012-02-19 03:01:35 +04:00
function noop() {}
2011-10-07 21:00:48 +04:00
// constructor for lazy loading
function createPipe() {
var Pipe = process.binding('pipe_wrap').Pipe;
return new Pipe();
}
// constructor for lazy loading
function createTCP() {
var TCP = process.binding('tcp_wrap').TCP;
return new TCP();
}
2011-06-16 23:11:05 +04:00
var debug;
if (process.env.NODE_DEBUG && /net/.test(process.env.NODE_DEBUG)) {
var pid = process.pid;
debug = function(x) {
// if console is not set up yet, then skip this.
if (!console.error)
return;
console.error('NET: %d', pid,
util.format.apply(util, arguments).slice(0, 500));
};
2011-06-16 23:11:05 +04:00
} else {
debug = function() { };
}
2011-07-20 00:49:42 +04:00
function isPipeName(s) {
return typeof s === 'string' && toNumber(s) === false;
2011-07-20 00:49:42 +04:00
}
2011-06-17 16:27:02 +04:00
exports.createServer = function() {
return new Server(arguments[0], arguments[1]);
};
// Target API:
//
// var s = net.connect({port: 80, host: 'google.com'}, function() {
// ...
// });
//
// There are various forms:
//
// connect(options, [cb])
// connect(port, [host], [cb])
// connect(path, [cb]);
//
exports.connect = exports.createConnection = function() {
var args = normalizeConnectArgs(arguments);
var s = new Socket(args[0]);
return Socket.prototype.connect.apply(s, args);
};
2011-07-20 00:49:42 +04:00
// Returns an array [options] or [options, cb]
// It is the same as the argument of Socket.prototype.connect().
function normalizeConnectArgs(args) {
var options = {};
if (typeof args[0] === 'object') {
// connect(options, [cb])
options = args[0];
} else if (isPipeName(args[0])) {
// connect(path, [cb]);
options.path = args[0];
2011-07-20 00:49:42 +04:00
} else {
// connect(port, [host], [cb])
options.port = args[0];
if (typeof args[1] === 'string') {
options.host = args[1];
}
2011-07-20 00:49:42 +04:00
}
var cb = args[args.length - 1];
return (typeof cb === 'function') ? [options, cb] : [options];
}
exports._normalizeConnectArgs = normalizeConnectArgs;
2011-06-17 16:27:02 +04:00
// called when creating new Socket, or when re-using a closed Socket
2011-07-01 03:37:30 +04:00
function initSocketHandle(self) {
self.destroyed = false;
2012-03-20 11:21:38 +04:00
self.errorEmitted = false;
self.bytesRead = 0;
self._bytesDispatched = 0;
// Handle creation may be deferred to bind() or connect() time.
if (self._handle) {
self._handle.owner = self;
self._handle.onread = onread;
}
2011-07-01 03:37:30 +04:00
}
2011-06-17 16:27:02 +04:00
2011-06-16 23:11:05 +04:00
function Socket(options) {
if (!(this instanceof Socket)) return new Socket(options);
this._connecting = false;
this._handle = null;
switch (typeof options) {
2012-07-25 04:05:22 +04:00
case 'number':
options = { fd: options }; // Legacy interface.
break;
case 'undefined':
options = {};
break;
}
stream.Duplex.call(this, options);
if (options.handle) {
this._handle = options.handle; // private
} else if (typeof options.fd !== 'undefined') {
this._handle = createPipe();
this._handle.open(options.fd);
this.readable = options.readable !== false;
this.writable = options.writable !== false;
} else {
// these will be set once there is a connection
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);
initSocketHandle(this);
this._pendingWrite = null;
// handle strings directly
this._writableState.decodeStrings = false;
// default to *not* allowing half open sockets
this.allowHalfOpen = options && options.allowHalfOpen || false;
// if we have a handle, then start the flow of data into the
// buffer. if not, then this will happen when we connect
if (this._handle && options.readable !== false)
this.read(0);
}
util.inherits(Socket, stream.Duplex);
// the user has called .end(), and all the bytes have been
// sent out to the other side.
// If allowHalfOpen is false, or if the readable side has
// ended already, then destroy.
// If allowHalfOpen is true, then we need to do a shutdown,
// so that only the writable side will be cleaned up.
function onSocketFinish() {
debug('onSocketFinish');
if (!this.readable || this._readableState.ended) {
debug('oSF: ended, destroy', this._readableState);
return this.destroy();
}
debug('oSF: not ended, call shutdown()');
// otherwise, just shutdown, or destroy() if not possible
if (!this._handle || !this._handle.shutdown)
return this.destroy();
var shutdownReq = this._handle.shutdown();
if (!shutdownReq)
return this._destroy(errnoException(errno, 'shutdown'));
shutdownReq.oncomplete = afterShutdown;
}
function afterShutdown(status, handle, req) {
var self = handle.owner;
debug('afterShutdown destroyed=%j', self.destroyed,
self._readableState);
// callback may come after call to destroy.
if (self.destroyed)
return;
if (self._readableState.ended) {
debug('readableState ended, destroying');
self.destroy();
} else {
self.once('_socketEnd', self.destroy);
}
2011-06-16 23:11:05 +04:00
}
// the EOF has been received, and no more bytes are coming.
// if the writable side has ended already, then clean everything
// up.
function onSocketEnd() {
// XXX Should not have to do as much crap in this function.
// ended should already be true, since this is called *after*
// the EOF errno and onread has returned null to the _read cb.
debug('onSocketEnd', this._readableState);
this._readableState.ended = true;
if (this._readableState.endEmitted) {
this.readable = false;
} else {
this.once('end', function() {
this.readable = false;
});
this.read(0);
}
if (!this.allowHalfOpen)
this.destroySoon();
}
2011-06-16 23:11:05 +04:00
2011-06-17 20:09:15 +04:00
exports.Socket = Socket;
exports.Stream = Socket; // Legacy naming.
Socket.prototype.read = function(n) {
if (n === 0)
return stream.Readable.prototype.read.call(this, n);
this.read = stream.Readable.prototype.read;
this._consuming = true;
return this.read(n);
};
2011-06-17 20:09:15 +04:00
Socket.prototype.listen = function() {
debug('socket.listen');
var self = this;
self.on('connection', arguments[0]);
listen(self, null, null, null);
};
2011-06-16 23:11:05 +04:00
Socket.prototype.setTimeout = function(msecs, callback) {
if (msecs > 0 && !isNaN(msecs) && isFinite(msecs)) {
2011-06-16 23:11:05 +04:00
timers.enroll(this, msecs);
timers.active(this);
2011-06-16 23:11:05 +04:00
if (callback) {
this.once('timeout', callback);
}
} else if (msecs === 0) {
timers.unenroll(this);
if (callback) {
this.removeListener('timeout', callback);
}
2011-06-16 23:11:05 +04:00
}
};
Socket.prototype._onTimeout = function() {
2012-02-19 03:01:35 +04:00
debug('_onTimeout');
this.emit('timeout');
};
Socket.prototype.setNoDelay = function(enable) {
// backwards compatibility: assume true when `enable` is omitted
if (this._handle && this._handle.setNoDelay)
this._handle.setNoDelay(typeof enable === 'undefined' ? true : !!enable);
2011-06-17 16:27:02 +04:00
};
2011-07-01 01:53:22 +04:00
Socket.prototype.setKeepAlive = function(setting, msecs) {
if (this._handle && this._handle.setKeepAlive)
this._handle.setKeepAlive(setting, ~~(msecs / 1000));
2011-07-01 01:53:22 +04:00
};
2011-07-16 02:06:10 +04:00
Socket.prototype.address = function() {
if (this._handle && this._handle.getsockname) {
return this._handle.getsockname();
}
return null;
2011-07-16 02:06:10 +04:00
};
2011-06-17 16:27:02 +04:00
Object.defineProperty(Socket.prototype, 'readyState', {
get: function() {
if (this._connecting) {
return 'opening';
} else if (this.readable && this.writable) {
return 'open';
} else if (this.readable && !this.writable) {
return 'readOnly';
} else if (!this.readable && this.writable) {
return 'writeOnly';
} else {
return 'closed';
}
}
});
Object.defineProperty(Socket.prototype, 'bufferSize', {
get: function() {
if (this._handle) {
return this._handle.writeQueueSize + this._writableState.length;
}
2011-06-17 16:27:02 +04:00
}
});
// Just call handle.readStart until we have enough in the buffer
Socket.prototype._read = function(n, callback) {
debug('_read');
if (this._connecting || !this._handle) {
debug('_read wait for connection');
this.once('connect', this._read.bind(this, n, callback));
return;
}
2011-06-16 23:11:05 +04:00
assert(callback === this._readableState.onread);
assert(this._readableState.reading = true);
2011-06-16 23:11:05 +04:00
if (!this._handle.reading) {
debug('Socket._read readStart');
this._handle.reading = true;
var r = this._handle.readStart();
if (r)
this._destroy(errnoException(errno, 'read'));
} else {
debug('readStart already has been called.');
}
2011-06-16 23:11:05 +04:00
};
2011-06-17 19:10:12 +04:00
Socket.prototype.end = function(data, encoding) {
stream.Duplex.prototype.end.call(this, data, encoding);
2011-06-17 19:10:12 +04:00
this.writable = false;
DTRACE_NET_STREAM_END(this);
// just in case we're waiting for an EOF.
if (this.readable && !this._readableState.endEmitted)
this.read(0);
return;
2011-06-16 23:11:05 +04:00
};
Socket.prototype.destroySoon = function() {
if (this.writable)
this.end();
2011-06-16 23:11:05 +04:00
if (this._writableState.finishing || this._writableState.finished)
this.destroy();
else
this.once('finish', this.destroy);
};
2012-03-20 11:21:38 +04:00
Socket.prototype._destroy = function(exception, cb) {
debug('destroy');
2011-06-16 23:11:05 +04:00
var self = this;
2012-03-20 11:21:38 +04:00
function fireErrorCallbacks() {
if (cb) cb(exception);
if (exception && !self.errorEmitted) {
process.nextTick(function() {
self.emit('error', exception);
});
self.errorEmitted = true;
}
};
if (this.destroyed) {
debug('already destroyed, fire error callbacks');
2012-03-20 11:21:38 +04:00
fireErrorCallbacks();
return;
}
self._connecting = false;
2011-06-16 23:11:05 +04:00
this.readable = this.writable = false;
timers.unenroll(this);
debug('close');
if (this._handle) {
if (this !== process.stderr)
debug('close handle');
this._handle.close();
2011-10-07 21:00:48 +04:00
this._handle.onread = noop;
this._handle = null;
}
2011-06-16 23:11:05 +04:00
2012-03-20 11:21:38 +04:00
fireErrorCallbacks();
2011-06-16 23:11:05 +04:00
process.nextTick(function() {
debug('emit close');
2011-06-16 23:11:05 +04:00
self.emit('close', exception ? true : false);
});
this.destroyed = true;
if (this.server) {
COUNTER_NET_SERVER_CONNECTION_CLOSE(this);
debug('has server');
this.server._connections--;
if (this.server._emitCloseIfDrained) {
this.server._emitCloseIfDrained();
}
}
2011-06-16 23:11:05 +04:00
};
2012-03-20 11:21:38 +04:00
Socket.prototype.destroy = function(exception) {
debug('destroy', exception);
2012-03-20 11:21:38 +04:00
this._destroy(exception);
2012-04-18 23:24:41 +04:00
};
2012-03-20 11:21:38 +04:00
// This function is called whenever the handle gets a
// buffer, or when there's an error reading.
2011-06-16 23:11:05 +04:00
function onread(buffer, offset, length) {
var handle = this;
var self = handle.owner;
assert(handle === self._handle, 'handle != self._handle');
2011-06-16 23:11:05 +04:00
timers.active(self);
var end = offset + length;
debug('onread', global.errno, offset, length, end);
2011-06-16 23:11:05 +04:00
2011-06-17 15:36:16 +04:00
if (buffer) {
debug('got data');
2011-06-16 23:11:05 +04:00
// read success.
// In theory (and in practice) calling readStop right now
// will prevent this from being called again until _read() gets
// called again.
// if we didn't get any bytes, that doesn't necessarily mean EOF.
// wait for the next one.
if (offset === end) {
debug('not any data, keep waiting');
return;
2011-06-16 23:11:05 +04:00
}
// if it's not enough data, we'll just call handle.readStart()
// again right away.
self.bytesRead += length;
// Optimization: emit the original buffer with end points
var ret = true;
if (self.ondata) self.ondata(buffer, offset, end);
else ret = self.push(buffer.slice(offset, end));
if (handle.reading && !ret) {
handle.reading = false;
debug('readStop');
var r = handle.readStop();
if (r)
self._destroy(errnoException(errno, 'read'));
}
2011-08-23 02:03:27 +04:00
} else if (errno == 'EOF') {
debug('EOF');
2011-06-17 15:36:16 +04:00
if (self._readableState.length === 0)
self.readable = false;
2011-06-17 19:10:12 +04:00
if (self.onend) self.once('end', self.onend);
2011-06-17 15:36:16 +04:00
// send a null to the _read cb to signal the end of data.
self.push(null);
// internal end event so that we know that the actual socket
// is no longer readable, and we can start the shutdown
// procedure. No need to wait for all the data to be consumed.
self.emit('_socketEnd');
2011-08-23 02:03:27 +04:00
} else {
debug('error', errno);
2011-08-23 02:03:27 +04:00
// Error
self._destroy(errnoException(errno, 'read'));
2011-06-17 15:36:16 +04:00
}
2011-06-16 23:11:05 +04:00
}
Socket.prototype._getpeername = function() {
if (!this._handle || !this._handle.getpeername) {
return {};
}
if (!this._peername) {
this._peername = this._handle.getpeername();
// getpeername() returns null on error
if (this._peername === null) {
return {};
}
}
return this._peername;
};
Socket.prototype.__defineGetter__('remoteAddress', function() {
return this._getpeername().address;
});
Socket.prototype.__defineGetter__('remotePort', function() {
return this._getpeername().port;
});
Socket.prototype._getsockname = function() {
if (!this._handle || !this._handle.getsockname) {
return {};
}
if (!this._sockname) {
this._sockname = this._handle.getsockname();
if (this._sockname === null) {
return {};
}
}
return this._sockname;
};
Socket.prototype.__defineGetter__('localAddress', function() {
return this._getsockname().address;
});
Socket.prototype.__defineGetter__('localPort', function() {
return this._getsockname().port;
});
Socket.prototype.write = function(chunk, encoding, cb) {
if (typeof chunk !== 'string' && !Buffer.isBuffer(chunk))
throw new TypeError('invalid data');
return stream.Duplex.prototype.write.apply(this, arguments);
};
2011-06-16 23:11:05 +04:00
Socket.prototype._write = function(dataEncoding, cb) {
// assert(Array.isArray(dataEncoding));
var data = dataEncoding[0];
var encoding = dataEncoding[1] || 'utf8';
// If we are still connecting, then buffer this for later.
// The Writable logic will buffer up any more writes while
// waiting for this one to be done.
if (this._connecting) {
this._pendingWrite = dataEncoding;
this.once('connect', function() {
this._write(dataEncoding, cb);
});
return;
}
this._pendingWrite = null;
timers.active(this);
2012-03-20 11:21:38 +04:00
if (!this._handle) {
this._destroy(new Error('This socket is closed.'), cb);
return false;
}
var enc = Buffer.isBuffer(data) ? 'buffer' : encoding;
var writeReq = createWriteReq(this._handle, data, enc);
if (!writeReq || typeof writeReq !== 'object')
return this._destroy(errnoException(errno, 'write'), cb);
2011-06-16 23:11:05 +04:00
writeReq.oncomplete = afterWrite;
this._bytesDispatched += writeReq.bytes;
// If it was entirely flushed, we can write some more right now.
// However, if more is left in the queue, then wait until that clears.
if (this._handle.writeQueueSize === 0)
cb();
else
writeReq.cb = cb;
2011-06-16 23:11:05 +04:00
};
function createWriteReq(handle, data, encoding) {
switch (encoding) {
case 'buffer':
return handle.writeBuffer(data);
case 'utf8':
case 'utf-8':
return handle.writeUtf8String(data);
case 'ascii':
return handle.writeAsciiString(data);
case 'ucs2':
case 'ucs-2':
case 'utf16le':
case 'utf-16le':
return handle.writeUcs2String(data);
default:
return handle.writeBuffer(new Buffer(data, encoding));
}
}
2011-06-16 23:11:05 +04:00
Socket.prototype.__defineGetter__('bytesWritten', function() {
var bytes = this._bytesDispatched,
state = this._writableState,
pending = this._pendingWrite;
state.buffer.forEach(function(el) {
el = el[0];
bytes += Buffer.byteLength(el[0], el[1]);
});
if (pending)
bytes += Buffer.byteLength(pending[0], pending[1]);
return bytes;
});
function afterWrite(status, handle, req) {
var self = handle.owner;
var state = self._writableState;
if (self !== process.stderr && self !== process.stdout)
debug('afterWrite', status, req);
2011-06-16 23:11:05 +04:00
2011-07-16 00:20:24 +04:00
// callback may come after call to destroy.
if (self.destroyed) {
debug('afterWrite destroyed');
2011-07-16 00:20:24 +04:00
return;
}
if (status) {
debug('write failure', errnoException(errno, 'write'));
2012-03-20 11:21:38 +04:00
self._destroy(errnoException(errno, 'write'), req.cb);
return;
}
2011-06-16 23:11:05 +04:00
timers.active(self);
if (self !== process.stderr && self !== process.stdout)
debug('afterWrite call cb');
2011-06-17 19:10:12 +04:00
if (req.cb)
req.cb.call(self);
2011-06-16 23:11:05 +04:00
}
function connect(self, address, port, addressType, localAddress) {
2011-06-24 23:32:09 +04:00
// TODO return promise from Socket.prototype.connect which
// wraps _connectReq.
assert.ok(self._connecting);
2011-06-24 23:32:09 +04:00
if (localAddress) {
var r;
if (addressType == 6) {
r = self._handle.bind6(localAddress);
} else {
r = self._handle.bind(localAddress);
}
if (r) {
self._destroy(errnoException(errno, 'bind'));
return;
}
}
2011-07-20 00:49:42 +04:00
var connectReq;
2011-07-07 00:34:04 +04:00
if (addressType == 6) {
2011-07-20 00:49:42 +04:00
connectReq = self._handle.connect6(address, port);
} else if (addressType == 4) {
connectReq = self._handle.connect(address, port);
2011-07-07 00:34:04 +04:00
} else {
2011-07-20 00:49:42 +04:00
connectReq = self._handle.connect(address, afterConnect);
2011-07-07 00:34:04 +04:00
}
2011-06-24 23:32:09 +04:00
2011-07-20 00:49:42 +04:00
if (connectReq !== null) {
2011-06-24 23:32:09 +04:00
connectReq.oncomplete = afterConnect;
} else {
2012-03-20 11:21:38 +04:00
self._destroy(errnoException(errno, 'connect'));
2011-06-24 23:32:09 +04:00
}
}
Socket.prototype.connect = function(options, cb) {
if (typeof options !== 'object') {
// Old API:
// connect(port, [host], [cb])
// connect(path, [cb]);
var args = normalizeConnectArgs(arguments);
return Socket.prototype.connect.apply(this, args);
}
2011-06-16 23:11:05 +04:00
if (this.destroyed) {
this._readableState.reading = false;
this._readableState.ended = false;
this._writableState.ended = false;
this._writableState.ending = false;
this._writableState.finished = false;
this._writableState.finishing = false;
this.destroyed = false;
this._handle = null;
}
var self = this;
var pipe = !!options.path;
2011-07-20 00:49:42 +04:00
if (!this._handle) {
this._handle = pipe ? createPipe() : createTCP();
2011-07-01 03:37:30 +04:00
initSocketHandle(this);
}
2011-07-20 00:49:42 +04:00
if (typeof cb === 'function') {
self.once('connect', cb);
}
2011-06-16 23:11:05 +04:00
timers.active(this);
self._connecting = true;
self.writable = true;
2011-07-20 00:49:42 +04:00
if (pipe) {
connect(self, options.path);
} else if (!options.host) {
debug('connect: missing host');
connect(self, '127.0.0.1', options.port, 4);
2011-07-20 00:49:42 +04:00
} else {
var host = options.host;
debug('connect: find host ' + host);
2011-06-24 23:32:09 +04:00
require('dns').lookup(host, function(err, ip, addressType) {
// It's possible we were destroyed while looking this up.
// XXX it would be great if we could cancel the promise returned by
// the look up.
if (!self._connecting) return;
2011-06-24 23:32:09 +04:00
if (err) {
// net.createConnection() creates a net.Socket object and
// immediately calls net.Socket.connect() on it (that's us).
// There are no event listeners registered yet so defer the
// error event to the next tick.
process.nextTick(function() {
self.emit('error', err);
2012-03-20 11:21:38 +04:00
self._destroy();
});
2011-06-24 23:32:09 +04:00
} else {
timers.active(self);
2011-06-16 23:11:05 +04:00
addressType = addressType || 4;
// node_net.cc handles null host names graciously but user land
// expects remoteAddress to have a meaningful value
ip = ip || (addressType === 4 ? '127.0.0.1' : '0:0:0:0:0:0:0:1');
connect(self, ip, options.port, addressType, options.localAddress);
2011-06-16 23:11:05 +04:00
}
2011-06-24 23:32:09 +04:00
});
}
return self;
2011-06-16 23:11:05 +04:00
};
Socket.prototype.ref = function() {
if (this._handle)
this._handle.ref();
};
Socket.prototype.unref = function() {
if (this._handle)
this._handle.unref();
};
2012-02-10 13:35:35 +04:00
function afterConnect(status, handle, req, readable, writable) {
var self = handle.owner;
2011-07-16 00:20:24 +04:00
// callback may come after call to destroy
if (self.destroyed) {
return;
}
assert(handle === self._handle, 'handle != self._handle');
debug('afterConnect');
2011-06-17 19:10:12 +04:00
assert.ok(self._connecting);
self._connecting = false;
2011-06-16 23:11:05 +04:00
if (status == 0) {
2012-02-10 13:35:35 +04:00
self.readable = readable;
self.writable = writable;
2011-06-16 23:11:05 +04:00
timers.active(self);
self.emit('connect');
// start the first read, or get an immediate EOF.
// this doesn't actually consume any bytes, because len=0.
if (readable)
self.read(0);
2011-06-16 23:11:05 +04:00
} else {
self._connecting = false;
2012-03-20 11:21:38 +04:00
self._destroy(errnoException(errno, 'connect'));
2011-06-16 23:11:05 +04:00
}
}
function errnoException(errorno, syscall) {
// TODO make this more compatible with ErrnoException from src/node.cc
// Once all of Node is using this function the ErrnoException from
// src/node.cc should be removed.
var e = new Error(syscall + ' ' + errorno);
e.errno = e.code = errorno;
2011-06-16 23:11:05 +04:00
e.syscall = syscall;
return e;
}
function Server(/* [ options, ] listener */) {
if (!(this instanceof Server)) return new Server(arguments[0], arguments[1]);
events.EventEmitter.call(this);
var self = this;
var options;
2011-06-16 23:11:05 +04:00
if (typeof arguments[0] == 'function') {
options = {};
self.on('connection', arguments[0]);
2011-06-16 23:11:05 +04:00
} else {
options = arguments[0] || {};
2011-06-16 23:11:05 +04:00
if (typeof arguments[1] == 'function') {
self.on('connection', arguments[1]);
2011-06-16 23:11:05 +04:00
}
}
this._connections = 0;
Object.defineProperty(this, 'connections', {
get: util.deprecate(function() {
if (self._usingSlaves) {
return null;
}
return self._connections;
}, 'connections property is deprecated. Use getConnections() method'),
set: util.deprecate(function(val) {
return (self._connections = val);
}, 'connections property is deprecated. Use getConnections() method'),
configurable: true, enumerable: true
});
2011-07-07 04:13:17 +04:00
this._handle = null;
this._usingSlaves = false;
this._slaves = [];
this.allowHalfOpen = options.allowHalfOpen || false;
2011-06-16 23:11:05 +04:00
}
util.inherits(Server, events.EventEmitter);
exports.Server = Server;
function toNumber(x) { return (x = Number(x)) >= 0 ? x : false; }
2011-06-16 23:11:05 +04:00
var createServerHandle = exports._createServerHandle =
function(address, port, addressType, fd) {
2011-10-12 13:56:29 +04:00
var r = 0;
// assign handle in listen, and clean up if bind or listen fails
2011-12-02 01:24:28 +04:00
var handle;
if (typeof fd === 'number' && fd >= 0) {
var tty_wrap = process.binding('tty_wrap');
var type = tty_wrap.guessHandleType(fd);
switch (type) {
case 'PIPE':
debug('listen pipe fd=' + fd);
// create a PipeWrap
handle = createPipe();
handle.open(fd);
handle.readable = true;
handle.writable = true;
break;
default:
// Not a fd we can listen on. This will trigger an error.
debug('listen invalid fd=' + fd + ' type=' + type);
global.errno = 'EINVAL'; // hack, callers expect that errno is set
handle = null;
break;
}
return handle;
} else if (port == -1 && addressType == -1) {
2011-12-02 01:24:28 +04:00
handle = createPipe();
if (process.platform === 'win32') {
var instances = parseInt(process.env.NODE_PENDING_PIPE_INSTANCES);
if (!isNaN(instances)) {
handle.setPendingInstances(instances);
}
}
} else {
handle = createTCP();
}
2011-10-12 13:56:29 +04:00
if (address || port) {
debug('bind to ' + address);
if (addressType == 6) {
r = handle.bind6(address, port);
} else {
r = handle.bind(address, port);
}
}
if (r) {
handle.close();
handle = null;
}
return handle;
};
Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
debug('listen2', address, port, addressType, backlog);
2011-10-12 13:56:29 +04:00
var self = this;
var r = 0;
// If there is not yet a handle, we need to create one and bind.
// In the case of a server sent via IPC, we don't need to do this.
if (!self._handle) {
debug('_listen2: create a handle');
self._handle = createServerHandle(address, port, addressType, fd);
if (!self._handle) {
var error = errnoException(errno, 'listen');
process.nextTick(function() {
self.emit('error', error);
});
return;
}
} else {
debug('_listen2: have a handle already');
}
self._handle.onconnection = onconnection;
self._handle.owner = self;
// Use a backlog of 512 entries. We pass 511 to the listen() call because
// the kernel does: backlogsize = roundup_pow_of_two(backlogsize + 1);
// which will thus give us a backlog of 512 entries.
r = self._handle.listen(backlog || 511);
2011-06-24 23:32:09 +04:00
if (r) {
var ex = errnoException(errno, 'listen');
2011-07-07 04:13:17 +04:00
self._handle.close();
self._handle = null;
2011-06-29 01:32:01 +04:00
process.nextTick(function() {
self.emit('error', ex);
2011-06-29 01:32:01 +04:00
});
return;
2011-06-24 23:32:09 +04:00
}
// generate connection key, this should be unique to the connection
this._connectionKey = addressType + ':' + address + ':' + port;
process.nextTick(function() {
self.emit('listening');
});
2012-02-19 03:01:35 +04:00
};
2011-06-24 23:32:09 +04:00
function listen(self, address, port, addressType, backlog, fd) {
if (!cluster) cluster = require('cluster');
if (cluster.isWorker) {
cluster._getServer(self, address, port, addressType, fd, function(handle) {
self._handle = handle;
self._listen2(address, port, addressType, backlog, fd);
});
} else {
self._listen2(address, port, addressType, backlog, fd);
}
}
2011-10-12 13:56:29 +04:00
Server.prototype.listen = function() {
2011-06-16 23:11:05 +04:00
var self = this;
var lastArg = arguments[arguments.length - 1];
if (typeof lastArg == 'function') {
self.once('listening', lastArg);
2011-06-16 23:11:05 +04:00
}
var port = toNumber(arguments[0]);
// The third optional argument is the backlog size.
// When the ip is omitted it can be the second argument.
var backlog = toNumber(arguments[1]) || toNumber(arguments[2]);
2011-06-16 23:11:05 +04:00
var TCP = process.binding('tcp_wrap').TCP;
2011-06-16 23:11:05 +04:00
if (arguments.length == 0 || typeof arguments[0] == 'function') {
// Bind to a random port.
listen(self, '0.0.0.0', 0, null, backlog);
2011-07-20 00:49:42 +04:00
} else if (arguments[0] && typeof arguments[0] === 'object') {
var h = arguments[0];
if (h._handle) {
h = h._handle;
} else if (h.handle) {
h = h.handle;
}
if (h instanceof TCP) {
self._handle = h;
listen(self, null, -1, -1, backlog);
2012-08-02 15:37:02 +04:00
} else if (typeof h.fd === 'number' && h.fd >= 0) {
listen(self, null, null, null, backlog, h.fd);
} else {
throw new Error('Invalid listen argument: ' + h);
}
2011-07-20 00:49:42 +04:00
} else if (isPipeName(arguments[0])) {
// UNIX socket or Windows pipe.
var pipeName = self._pipeName = arguments[0];
listen(self, pipeName, -1, -1, backlog);
2011-06-24 23:32:09 +04:00
} else if (typeof arguments[1] == 'undefined' ||
typeof arguments[1] == 'function' ||
typeof arguments[1] == 'number') {
2011-06-24 23:32:09 +04:00
// The first argument is the port, no IP given.
listen(self, '0.0.0.0', port, 4, backlog);
2011-06-24 23:32:09 +04:00
2011-06-16 23:11:05 +04:00
} else {
// The first argument is the port, the second an IP.
2011-06-16 23:11:05 +04:00
require('dns').lookup(arguments[1], function(err, ip, addressType) {
if (err) {
self.emit('error', err);
} else {
listen(self, ip || '0.0.0.0', port, ip ? addressType : 4, backlog);
2011-06-16 23:11:05 +04:00
}
});
}
return self;
2011-06-16 23:11:05 +04:00
};
2011-07-16 02:06:10 +04:00
Server.prototype.address = function() {
if (this._handle && this._handle.getsockname) {
return this._handle.getsockname();
} else if (this._pipeName) {
return this._pipeName;
} else {
return null;
}
2011-07-16 02:06:10 +04:00
};
2011-06-16 23:11:05 +04:00
function onconnection(clientHandle) {
var handle = this;
var self = handle.owner;
2011-06-16 23:11:05 +04:00
debug('onconnection');
if (!clientHandle) {
self.emit('error', errnoException(errno, 'accept'));
return;
}
if (self.maxConnections && self._connections >= self.maxConnections) {
2011-07-15 00:18:28 +04:00
clientHandle.close();
return;
}
2011-06-17 19:10:12 +04:00
var socket = new Socket({
handle: clientHandle,
allowHalfOpen: self.allowHalfOpen
});
2011-06-17 16:27:02 +04:00
socket.readable = socket.writable = true;
2011-06-16 23:11:05 +04:00
self._connections++;
2011-06-17 16:27:02 +04:00
socket.server = self;
2011-06-16 23:11:05 +04:00
DTRACE_NET_SERVER_CONNECTION(socket);
COUNTER_NET_SERVER_CONNECTION(socket);
2011-06-16 23:11:05 +04:00
self.emit('connection', socket);
socket.emit('connect');
2011-06-16 23:11:05 +04:00
}
2011-06-17 19:10:12 +04:00
Server.prototype.getConnections = function(cb) {
function end(err, connections) {
process.nextTick(function() {
cb(err, connections);
});
}
if (!this._usingSlaves) {
return end(null, this._connections);
}
// Poll slaves
var left = this._slaves.length,
total = this._connections;
function oncount(err, count) {
if (err) {
left = -1;
return end(err);
}
total += count;
if (--left === 0) return end(null, total);
}
this._slaves.forEach(function(slave) {
slave.getConnections(oncount);
});
};
Server.prototype.close = function(cb) {
function onSlaveClose() {
if (--left !== 0) return;
self._connections = 0;
self._emitCloseIfDrained();
}
if (!this._handle) {
// Throw error. Follows net_legacy behaviour.
throw new Error('Not running');
2011-07-07 04:13:17 +04:00
}
if (cb) {
this.once('close', cb);
}
this._handle.close();
this._handle = null;
if (this._usingSlaves) {
var self = this,
left = this._slaves.length;
// Increment connections to be sure that, even if all sockets will be closed
// during polling of slaves, `close` event will be emitted only once.
this._connections++;
// Poll slaves
this._slaves.forEach(function(slave) {
slave.close(onSlaveClose);
});
} else {
this._emitCloseIfDrained();
}
return this;
};
Server.prototype._emitCloseIfDrained = function() {
debug('SERVER _emitCloseIfDrained');
var self = this;
if (self._handle || self._connections) {
debug('SERVER handle? %j connections? %d',
!!self._handle, self._connections);
return;
}
process.nextTick(function() {
debug('SERVER: emit close');
self.emit('close');
});
2011-06-17 19:10:12 +04:00
};
Server.prototype.listenFD = util.deprecate(function(fd, type) {
return this.listen({ fd: fd });
}, 'listenFD is deprecated. Use listen({fd: <number>}).');
Server.prototype._setupSlave = function(socketList) {
this._usingSlaves = true;
this._slaves.push(socketList);
};
Server.prototype.ref = function() {
if (this._handle)
this._handle.ref();
};
Server.prototype.unref = function() {
if (this._handle)
this._handle.unref();
};
// TODO: isIP should be moved to the DNS code. Putting it here now because
// this is what the legacy system did.
exports.isIP = cares.isIP;
exports.isIPv4 = function(input) {
2011-07-05 02:03:21 +04:00
return exports.isIP(input) === 4;
};
exports.isIPv6 = function(input) {
2011-07-05 02:03:21 +04:00
return exports.isIP(input) === 6;
};
if (process.platform === 'win32') {
var simultaneousAccepts;
exports._setSimultaneousAccepts = function(handle) {
if (typeof handle === 'undefined') {
return;
}
if (typeof simultaneousAccepts === 'undefined') {
simultaneousAccepts = (process.env.NODE_MANY_ACCEPTS &&
2012-02-19 03:01:35 +04:00
process.env.NODE_MANY_ACCEPTS !== '0');
}
2012-02-19 03:01:35 +04:00
if (handle._simultaneousAccepts !== simultaneousAccepts) {
handle.setSimultaneousAccepts(simultaneousAccepts);
handle._simultaneousAccepts = simultaneousAccepts;
}
2012-02-19 03:01:35 +04:00
};
} else {
2012-02-19 03:01:35 +04:00
exports._setSimultaneousAccepts = function(handle) {};
}