lib/net.js

   1 // Copyright Joyent, Inc. and other Node contributors.
   2 //
   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:
  10 //
  11 // The above copyright notice and this permission notice shall be included
  12 // in all copies or substantial portions of the Software.
  13 //
  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.
  21
  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');
  27 var cares = process.binding('cares_wrap');
  28 var uv = process.binding('uv');
  29
  30 var cluster;
  31 var errnoException = util._errnoException;
  32
  33 function noop() {}
  34
  35 // constructor for lazy loading
  36 function createPipe() {
  37   var Pipe = process.binding('pipe_wrap').Pipe;
  38   return new Pipe();
  39 }
  40
  41 // constructor for lazy loading
  42 function createTCP() {
  43   var TCP = process.binding('tcp_wrap').TCP;
  44   return new TCP();
  45 }
  46
  47
  48 function createHandle(fd) {
  49   var tty = process.binding('tty_wrap');
  50   var type = tty.guessHandleType(fd);
  51   if (type === 'PIPE') return createPipe();
  52   if (type === 'TCP') return createTCP();
  53   throw new TypeError('Unsupported fd type: ' + type);
  54 }
  55
  56
  57 var debug = util.debuglog('net');
  58
  59 function isPipeName(s) {
  60   return util.isString(s) && toNumber(s) === false;
  61 }
  62
  63
  64 exports.createServer = function() {
  65   return new Server(arguments[0], arguments[1]);
  66 };
  67
  68
  69 // Target API:
  70 //
  71 // var s = net.connect({port: 80, host: 'google.com'}, function() {
  72 //   ...
  73 // });
  74 //
  75 // There are various forms:
  76 //
  77 // connect(options, [cb])
  78 // connect(port, [host], [cb])
  79 // connect(path, [cb]);
  80 //
  81 exports.connect = exports.createConnection = function() {
  82   var args = normalizeConnectArgs(arguments);
  83   debug('createConnection', args);
  84   var s = new Socket(args[0]);
  85   return Socket.prototype.connect.apply(s, args);
  86 };
  87
  88 // Returns an array [options] or [options, cb]
  89 // It is the same as the argument of Socket.prototype.connect().
  90 function normalizeConnectArgs(args) {
  91   var options = {};
  92
  93   if (util.isObject(args[0])) {
  94     // connect(options, [cb])
  95     options = args[0];
  96   } else if (isPipeName(args[0])) {
  97     // connect(path, [cb]);
  98     options.path = args[0];
  99   } else {
 100     // connect(port, [host], [cb])
 101     options.port = args[0];
 102     if (util.isString(args[1])) {
 103       options.host = args[1];
 104     }
 105   }
 106
 107   var cb = args[args.length - 1];
 108   return util.isFunction(cb) ? [options, cb] : [options];
 109 }
 110 exports._normalizeConnectArgs = normalizeConnectArgs;
 111
 112
 113 // called when creating new Socket, or when re-using a closed Socket
 114 function initSocketHandle(self) {
 115   self.destroyed = false;
 116   self.errorEmitted = false;
 117   self.bytesRead = 0;
 118   self._bytesDispatched = 0;
 119
 120   // Handle creation may be deferred to bind() or connect() time.
 121   if (self._handle) {
 122     self._handle.owner = self;
 123     self._handle.onread = onread;
 124
 125     // If handle doesn't support writev - neither do we
 126     if (!self._handle.writev)
 127       self._writev = null;
 128   }
 129 }
 130
 131 function Socket(options) {
 132   if (!(this instanceof Socket)) return new Socket(options);
 133
 134   this._connecting = false;
 135   this._hadError = false;
 136   this._handle = null;
 137
 138   if (util.isNumber(options))
 139     options = { fd: options }; // Legacy interface.
 140   else if (util.isUndefined(options))
 141     options = {};
 142
 143   stream.Duplex.call(this, options);
 144
 145   if (options.handle) {
 146     this._handle = options.handle; // private
 147   } else if (!util.isUndefined(options.fd)) {
 148     this._handle = createHandle(options.fd);
 149     this._handle.open(options.fd);
 150     this.readable = options.readable !== false;
 151     this.writable = options.writable !== false;
 152   } else {
 153     // these will be set once there is a connection
 154     this.readable = this.writable = false;
 155   }
 156
 157   // shut down the socket when we're finished with it.
 158   this.on('finish', onSocketFinish);
 159   this.on('_socketEnd', onSocketEnd);
 160
 161   initSocketHandle(this);
 162
 163   this._pendingData = null;
 164   this._pendingEncoding = '';
 165
 166   // handle strings directly
 167   this._writableState.decodeStrings = false;
 168
 169   // default to *not* allowing half open sockets
 170   this.allowHalfOpen = options && options.allowHalfOpen || false;
 171
 172   // if we have a handle, then start the flow of data into the
 173   // buffer.  if not, then this will happen when we connect
 174   if (this._handle && options.readable !== false)
 175     this.read(0);
 176 }
 177 util.inherits(Socket, stream.Duplex);
 178
 179 // the user has called .end(), and all the bytes have been
 180 // sent out to the other side.
 181 // If allowHalfOpen is false, or if the readable side has
 182 // ended already, then destroy.
 183 // If allowHalfOpen is true, then we need to do a shutdown,
 184 // so that only the writable side will be cleaned up.
 185 function onSocketFinish() {
 186   // If still connecting - defer handling 'finish' until 'connect' will happen
 187   if (this._connecting) {
 188     debug('osF: not yet connected');
 189     return this.once('connect', onSocketFinish);
 190   }
 191
 192   debug('onSocketFinish');
 193   if (!this.readable || this._readableState.ended) {
 194     debug('oSF: ended, destroy', this._readableState);
 195     return this.destroy();
 196   }
 197
 198   debug('oSF: not ended, call shutdown()');
 199
 200   // otherwise, just shutdown, or destroy() if not possible
 201   if (!this._handle || !this._handle.shutdown)
 202     return this.destroy();
 203
 204   var req = { oncomplete: afterShutdown };
 205   var err = this._handle.shutdown(req);
 206
 207   if (err)
 208     return this._destroy(errnoException(err, 'shutdown'));
 209 }
 210
 211
 212 function afterShutdown(status, handle, req) {
 213   var self = handle.owner;
 214
 215   debug('afterShutdown destroyed=%j', self.destroyed,
 216         self._readableState);
 217
 218   // callback may come after call to destroy.
 219   if (self.destroyed)
 220     return;
 221
 222   if (self._readableState.ended) {
 223     debug('readableState ended, destroying');
 224     self.destroy();
 225   } else {
 226     self.once('_socketEnd', self.destroy);
 227   }
 228 }
 229
 230 // the EOF has been received, and no more bytes are coming.
 231 // if the writable side has ended already, then clean everything
 232 // up.
 233 function onSocketEnd() {
 234   // XXX Should not have to do as much crap in this function.
 235   // ended should already be true, since this is called *after*
 236   // the EOF errno and onread has eof'ed
 237   debug('onSocketEnd', this._readableState);
 238   this._readableState.ended = true;
 239   if (this._readableState.endEmitted) {
 240     this.readable = false;
 241     maybeDestroy(this);
 242   } else {
 243     this.once('end', function() {
 244       this.readable = false;
 245       maybeDestroy(this);
 246     });
 247     this.read(0);
 248   }
 249
 250   if (!this.allowHalfOpen) {
 251     this.write = writeAfterFIN;
 252     this.destroySoon();
 253   }
 254 }
 255
 256 // Provide a better error message when we call end() as a result
 257 // of the other side sending a FIN.  The standard 'write after end'
 258 // is overly vague, and makes it seem like the user's code is to blame.
 259 function writeAfterFIN(chunk, encoding, cb) {
 260   if (util.isFunction(encoding)) {
 261     cb = encoding;
 262     encoding = null;
 263   }
 264
 265   var er = new Error('This socket has been ended by the other party');
 266   er.code = 'EPIPE';
 267   var self = this;
 268   // TODO: defer error events consistently everywhere, not just the cb
 269   self.emit('error', er);
 270   if (util.isFunction(cb)) {
 271     process.nextTick(function() {
 272       cb(er);
 273     });
 274   }
 275 }
 276
 277 exports.Socket = Socket;
 278 exports.Stream = Socket; // Legacy naming.
 279
 280 Socket.prototype.read = function(n) {
 281   if (n === 0)
 282     return stream.Readable.prototype.read.call(this, n);
 283
 284   this.read = stream.Readable.prototype.read;
 285   this._consuming = true;
 286   return this.read(n);
 287 };
 288
 289
 290 Socket.prototype.listen = function() {
 291   debug('socket.listen');
 292   var self = this;
 293   self.on('connection', arguments[0]);
 294   listen(self, null, null, null);
 295 };
 296
 297
 298 Socket.prototype.setTimeout = function(msecs, callback) {
 299   if (msecs > 0 && !isNaN(msecs) && isFinite(msecs)) {
 300     timers.enroll(this, msecs);
 301     timers._unrefActive(this);
 302     if (callback) {
 303       this.once('timeout', callback);
 304     }
 305   } else if (msecs === 0) {
 306     timers.unenroll(this);
 307     if (callback) {
 308       this.removeListener('timeout', callback);
 309     }
 310   }
 311 };
 312
 313
 314 Socket.prototype._onTimeout = function() {
 315   debug('_onTimeout');
 316   this.emit('timeout');
 317 };
 318
 319
 320 Socket.prototype.setNoDelay = function(enable) {
 321   // backwards compatibility: assume true when `enable` is omitted
 322   if (this._handle && this._handle.setNoDelay)
 323     this._handle.setNoDelay(util.isUndefined(enable) ? true : !!enable);
 324 };
 325
 326
 327 Socket.prototype.setKeepAlive = function(setting, msecs) {
 328   if (this._handle && this._handle.setKeepAlive)
 329     this._handle.setKeepAlive(setting, ~~(msecs / 1000));
 330 };
 331
 332
 333 Socket.prototype.address = function() {
 334   if (this._handle && this._handle.getsockname) {
 335     var out = {};
 336     var err = this._handle.getsockname(out);
 337     // TODO(bnoordhuis) Check err and throw?
 338     return out;
 339   }
 340   return null;
 341 };
 342
 343
 344 Object.defineProperty(Socket.prototype, 'readyState', {
 345   get: function() {
 346     if (this._connecting) {
 347       return 'opening';
 348     } else if (this.readable && this.writable) {
 349       return 'open';
 350     } else if (this.readable && !this.writable) {
 351       return 'readOnly';
 352     } else if (!this.readable && this.writable) {
 353       return 'writeOnly';
 354     } else {
 355       return 'closed';
 356     }
 357   }
 358 });
 359
 360
 361 Object.defineProperty(Socket.prototype, 'bufferSize', {
 362   get: function() {
 363     if (this._handle) {
 364       return this._handle.writeQueueSize + this._writableState.length;
 365     }
 366   }
 367 });
 368
 369
 370 // Just call handle.readStart until we have enough in the buffer
 371 Socket.prototype._read = function(n) {
 372   debug('_read');
 373
 374   if (this._connecting || !this._handle) {
 375     debug('_read wait for connection');
 376     this.once('connect', this._read.bind(this, n));
 377   } else if (!this._handle.reading) {
 378     // not already reading, start the flow
 379     debug('Socket._read readStart');
 380     this._handle.reading = true;
 381     var err = this._handle.readStart();
 382     if (err)
 383       this._destroy(errnoException(err, 'read'));
 384   }
 385 };
 386
 387
 388 Socket.prototype.end = function(data, encoding) {
 389   stream.Duplex.prototype.end.call(this, data, encoding);
 390   this.writable = false;
 391   DTRACE_NET_STREAM_END(this);
 392
 393   // just in case we're waiting for an EOF.
 394   if (this.readable && !this._readableState.endEmitted)
 395     this.read(0);
 396   else
 397     maybeDestroy(this);
 398 };
 399
 400
 401 // Call whenever we set writable=false or readable=false
 402 function maybeDestroy(socket) {
 403   if (!socket.readable &&
 404       !socket.writable &&
 405       !socket.destroyed &&
 406       !socket._connecting &&
 407       !socket._writableState.length) {
 408     socket.destroy();
 409   }
 410 }
 411
 412
 413 Socket.prototype.destroySoon = function() {
 414   if (this.writable)
 415     this.end();
 416
 417   if (this._writableState.finished)
 418     this.destroy();
 419   else
 420     this.once('finish', this.destroy);
 421 };
 422
 423
 424 Socket.prototype._destroy = function(exception, cb) {
 425   debug('destroy');
 426
 427   var self = this;
 428
 429   function fireErrorCallbacks() {
 430     if (cb) cb(exception);
 431     if (exception && !self.errorEmitted) {
 432       process.nextTick(function() {
 433         self.emit('error', exception);
 434       });
 435       self.errorEmitted = true;
 436     }
 437   };
 438
 439   if (this.destroyed) {
 440     debug('already destroyed, fire error callbacks');
 441     fireErrorCallbacks();
 442     return;
 443   }
 444
 445   self._connecting = false;
 446
 447   this.readable = this.writable = false;
 448
 449   timers.unenroll(this);
 450
 451   debug('close');
 452   if (this._handle) {
 453     if (this !== process.stderr)
 454       debug('close handle');
 455     var isException = exception ? true : false;
 456     this._handle.close(function() {
 457       debug('emit close');
 458       self.emit('close', isException);
 459     });
 460     this._handle.onread = noop;
 461     this._handle = null;
 462   }
 463
 464   fireErrorCallbacks();
 465   this.destroyed = true;
 466
 467   if (this.server) {
 468     COUNTER_NET_SERVER_CONNECTION_CLOSE(this);
 469     debug('has server');
 470     this.server._connections--;
 471     if (this.server._emitCloseIfDrained) {
 472       this.server._emitCloseIfDrained();
 473     }
 474   }
 475 };
 476
 477
 478 Socket.prototype.destroy = function(exception) {
 479   debug('destroy', exception);
 480   this._destroy(exception);
 481 };
 482
 483
 484 // This function is called whenever the handle gets a
 485 // buffer, or when there's an error reading.
 486 function onread(nread, buffer) {
 487   var handle = this;
 488   var self = handle.owner;
 489   assert(handle === self._handle, 'handle != self._handle');
 490
 491   timers._unrefActive(self);
 492
 493   debug('onread', nread);
 494
 495   if (nread > 0) {
 496     debug('got data');
 497
 498     // read success.
 499     // In theory (and in practice) calling readStop right now
 500     // will prevent this from being called again until _read() gets
 501     // called again.
 502
 503     // if it's not enough data, we'll just call handle.readStart()
 504     // again right away.
 505     self.bytesRead += nread;
 506
 507     // Optimization: emit the original buffer with end points
 508     var ret = self.push(buffer);
 509
 510     if (handle.reading && !ret) {
 511       handle.reading = false;
 512       debug('readStop');
 513       var err = handle.readStop();
 514       if (err)
 515         self._destroy(errnoException(err, 'read'));
 516     }
 517     return;
 518   }
 519
 520   // if we didn't get any bytes, that doesn't necessarily mean EOF.
 521   // wait for the next one.
 522   if (nread === 0) {
 523     debug('not any data, keep waiting');
 524     return;
 525   }
 526
 527   // Error, possibly EOF.
 528   if (nread !== uv.UV_EOF) {
 529     return self._destroy(errnoException(nread, 'read'));
 530   }
 531
 532   debug('EOF');
 533
 534   if (self._readableState.length === 0) {
 535     self.readable = false;
 536     maybeDestroy(self);
 537   }
 538
 539   // push a null to signal the end of data.
 540   self.push(null);
 541
 542   // internal end event so that we know that the actual socket
 543   // is no longer readable, and we can start the shutdown
 544   // procedure. No need to wait for all the data to be consumed.
 545   self.emit('_socketEnd');
 546 }
 547
 548
 549 Socket.prototype._getpeername = function() {
 550   if (!this._handle || !this._handle.getpeername) {
 551     return {};
 552   }
 553   if (!this._peername) {
 554     var out = {};
 555     var err = this._handle.getpeername(out);
 556     if (err) return {};  // FIXME(bnoordhuis) Throw?
 557     this._peername = out;
 558   }
 559   return this._peername;
 560 };
 561
 562
 563 Socket.prototype.__defineGetter__('remoteAddress', function() {
 564   return this._getpeername().address;
 565 });
 566
 567
 568 Socket.prototype.__defineGetter__('remotePort', function() {
 569   return this._getpeername().port;
 570 });
 571
 572
 573 Socket.prototype._getsockname = function() {
 574   if (!this._handle || !this._handle.getsockname) {
 575     return {};
 576   }
 577   if (!this._sockname) {
 578     var out = {};
 579     var err = this._handle.getsockname(out);
 580     if (err) return {};  // FIXME(bnoordhuis) Throw?
 581     this._sockname = out;
 582   }
 583   return this._sockname;
 584 };
 585
 586
 587 Socket.prototype.__defineGetter__('localAddress', function() {
 588   return this._getsockname().address;
 589 });
 590
 591
 592 Socket.prototype.__defineGetter__('localPort', function() {
 593   return this._getsockname().port;
 594 });
 595
 596
 597 Socket.prototype.write = function(chunk, encoding, cb) {
 598   if (!util.isString(chunk) && !util.isBuffer(chunk))
 599     throw new TypeError('invalid data');
 600   return stream.Duplex.prototype.write.apply(this, arguments);
 601 };
 602
 603
 604 Socket.prototype._writeGeneric = function(writev, data, encoding, cb) {
 605   // If we are still connecting, then buffer this for later.
 606   // The Writable logic will buffer up any more writes while
 607   // waiting for this one to be done.
 608   if (this._connecting) {
 609     this._pendingData = data;
 610     this._pendingEncoding = encoding;
 611     this.once('connect', function() {
 612       this._writeGeneric(writev, data, encoding, cb);
 613     });
 614     return;
 615   }
 616   this._pendingData = null;
 617   this._pendingEncoding = '';
 618
 619   timers._unrefActive(this);
 620
 621   if (!this._handle) {
 622     this._destroy(new Error('This socket is closed.'), cb);
 623     return false;
 624   }
 625
 626   var req = { oncomplete: afterWrite };
 627   var err;
 628
 629   if (writev) {
 630     var chunks = new Array(data.length << 1);
 631     for (var i = 0; i < data.length; i++) {
 632       var entry = data[i];
 633       var chunk = entry.chunk;
 634       var enc = entry.encoding;
 635       chunks[i * 2] = chunk;
 636       chunks[i * 2 + 1] = enc;
 637     }
 638     err = this._handle.writev(req, chunks);
 639
 640     // Retain chunks
 641     if (err === 0) req._chunks = chunks;
 642   } else {
 643     var enc;
 644     if (util.isBuffer(data)) {
 645       req.buffer = data;  // Keep reference alive.
 646       enc = 'buffer';
 647     } else {
 648       enc = encoding;
 649     }
 650     err = createWriteReq(req, this._handle, data, enc);
 651   }
 652
 653   if (err)
 654     return this._destroy(errnoException(err, 'write'), cb);
 655
 656   this._bytesDispatched += req.bytes;
 657
 658   // If it was entirely flushed, we can write some more right now.
 659   // However, if more is left in the queue, then wait until that clears.
 660   if (this._handle.writeQueueSize === 0)
 661     cb();
 662   else
 663     req.cb = cb;
 664 };
 665
 666
 667 Socket.prototype._writev = function(chunks, cb) {
 668   this._writeGeneric(true, chunks, '', cb);
 669 };
 670
 671
 672 Socket.prototype._write = function(data, encoding, cb) {
 673   this._writeGeneric(false, data, encoding, cb);
 674 };
 675
 676 // Important: this should have the same values as in src/stream_wrap.h
 677 function getEncodingId(encoding) {
 678   switch (encoding) {
 679     case 'buffer':
 680       return 0;
 681
 682     case 'utf8':
 683     case 'utf-8':
 684       return 1;
 685
 686     case 'ascii':
 687       return 2;
 688
 689     case 'ucs2':
 690     case 'ucs-2':
 691     case 'utf16le':
 692     case 'utf-16le':
 693       return 3;
 694
 695     default:
 696       return 0;
 697   }
 698 }
 699
 700 function createWriteReq(req, handle, data, encoding) {
 701   switch (encoding) {
 702     case 'buffer':
 703       return handle.writeBuffer(req, data);
 704
 705     case 'utf8':
 706     case 'utf-8':
 707       return handle.writeUtf8String(req, data);
 708
 709     case 'ascii':
 710       return handle.writeAsciiString(req, data);
 711
 712     case 'ucs2':
 713     case 'ucs-2':
 714     case 'utf16le':
 715     case 'utf-16le':
 716       return handle.writeUcs2String(req, data);
 717
 718     default:
 719       return handle.writeBuffer(req, new Buffer(data, encoding));
 720   }
 721 }
 722
 723
 724 Socket.prototype.__defineGetter__('bytesWritten', function() {
 725   var bytes = this._bytesDispatched,
 726       state = this._writableState,
 727       data = this._pendingData,
 728       encoding = this._pendingEncoding;
 729
 730   state.buffer.forEach(function(el) {
 731     if (util.isBuffer(el.chunk))
 732       bytes += el.chunk.length;
 733     else
 734       bytes += Buffer.byteLength(el.chunk, el.encoding);
 735   });
 736
 737   if (data) {
 738     if (util.isBuffer(data))
 739       bytes += data.length;
 740     else
 741       bytes += Buffer.byteLength(data, encoding);
 742   }
 743
 744   return bytes;
 745 });
 746
 747
 748 function afterWrite(status, handle, req) {
 749   var self = handle.owner;
 750   if (self !== process.stderr && self !== process.stdout)
 751     debug('afterWrite', status);
 752
 753   // callback may come after call to destroy.
 754   if (self.destroyed) {
 755     debug('afterWrite destroyed');
 756     return;
 757   }
 758
 759   if (status < 0) {
 760     var ex = errnoException(status, 'write');
 761     debug('write failure', ex);
 762     self._destroy(ex, req.cb);
 763     return;
 764   }
 765
 766   timers._unrefActive(self);
 767
 768   if (self !== process.stderr && self !== process.stdout)
 769     debug('afterWrite call cb');
 770
 771   if (req.cb)
 772     req.cb.call(self);
 773 }
 774
 775
 776 function connect(self, address, port, addressType, localAddress) {
 777   // TODO return promise from Socket.prototype.connect which
 778   // wraps _connectReq.
 779
 780   assert.ok(self._connecting);
 781
 782   var err;
 783   if (localAddress) {
 784     if (addressType == 6) {
 785       err = self._handle.bind6(localAddress);
 786     } else {
 787       err = self._handle.bind(localAddress);
 788     }
 789
 790     if (err) {
 791       self._destroy(errnoException(err, 'bind'));
 792       return;
 793     }
 794   }
 795
 796   var req = { oncomplete: afterConnect };
 797   if (addressType == 6) {
 798     err = self._handle.connect6(req, address, port);
 799   } else if (addressType == 4) {
 800     err = self._handle.connect(req, address, port);
 801   } else {
 802     err = self._handle.connect(req, address, afterConnect);
 803   }
 804
 805   if (err) {
 806     self._destroy(errnoException(err, 'connect'));
 807   }
 808 }
 809
 810
 811 Socket.prototype.connect = function(options, cb) {
 812   if (this.write !== Socket.prototype.write)
 813     this.write = Socket.prototype.write;
 814
 815   if (!util.isObject(options)) {
 816     // Old API:
 817     // connect(port, [host], [cb])
 818     // connect(path, [cb]);
 819     var args = normalizeConnectArgs(arguments);
 820     return Socket.prototype.connect.apply(this, args);
 821   }
 822
 823   if (this.destroyed) {
 824     this._readableState.reading = false;
 825     this._readableState.ended = false;
 826     this._writableState.ended = false;
 827     this._writableState.ending = false;
 828     this._writableState.finished = false;
 829     this.destroyed = false;
 830     this._handle = null;
 831   }
 832
 833   var self = this;
 834   var pipe = !!options.path;
 835   debug('pipe', pipe, options.path);
 836
 837   if (!this._handle) {
 838     this._handle = pipe ? createPipe() : createTCP();
 839     initSocketHandle(this);
 840   }
 841
 842   if (util.isFunction(cb)) {
 843     self.once('connect', cb);
 844   }
 845
 846   timers._unrefActive(this);
 847
 848   self._connecting = true;
 849   self.writable = true;
 850
 851   if (pipe) {
 852     connect(self, options.path);
 853
 854   } else if (!options.host) {
 855     debug('connect: missing host');
 856     connect(self, '127.0.0.1', options.port, 4);
 857
 858   } else {
 859     var host = options.host;
 860     var family = options.family || 4;
 861     debug('connect: find host ' + host);
 862     require('dns').lookup(host, family, function(err, ip, addressType) {
 863       self.emit('lookup', err, ip, addressType);
 864
 865       // It's possible we were destroyed while looking this up.
 866       // XXX it would be great if we could cancel the promise returned by
 867       // the look up.
 868       if (!self._connecting) return;
 869
 870       if (err) {
 871         // net.createConnection() creates a net.Socket object and
 872         // immediately calls net.Socket.connect() on it (that's us).
 873         // There are no event listeners registered yet so defer the
 874         // error event to the next tick.
 875         process.nextTick(function() {
 876           self.emit('error', err);
 877           self._destroy();
 878         });
 879       } else {
 880         timers._unrefActive(self);
 881
 882         addressType = addressType || 4;
 883
 884         // node_net.cc handles null host names graciously but user land
 885         // expects remoteAddress to have a meaningful value
 886         ip = ip || (addressType === 4 ? '127.0.0.1' : '0:0:0:0:0:0:0:1');
 887
 888         connect(self, ip, options.port, addressType, options.localAddress);
 889       }
 890     });
 891   }
 892   return self;
 893 };
 894
 895
 896 Socket.prototype.ref = function() {
 897   if (this._handle)
 898     this._handle.ref();
 899 };
 900
 901
 902 Socket.prototype.unref = function() {
 903   if (this._handle)
 904     this._handle.unref();
 905 };
 906
 907
 908
 909 function afterConnect(status, handle, req, readable, writable) {
 910   var self = handle.owner;
 911
 912   // callback may come after call to destroy
 913   if (self.destroyed) {
 914     return;
 915   }
 916
 917   assert(handle === self._handle, 'handle != self._handle');
 918
 919   debug('afterConnect');
 920
 921   assert.ok(self._connecting);
 922   self._connecting = false;
 923
 924   if (status == 0) {
 925     self.readable = readable;
 926     self.writable = writable;
 927     timers._unrefActive(self);
 928
 929     self.emit('connect');
 930
 931     // start the first read, or get an immediate EOF.
 932     // this doesn't actually consume any bytes, because len=0.
 933     if (readable)
 934       self.read(0);
 935
 936   } else {
 937     self._connecting = false;
 938     self._destroy(errnoException(status, 'connect'));
 939   }
 940 }
 941
 942
 943 function Server(/* [ options, ] listener */) {
 944   if (!(this instanceof Server)) return new Server(arguments[0], arguments[1]);
 945   events.EventEmitter.call(this);
 946
 947   var self = this;
 948
 949   var options;
 950
 951   if (util.isFunction(arguments[0])) {
 952     options = {};
 953     self.on('connection', arguments[0]);
 954   } else {
 955     options = arguments[0] || {};
 956
 957     if (util.isFunction(arguments[1])) {
 958       self.on('connection', arguments[1]);
 959     }
 960   }
 961
 962   this._connections = 0;
 963
 964   Object.defineProperty(this, 'connections', {
 965     get: util.deprecate(function() {
 966
 967       if (self._usingSlaves) {
 968         return null;
 969       }
 970       return self._connections;
 971     }, 'connections property is deprecated. Use getConnections() method'),
 972     set: util.deprecate(function(val) {
 973       return (self._connections = val);
 974     }, 'connections property is deprecated. Use getConnections() method'),
 975     configurable: true, enumerable: true
 976   });
 977
 978   this._handle = null;
 979   this._usingSlaves = false;
 980   this._slaves = [];
 981
 982   this.allowHalfOpen = options.allowHalfOpen || false;
 983 }
 984 util.inherits(Server, events.EventEmitter);
 985 exports.Server = Server;
 986
 987
 988 function toNumber(x) { return (x = Number(x)) >= 0 ? x : false; }
 989
 990
 991 var createServerHandle = exports._createServerHandle =
 992     function(address, port, addressType, fd) {
 993   var err = 0;
 994   // assign handle in listen, and clean up if bind or listen fails
 995   var handle;
 996
 997   if (util.isNumber(fd) && fd >= 0) {
 998     try {
 999       handle = createHandle(fd);
1000     }
1001     catch (e) {
1002       // Not a fd we can listen on.  This will trigger an error.
1003       debug('listen invalid fd=' + fd + ': ' + e.message);
1004       return uv.UV_EINVAL;
1005     }
1006     handle.open(fd);
1007     handle.readable = true;
1008     handle.writable = true;
1009     return handle;
1010
1011   } else if (port == -1 && addressType == -1) {
1012     handle = createPipe();
1013     if (process.platform === 'win32') {
1014       var instances = parseInt(process.env.NODE_PENDING_PIPE_INSTANCES);
1015       if (!isNaN(instances)) {
1016         handle.setPendingInstances(instances);
1017       }
1018     }
1019   } else {
1020     handle = createTCP();
1021   }
1022
1023   if (address || port) {
1024     debug('bind to ' + address);
1025     if (addressType == 6) {
1026       err = handle.bind6(address, port);
1027     } else {
1028       err = handle.bind(address, port);
1029     }
1030   }
1031
1032   if (err) {
1033     handle.close();
1034     return err;
1035   }
1036
1037   return handle;
1038 };
1039
1040
1041 Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
1042   debug('listen2', address, port, addressType, backlog);
1043   var self = this;
1044
1045   // If there is not yet a handle, we need to create one and bind.
1046   // In the case of a server sent via IPC, we don't need to do this.
1047   if (!self._handle) {
1048     debug('_listen2: create a handle');
1049     var rval = createServerHandle(address, port, addressType, fd);
1050     if (util.isNumber(rval)) {
1051       var error = errnoException(rval, 'listen');
1052       process.nextTick(function() {
1053         self.emit('error', error);
1054       });
1055       return;
1056     }
1057     self._handle = rval;
1058   } else {
1059     debug('_listen2: have a handle already');
1060   }
1061
1062   self._handle.onconnection = onconnection;
1063   self._handle.owner = self;
1064
1065   // Use a backlog of 512 entries. We pass 511 to the listen() call because
1066   // the kernel does: backlogsize = roundup_pow_of_two(backlogsize + 1);
1067   // which will thus give us a backlog of 512 entries.
1068   var err = self._handle.listen(backlog || 511);
1069
1070   if (err) {
1071     var ex = errnoException(err, 'listen');
1072     self._handle.close();
1073     self._handle = null;
1074     process.nextTick(function() {
1075       self.emit('error', ex);
1076     });
1077     return;
1078   }
1079
1080   // generate connection key, this should be unique to the connection
1081   this._connectionKey = addressType + ':' + address + ':' + port;
1082
1083   process.nextTick(function() {
1084     self.emit('listening');
1085   });
1086 };
1087
1088
1089 function listen(self, address, port, addressType, backlog, fd) {
1090   if (!cluster) cluster = require('cluster');
1091
1092   if (cluster.isMaster) {
1093     self._listen2(address, port, addressType, backlog, fd);
1094     return;
1095   }
1096
1097   cluster._getServer(self, address, port, addressType, fd, cb);
1098
1099   function cb(err, handle) {
1100     // EADDRINUSE may not be reported until we call listen(). To complicate
1101     // matters, a failed bind() followed by listen() will implicitly bind to
1102     // a random port. Ergo, check that the socket is bound to the expected
1103     // port before calling listen().
1104     //
1105     // FIXME(bnoordhuis) Doesn't work for pipe handles, they don't have a
1106     // getsockname() method. Non-issue for now, the cluster module doesn't
1107     // really support pipes anyway.
1108     if (err === 0 && port > 0 && handle.getsockname) {
1109       var out = {};
1110       err = handle.getsockname(out);
1111       if (err === 0 && port !== out.port)
1112         err = uv.UV_EADDRINUSE;
1113     }
1114
1115     if (err)
1116       return self.emit('error', errnoException(err, 'bind'));
1117
1118     self._handle = handle;
1119     self._listen2(address, port, addressType, backlog, fd);
1120   }
1121 }
1122
1123
1124 Server.prototype.listen = function() {
1125   var self = this;
1126
1127   var lastArg = arguments[arguments.length - 1];
1128   if (util.isFunction(lastArg)) {
1129     self.once('listening', lastArg);
1130   }
1131
1132   var port = toNumber(arguments[0]);
1133
1134   // The third optional argument is the backlog size.
1135   // When the ip is omitted it can be the second argument.
1136   var backlog = toNumber(arguments[1]) || toNumber(arguments[2]);
1137
1138   var TCP = process.binding('tcp_wrap').TCP;
1139
1140   if (arguments.length == 0 || util.isFunction(arguments[0])) {
1141     // Bind to a random port.
1142     listen(self, '0.0.0.0', 0, null, backlog);
1143
1144   } else if (arguments[0] && util.isObject(arguments[0])) {
1145     var h = arguments[0];
1146     if (h._handle) {
1147       h = h._handle;
1148     } else if (h.handle) {
1149       h = h.handle;
1150     }
1151     if (h instanceof TCP) {
1152       self._handle = h;
1153       listen(self, null, -1, -1, backlog);
1154     } else if (util.isNumber(h.fd) && h.fd >= 0) {
1155       listen(self, null, null, null, backlog, h.fd);
1156     } else {
1157       throw new Error('Invalid listen argument: ' + h);
1158     }
1159   } else if (isPipeName(arguments[0])) {
1160     // UNIX socket or Windows pipe.
1161     var pipeName = self._pipeName = arguments[0];
1162     listen(self, pipeName, -1, -1, backlog);
1163
1164   } else if (util.isUndefined(arguments[1]) ||
1165              util.isFunction(arguments[1]) ||
1166              util.isNumber(arguments[1])) {
1167     // The first argument is the port, no IP given.
1168     listen(self, '0.0.0.0', port, 4, backlog);
1169
1170   } else {
1171     // The first argument is the port, the second an IP.
1172     require('dns').lookup(arguments[1], function(err, ip, addressType) {
1173       if (err) {
1174         self.emit('error', err);
1175       } else {
1176         listen(self, ip || '0.0.0.0', port, ip ? addressType : 4, backlog);
1177       }
1178     });
1179   }
1180   return self;
1181 };
1182
1183 Server.prototype.address = function() {
1184   if (this._handle && this._handle.getsockname) {
1185     var out = {};
1186     var err = this._handle.getsockname(out);
1187     // TODO(bnoordhuis) Check err and throw?
1188     return out;
1189   } else if (this._pipeName) {
1190     return this._pipeName;
1191   } else {
1192     return null;
1193   }
1194 };
1195
1196 function onconnection(err, clientHandle) {
1197   var handle = this;
1198   var self = handle.owner;
1199
1200   debug('onconnection');
1201
1202   if (err) {
1203     self.emit('error', errnoException(err, 'accept'));
1204     return;
1205   }
1206
1207   if (self.maxConnections && self._connections >= self.maxConnections) {
1208     clientHandle.close();
1209     return;
1210   }
1211
1212   var socket = new Socket({
1213     handle: clientHandle,
1214     allowHalfOpen: self.allowHalfOpen
1215   });
1216   socket.readable = socket.writable = true;
1217
1218
1219   self._connections++;
1220   socket.server = self;
1221
1222   DTRACE_NET_SERVER_CONNECTION(socket);
1223   COUNTER_NET_SERVER_CONNECTION(socket);
1224   self.emit('connection', socket);
1225 }
1226
1227
1228 Server.prototype.getConnections = function(cb) {
1229   function end(err, connections) {
1230     process.nextTick(function() {
1231       cb(err, connections);
1232     });
1233   }
1234
1235   if (!this._usingSlaves) {
1236     return end(null, this._connections);
1237   }
1238
1239   // Poll slaves
1240   var left = this._slaves.length,
1241       total = this._connections;
1242
1243   function oncount(err, count) {
1244     if (err) {
1245       left = -1;
1246       return end(err);
1247     }
1248
1249     total += count;
1250     if (--left === 0) return end(null, total);
1251   }
1252
1253   this._slaves.forEach(function(slave) {
1254     slave.getConnections(oncount);
1255   });
1256 };
1257
1258
1259 Server.prototype.close = function(cb) {
1260   function onSlaveClose() {
1261     if (--left !== 0) return;
1262
1263     self._connections = 0;
1264     self._emitCloseIfDrained();
1265   }
1266
1267   if (!this._handle) {
1268     // Throw error. Follows net_legacy behaviour.
1269     throw new Error('Not running');
1270   }
1271
1272   if (cb) {
1273     this.once('close', cb);
1274   }
1275   this._handle.close();
1276   this._handle = null;
1277
1278   if (this._usingSlaves) {
1279     var self = this,
1280         left = this._slaves.length;
1281
1282     // Increment connections to be sure that, even if all sockets will be closed
1283     // during polling of slaves, `close` event will be emitted only once.
1284     this._connections++;
1285
1286     // Poll slaves
1287     this._slaves.forEach(function(slave) {
1288       slave.close(onSlaveClose);
1289     });
1290   } else {
1291     this._emitCloseIfDrained();
1292   }
1293
1294   return this;
1295 };
1296
1297 Server.prototype._emitCloseIfDrained = function() {
1298   debug('SERVER _emitCloseIfDrained');
1299   var self = this;
1300
1301   if (self._handle || self._connections) {
1302     debug('SERVER handle? %j   connections? %d',
1303           !!self._handle, self._connections);
1304     return;
1305   }
1306
1307   process.nextTick(function() {
1308     debug('SERVER: emit close');
1309     self.emit('close');
1310   });
1311 };
1312
1313
1314 Server.prototype.listenFD = util.deprecate(function(fd, type) {
1315   return this.listen({ fd: fd });
1316 }, 'listenFD is deprecated. Use listen({fd: <number>}).');
1317
1318 Server.prototype._setupSlave = function(socketList) {
1319   this._usingSlaves = true;
1320   this._slaves.push(socketList);
1321 };
1322
1323 Server.prototype.ref = function() {
1324   if (this._handle)
1325     this._handle.ref();
1326 };
1327
1328 Server.prototype.unref = function() {
1329   if (this._handle)
1330     this._handle.unref();
1331 };
1332
1333
1334 // TODO: isIP should be moved to the DNS code. Putting it here now because
1335 // this is what the legacy system did.
1336 exports.isIP = cares.isIP;
1337
1338
1339 exports.isIPv4 = function(input) {
1340   return exports.isIP(input) === 4;
1341 };
1342
1343
1344 exports.isIPv6 = function(input) {
1345   return exports.isIP(input) === 6;
1346 };
1347
1348
1349 if (process.platform === 'win32') {
1350   var simultaneousAccepts;
1351
1352   exports._setSimultaneousAccepts = function(handle) {
1353     if (util.isUndefined(handle)) {
1354       return;
1355     }
1356
1357     if (util.isUndefined(simultaneousAccepts)) {
1358       simultaneousAccepts = (process.env.NODE_MANY_ACCEPTS &&
1359                              process.env.NODE_MANY_ACCEPTS !== '0');
1360     }
1361
1362     if (handle._simultaneousAccepts !== simultaneousAccepts) {
1363       handle.setSimultaneousAccepts(simultaneousAccepts);
1364       handle._simultaneousAccepts = simultaneousAccepts;
1365     }
1366   };
1367 } else {
1368   exports._setSimultaneousAccepts = function(handle) {};
1369 }