net.js

// Copyright Joyent, Inc. and other Node contributors.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the
// "Software"), to deal in the Software without restriction, including
// without limitation the rights to use, copy, modify, merge, publish,
// distribute, sublicense, and/or sell copies of the Software, and to permit
// persons to whom the Software is furnished to do so, subject to the
// following conditions:
//
// The above copyright notice and this permission notice shall be included
// in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
// USE OR OTHER DEALINGS IN THE SOFTWARE.

'use strict';

const {
  ArrayIsArray,
  ArrayPrototypeIncludes,
  ArrayPrototypeIndexOf,
  ArrayPrototypePush,
  Boolean,
  FunctionPrototypeBind,
  FunctionPrototypeCall,
  MathMax,
  Number,
  NumberIsNaN,
  NumberParseInt,
  ObjectDefineProperty,
  ObjectSetPrototypeOf,
  Symbol,
  SymbolAsyncDispose,
  SymbolDispose,
} = primordials;

const EventEmitter = require('events');
const stream = require('stream');
let debug = require('internal/util/debuglog').debuglog('net', (fn) => {
  debug = fn;
});
const {
  kReinitializeHandle,
  isIP,
  isIPv4,
  isIPv6,
  normalizedArgsSymbol,
  makeSyncWrite,
} = require('internal/net');
const assert = require('internal/assert');
const {
  UV_EADDRINUSE,
  UV_EINVAL,
  UV_ENOTCONN,
  UV_ECANCELED,
  UV_ETIMEDOUT,
} = internalBinding('uv');

const { Buffer } = require('buffer');
const { ShutdownWrap } = internalBinding('stream_wrap');
const {
  TCP,
  TCPConnectWrap,
  constants: TCPConstants,
} = internalBinding('tcp_wrap');
const {
  Pipe,
  PipeConnectWrap,
  constants: PipeConstants,
} = internalBinding('pipe_wrap');
const {
  newAsyncId,
  defaultTriggerAsyncIdScope,
  symbols: { async_id_symbol, owner_symbol },
} = require('internal/async_hooks');
const {
  writevGeneric,
  writeGeneric,
  onStreamRead,
  kAfterAsyncWrite,
  kHandle,
  kUpdateTimer,
  setStreamTimeout,
  kBuffer,
  kBufferCb,
  kBufferGen,
} = require('internal/stream_base_commons');
const {
  codes: {
    ERR_INVALID_ADDRESS_FAMILY,
    ERR_INVALID_ARG_TYPE,
    ERR_INVALID_ARG_VALUE,
    ERR_INVALID_FD_TYPE,
    ERR_INVALID_IP_ADDRESS,
    ERR_INVALID_HANDLE_TYPE,
    ERR_SERVER_ALREADY_LISTEN,
    ERR_SERVER_NOT_RUNNING,
    ERR_SOCKET_CONNECTION_TIMEOUT,
    ERR_SOCKET_CLOSED,
    ERR_SOCKET_CLOSED_BEFORE_CONNECTION,
    ERR_MISSING_ARGS,
  },
  ErrnoException,
  ExceptionWithHostPort,
  genericNodeError,
  NodeAggregateError,
  UVExceptionWithHostPort,
} = require('internal/errors');
const { isUint8Array } = require('internal/util/types');
const { queueMicrotask } = require('internal/process/task_queues');
const { kEmptyObject, guessHandleType, promisify } = require('internal/util');
const {
  validateAbortSignal,
  validateBoolean,
  validateFunction,
  validateInt32,
  validateNumber,
  validatePort,
  validateString,
} = require('internal/validators');
const kLastWriteQueueSize = Symbol('lastWriteQueueSize');
const { getOptionValue } = require('internal/options');

// Lazy loaded to improve startup performance.
let cluster;
let dns;
let BlockList;
let SocketAddress;
let autoSelectFamilyDefault = getOptionValue('--network-family-autoselection');
let autoSelectFamilyAttemptTimeoutDefault = 250;

const { clearTimeout, setTimeout } = require('timers');
const { kTimeout } = require('internal/timers');

const DEFAULT_IPV4_ADDR = '0.0.0.0';
const DEFAULT_IPV6_ADDR = '::';

const isWindows = process.platform === 'win32';

const noop = () => {};

const kPerfHooksNetConnectContext = Symbol('kPerfHooksNetConnectContext');

const dc = require('diagnostics_channel');
const netClientSocketChannel = dc.channel('net.client.socket');
const netServerSocketChannel = dc.channel('net.server.socket');

const {
  hasObserver,
  startPerf,
  stopPerf,
} = require('internal/perf/observe');
const { getDefaultHighWaterMark } = require('internal/streams/state');

function getFlags(ipv6Only) {
  return ipv6Only === true ? TCPConstants.UV_TCP_IPV6ONLY : 0;
}

function createHandle(fd, is_server) {
  validateInt32(fd, 'fd', 0);
  const type = guessHandleType(fd);
  if (type === 'PIPE') {
    return new Pipe(
      is_server ? PipeConstants.SERVER : PipeConstants.SOCKET,
    );
  }

  if (type === 'TCP') {
    return new TCP(
      is_server ? TCPConstants.SERVER : TCPConstants.SOCKET,
    );
  }

  throw new ERR_INVALID_FD_TYPE(type);
}


function getNewAsyncId(handle) {
  return (!handle || typeof handle.getAsyncId !== 'function') ?
    newAsyncId() : handle.getAsyncId();
}


function isPipeName(s) {
  return typeof s === 'string' && toNumber(s) === false;
}

/**
 * Creates a new TCP or IPC server
 * @param {{
 *   allowHalfOpen?: boolean;
 *   pauseOnConnect?: boolean;
 *   }} [options]
 * @param {Function} [connectionListener]
 * @returns {Server}
 */

function createServer(options, connectionListener) {
  return new Server(options, connectionListener);
}


// Target API:
//
// let s = net.connect({port: 80, host: 'google.com'}, function() {
//   ...
// });
//
// There are various forms:
//
// connect(options, [cb])
// connect(port, [host], [cb])
// connect(path, [cb]);
//
function connect(...args) {
  const normalized = normalizeArgs(args);
  const options = normalized[0];
  debug('createConnection', normalized);
  const socket = new Socket(options);

  if (netClientSocketChannel.hasSubscribers) {
    netClientSocketChannel.publish({
      socket,
    });
  }
  if (options.timeout) {
    socket.setTimeout(options.timeout);
  }

  return socket.connect(normalized);
}

function getDefaultAutoSelectFamily() {
  return autoSelectFamilyDefault;
}

function setDefaultAutoSelectFamily(value) {
  validateBoolean(value, 'value');
  autoSelectFamilyDefault = value;
}

function getDefaultAutoSelectFamilyAttemptTimeout() {
  return autoSelectFamilyAttemptTimeoutDefault;
}

function setDefaultAutoSelectFamilyAttemptTimeout(value) {
  validateInt32(value, 'value', 1);

  if (value < 10) {
    value = 10;
  }

  autoSelectFamilyAttemptTimeoutDefault = value;
}

// Returns an array [options, cb], where options is an object,
// cb is either a function or null.
// Used to normalize arguments of Socket.prototype.connect() and
// Server.prototype.listen(). Possible combinations of parameters:
//   (options[...][, cb])
//   (path[...][, cb])
//   ([port][, host][...][, cb])
// For Socket.prototype.connect(), the [...] part is ignored
// For Server.prototype.listen(), the [...] part is [, backlog]
// but will not be handled here (handled in listen())
function normalizeArgs(args) {
  let arr;

  if (args.length === 0) {
    arr = [{}, null];
    arr[normalizedArgsSymbol] = true;
    return arr;
  }

  const arg0 = args[0];
  let options = {};
  if (typeof arg0 === 'object' && arg0 !== null) {
    // (options[...][, cb])
    options = arg0;
  } else if (isPipeName(arg0)) {
    // (path[...][, cb])
    options.path = arg0;
  } else {
    // ([port][, host][...][, cb])
    options.port = arg0;
    if (args.length > 1 && typeof args[1] === 'string') {
      options.host = args[1];
    }
  }

  const cb = args[args.length - 1];
  if (typeof cb !== 'function')
    arr = [options, null];
  else
    arr = [options, cb];

  arr[normalizedArgsSymbol] = true;
  return arr;
}


// Called when creating new Socket, or when re-using a closed Socket
function initSocketHandle(self) {
  self._undestroy();
  self._sockname = null;

  // Handle creation may be deferred to bind() or connect() time.
  if (self._handle) {
    self._handle[owner_symbol] = self;
    self._handle.onread = onStreamRead;
    self[async_id_symbol] = getNewAsyncId(self._handle);

    let userBuf = self[kBuffer];
    if (userBuf) {
      const bufGen = self[kBufferGen];
      if (bufGen !== null) {
        userBuf = bufGen();
        if (!isUint8Array(userBuf))
          return;
        self[kBuffer] = userBuf;
      }
      self._handle.useUserBuffer(userBuf);
    }
  }
}

function closeSocketHandle(self, isException, isCleanupPending = false) {
  if (self._handle) {
    self._handle.close(() => {
      debug('emit close');
      self.emit('close', isException);
      if (isCleanupPending) {
        self._handle.onread = noop;
        self._handle = null;
        self._sockname = null;
      }
    });
  }
}

const kBytesRead = Symbol('kBytesRead');
const kBytesWritten = Symbol('kBytesWritten');
const kSetNoDelay = Symbol('kSetNoDelay');
const kSetKeepAlive = Symbol('kSetKeepAlive');
const kSetKeepAliveInitialDelay = Symbol('kSetKeepAliveInitialDelay');

function Socket(options) {
  if (!(this instanceof Socket)) return new Socket(options);
  if (options?.objectMode) {
    throw new ERR_INVALID_ARG_VALUE(
      'options.objectMode',
      options.objectMode,
      'is not supported',
    );
  } else if (options?.readableObjectMode || options?.writableObjectMode) {
    throw new ERR_INVALID_ARG_VALUE(
      `options.${
        options.readableObjectMode ? 'readableObjectMode' : 'writableObjectMode'
      }`,
      options.readableObjectMode || options.writableObjectMode,
      'is not supported',
    );
  }
  if (typeof options?.keepAliveInitialDelay !== 'undefined') {
    validateNumber(
      options?.keepAliveInitialDelay, 'options.keepAliveInitialDelay',
    );

    if (options.keepAliveInitialDelay < 0) {
      options.keepAliveInitialDelay = 0;
    }
  }

  this.connecting = false;
  // Problem with this is that users can supply their own handle, that may not
  // have _handle.getAsyncId(). In this case an[async_id_symbol] should
  // probably be supplied by async_hooks.
  this[async_id_symbol] = -1;
  this._hadError = false;
  this[kHandle] = null;
  this._parent = null;
  this._host = null;
  this[kLastWriteQueueSize] = 0;
  this[kTimeout] = null;
  this[kBuffer] = null;
  this[kBufferCb] = null;
  this[kBufferGen] = null;
  this._closeAfterHandlingError = false;

  if (typeof options === 'number')
    options = { fd: options }; // Legacy interface.
  else
    options = { ...options };

  // Default to *not* allowing half open sockets.
  options.allowHalfOpen = Boolean(options.allowHalfOpen);
  // For backwards compat do not emit close on destroy.
  options.emitClose = false;
  options.autoDestroy = true;
  // Handle strings directly.
  options.decodeStrings = false;
  stream.Duplex.call(this, options);

  if (options.handle) {
    this._handle = options.handle; // private
    this[async_id_symbol] = getNewAsyncId(this._handle);
  } else if (options.fd !== undefined) {
    const { fd } = options;
    let err;

    // createHandle will throw ERR_INVALID_FD_TYPE if `fd` is not
    // a valid `PIPE` or `TCP` descriptor
    this._handle = createHandle(fd, false);

    err = this._handle.open(fd);

    // While difficult to fabricate, in some architectures
    // `open` may return an error code for valid file descriptors
    // which cannot be opened. This is difficult to test as most
    // un-openable fds will throw on `createHandle`
    if (err)
      throw new ErrnoException(err, 'open');

    this[async_id_symbol] = this._handle.getAsyncId();

    if ((fd === 1 || fd === 2) &&
        (this._handle instanceof Pipe) && isWindows) {
      // Make stdout and stderr blocking on Windows
      err = this._handle.setBlocking(true);
      if (err)
        throw new ErrnoException(err, 'setBlocking');

      this._writev = null;
      this._write = makeSyncWrite(fd);
      // makeSyncWrite adjusts this value like the original handle would, so
      // we need to let it do that by turning it into a writable, own
      // property.
      ObjectDefineProperty(this._handle, 'bytesWritten', {
        __proto__: null,
        value: 0, writable: true,
      });
    }
  }

  const onread = options.onread;
  if (onread !== null && typeof onread === 'object' &&
      (isUint8Array(onread.buffer) || typeof onread.buffer === 'function') &&
      typeof onread.callback === 'function') {
    if (typeof onread.buffer === 'function') {
      this[kBuffer] = true;
      this[kBufferGen] = onread.buffer;
    } else {
      this[kBuffer] = onread.buffer;
    }
    this[kBufferCb] = onread.callback;
  }

  this[kSetNoDelay] = Boolean(options.noDelay);
  this[kSetKeepAlive] = Boolean(options.keepAlive);
  this[kSetKeepAliveInitialDelay] = ~~(options.keepAliveInitialDelay / 1000);

  // Shut down the socket when we're finished with it.
  this.on('end', onReadableStreamEnd);

  initSocketHandle(this);

  this._pendingData = null;
  this._pendingEncoding = '';

  // If we have a handle, then start the flow of data into the
  // buffer.  if not, then this will happen when we connect
  if (this._handle && options.readable !== false) {
    if (options.pauseOnCreate) {
      // Stop the handle from reading and pause the stream
      this._handle.reading = false;
      this._handle.readStop();
      this.readableFlowing = false;
    } else if (!options.manualStart) {
      this.read(0);
    }
  }

  if (options.signal) {
    addClientAbortSignalOption(this, options);
  }

  // Reserve properties
  this.server = null;
  this._server = null;

  // Used after `.destroy()`
  this[kBytesRead] = 0;
  this[kBytesWritten] = 0;
}
ObjectSetPrototypeOf(Socket.prototype, stream.Duplex.prototype);
ObjectSetPrototypeOf(Socket, stream.Duplex);

// Refresh existing timeouts.
Socket.prototype._unrefTimer = function _unrefTimer() {
  for (let s = this; s !== null; s = s._parent) {
    if (s[kTimeout])
      s[kTimeout].refresh();
  }
};


// The user has called .end(), and all the bytes have been
// sent out to the other side.
Socket.prototype._final = function(cb) {
  // If still connecting - defer handling `_final` until 'connect' will happen
  if (this.connecting) {
    debug('_final: not yet connected');
    return this.once('connect', () => this._final(cb));
  }

  if (!this._handle)
    return cb();

  debug('_final: not ended, call shutdown()');

  const req = new ShutdownWrap();
  req.oncomplete = afterShutdown;
  req.handle = this._handle;
  req.callback = cb;
  const err = this._handle.shutdown(req);

  if (err === 1 || err === UV_ENOTCONN)  // synchronous finish
    return cb();
  else if (err !== 0)
    return cb(new ErrnoException(err, 'shutdown'));
};

function afterShutdown() {
  const self = this.handle[owner_symbol];

  debug('afterShutdown destroyed=%j', self.destroyed);

  this.callback();
}

// Provide a better error message when we call end() as a result
// of the other side sending a FIN.  The standard 'write after end'
// is overly vague, and makes it seem like the user's code is to blame.
function writeAfterFIN(chunk, encoding, cb) {
  if (!this.writableEnded) {
    return stream.Duplex.prototype.write.call(this, chunk, encoding, cb);
  }

  if (typeof encoding === 'function') {
    cb = encoding;
    encoding = null;
  }

  const er = genericNodeError(
    'This socket has been ended by the other party',
    { code: 'EPIPE' },
  );
  if (typeof cb === 'function') {
    defaultTriggerAsyncIdScope(this[async_id_symbol], process.nextTick, cb, er);
  }
  this.destroy(er);

  return false;
}

Socket.prototype.setTimeout = setStreamTimeout;


Socket.prototype._onTimeout = function() {
  const handle = this._handle;
  const lastWriteQueueSize = this[kLastWriteQueueSize];
  if (lastWriteQueueSize > 0 && handle) {
    // `lastWriteQueueSize !== writeQueueSize` means there is
    // an active write in progress, so we suppress the timeout.
    const { writeQueueSize } = handle;
    if (lastWriteQueueSize !== writeQueueSize) {
      this[kLastWriteQueueSize] = writeQueueSize;
      this._unrefTimer();
      return;
    }
  }
  debug('_onTimeout');
  this.emit('timeout');
};


Socket.prototype.setNoDelay = function(enable) {
  // Backwards compatibility: assume true when `enable` is omitted
  enable = Boolean(enable === undefined ? true : enable);

  if (!this._handle) {
    this[kSetNoDelay] = enable;
    return this;
  }

  if (this._handle.setNoDelay && enable !== this[kSetNoDelay]) {
    this[kSetNoDelay] = enable;
    this._handle.setNoDelay(enable);
  }

  return this;
};


Socket.prototype.setKeepAlive = function(enable, initialDelayMsecs) {
  enable = Boolean(enable);
  const initialDelay = ~~(initialDelayMsecs / 1000);

  if (!this._handle) {
    this[kSetKeepAlive] = enable;
    this[kSetKeepAliveInitialDelay] = initialDelay;
    return this;
  }

  if (!this._handle.setKeepAlive) {
    return this;
  }

  if (enable !== this[kSetKeepAlive] ||
      (
        enable &&
        this[kSetKeepAliveInitialDelay] !== initialDelay
      )
  ) {
    this[kSetKeepAlive] = enable;
    this[kSetKeepAliveInitialDelay] = initialDelay;
    this._handle.setKeepAlive(enable, initialDelay);
  }

  return this;
};


Socket.prototype.address = function() {
  return this._getsockname();
};


ObjectDefineProperty(Socket.prototype, '_connecting', {
  __proto__: null,
  get: function() {
    return this.connecting;
  },
});

ObjectDefineProperty(Socket.prototype, 'pending', {
  __proto__: null,
  get() {
    return !this._handle || this.connecting;
  },
  configurable: true,
});


ObjectDefineProperty(Socket.prototype, 'readyState', {
  __proto__: null,
  get: function() {
    if (this.connecting) {
      return 'opening';
    } else if (this.readable && this.writable) {
      return 'open';
    } else if (this.readable && !this.writable) {
      return 'readOnly';
    } else if (!this.readable && this.writable) {
      return 'writeOnly';
    }
    return 'closed';
  },
});


ObjectDefineProperty(Socket.prototype, 'bufferSize', {
  __proto__: null,
  get: function() {
    if (this._handle) {
      return this.writableLength;
    }
  },
});

ObjectDefineProperty(Socket.prototype, kUpdateTimer, {
  __proto__: null,
  get: function() {
    return this._unrefTimer;
  },
});


function tryReadStart(socket) {
  // Not already reading, start the flow
  debug('Socket._handle.readStart');
  socket._handle.reading = true;
  const err = socket._handle.readStart();
  if (err)
    socket.destroy(new ErrnoException(err, 'read'));
}

// Just call handle.readStart until we have enough in the buffer
Socket.prototype._read = function(n) {
  debug(
    '_read - n', n,
    'isConnecting?', !!this.connecting,
    'hasHandle?', !!this._handle,
  );

  if (this.connecting || !this._handle) {
    debug('_read wait for connection');
    this.once('connect', () => this._read(n));
  } else if (!this._handle.reading) {
    tryReadStart(this);
  }
};


Socket.prototype.end = function(data, encoding, callback) {
  stream.Duplex.prototype.end.call(this,
                                   data, encoding, callback);
  return this;
};

Socket.prototype.resetAndDestroy = function() {
  if (this._handle) {
    if (!(this._handle instanceof TCP))
      throw new ERR_INVALID_HANDLE_TYPE();
    if (this.connecting) {
      debug('reset wait for connection');
      this.once('connect', () => this._reset());
    } else {
      this._reset();
    }
  } else {
    this.destroy(new ERR_SOCKET_CLOSED());
  }
  return this;
};

Socket.prototype.pause = function() {
  if (this[kBuffer] && !this.connecting && this._handle &&
      this._handle.reading) {
    this._handle.reading = false;
    if (!this.destroyed) {
      const err = this._handle.readStop();
      if (err)
        this.destroy(new ErrnoException(err, 'read'));
    }
  }
  return stream.Duplex.prototype.pause.call(this);
};


Socket.prototype.resume = function() {
  if (this[kBuffer] && !this.connecting && this._handle &&
      !this._handle.reading) {
    tryReadStart(this);
  }
  return stream.Duplex.prototype.resume.call(this);
};


Socket.prototype.read = function(n) {
  if (this[kBuffer] && !this.connecting && this._handle &&
      !this._handle.reading) {
    tryReadStart(this);
  }
  return stream.Duplex.prototype.read.call(this, n);
};


// Called when the 'end' event is emitted.
function onReadableStreamEnd() {
  if (!this.allowHalfOpen) {
    this.write = writeAfterFIN;
  }
}


Socket.prototype.destroySoon = function() {
  if (this.writable)
    this.end();

  if (this.writableFinished)
    this.destroy();
  else
    this.once('finish', this.destroy);
};


Socket.prototype._destroy = function(exception, cb) {
  debug('destroy');

  this.connecting = false;

  for (let s = this; s !== null; s = s._parent) {
    clearTimeout(s[kTimeout]);
  }

  debug('close');
  if (this._handle) {
    if (this !== process.stderr)
      debug('close handle');
    const isException = exception ? true : false;
    // `bytesRead` and `kBytesWritten` should be accessible after `.destroy()`
    this[kBytesRead] = this._handle.bytesRead;
    this[kBytesWritten] = this._handle.bytesWritten;

    if (this.resetAndClosing) {
      this.resetAndClosing = false;
      const err = this._handle.reset(() => {
        debug('emit close');
        this.emit('close', isException);
      });
      if (err)
        this.emit('error', new ErrnoException(err, 'reset'));
    } else if (this._closeAfterHandlingError) {
      // Enqueue closing the socket as a microtask, so that the socket can be
      // accessible when an `error` event is handled in the `next tick queue`.
      queueMicrotask(() => closeSocketHandle(this, isException, true));
    } else {
      closeSocketHandle(this, isException);
    }

    if (!this._closeAfterHandlingError) {
      this._handle.onread = noop;
      this._handle = null;
      this._sockname = null;
    }
    cb(exception);
  } else {
    cb(exception);
    process.nextTick(emitCloseNT, this);
  }

  if (this._server) {
    debug('has server');
    this._server._connections--;
    if (this._server._emitCloseIfDrained) {
      this._server._emitCloseIfDrained();
    }
  }
};

Socket.prototype._reset = function() {
  debug('reset connection');
  this.resetAndClosing = true;
  return this.destroy();
};

Socket.prototype._getpeername = function() {
  if (!this._handle || !this._handle.getpeername || this.connecting) {
    return this._peername || {};
  } else if (!this._peername) {
    const out = {};
    const err = this._handle.getpeername(out);
    if (err) return out;
    this._peername = out;
  }
  return this._peername;
};

function protoGetter(name, callback) {
  ObjectDefineProperty(Socket.prototype, name, {
    __proto__: null,
    configurable: false,
    enumerable: true,
    get: callback,
  });
}

protoGetter('bytesRead', function bytesRead() {
  return this._handle ? this._handle.bytesRead : this[kBytesRead];
});

protoGetter('remoteAddress', function remoteAddress() {
  return this._getpeername().address;
});

protoGetter('remoteFamily', function remoteFamily() {
  return this._getpeername().family;
});

protoGetter('remotePort', function remotePort() {
  return this._getpeername().port;
});


Socket.prototype._getsockname = function() {
  if (!this._handle || !this._handle.getsockname) {
    return {};
  } else if (!this._sockname) {
    this._sockname = {};
    // FIXME(bnoordhuis) Throw when the return value is not 0?
    this._handle.getsockname(this._sockname);
  }
  return this._sockname;
};


protoGetter('localAddress', function localAddress() {
  return this._getsockname().address;
});


protoGetter('localPort', function localPort() {
  return this._getsockname().port;
});

protoGetter('localFamily', function localFamily() {
  return this._getsockname().family;
});

Socket.prototype[kAfterAsyncWrite] = function() {
  this[kLastWriteQueueSize] = 0;
};

Socket.prototype._writeGeneric = function(writev, data, encoding, cb) {
  // If we are still connecting, then buffer this for later.
  // The Writable logic will buffer up any more writes while
  // waiting for this one to be done.
  if (this.connecting) {
    this._pendingData = data;
    this._pendingEncoding = encoding;
    this.once('connect', function connect() {
      this.off('close', onClose);
      this._writeGeneric(writev, data, encoding, cb);
    });
    function onClose() {
      cb(new ERR_SOCKET_CLOSED_BEFORE_CONNECTION());
    }
    this.once('close', onClose);
    return;
  }
  this._pendingData = null;
  this._pendingEncoding = '';

  if (!this._handle) {
    cb(new ERR_SOCKET_CLOSED());
    return false;
  }

  this._unrefTimer();

  let req;
  if (writev)
    req = writevGeneric(this, data, cb);
  else
    req = writeGeneric(this, data, encoding, cb);
  if (req.async)
    this[kLastWriteQueueSize] = req.bytes;
};


Socket.prototype._writev = function(chunks, cb) {
  this._writeGeneric(true, chunks, '', cb);
};


Socket.prototype._write = function(data, encoding, cb) {
  this._writeGeneric(false, data, encoding, cb);
};


// Legacy alias. Having this is probably being overly cautious, but it doesn't
// really hurt anyone either. This can probably be removed safely if desired.
protoGetter('_bytesDispatched', function _bytesDispatched() {
  return this._handle ? this._handle.bytesWritten : this[kBytesWritten];
});

protoGetter('bytesWritten', function bytesWritten() {
  let bytes = this._bytesDispatched;
  const data = this._pendingData;
  const encoding = this._pendingEncoding;
  const writableBuffer = this.writableBuffer;

  if (!writableBuffer)
    return undefined;

  for (const el of writableBuffer) {
    bytes += el.chunk instanceof Buffer ?
      el.chunk.length :
      Buffer.byteLength(el.chunk, el.encoding);
  }

  if (ArrayIsArray(data)) {
    // Was a writev, iterate over chunks to get total length
    for (let i = 0; i < data.length; i++) {
      const chunk = data[i];

      if (data.allBuffers || chunk instanceof Buffer)
        bytes += chunk.length;
      else
        bytes += Buffer.byteLength(chunk.chunk, chunk.encoding);
    }
  } else if (data) {
    // Writes are either a string or a Buffer.