1 // Copyright Joyent, Inc. and other Node contributors.
3 // Permission is hereby granted, free of charge, to any person obtaining a
4 // copy of this software and associated documentation files (the
5 // "Software"), to deal in the Software without restriction, including
6 // without limitation the rights to use, copy, modify, merge, publish,
7 // distribute, sublicense, and/or sell copies of the Software, and to permit
8 // persons to whom the Software is furnished to do so, subject to the
9 // following conditions:
11 // The above copyright notice and this permission notice shall be included
12 // in all copies or substantial portions of the Software.
14 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
15 // OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
16 // MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
17 // NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
18 // DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
19 // OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
20 // USE OR OTHER DEALINGS IN THE SOFTWARE.
22 var events = require('events');
23 var Stream = require('stream');
24 var timers = require('timers');
25 var util = require('util');
26 var assert = require('assert');
31 // constructor for lazy loading
32 function createPipe() {
33 var Pipe = process.binding('pipe_wrap').Pipe;
37 // constructor for lazy loading
38 function createTCP() {
39 var TCP = process.binding('tcp_wrap').TCP;
44 /* Bit flags for socket._flags */
45 var FLAG_GOT_EOF = 1 << 0;
46 var FLAG_SHUTDOWN = 1 << 1;
47 var FLAG_DESTROY_SOON = 1 << 2;
48 var FLAG_SHUTDOWN_QUEUED = 1 << 3;
52 if (process.env.NODE_DEBUG && /net/.test(process.env.NODE_DEBUG)) {
53 debug = function(x) { console.error('NET:', x); };
55 debug = function() { };
59 function isPipeName(s) {
60 return typeof s === 'string' && toNumber(s) === false;
64 exports.createServer = function() {
65 return new Server(arguments[0], arguments[1]);
71 // var s = net.connect({port: 80, host: 'google.com'}, function() {
75 // There are various forms:
77 // connect(options, [cb])
78 // connect(port, [host], [cb])
79 // connect(path, [cb]);
81 exports.connect = exports.createConnection = function() {
82 var args = normalizeConnectArgs(arguments);
83 var s = new Socket(args[0]);
84 return Socket.prototype.connect.apply(s, args);
87 // Returns an array [options] or [options, cb]
88 // It is the same as the argument of Socket.prototype.connect().
89 function normalizeConnectArgs(args) {
92 if (typeof args[0] === 'object') {
93 // connect(options, [cb])
95 } else if (isPipeName(args[0])) {
96 // connect(path, [cb]);
97 options.path = args[0];
99 // connect(port, [host], [cb])
100 options.port = args[0];
101 if (typeof args[1] === 'string') {
102 options.host = args[1];
106 var cb = args[args.length - 1];
107 return (typeof cb === 'function') ? [options, cb] : [options];
111 /* called when creating new Socket, or when re-using a closed Socket */
112 function initSocketHandle(self) {
113 self._pendingWriteReqs = 0;
116 self._connectQueueSize = 0;
117 self.destroyed = false;
118 self.errorEmitted = false;
120 self._bytesDispatched = 0;
122 // Handle creation may be deferred to bind() or connect() time.
124 self._handle.owner = self;
125 self._handle.onread = onread;
129 function Socket(options) {
130 if (!(this instanceof Socket)) return new Socket(options);
134 if (typeof options == 'number') {
137 this._handle = createPipe();
138 this._handle.open(fd);
139 this.readable = this.writable = true;
140 initSocketHandle(this);
143 this._handle = options && options.handle;
144 initSocketHandle(this);
145 this.allowHalfOpen = options && options.allowHalfOpen;
148 util.inherits(Socket, Stream);
151 exports.Socket = Socket;
152 exports.Stream = Socket; // Legacy naming.
155 Socket.prototype.listen = function() {
157 self.on('connection', arguments[0]);
158 listen(self, null, null, null);
162 Socket.prototype.setTimeout = function(msecs, callback) {
164 timers.enroll(this, msecs);
167 this.once('timeout', callback);
169 } else if (msecs === 0) {
170 timers.unenroll(this);
172 this.removeListener('timeout', callback);
178 Socket.prototype._onTimeout = function() {
180 this.emit('timeout');
184 Socket.prototype.setNoDelay = function(enable) {
185 // backwards compatibility: assume true when `enable` is omitted
186 if (this._handle && this._handle.setNoDelay)
187 this._handle.setNoDelay(typeof enable === 'undefined' ? true : !!enable);
191 Socket.prototype.setKeepAlive = function(setting, msecs) {
192 if (this._handle && this._handle.setKeepAlive)
193 this._handle.setKeepAlive(setting, ~~(msecs / 1000));
197 Socket.prototype.address = function() {
198 if (this._handle && this._handle.getsockname) {
199 return this._handle.getsockname();
205 Object.defineProperty(Socket.prototype, 'readyState', {
207 if (this._connecting) {
209 } else if (this.readable && this.writable) {
211 } else if (this.readable && !this.writable) {
213 } else if (!this.readable && this.writable) {
222 Object.defineProperty(Socket.prototype, 'bufferSize', {
225 return this._handle.writeQueueSize + this._connectQueueSize;
231 Socket.prototype.pause = function() {
233 if (this._connecting) {
234 // will actually pause once the handle is established.
238 this._handle.readStop();
243 Socket.prototype.resume = function() {
244 this._paused = false;
245 if (this._connecting) {
246 // will actually resume once the handle is established.
250 this._handle.readStart();
255 Socket.prototype.end = function(data, encoding) {
256 if (this._connecting && ((this._flags & FLAG_SHUTDOWN_QUEUED) == 0)) {
257 // still connecting, add data to buffer
258 if (data) this.write(data, encoding);
259 this.writable = false;
260 this._flags |= FLAG_SHUTDOWN_QUEUED;
263 if (!this.writable) return;
264 this.writable = false;
266 if (data) this.write(data, encoding);
267 DTRACE_NET_STREAM_END(this);
269 if (!this.readable) {
272 this._flags |= FLAG_SHUTDOWN;
273 var shutdownReq = this._handle.shutdown();
276 this._destroy(errnoException(errno, 'shutdown'));
280 shutdownReq.oncomplete = afterShutdown;
287 function afterShutdown(status, handle, req) {
288 var self = handle.owner;
290 assert.ok(self._flags & FLAG_SHUTDOWN);
291 assert.ok(!self.writable);
293 // callback may come after call to destroy.
294 if (self.destroyed) {
298 if (self._flags & FLAG_GOT_EOF || !self.readable) {
305 Socket.prototype.destroySoon = function() {
306 this.writable = false;
307 this._flags |= FLAG_DESTROY_SOON;
309 if (this._pendingWriteReqs == 0) {
315 Socket.prototype._connectQueueCleanUp = function(exception) {
316 this._connecting = false;
317 this._connectQueueSize = 0;
318 this._connectQueue = null;
322 Socket.prototype._destroy = function(exception, cb) {
325 function fireErrorCallbacks() {
326 if (cb) cb(exception);
327 if (exception && !self.errorEmitted) {
328 process.nextTick(function() {
329 self.emit('error', exception);
331 self.errorEmitted = true;
335 if (this.destroyed) {
336 fireErrorCallbacks();
340 self._connectQueueCleanUp();
344 this.readable = this.writable = false;
346 timers.unenroll(this);
350 this._handle.close();
351 this._handle.onread = noop;
355 fireErrorCallbacks();
357 process.nextTick(function() {
358 self.emit('close', exception ? true : false);
361 this.destroyed = true;
364 this.server._connections--;
365 if (this.server._emitCloseIfDrained) {
366 this.server._emitCloseIfDrained();
372 Socket.prototype.destroy = function(exception) {
373 this._destroy(exception);
377 function onread(buffer, offset, length) {
379 var self = handle.owner;
380 assert.equal(handle, self._handle);
384 var end = offset + length;
387 // Emit 'data' event.
391 var string = self._decoder.write(buffer.slice(offset, end));
392 if (string.length) self.emit('data', string);
394 // Emit a slice. Attempt to avoid slicing the buffer if no one is
395 // listening for 'data'.
396 if (self._events && self._events['data']) {
397 self.emit('data', buffer.slice(offset, end));
401 self.bytesRead += length;
403 // Optimization: emit the original buffer with end points
404 if (self.ondata) self.ondata(buffer, offset, end);
406 } else if (errno == 'EOF') {
408 self.readable = false;
410 assert.ok(!(self._flags & FLAG_GOT_EOF));
411 self._flags |= FLAG_GOT_EOF;
413 // We call destroy() before end(). 'close' not emitted until nextTick so
414 // the 'end' event will come first as required.
415 if (!self.writable) self._destroy();
417 if (!self.allowHalfOpen) self.end();
418 if (self._events && self._events['end']) self.emit('end');
419 if (self.onend) self.onend();
422 if (errno == 'ECONNRESET') {
425 self._destroy(errnoException(errno, 'read'));
431 Socket.prototype.setEncoding = function(encoding) {
432 var StringDecoder = require('string_decoder').StringDecoder; // lazy load
433 this._decoder = new StringDecoder(encoding);
437 Socket.prototype._getpeername = function() {
438 if (!this._handle || !this._handle.getpeername) {
441 if (!this._peername) {
442 this._peername = this._handle.getpeername();
443 // getpeername() returns null on error
444 if (this._peername === null) {
448 return this._peername;
452 Socket.prototype.__defineGetter__('remoteAddress', function() {
453 return this._getpeername().address;
457 Socket.prototype.__defineGetter__('remotePort', function() {
458 return this._getpeername().port;
463 * Arguments data, [encoding], [cb]
465 Socket.prototype.write = function(data, arg1, arg2) {
470 if (typeof arg1 === 'string') {
473 } else if (typeof arg1 === 'function') {
476 throw new Error('bad arg');
480 if (typeof data === 'string') {
481 encoding = (encoding || 'utf8').toLowerCase();
490 // This encoding can be handled in the binding layer.
494 data = new Buffer(data, encoding);
496 } else if (!Buffer.isBuffer(data)) {
497 throw new TypeError('First argument must be a buffer or a string.');
500 // If we are still connecting, then buffer this for later.
501 if (this._connecting) {
502 this._connectQueueSize += data.length;
503 if (this._connectQueue) {
504 this._connectQueue.push([data, encoding, cb]);
506 this._connectQueue = [[data, encoding, cb]];
511 return this._write(data, encoding, cb);
515 Socket.prototype._write = function(data, encoding, cb) {
519 this._destroy(new Error('This socket is closed.'), cb);
525 if (Buffer.isBuffer(data)) {
526 writeReq = this._handle.writeBuffer(data);
532 writeReq = this._handle.writeUtf8String(data);
536 writeReq = this._handle.writeAsciiString(data);
543 writeReq = this._handle.writeUcs2String(data);
551 if (!writeReq || typeof writeReq !== 'object') {
552 this._destroy(errnoException(errno, 'write'), cb);
556 writeReq.oncomplete = afterWrite;
559 this._pendingWriteReqs++;
560 this._bytesDispatched += writeReq.bytes;
562 return this._handle.writeQueueSize == 0;
566 Socket.prototype.__defineGetter__('bytesWritten', function() {
567 var bytes = this._bytesDispatched,
568 connectQueue = this._connectQueue;
571 connectQueue.forEach(function(el) {
573 if (Buffer.isBuffer(data)) {
574 bytes += data.length;
576 bytes += Buffer.byteLength(data, el[1]);
585 function afterWrite(status, handle, req) {
586 var self = handle.owner;
588 // callback may come after call to destroy.
589 if (self.destroyed) {
594 self._destroy(errnoException(errno, 'write'), req.cb);
600 self._pendingWriteReqs--;
602 if (self._pendingWriteReqs == 0) {
606 if (req.cb) req.cb();
608 if (self._pendingWriteReqs == 0 && self._flags & FLAG_DESTROY_SOON) {
614 function connect(self, address, port, addressType, localAddress) {
615 // TODO return promise from Socket.prototype.connect which
616 // wraps _connectReq.
618 assert.ok(self._connecting);
622 if (addressType == 6) {
623 r = self._handle.bind6(localAddress);
625 r = self._handle.bind(localAddress);
629 self._destroy(errnoException(errno, 'bind'));
635 if (addressType == 6) {
636 connectReq = self._handle.connect6(address, port);
637 } else if (addressType == 4) {
638 connectReq = self._handle.connect(address, port);
640 connectReq = self._handle.connect(address, afterConnect);
643 if (connectReq !== null) {
644 connectReq.oncomplete = afterConnect;
646 self._destroy(errnoException(errno, 'connect'));
651 Socket.prototype.connect = function(options, cb) {
652 if (typeof options !== 'object') {
654 // connect(port, [host], [cb])
655 // connect(path, [cb]);
656 var args = normalizeConnectArgs(arguments);
657 return Socket.prototype.connect.apply(this, args);
661 var pipe = !!options.path;
663 if (this.destroyed || !this._handle) {
664 this._handle = pipe ? createPipe() : createTCP();
665 initSocketHandle(this);
668 if (typeof cb === 'function') {
669 self.on('connect', cb);
674 self._connecting = true;
675 self.writable = true;
678 connect(self, options.path);
680 } else if (!options.host) {
681 debug('connect: missing host');
682 connect(self, '127.0.0.1', options.port, 4);
685 var host = options.host;
686 debug('connect: find host ' + host);
687 require('dns').lookup(host, function(err, ip, addressType) {
688 // It's possible we were destroyed while looking this up.
689 // XXX it would be great if we could cancel the promise returned by
691 if (!self._connecting) return;
694 // net.createConnection() creates a net.Socket object and
695 // immediately calls net.Socket.connect() on it (that's us).
696 // There are no event listeners registered yet so defer the
697 // error event to the next tick.
698 process.nextTick(function() {
699 self.emit('error', err);
705 addressType = addressType || 4;
707 // node_net.cc handles null host names graciously but user land
708 // expects remoteAddress to have a meaningful value
709 ip = ip || (addressType === 4 ? '127.0.0.1' : '0:0:0:0:0:0:0:1');
711 connect(self, ip, options.port, addressType, options.localAddress);
719 function afterConnect(status, handle, req, readable, writable) {
720 var self = handle.owner;
722 // callback may come after call to destroy
723 if (self.destroyed) {
727 assert.equal(handle, self._handle);
729 debug('afterConnect');
731 assert.ok(self._connecting);
732 self._connecting = false;
734 // now that we're connected, process any pending pause state.
736 self._paused = false;
741 self.readable = readable;
742 self.writable = writable;
749 if (self._connectQueue) {
750 debug('Drain the connect queue');
751 var connectQueue = self._connectQueue;
752 for (var i = 0; i < connectQueue.length; i++) {
753 self._write.apply(self, connectQueue[i]);
755 self._connectQueueCleanUp();
758 self.emit('connect');
760 if (self._flags & FLAG_SHUTDOWN_QUEUED) {
761 // end called before connected - call end now with no data
762 self._flags &= ~FLAG_SHUTDOWN_QUEUED;
766 self._connectQueueCleanUp();
767 self._destroy(errnoException(errno, 'connect'));
772 function errnoException(errorno, syscall) {
773 // TODO make this more compatible with ErrnoException from src/node.cc
774 // Once all of Node is using this function the ErrnoException from
775 // src/node.cc should be removed.
776 var e = new Error(syscall + ' ' + errorno);
777 e.errno = e.code = errorno;
785 function Server(/* [ options, ] listener */) {
786 if (!(this instanceof Server)) return new Server(arguments[0], arguments[1]);
787 events.EventEmitter.call(this);
793 if (typeof arguments[0] == 'function') {
795 self.on('connection', arguments[0]);
797 options = arguments[0] || {};
799 if (typeof arguments[1] == 'function') {
800 self.on('connection', arguments[1]);
804 this._connections = 0;
806 // when server is using slaves .connections is not reliable
807 // so null will be return if thats the case
808 Object.defineProperty(this, 'connections', {
810 if (self._usingSlaves) {
813 return self._connections;
816 return (self._connections = val);
818 configurable: true, enumerable: true
821 this.allowHalfOpen = options.allowHalfOpen || false;
825 util.inherits(Server, events.EventEmitter);
826 exports.Server = Server;
829 function toNumber(x) { return (x = Number(x)) >= 0 ? x : false; }
832 var createServerHandle = exports._createServerHandle =
833 function(address, port, addressType, fd) {
835 // assign handle in listen, and clean up if bind or listen fails
838 if (typeof fd === 'number' && fd >= 0) {
839 var tty_wrap = process.binding('tty_wrap');
840 var type = tty_wrap.guessHandleType(fd);
843 debug('listen pipe fd=' + fd);
845 handle = createPipe();
847 handle.readable = true;
848 handle.writable = true;
852 // Not a fd we can listen on. This will trigger an error.
853 debug('listen invalid fd=' + fd + ' type=' + type);
854 global.errno = 'EINVAL'; // hack, callers expect that errno is set
860 } else if (port == -1 && addressType == -1) {
861 handle = createPipe();
862 if (process.platform === 'win32') {
863 var instances = parseInt(process.env.NODE_PENDING_PIPE_INSTANCES);
864 if (!isNaN(instances)) {
865 handle.setPendingInstances(instances);
869 handle = createTCP();
872 if (address || port) {
873 debug('bind to ' + address);
874 if (addressType == 6) {
875 r = handle.bind6(address, port);
877 r = handle.bind(address, port);
890 Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
894 // If there is not yet a handle, we need to create one and bind.
895 // In the case of a server sent via IPC, we don't need to do this.
897 self._handle = createServerHandle(address, port, addressType, fd);
899 process.nextTick(function() {
900 self.emit('error', errnoException(errno, 'listen'));
906 self._handle.onconnection = onconnection;
907 self._handle.owner = self;
909 // Use a backlog of 512 entries. We pass 511 to the listen() call because
910 // the kernel does: backlogsize = roundup_pow_of_two(backlogsize + 1);
911 // which will thus give us a backlog of 512 entries.
912 r = self._handle.listen(backlog || 511);
915 var ex = errnoException(errno, 'listen');
916 self._handle.close();
918 process.nextTick(function() {
919 self.emit('error', ex);
924 // generate connection key, this should be unique to the connection
925 this._connectionKey = addressType + ':' + address + ':' + port;
927 process.nextTick(function() {
928 self.emit('listening');
933 function listen(self, address, port, addressType, backlog, fd) {
934 if (!cluster) cluster = require('cluster');
936 if (cluster.isWorker) {
937 cluster._getServer(self, address, port, addressType, fd, function(handle) {
938 self._handle = handle;
939 self._listen2(address, port, addressType, backlog, fd);
942 self._listen2(address, port, addressType, backlog, fd);
947 Server.prototype.listen = function() {
950 var lastArg = arguments[arguments.length - 1];
951 if (typeof lastArg == 'function') {
952 self.once('listening', lastArg);
955 var port = toNumber(arguments[0]);
957 // The third optional argument is the backlog size.
958 // When the ip is omitted it can be the second argument.
959 var backlog = toNumber(arguments[1]) || toNumber(arguments[2]);
961 var TCP = process.binding('tcp_wrap').TCP;
963 if (arguments.length == 0 || typeof arguments[0] == 'function') {
964 // Don't bind(). OS will assign a port with INADDR_ANY.
965 // The port can be found with server.address()
966 listen(self, null, null, backlog);
968 } else if (arguments[0] && typeof arguments[0] === 'object') {
969 var h = arguments[0];
972 } else if (h.handle) {
975 if (h instanceof TCP) {
977 listen(self, null, -1, -1, backlog);
978 } else if (h.fd && typeof h.fd === 'number' && h.fd >= 0) {
979 listen(self, null, null, null, backlog, h.fd);
981 throw new Error('Invalid listen argument: ' + h);
983 } else if (isPipeName(arguments[0])) {
984 // UNIX socket or Windows pipe.
985 var pipeName = self._pipeName = arguments[0];
986 listen(self, pipeName, -1, -1, backlog);
988 } else if (typeof arguments[1] == 'undefined' ||
989 typeof arguments[1] == 'function' ||
990 typeof arguments[1] == 'number') {
991 // The first argument is the port, no IP given.
992 listen(self, '0.0.0.0', port, 4, backlog);
995 // The first argument is the port, the second an IP.
996 require('dns').lookup(arguments[1], function(err, ip, addressType) {
998 self.emit('error', err);
1000 listen(self, ip || '0.0.0.0', port, ip ? addressType : 4, backlog);
1007 Server.prototype.address = function() {
1008 if (this._handle && this._handle.getsockname) {
1009 return this._handle.getsockname();
1010 } else if (this._pipeName) {
1011 return this._pipeName;
1017 function onconnection(clientHandle) {
1019 var self = handle.owner;
1021 debug('onconnection');
1023 if (!clientHandle) {
1024 self.emit('error', errnoException(errno, 'accept'));
1028 if (self.maxConnections && self._connections >= self.maxConnections) {
1029 clientHandle.close();
1033 var socket = new Socket({
1034 handle: clientHandle,
1035 allowHalfOpen: self.allowHalfOpen
1037 socket.readable = socket.writable = true;
1041 self._connections++;
1042 socket.server = self;
1044 DTRACE_NET_SERVER_CONNECTION(socket);
1045 self.emit('connection', socket);
1046 socket.emit('connect');
1050 Server.prototype.close = function(cb) {
1051 if (!this._handle) {
1052 // Throw error. Follows net_legacy behaviour.
1053 throw new Error('Not running');
1057 this.once('close', cb);
1059 this._handle.close();
1060 this._handle = null;
1061 this._emitCloseIfDrained();
1063 // fetch new socket lists
1064 if (this._usingSlaves) {
1065 this._slaves.forEach(function(socketList) {
1066 if (socketList.list.length === 0) return;
1067 socketList.update();
1074 Server.prototype._emitCloseIfDrained = function() {
1077 if (self._handle || self._connections) return;
1079 process.nextTick(function() {
1085 Server.prototype.listenFD = util.deprecate(function(fd, type) {
1086 return this.listen({ fd: fd });
1087 }, 'listenFD is deprecated. Use listen({fd: <number>}).');
1089 // when sending a socket using fork IPC this function is executed
1090 Server.prototype._setupSlave = function(socketList) {
1091 if (!this._usingSlaves) {
1092 this._usingSlaves = true;
1095 this._slaves.push(socketList);
1099 // TODO: isIP should be moved to the DNS code. Putting it here now because
1100 // this is what the legacy system did.
1101 // NOTE: This does not accept IPv6 with an IPv4 dotted address at the end,
1102 // and it does not detect more than one double : in a string.
1103 exports.isIP = function(input) {
1106 } else if (/^(\d?\d?\d)\.(\d?\d?\d)\.(\d?\d?\d)\.(\d?\d?\d)$/.test(input)) {
1107 var parts = input.split('.');
1108 for (var i = 0; i < parts.length; i++) {
1109 var part = parseInt(parts[i]);
1110 if (part < 0 || 255 < part) {
1115 } else if (/^::|^::1|^([a-fA-F0-9]{1,4}::?){1,7}([a-fA-F0-9]{1,4})$/.test(
1124 exports.isIPv4 = function(input) {
1125 return exports.isIP(input) === 4;
1129 exports.isIPv6 = function(input) {
1130 return exports.isIP(input) === 6;
1134 if (process.platform === 'win32') {
1135 var simultaneousAccepts;
1137 exports._setSimultaneousAccepts = function(handle) {
1138 if (typeof handle === 'undefined') {
1142 if (typeof simultaneousAccepts === 'undefined') {
1143 simultaneousAccepts = (process.env.NODE_MANY_ACCEPTS &&
1144 process.env.NODE_MANY_ACCEPTS !== '0');
1147 if (handle._simultaneousAccepts !== simultaneousAccepts) {
1148 handle.setSimultaneousAccepts(simultaneousAccepts);
1149 handle._simultaneousAccepts = simultaneousAccepts;
1153 exports._setSimultaneousAccepts = function(handle) {};