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.bytesRead = 0;
 117   self._bytesDispatched = 0;
 118
 119   // Handle creation may be deferred to bind() or connect() time.
 120   if (self._handle) {
 121     self._handle.owner = self;
 122     self._handle.onread = onread;
 123
 124     // If handle doesn't support writev - neither do we
 125     if (!self._handle.writev)
 126       self._writev = null;
 127   }
 128 }
 129
 130 function Socket(options) {
 131   if (!(this instanceof Socket)) return new Socket(options);
 132
 133   this._connecting = false;
 134   this._hadError = false;
 135   this._handle = null;
 136   this._host = 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   // we set destroyed to true before firing error callbacks in order
 465   // to make it re-entrance safe in case Socket.prototype.destroy()
 466   // is called within callbacks
 467   this.destroyed = true;
 468   fireErrorCallbacks();
 469
 470   if (this.server) {
 471     COUNTER_NET_SERVER_CONNECTION_CLOSE(this);
 472     debug('has server');
 473     this.server._connections--;
 474     if (this.server._emitCloseIfDrained) {
 475       this.server._emitCloseIfDrained();
 476     }
 477   }
 478 };
 479
 480
 481 Socket.prototype.destroy = function(exception) {
 482   debug('destroy', exception);
 483   this._destroy(exception);
 484 };
 485
 486
 487 // This function is called whenever the handle gets a
 488 // buffer, or when there's an error reading.
 489 function onread(nread, buffer) {
 490   var handle = this;
 491   var self = handle.owner;
 492   assert(handle === self._handle, 'handle != self._handle');
 493
 494   timers._unrefActive(self);
 495
 496   debug('onread', nread);
 497
 498   if (nread > 0) {
 499     debug('got data');
 500
 501     // read success.
 502     // In theory (and in practice) calling readStop right now
 503     // will prevent this from being called again until _read() gets
 504     // called again.
 505
 506     // if it's not enough data, we'll just call handle.readStart()
 507     // again right away.
 508     self.bytesRead += nread;
 509
 510     // Optimization: emit the original buffer with end points
 511     var ret = self.push(buffer);
 512
 513     if (handle.reading && !ret) {
 514       handle.reading = false;
 515       debug('readStop');
 516       var err = handle.readStop();
 517       if (err)
 518         self._destroy(errnoException(err, 'read'));
 519     }
 520     return;
 521   }
 522
 523   // if we didn't get any bytes, that doesn't necessarily mean EOF.
 524   // wait for the next one.
 525   if (nread === 0) {
 526     debug('not any data, keep waiting');
 527     return;
 528   }
 529
 530   // Error, possibly EOF.
 531   if (nread !== uv.UV_EOF) {
 532     return self._destroy(errnoException(nread, 'read'));
 533   }
 534
 535   debug('EOF');
 536
 537   if (self._readableState.length === 0) {
 538     self.readable = false;
 539     maybeDestroy(self);
 540   }
 541
 542   // push a null to signal the end of data.
 543   self.push(null);
 544
 545   // internal end event so that we know that the actual socket
 546   // is no longer readable, and we can start the shutdown
 547   // procedure. No need to wait for all the data to be consumed.
 548   self.emit('_socketEnd');
 549 }
 550
 551
 552 Socket.prototype._getpeername = function() {
 553   if (!this._handle || !this._handle.getpeername) {
 554     return {};
 555   }
 556   if (!this._peername) {
 557     var out = {};
 558     var err = this._handle.getpeername(out);
 559     if (err) return {};  // FIXME(bnoordhuis) Throw?
 560     this._peername = out;
 561   }
 562   return this._peername;
 563 };
 564
 565
 566 Socket.prototype.__defineGetter__('remoteAddress', function() {
 567   return this._getpeername().address;
 568 });
 569
 570
 571 Socket.prototype.__defineGetter__('remotePort', function() {
 572   return this._getpeername().port;
 573 });
 574
 575
 576 Socket.prototype._getsockname = function() {
 577   if (!this._handle || !this._handle.getsockname) {
 578     return {};
 579   }
 580   if (!this._sockname) {
 581     var out = {};
 582     var err = this._handle.getsockname(out);
 583     if (err) return {};  // FIXME(bnoordhuis) Throw?
 584     this._sockname = out;
 585   }
 586   return this._sockname;
 587 };
 588
 589
 590 Socket.prototype.__defineGetter__('localAddress', function() {
 591   return this._getsockname().address;
 592 });
 593
 594
 595 Socket.prototype.__defineGetter__('localPort', function() {
 596   return this._getsockname().port;
 597 });
 598
 599
 600 Socket.prototype.write = function(chunk, encoding, cb) {
 601   if (!util.isString(chunk) && !util.isBuffer(chunk))
 602     throw new TypeError('invalid data');
 603   return stream.Duplex.prototype.write.apply(this, arguments);
 604 };
 605
 606
 607 Socket.prototype._writeGeneric = function(writev, data, encoding, cb) {
 608   // If we are still connecting, then buffer this for later.
 609   // The Writable logic will buffer up any more writes while
 610   // waiting for this one to be done.
 611   if (this._connecting) {
 612     this._pendingData = data;
 613     this._pendingEncoding = encoding;
 614     this.once('connect', function() {
 615       this._writeGeneric(writev, data, encoding, cb);
 616     });
 617     return;
 618   }
 619   this._pendingData = null;
 620   this._pendingEncoding = '';
 621
 622   timers._unrefActive(this);
 623
 624   if (!this._handle) {
 625     this._destroy(new Error('This socket is closed.'), cb);
 626     return false;
 627   }
 628
 629   var req = { oncomplete: afterWrite, async: false };
 630   var err;
 631
 632   if (writev) {
 633     var chunks = new Array(data.length << 1);
 634     for (var i = 0; i < data.length; i++) {
 635       var entry = data[i];
 636       var chunk = entry.chunk;
 637       var enc = entry.encoding;
 638       chunks[i * 2] = chunk;
 639       chunks[i * 2 + 1] = enc;
 640     }
 641     err = this._handle.writev(req, chunks);
 642
 643     // Retain chunks
 644     if (err === 0) req._chunks = chunks;
 645   } else {
 646     var enc;
 647     if (util.isBuffer(data)) {
 648       req.buffer = data;  // Keep reference alive.
 649       enc = 'buffer';
 650     } else {
 651       enc = encoding;
 652     }
 653     err = createWriteReq(req, this._handle, data, enc);
 654   }
 655
 656   if (err)
 657     return this._destroy(errnoException(err, 'write', req.error), cb);
 658
 659   this._bytesDispatched += req.bytes;
 660
 661   // If it was entirely flushed, we can write some more right now.
 662   // However, if more is left in the queue, then wait until that clears.
 663   if (req.async && this._handle.writeQueueSize != 0)
 664     req.cb = cb;
 665   else
 666     cb();
 667 };
 668
 669
 670 Socket.prototype._writev = function(chunks, cb) {
 671   this._writeGeneric(true, chunks, '', cb);
 672 };
 673
 674
 675 Socket.prototype._write = function(data, encoding, cb) {
 676   this._writeGeneric(false, data, encoding, cb);
 677 };
 678
 679 // Important: this should have the same values as in src/stream_wrap.h
 680 function getEncodingId(encoding) {
 681   switch (encoding) {
 682     case 'buffer':
 683       return 0;
 684
 685     case 'utf8':
 686     case 'utf-8':
 687       return 1;
 688
 689     case 'ascii':
 690       return 2;
 691
 692     case 'ucs2':
 693     case 'ucs-2':
 694     case 'utf16le':
 695     case 'utf-16le':
 696       return 3;
 697
 698     default:
 699       return 0;
 700   }
 701 }
 702
 703 function createWriteReq(req, handle, data, encoding) {
 704   switch (encoding) {
 705     case 'buffer':
 706       return handle.writeBuffer(req, data);
 707
 708     case 'utf8':
 709     case 'utf-8':
 710       return handle.writeUtf8String(req, data);
 711
 712     case 'ascii':
 713       return handle.writeAsciiString(req, data);
 714
 715     case 'ucs2':
 716     case 'ucs-2':
 717     case 'utf16le':
 718     case 'utf-16le':
 719       return handle.writeUcs2String(req, data);
 720
 721     default:
 722       return handle.writeBuffer(req, new Buffer(data, encoding));
 723   }
 724 }
 725
 726
 727 Socket.prototype.__defineGetter__('bytesWritten', function() {
 728   var bytes = this._bytesDispatched,
 729       state = this._writableState,
 730       data = this._pendingData,
 731       encoding = this._pendingEncoding;
 732
 733   state.buffer.forEach(function(el) {
 734     if (util.isBuffer(el.chunk))
 735       bytes += el.chunk.length;
 736     else
 737       bytes += Buffer.byteLength(el.chunk, el.encoding);
 738   });
 739
 740   if (data) {
 741     if (util.isBuffer(data))
 742       bytes += data.length;
 743     else
 744       bytes += Buffer.byteLength(data, encoding);
 745   }
 746
 747   return bytes;
 748 });
 749
 750
 751 function afterWrite(status, handle, req, err) {
 752   var self = handle.owner;
 753   if (self !== process.stderr && self !== process.stdout)
 754     debug('afterWrite', status);
 755
 756   // callback may come after call to destroy.
 757   if (self.destroyed) {
 758     debug('afterWrite destroyed');
 759     return;
 760   }
 761
 762   if (status < 0) {
 763     var ex = errnoException(status, 'write', err);
 764     debug('write failure', ex);
 765     self._destroy(ex, req.cb);
 766     return;
 767   }
 768
 769   timers._unrefActive(self);
 770
 771   if (self !== process.stderr && self !== process.stdout)
 772     debug('afterWrite call cb');
 773
 774   if (req.cb)
 775     req.cb.call(self);
 776 }
 777
 778
 779 function connect(self, address, port, addressType, localAddress) {
 780   // TODO return promise from Socket.prototype.connect which
 781   // wraps _connectReq.
 782
 783   assert.ok(self._connecting);
 784
 785   var err;
 786   if (localAddress) {
 787     if (addressType === 6) {
 788       err = self._handle.bind6(localAddress);
 789     } else {
 790       err = self._handle.bind(localAddress);
 791     }
 792
 793     if (err) {
 794       self._destroy(errnoException(err, 'bind'));
 795       return;
 796     }
 797   }
 798
 799   var req = { oncomplete: afterConnect };
 800   if (addressType === 6 || addressType === 4) {
 801     port = port | 0;
 802     if (port <= 0 || port > 65535)
 803       throw new RangeError('Port should be > 0 and < 65536');
 804
 805     if (addressType === 6) {
 806       err = self._handle.connect6(req, address, port);
 807     } else if (addressType === 4) {
 808       err = self._handle.connect(req, address, port);
 809     }
 810   } else {
 811     err = self._handle.connect(req, address, afterConnect);
 812   }
 813
 814   if (err) {
 815     self._destroy(errnoException(err, 'connect'));
 816   }
 817 }
 818
 819
 820 Socket.prototype.connect = function(options, cb) {
 821   if (this.write !== Socket.prototype.write)
 822     this.write = Socket.prototype.write;
 823
 824   if (!util.isObject(options)) {
 825     // Old API:
 826     // connect(port, [host], [cb])
 827     // connect(path, [cb]);
 828     var args = normalizeConnectArgs(arguments);
 829     return Socket.prototype.connect.apply(this, args);
 830   }
 831
 832   if (this.destroyed) {
 833     this._readableState.reading = false;
 834     this._readableState.ended = false;
 835     this._readableState.endEmitted = false;
 836     this._writableState.ended = false;
 837     this._writableState.ending = false;
 838     this._writableState.finished = false;
 839     this.destroyed = false;
 840     this._handle = null;
 841   }
 842
 843   var self = this;
 844   var pipe = !!options.path;
 845   debug('pipe', pipe, options.path);
 846
 847   if (!this._handle) {
 848     this._handle = pipe ? createPipe() : createTCP();
 849     initSocketHandle(this);
 850   }
 851
 852   if (util.isFunction(cb)) {
 853     self.once('connect', cb);
 854   }
 855
 856   timers._unrefActive(this);
 857
 858   self._connecting = true;
 859   self.writable = true;
 860
 861   if (pipe) {
 862     connect(self, options.path);
 863
 864   } else if (!options.host) {
 865     debug('connect: missing host');
 866     self._host = '127.0.0.1';
 867     connect(self, self._host, options.port, 4);
 868
 869   } else {
 870     var host = options.host;
 871     var family = options.family || 4;
 872     debug('connect: find host ' + host);
 873     self._host = host;
 874     require('dns').lookup(host, family, function(err, ip, addressType) {
 875       self.emit('lookup', err, ip, addressType);
 876
 877       // It's possible we were destroyed while looking this up.
 878       // XXX it would be great if we could cancel the promise returned by
 879       // the look up.
 880       if (!self._connecting) return;
 881
 882       if (err) {
 883         // net.createConnection() creates a net.Socket object and
 884         // immediately calls net.Socket.connect() on it (that's us).
 885         // There are no event listeners registered yet so defer the
 886         // error event to the next tick.
 887         process.nextTick(function() {
 888           self.emit('error', err);
 889           self._destroy();
 890         });
 891       } else {
 892         timers._unrefActive(self);
 893
 894         addressType = addressType || 4;
 895
 896         // node_net.cc handles null host names graciously but user land
 897         // expects remoteAddress to have a meaningful value
 898         ip = ip || (addressType === 4 ? '127.0.0.1' : '0:0:0:0:0:0:0:1');
 899
 900         connect(self, ip, options.port, addressType, options.localAddress);
 901       }
 902     });
 903   }
 904   return self;
 905 };
 906
 907
 908 Socket.prototype.ref = function() {
 909   if (this._handle)
 910     this._handle.ref();
 911 };
 912
 913
 914 Socket.prototype.unref = function() {
 915   if (this._handle)
 916     this._handle.unref();
 917 };
 918
 919
 920
 921 function afterConnect(status, handle, req, readable, writable) {
 922   var self = handle.owner;
 923
 924   // callback may come after call to destroy
 925   if (self.destroyed) {
 926     return;
 927   }
 928
 929   assert(handle === self._handle, 'handle != self._handle');
 930
 931   debug('afterConnect');
 932
 933   assert.ok(self._connecting);
 934   self._connecting = false;
 935
 936   if (status == 0) {
 937     self.readable = readable;
 938     self.writable = writable;
 939     timers._unrefActive(self);
 940
 941     self.emit('connect');
 942
 943     // start the first read, or get an immediate EOF.
 944     // this doesn't actually consume any bytes, because len=0.
 945     if (readable)
 946       self.read(0);
 947
 948   } else {
 949     self._connecting = false;
 950     self._destroy(errnoException(status, 'connect'));
 951   }
 952 }
 953
 954
 955 function Server(/* [ options, ] listener */) {
 956   if (!(this instanceof Server)) return new Server(arguments[0], arguments[1]);
 957   events.EventEmitter.call(this);
 958
 959   var self = this;
 960
 961   var options;
 962
 963   if (util.isFunction(arguments[0])) {
 964     options = {};
 965     self.on('connection', arguments[0]);
 966   } else {
 967     options = arguments[0] || {};
 968
 969     if (util.isFunction(arguments[1])) {
 970       self.on('connection', arguments[1]);
 971     }
 972   }
 973
 974   this._connections = 0;
 975
 976   Object.defineProperty(this, 'connections', {
 977     get: util.deprecate(function() {
 978
 979       if (self._usingSlaves) {
 980         return null;
 981       }
 982       return self._connections;
 983     }, 'connections property is deprecated. Use getConnections() method'),
 984     set: util.deprecate(function(val) {
 985       return (self._connections = val);
 986     }, 'connections property is deprecated. Use getConnections() method'),
 987     configurable: true, enumerable: true
 988   });
 989
 990   this._handle = null;
 991   this._usingSlaves = false;
 992   this._slaves = [];
 993
 994   this.allowHalfOpen = options.allowHalfOpen || false;
 995 }
 996 util.inherits(Server, events.EventEmitter);
 997 exports.Server = Server;
 998
 999
1000 function toNumber(x) { return (x = Number(x)) >= 0 ? x : false; }
1001
1002
1003 var createServerHandle = exports._createServerHandle =
1004     function(address, port, addressType, fd) {
1005   var err = 0;
1006   // assign handle in listen, and clean up if bind or listen fails
1007   var handle;
1008
1009   if (util.isNumber(fd) && fd >= 0) {
1010     try {
1011       handle = createHandle(fd);
1012     }
1013     catch (e) {
1014       // Not a fd we can listen on.  This will trigger an error.
1015       debug('listen invalid fd=' + fd + ': ' + e.message);
1016       return uv.UV_EINVAL;
1017     }
1018     handle.open(fd);
1019     handle.readable = true;
1020     handle.writable = true;
1021     return handle;
1022
1023   } else if (port === -1 && addressType === -1) {
1024     handle = createPipe();
1025     if (process.platform === 'win32') {
1026       var instances = parseInt(process.env.NODE_PENDING_PIPE_INSTANCES);
1027       if (!isNaN(instances)) {
1028         handle.setPendingInstances(instances);
1029       }
1030     }
1031   } else {
1032     handle = createTCP();
1033   }
1034
1035   if (address || port) {
1036     debug('bind to ' + address);
1037     if (addressType === 6) {
1038       err = handle.bind6(address, port);
1039     } else {
1040       err = handle.bind(address, port);
1041     }
1042   }
1043
1044   if (err) {
1045     handle.close();
1046     return err;
1047   }
1048
1049   return handle;
1050 };
1051
1052
1053 Server.prototype._listen2 = function(address, port, addressType, backlog, fd) {
1054   debug('listen2', address, port, addressType, backlog);
1055   var self = this;
1056
1057   // If there is not yet a handle, we need to create one and bind.
1058   // In the case of a server sent via IPC, we don't need to do this.
1059   if (!self._handle) {
1060     debug('_listen2: create a handle');
1061     var rval = createServerHandle(address, port, addressType, fd);
1062     if (util.isNumber(rval)) {
1063       var error = errnoException(rval, 'listen');
1064       process.nextTick(function() {
1065         self.emit('error', error);
1066       });
1067       return;
1068     }
1069     self._handle = rval;
1070   } else {
1071     debug('_listen2: have a handle already');
1072   }
1073
1074   self._handle.onconnection = onconnection;
1075   self._handle.owner = self;
1076
1077   // Use a backlog of 512 entries. We pass 511 to the listen() call because
1078   // the kernel does: backlogsize = roundup_pow_of_two(backlogsize + 1);
1079   // which will thus give us a backlog of 512 entries.
1080   var err = self._handle.listen(backlog || 511);
1081
1082   if (err) {
1083     var ex = errnoException(err, 'listen');
1084     self._handle.close();
1085     self._handle = null;
1086     process.nextTick(function() {
1087       self.emit('error', ex);
1088     });
1089     return;
1090   }
1091
1092   // generate connection key, this should be unique to the connection
1093   this._connectionKey = addressType + ':' + address + ':' + port;
1094
1095   process.nextTick(function() {
1096     self.emit('listening');
1097   });
1098 };
1099
1100
1101 function listen(self, address, port, addressType, backlog, fd) {
1102   if (!cluster) cluster = require('cluster');
1103
1104   if (cluster.isMaster) {
1105     self._listen2(address, port, addressType, backlog, fd);
1106     return;
1107   }
1108
1109   cluster._getServer(self, address, port, addressType, fd, cb);
1110
1111   function cb(err, handle) {
1112     // EADDRINUSE may not be reported until we call listen(). To complicate
1113     // matters, a failed bind() followed by listen() will implicitly bind to
1114     // a random port. Ergo, check that the socket is bound to the expected
1115     // port before calling listen().
1116     //
1117     // FIXME(bnoordhuis) Doesn't work for pipe handles, they don't have a
1118     // getsockname() method. Non-issue for now, the cluster module doesn't
1119     // really support pipes anyway.
1120     if (err === 0 && port > 0 && handle.getsockname) {
1121       var out = {};
1122       err = handle.getsockname(out);
1123       if (err === 0 && port !== out.port)
1124         err = uv.UV_EADDRINUSE;
1125     }
1126
1127     if (err)
1128       return self.emit('error', errnoException(err, 'bind'));
1129
1130     self._handle = handle;
1131     self._listen2(address, port, addressType, backlog, fd);
1132   }
1133 }
1134
1135
1136 Server.prototype.listen = function() {
1137   var self = this;
1138
1139   var lastArg = arguments[arguments.length - 1];
1140   if (util.isFunction(lastArg)) {
1141     self.once('listening', lastArg);
1142   }
1143
1144   var port = toNumber(arguments[0]);
1145
1146   // The third optional argument is the backlog size.
1147   // When the ip is omitted it can be the second argument.
1148   var backlog = toNumber(arguments[1]) || toNumber(arguments[2]);
1149
1150   var TCP = process.binding('tcp_wrap').TCP;
1151
1152   if (arguments.length == 0 || util.isFunction(arguments[0])) {
1153     // Bind to a random port.
1154     listen(self, '0.0.0.0', 0, null, backlog);
1155
1156   } else if (arguments[0] && util.isObject(arguments[0])) {
1157     var h = arguments[0];
1158     if (h._handle) {
1159       h = h._handle;
1160     } else if (h.handle) {
1161       h = h.handle;
1162     }
1163     if (h instanceof TCP) {
1164       self._handle = h;
1165       listen(self, null, -1, -1, backlog);
1166     } else if (util.isNumber(h.fd) && h.fd >= 0) {
1167       listen(self, null, null, null, backlog, h.fd);
1168     } else {
1169       throw new Error('Invalid listen argument: ' + h);
1170     }
1171   } else if (isPipeName(arguments[0])) {
1172     // UNIX socket or Windows pipe.
1173     var pipeName = self._pipeName = arguments[0];
1174     listen(self, pipeName, -1, -1, backlog);
1175
1176   } else if (util.isUndefined(arguments[1]) ||
1177              util.isFunction(arguments[1]) ||
1178              util.isNumber(arguments[1])) {
1179     // The first argument is the port, no IP given.
1180     listen(self, '0.0.0.0', port, 4, backlog);
1181
1182   } else {
1183     // The first argument is the port, the second an IP.
1184     require('dns').lookup(arguments[1], function(err, ip, addressType) {
1185       if (err) {
1186         self.emit('error', err);
1187       } else {
1188         listen(self, ip || '0.0.0.0', port, ip ? addressType : 4, backlog);
1189       }
1190     });
1191   }
1192   return self;
1193 };
1194
1195 Server.prototype.address = function() {
1196   if (this._handle && this._handle.getsockname) {
1197     var out = {};
1198     var err = this._handle.getsockname(out);
1199     // TODO(bnoordhuis) Check err and throw?
1200     return out;
1201   } else if (this._pipeName) {
1202     return this._pipeName;
1203   } else {
1204     return null;
1205   }
1206 };
1207
1208 function onconnection(err, clientHandle) {
1209   var handle = this;
1210   var self = handle.owner;
1211
1212   debug('onconnection');
1213
1214   if (err) {
1215     self.emit('error', errnoException(err, 'accept'));
1216     return;
1217   }
1218
1219   if (self.maxConnections && self._connections >= self.maxConnections) {
1220     clientHandle.close();
1221     return;
1222   }
1223
1224   var socket = new Socket({
1225     handle: clientHandle,
1226     allowHalfOpen: self.allowHalfOpen
1227   });
1228   socket.readable = socket.writable = true;
1229
1230
1231   self._connections++;
1232   socket.server = self;
1233
1234   DTRACE_NET_SERVER_CONNECTION(socket);
1235   COUNTER_NET_SERVER_CONNECTION(socket);
1236   self.emit('connection', socket);
1237 }
1238
1239
1240 Server.prototype.getConnections = function(cb) {
1241   function end(err, connections) {
1242     process.nextTick(function() {
1243       cb(err, connections);
1244     });
1245   }
1246
1247   if (!this._usingSlaves) {
1248     return end(null, this._connections);
1249   }
1250
1251   // Poll slaves
1252   var left = this._slaves.length,
1253       total = this._connections;
1254
1255   function oncount(err, count) {
1256     if (err) {
1257       left = -1;
1258       return end(err);
1259     }
1260
1261     total += count;
1262     if (--left === 0) return end(null, total);
1263   }
1264
1265   this._slaves.forEach(function(slave) {
1266     slave.getConnections(oncount);
1267   });
1268 };
1269
1270
1271 Server.prototype.close = function(cb) {
1272   function onSlaveClose() {
1273     if (--left !== 0) return;
1274
1275     self._connections = 0;
1276     self._emitCloseIfDrained();
1277   }
1278
1279   if (!this._handle) {
1280     // Throw error. Follows net_legacy behaviour.
1281     throw new Error('Not running');
1282   }
1283
1284   if (cb) {
1285     this.once('close', cb);
1286   }
1287   this._handle.close();
1288   this._handle = null;
1289
1290   if (this._usingSlaves) {
1291     var self = this,
1292         left = this._slaves.length;
1293
1294     // Increment connections to be sure that, even if all sockets will be closed
1295     // during polling of slaves, `close` event will be emitted only once.
1296     this._connections++;
1297
1298     // Poll slaves
1299     this._slaves.forEach(function(slave) {
1300       slave.close(onSlaveClose);
1301     });
1302   } else {
1303     this._emitCloseIfDrained();
1304   }
1305
1306   return this;
1307 };
1308
1309 Server.prototype._emitCloseIfDrained = function() {
1310   debug('SERVER _emitCloseIfDrained');
1311   var self = this;
1312
1313   if (self._handle || self._connections) {
1314     debug('SERVER handle? %j   connections? %d',
1315           !!self._handle, self._connections);
1316     return;
1317   }
1318
1319   process.nextTick(function() {
1320     debug('SERVER: emit close');
1321     self.emit('close');
1322   });
1323 };
1324
1325
1326 Server.prototype.listenFD = util.deprecate(function(fd, type) {
1327   return this.listen({ fd: fd });
1328 }, 'listenFD is deprecated. Use listen({fd: <number>}).');
1329
1330 Server.prototype._setupSlave = function(socketList) {
1331   this._usingSlaves = true;
1332   this._slaves.push(socketList);
1333 };
1334
1335 Server.prototype.ref = function() {
1336   if (this._handle)
1337     this._handle.ref();
1338 };
1339
1340 Server.prototype.unref = function() {
1341   if (this._handle)
1342     this._handle.unref();
1343 };
1344
1345
1346 // TODO: isIP should be moved to the DNS code. Putting it here now because
1347 // this is what the legacy system did.
1348 exports.isIP = cares.isIP;
1349
1350
1351 exports.isIPv4 = function(input) {
1352   return exports.isIP(input) === 4;
1353 };
1354
1355
1356 exports.isIPv6 = function(input) {
1357   return exports.isIP(input) === 6;
1358 };
1359
1360
1361 if (process.platform === 'win32') {
1362   var simultaneousAccepts;
1363
1364   exports._setSimultaneousAccepts = function(handle) {
1365     if (util.isUndefined(handle)) {
1366       return;
1367     }
1368
1369     if (util.isUndefined(simultaneousAccepts)) {
1370       simultaneousAccepts = (process.env.NODE_MANY_ACCEPTS &&
1371                              process.env.NODE_MANY_ACCEPTS !== '0');
1372     }
1373
1374     if (handle._simultaneousAccepts !== simultaneousAccepts) {
1375       handle.setSimultaneousAccepts(simultaneousAccepts);
1376       handle._simultaneousAccepts = simultaneousAccepts;
1377     }
1378   };
1379 } else {
1380   exports._setSimultaneousAccepts = function(handle) {};
1381 }