// 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,
ArrayPrototypeSlice,
ArrayPrototypeSplice,
Boolean,
FunctionPrototypeBind,
FunctionPrototypeCall,
MathMax,
Number,
NumberIsNaN,
NumberParseInt,
ObjectDefineProperty,
ObjectSetPrototypeOf,
Symbol,
SymbolAsyncDispose,
SymbolAsyncIterator,
SymbolDispose,
} = primordials;
const EventEmitter = require('events');
const { addAbortListener } = require('internal/events/abort_listener');
const stream = require('stream');
let debug = require('internal/util/debuglog').debuglog('net', (fn) => {
debug = fn;
});
const {
kReinitializeHandle,
kSetNoDelay,
kSetKeepAlive,
kSetKeepAliveInitialDelay,
kSetKeepAliveInterval,
kSetKeepAliveCount,
isIP,
isIPv4,
isIPv6,
normalizedArgsSymbol,
makeSyncWrite,
} = require('internal/net');
const assert = require('internal/assert');
const {
UV_EADDRINUSE,
UV_EBADF,
UV_EINVAL,
UV_ENOTCONN,
UV_ETIMEDOUT,
} = internalBinding('uv');
const { convertIpv6StringToBuffer } = internalBinding('cares_wrap');
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 {
ErrnoException,
ExceptionWithHostPort,
NodeAggregateError,
UVExceptionWithHostPort,
codes: {
ERR_INVALID_ADDRESS_FAMILY,
ERR_INVALID_ARG_TYPE,
ERR_INVALID_ARG_VALUE,
ERR_INVALID_FD_TYPE,
ERR_INVALID_HANDLE_TYPE,
ERR_INVALID_IP_ADDRESS,
ERR_IP_BLOCKED,
ERR_MISSING_ARGS,
ERR_SERVER_ALREADY_LISTEN,
ERR_SERVER_NOT_RUNNING,
ERR_SOCKET_CLOSED,
ERR_SOCKET_CLOSED_BEFORE_CONNECTION,
ERR_SOCKET_CONNECTION_TIMEOUT,
ERR_SOCKET_HANDLE_ADOPTED,
ERR_WORKER_HANDLE_NOT_TRANSFERABLE,
},
genericNodeError,
} = require('internal/errors');
const {
markTransferMode,
kDeserialize,
kTransfer,
kTransferList,
} = require('internal/worker/js_transferable');
const { isUint8Array } = require('internal/util/types');
const { queueMicrotask } = require('internal/process/task_queues');
const {
deprecate,
guessHandleType,
isWindows,
kEmptyObject,
promisify,
} = require('internal/util');
const {
validateAbortSignal,
validateBoolean,
validateFunction,
validateInt32,
validateNumber,
validateObject,
validatePort,
validateString,
validateStringWithoutNullBytes,
} = 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 netPromises;
let kFirstEventParam;
let autoSelectFamilyDefault = getOptionValue('--network-family-autoselection');
let autoSelectFamilyAttemptTimeoutDefault = getOptionValue('--network-family-autoselection-attempt-timeout');
const { clearTimeout, setTimeout } = require('timers');
const { kTimeout } = require('internal/timers');
const DEFAULT_IPV4_ADDR = '0.0.0.0';
const DEFAULT_IPV6_ADDR = '::';
const noop = () => {};
const kPerfHooksNetConnectContext = Symbol('kPerfHooksNetConnectContext');
const kAutoSelectFamilyContext = Symbol('kAutoSelectFamilyContext');
const kSocketHandleUnused = 0;
const kSocketHandlePending = 1;
const kSocketHandleFailed = 2;
const dc = require('diagnostics_channel');
const netClientSocketChannel = dc.channel('net.client.socket');
const netServerSocketChannel = dc.channel('net.server.socket');
const netServerListen = dc.tracingChannel('net.server.listen');
const {
hasObserver,
startPerf,
stopPerf,
} = require('internal/perf/observe');
const { getDefaultHighWaterMark } = require('internal/streams/state');
function getFlags(options) {
let flags = 0;
if (options.ipv6Only === true) {
flags |= TCPConstants.UV_TCP_IPV6ONLY;
}
if (options.reusePort === true) {
flags |= TCPConstants.UV_TCP_REUSEPORT;
}
return flags;
}
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 (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 kSetTOS = Symbol('kSetTOS');
// Marks a Socket whose handle is an adopted, already-bound BoundSocket.
const kBoundSource = Symbol('kBoundSource');
// Internal: adopt the underlying handle, transferring ownership to a
// Server/Socket.
const kBoundSocketConsume = Symbol('kBoundSocketConsume');
// Internal: read a pipe BoundSocket's bound path before adoption (undefined for
// a TCP BoundSocket).
const kBoundSocketPath = Symbol('kBoundSocketPath');
// The source path of an adopted, bound client pipe, surfaced as localAddress.
const kBoundPath = Symbol('kBoundPath');
const isLinux = process.platform === 'linux';
// Internal: construct an empty BoundSocket shell during postMessage()
// deserialization; the transferred handle is installed by [kDeserialize].
const kBoundSocketDeserialize = Symbol('kBoundSocketDeserialize');
// A role-neutral wrapper over a synchronously bound libuv handle: bound to a
// local address (a numeric IP literal for TCP, or a filesystem/abstract path
// for a unix-domain socket via { path }) but neither listening nor connecting
// until adopted by exactly one Server (server.listen) or Socket
// (new net.Socket({ handle })). Adoption transfers ownership; an un-adopted
// handle must be closed by the caller. bind(2) is non-blocking, so binding
// happens inline and errors throw synchronously. No DNS is performed.
class BoundSocket {
#handle = null;
#address = {};
#path;
constructor(options = kEmptyObject) {
// An un-adopted BoundSocket can be moved to another thread by listing it
// in the transferList of a worker_threads postMessage() call. See
// [kTransfer]().
markTransferMode(this, false, true);
if (options === kBoundSocketDeserialize) {
return;
}
validateObject(options, 'options');
if (options.path !== undefined) {
this.#bindPipe(options);
return;
}
const port = validatePort(options.port ?? 0, 'options.port');
const ipv6Only = options.ipv6Only ?? false;
validateBoolean(ipv6Only, 'options.ipv6Only');
const reusePort = options.reusePort ?? false;
validateBoolean(reusePort, 'options.reusePort');
let { host } = options;
let addressType;
if (host === undefined || host === null) {
host = ipv6Only ? DEFAULT_IPV6_ADDR : DEFAULT_IPV4_ADDR;
addressType = ipv6Only ? 6 : 4;
} else {
validateString(host, 'options.host');
addressType = isIP(host);
if (addressType === 0) {
throw new ERR_INVALID_ARG_VALUE(
'options.host', host,
'must be a numeric IP address; net.BoundSocket does not perform DNS resolution');
}
}
let flags = 0;
if (ipv6Only) {
flags |= TCPConstants.UV_TCP_IPV6ONLY;
}
if (reusePort) {
flags |= TCPConstants.UV_TCP_REUSEPORT;
}
const handle = new TCP(TCPConstants.SOCKET);
let err = addressType === 6 ?
handle.bind6(host, port, flags) :
handle.bind(host, port, flags);
// EADDRINUSE is deferred by libuv's uv_tcp_bind(), which returns 0 and
// surfaces it on the next getsockname/listen/connect. Force getsockname()
// now so conflicts throw synchronously; the address is cached since the
// handle is immutable.
if (err === 0) {
err = handle.getsockname(this.#address);
}
if (err) {
handle.close();
throw new ExceptionWithHostPort(err, 'bind', host, port);
}
this.#handle = handle;
}
// Bind a named unix-domain socket (or Windows named pipe). A leading '\0'
// selects the Linux abstract namespace. path is mutually exclusive with the
// TCP options; uv_pipe_bind is synchronous so conflicts throw here.
#bindPipe(options) {
const { path, host, port, ipv6Only, reusePort } = options;
if (host !== undefined || port !== undefined ||
ipv6Only !== undefined || reusePort !== undefined) {
throw new ERR_INVALID_ARG_VALUE(
'options', options,
'path is mutually exclusive with host, port, ipv6Only, and reusePort');
}
validateString(path, 'options.path');
if (path[0] === '\0' && !isLinux) {
throw new ERR_INVALID_ARG_VALUE(
'options.path', path,
'abstract socket paths are only supported on Linux');
}
const handle = new Pipe(PipeConstants.SOCKET);
const err = handle.bind(path);
if (err) {
handle.close();
throw new ErrnoException(err, 'bind');
}
this.#handle = handle;
this.#path = path;
}
// The bound local endpoint: an { address, family, port } object for TCP, or
// the path string for a pipe (matching net.Server.address()). Resolved at
// construction; a TCP bind of port 0 reflects the OS-assigned port.
address() {
if (this.#handle === null) {
throw new ERR_SOCKET_HANDLE_ADOPTED();
}
return this.#path ?? this.#address;
}
get [kBoundSocketPath]() {
return this.#path;
}
// The underlying OS file descriptor, or -1 where sockets have none (Windows).
// Ownership stays with the BoundSocket until it is adopted, so the caller must
// not close it.
fd() {
if (this.#handle === null) {
throw new ERR_SOCKET_HANDLE_ADOPTED();
}
return this.#handle.fd;
}
// Release the socket if it is never adopted, preventing an fd/handle leak.
close() {
if (this.#handle === null) {
throw new ERR_SOCKET_HANDLE_ADOPTED();
}
this.#handle.close();
this.#handle = null;
}
// Enables `using bound = new net.BoundSocket(...)`: closes an un-adopted
// handle and is a no-op once the handle has been adopted or closed.
[SymbolDispose]() {
if (this.#handle !== null) {
this.#handle.close();
this.#handle = null;
}
}
[kBoundSocketConsume]() {
if (this.#handle === null) {
throw new ERR_SOCKET_HANDLE_ADOPTED();
}
const handle = this.#handle;
this.#handle = null;
return handle;
}
// Reports whether this is a pipe (unix-domain) bind rather than TCP. Its mere
// presence on the prototype ('isPipe' in net.BoundSocket.prototype) is the
// capability signal that this build honors { path } instead of a TCP port.
get isPipe() {
return this.#path !== undefined;
}
// A BoundSocket can be transferred only while it still owns its handle,
// i.e. before it has been adopted, closed or already transferred. Only TCP
// binds are transferable; pipe handles cannot move between event loops.
#assertTransferable() {
if (this.#handle === null || !(this.#handle instanceof TCP)) {
throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.BoundSocket');
}
}
[kTransferList]() {
this.#assertTransferable();
return [this.#handle];
}
[kTransfer]() {
this.#assertTransferable();
const handle = this.#handle;
// Detach the handle; the messaging layer takes ownership of it via
// TCPWrap::TransferForMessaging(). Further use on the sending side throws
// ERR_SOCKET_HANDLE_ADOPTED, as after adoption.
this.#handle = null;
return {
data: { handle },
deserializeInfo: 'net:_TransferredBoundSocket',
};
}
[kDeserialize](data) {
const handle = data?.handle;
if (handle == null || !(handle instanceof TCP)) {
throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.BoundSocket');
}
// Re-derive the bound address from the transferred handle rather than
// trusting serialized state.
const err = handle.getsockname(this.#address);
if (err) {
handle.close();
throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.BoundSocket');
}
this.#handle = handle;
}
}
// Deserialization target for a transferred BoundSocket: constructs an empty
// shell without binding a new socket. Internal, not part of the public API.
function _TransferredBoundSocket() {
return new BoundSocket(kBoundSocketDeserialize);
}
_TransferredBoundSocket.prototype = BoundSocket.prototype;
function Socket(options) {
if (!(this instanceof Socket)) return new Socket(options);
// A connected TCP Socket can be moved to another thread by listing it in the
// transferList of a worker_threads postMessage() call. See [kTransfer]().
markTransferMode(this, false, true);
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 (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);
// An adopted BoundSocket is bound but not connected: defer the read flow
// until connect() completes.
let boundNotConnected = false;
if (options.handle) {
if (options.handle instanceof BoundSocket) {
const boundPath = options.handle[kBoundSocketPath];
this._handle = options.handle[kBoundSocketConsume]();
this[kBoundSource] = true;
boundNotConnected = true;
// A bound client pipe owns a source path; surface it as localAddress.
if (boundPath !== undefined) {
this[kBoundPath] = boundPath;
// uv_pipe_bind() only assigns the fd; it does not open the stream, so a
// later connect() would leave the handle without its readable/writable
// flags and unusable. Re-open the already-bound fd (idempotent, sets the
// flags) so the adopted client pipe connects to a working stream.
if (this._handle.fd >= 0) {
const err = this._handle.open(this._handle.fd);
if (err) {
throw new ErrnoException(err, 'open');
}
}
}
} else {
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);
if (options.typeOfService !== undefined) {
validateInt32(options.typeOfService, 'options.typeOfService', 0, 255);
}
this[kSetTOS] = options.typeOfService;
// 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 && !boundNotConnected) {
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;
if (options.blockList) {
if (!module.exports.BlockList.isBlockList(options.blockList)) {
throw new ERR_INVALID_ARG_TYPE('options.blockList', 'net.BlockList', options.blockList);
}
this.blockList = options.blockList;
}
}
ObjectSetPrototypeOf(Socket.prototype, stream.Duplex.prototype);
ObjectSetPrototypeOf(Socket, stream.Duplex);
// Refresh existing timeouts.
Socket.prototype._unrefTimer = function _unrefTimer() {
// `_parent` may be null; we use a loose `!= null` check in case external
// code sets it to undefined.
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,
intervalMsecs, count) {
if (enable !== null && typeof enable === 'object') {
const options = enable;
enable = options.enable;
initialDelayMsecs = options.initialDelay;
intervalMsecs = options.interval;
count = options.count;
}
enable = Boolean(enable);
const initialDelay = ~~(initialDelayMsecs / 1000);
const interval = intervalMsecs === undefined ?
undefined : ~~(intervalMsecs / 1000);
if (!this._handle) {
this[kSetKeepAlive] = enable;
this[kSetKeepAliveInitialDelay] = initialDelay;
this[kSetKeepAliveInterval] = interval;
this[kSetKeepAliveCount] = count;
return this;
}
if (!this._handle.setKeepAlive) {
return this;
}
if (enable !== this[kSetKeepAlive] ||
(
enable &&
(this[kSetKeepAliveInitialDelay] !== initialDelay ||
this[kSetKeepAliveInterval] !== interval ||
this[kSetKeepAliveCount] !== count)
)
) {
this[kSetKeepAlive] = enable;
this[kSetKeepAliveInitialDelay] = initialDelay;
this[kSetKeepAliveInterval] = interval;
this[kSetKeepAliveCount] = count;
this._handle.setKeepAlive(enable, initialDelay, interval, count);
}
return this;
};
Socket.prototype.setTypeOfService = function(tos) {
if (NumberIsNaN(tos)) {
throw new ERR_INVALID_ARG_TYPE('tos', 'number', tos);
}
validateInt32(tos, 'tos', 0, 255);
if (!this._handle) {
this[kSetTOS] = tos;
return this;
}
if (!this._handle.setTypeOfService) {
this[kSetTOS] = tos;
return this;
}
if (tos !== this[kSetTOS]) {
this[kSetTOS] = tos;
const err = this._handle.setTypeOfService(tos);
// On Windows, setting TOS is often restricted or returns error codes even if partially applied.
// We treat this as a "best effort" operation and do not throw on Windows.
if (err && !isWindows) {
throw new ErrnoException(err, 'setTypeOfService');
}
}
return this;
};
Socket.prototype.getTypeOfService = function() {
if (!this._handle) {
// Return cached value if set, otherwise default to 0
return this[kSetTOS] !== undefined ? this[kSetTOS] : 0;
}
if (!this._handle.getTypeOfService) {
return this[kSetTOS] !== undefined ? this[kSetTOS] : 0;
}
const res = this._handle.getTypeOfService();
if (typeof res === 'number' && res < 0) {
// On Windows, getsockopt(IP_TOS) often fails. In that case, fall back
// to the cached value we attempted to set, or 0.
if (isWindows) {
return this[kSetTOS] !== undefined ? this[kSetTOS] : 0;
}
throw new ErrnoException(res, 'getTypeOfService');
}
return res;
};
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?.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;
const context = this[kAutoSelectFamilyContext];
if (context) {
closeConnectionAttempts(context);
this[kAutoSelectFamilyContext] = undefined;
}
// `_parent` may be null; we use a loose `!= null` check in case external
// code sets it to undefined.
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() {
// An adopted bound client pipe has no handle getsockname; its source path was
// captured at adoption and stays authoritative across the connect reset.
if (this[kBoundPath] !== undefined) {
return { address: this[kBoundPath] };
}
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.
if (typeof data !== 'string')
bytes += data.length;
else
bytes += Buffer.byteLength(data, encoding);
}
return bytes;
});
function checkBindError(err, port, handle) {
// EADDRINUSE may not be reported until we call listen() or connect().
// To complicate matters, a failed bind() followed by listen() or connect()
// will implicitly bind to a random port. Ergo, check that the socket is
// bound to the expected port before calling listen() or connect().
//
// FIXME(bnoordhuis) Doesn't work for pipe handles, they don't have a
// getsockname() method. Non-issue for now, the cluster module doesn't
// really support pipes anyway.
if (err === 0 && port > 0 && handle.getsockname) {
const out = {};
err = handle.getsockname(out);
if (err === 0 && port !== out.port) {
debug(`checkBindError, bound to ${out.port} instead of ${port}`);
err = UV_EADDRINUSE;
}
}
return err;
}
function internalConnect(
self, address, port, addressType, localAddress, localPort, flags) {
// TODO return promise from Socket.prototype.connect which
// wraps _connectReq.
assert(self.connecting);
let err;
if (localAddress || localPort) {
if (addressType === 4) {
localAddress ||= DEFAULT_IPV4_ADDR;
err = self._handle.bind(localAddress, localPort);
} else { // addressType === 6
localAddress ||= DEFAULT_IPV6_ADDR;
err = self._handle.bind6(localAddress, localPort, flags);
}
debug('connect: binding to localAddress: %s and localPort: %d (addressType: %d)',
localAddress, localPort, addressType);
err = checkBindError(err, localPort, self._handle);
if (err) {
const ex = new ExceptionWithHostPort(err, 'bind', localAddress, localPort);
self.destroy(ex);
return;
}
}
debug('connect: attempting to connect to %s:%d (addressType: %d)', address, port, addressType);
self.emit('connectionAttempt', address, port, addressType);
if (addressType === 6 || addressType === 4) {
if (self.blockList?.check(address, `ipv${addressType}`)) {
self.destroy(new ERR_IP_BLOCKED(address));
return;
}
const req = new TCPConnectWrap();
req.oncomplete = afterConnect;
req.address = address;
req.port = port;
req.localAddress = localAddress;
req.localPort = localPort;
req.addressType = addressType;
if (addressType === 4)
err = self._handle.connect(req, address, port);
else
err = self._handle.connect6(req, address, port);
} else {
const req = new PipeConnectWrap();
req.address = address;
req.oncomplete = afterConnect;
err = self._handle.connect(req, address);
}
if (err) {
const sockname = self._getsockname();
let details;
if (sockname) {
details = sockname.address + ':' + sockname.port;
}
const ex = new ExceptionWithHostPort(err, 'connect', address, port, details);
self.destroy(ex);
} else if ((addressType === 6 || addressType === 4) && hasObserver('net')) {
startPerf(self, kPerfHooksNetConnectContext, { type: 'net', name: 'connect', detail: { host: address, port } });
}
}
function closeConnectionAttempts(context) {
clearTimeout(context[kTimeout]);
context[kTimeout] = null;
context.done = true;
for (let i = 0; i < context.pending.length; i++) {
const { handle, req, own } = context.pending[i];
req.oncomplete = undefined;
// The socket's own handle is closed by whoever replaces or destroys it.
if (!own) handle.close();
}
context.pending.length = 0;
}
function removePendingAttempt(context, req) {
const { pending } = context;
for (let i = 0; i < pending.length; i++) {
if (pending[i].req === req) {
const attempt = pending[i];
ArrayPrototypeSplice(pending, i, 1);
return attempt;
}
}
}
function connectMultipleError(context) {
// Errors are stored by attempt index to preserve the attempt order.
const errors = [];
for (let i = 0; i < context.errors.length; i++) {
if (context.errors[i] !== undefined) ArrayPrototypePush(errors, context.errors[i]);
}
return errors.length === 0 ? new ERR_SOCKET_CONNECTION_TIMEOUT() : new NodeAggregateError(errors);
}
// Returns the handle for the next attempt. The first attempt uses the socket's
// own handle so that any state set on it before connecting is kept.
function acquireAttemptHandle(context) {
const self = context.socket;
if (context.socketHandleState === kSocketHandlePending) {
const handle = new TCP(TCPConstants.SOCKET);
if (self._handle.hasRef() === false) handle.unref();
return handle;
}
if (context.socketHandleState === kSocketHandleFailed) {
const unrefed = self._handle.hasRef() === false;
self[kReinitializeHandle](new TCP(TCPConstants.SOCKET));
if (unrefed) self._handle.unref();
}
context.socketHandleState = kSocketHandlePending;
return self._handle;
}
function releaseAttemptHandle(context, handle, own) {
if (own) {
context.socketHandleState = kSocketHandleFailed;
} else {
handle.close();
}
}
function connectionAttemptFailed(context, index) {
const self = context.socket;
if (context.done || !self.connecting) return;
// Start the next attempt right away, without waiting for the attempt timeout.
if (context.current < context.addresses.length) {
internalConnectMultiple(context);
return;
}
if (context.pending.length === 0) {
self.destroy(connectMultipleError(context));
return;
}
// The last attempt failed but earlier ones are still pending: give them one
// more attempt timeout instead of waiting for the operating system.
if (index === context.addresses.length - 1 && context[kTimeout] === null) {
context[kTimeout] = setTimeout(internalConnectMultipleFinalTimeout, context.timeout, context);
if (self._handle.hasRef() === false) context[kTimeout].unref();
}
}
function internalConnectMultipleTimeout(context, attempt) {
context[kTimeout] = null;
const self = context.socket;
if (context.done || !self.connecting) return;
const { handle, req, index, own } = attempt;
if (ArrayPrototypeIncludes(context.pending, attempt)) {
debug('connect/multiple: connection to %s:%s is still pending', req.address, req.port);
attempt.timedOut = true;
if (context.localPort) {
// Two attempts cannot portably bind to the same fixed source port.
removePendingAttempt(context, req);
req.oncomplete = undefined;
releaseAttemptHandle(context, handle, own);
context.errors[index] = createConnectionError(req, UV_ETIMEDOUT);
}
self.emit('connectionAttemptTimeout', req.address, req.port, req.addressType);
}
if (!context.done && self.connecting) internalConnectMultiple(context);
}
function internalConnectMultipleFinalTimeout(context) {
context[kTimeout] = null;
const self = context.socket;
if (context.done || !self.connecting) return;
const pending = ArrayPrototypeSlice(context.pending);
for (let i = 0; i < pending.length; i++) {
const { req, index, timedOut } = pending[i];
debug('connect/multiple: connection to %s:%s timed out', req.address, req.port);
context.errors[index] = createConnectionError(req, UV_ETIMEDOUT);
if (!timedOut) self.emit('connectionAttemptTimeout', req.address, req.port, req.addressType);
}
if (!context.done && self.connecting) self.destroy(connectMultipleError(context));
}
function internalConnectMultiple(context) {
clearTimeout(context[kTimeout]);
context[kTimeout] = null;
const self = context.socket;
if (context.done || !self.connecting || self._aborted) return;
const current = context.current++;
const { localPort, port, flags } = context;
const { address, family: addressType } = context.addresses[current];
const handle = acquireAttemptHandle(context);
const own = handle === self._handle;
let localAddress;
let err;
if (localPort) {
if (addressType === 4) {
localAddress = DEFAULT_IPV4_ADDR;
err = handle.bind(localAddress, localPort);
} else { // addressType === 6
localAddress = DEFAULT_IPV6_ADDR;
err = handle.bind6(localAddress, localPort, flags);
}
debug('connect/multiple: binding to localAddress: %s and localPort: %d (addressType: %d)',
localAddress, localPort, addressType);
err = checkBindError(err, localPort, handle);
if (err) {
releaseAttemptHandle(context, handle, own);
context.errors[current] = new ExceptionWithHostPort(err, 'bind', localAddress, localPort);
connectionAttemptFailed(context, current);
return;
}
}
if (self.blockList?.check(address, `ipv${addressType}`)) {
releaseAttemptHandle(context, handle, own);
const ex = new ERR_IP_BLOCKED(address);
context.errors[current] = ex;
self.emit('connectionAttemptFailed', address, port, addressType, ex);
connectionAttemptFailed(context, current);
return;
}
debug('connect/multiple: attempting to connect to %s:%d (addressType: %d)', address, port, addressType);
self.emit('connectionAttempt', address, port, addressType);
if (!self.connecting || context.done) {
if (!own) handle.close();
return;
}
const req = new TCPConnectWrap();
req.oncomplete = FunctionPrototypeBind(afterConnectMultiple, undefined, context);
req.address = address;
req.port = port;
req.localAddress = localAddress;
req.localPort = localPort;
req.addressType = addressType;
ArrayPrototypePush(self.autoSelectFamilyAttemptedAddresses, `${address}:${port}`);
err = addressType === 4 ? handle.connect(req, address, port) : handle.connect6(req, address, port);
if (err) {
releaseAttemptHandle(context, handle, own);
const ex = createConnectionError(req, err);
context.errors[current] = ex;
self.emit('connectionAttemptFailed', address, port, addressType, ex);
connectionAttemptFailed(context, current);
return;
}
const attempt = { handle, req, index: current, own, timedOut: false };
ArrayPrototypePush(context.pending, attempt);
if (context.current < context.addresses.length) {
debug('connect/multiple: setting the attempt timeout to %d ms', context.timeout);
context[kTimeout] = setTimeout(internalConnectMultipleTimeout, context.timeout, context, attempt);
if (self._handle.hasRef() === false) context[kTimeout].unref();
}
}
Socket.prototype.connect = function(...args) {
let normalized;
// If passed an array, it's treated as an array of arguments that have
// already been normalized (so we don't normalize more than once). This has
// been solved before in https://github.com/nodejs/node/pull/12342, but was
// reverted as it had unintended side effects.
if (ArrayIsArray(args[0]) && args[0][normalizedArgsSymbol]) {
normalized = args[0];
} else {
normalized = normalizeArgs(args);
}
const options = normalized[0];
const cb = normalized[1];
if (netClientSocketChannel.hasSubscribers) {
netClientSocketChannel.publish({
socket: this,
});
}
if (cb !== null) {
this.once('connect', cb);
}
// If the parent is already connecting, do not attempt to connect again
if (this._parent?.connecting) {
return this;
}
// options.port === null will be checked later.
if (options.port === undefined && options.path == null)
throw new ERR_MISSING_ARGS(['options', 'port', 'path']);
if (this.write !== Socket.prototype.write)
this.write = Socket.prototype.write;
if (this.destroyed) {
this._handle = null;
this._peername = null;
this._sockname = null;
}
const { path } = options;
// An adopted BoundSocket handle already fixes the transport; trust its type
// rather than inferring pipe-ness from a path option on the connect call.
// Once destroyed the adopted handle is gone (its reservation released), so
// fall back to the path option, as for other pre-existing handles (e.g. a
// TLSWrap) that are not transport handles.
const pipe = this[kBoundSource] && this._handle ?
this._handle instanceof Pipe : !!path;
debug('pipe', pipe, path);
if (!this._handle) {
this._handle = pipe ?
new Pipe(PipeConstants.SOCKET) :
new TCP(TCPConstants.SOCKET);
initSocketHandle(this);
}
this._unrefTimer();
this.connecting = true;
if (pipe) {
validateString(path, 'options.path');
defaultTriggerAsyncIdScope(
this[async_id_symbol], internalConnect, this, path,
);
} else {
lookupAndConnect(this, options);
}
return this;
};
Socket.prototype[kReinitializeHandle] = function reinitializeHandle(handle) {
this._handle?.close();
this._handle = handle;
this._handle[owner_symbol] = this;
initSocketHandle(this);
};
// A Socket can be transferred to another thread only while it is a freshly
// accepted/created TCP connection: still attached to a live handle, not
// connecting or destroyed, and with no data already buffered in either
// direction (which would otherwise be lost on the sending side).
function assertTransferableSocket(socket) {
const handle = socket._handle;
if (handle == null || !(handle instanceof TCP) ||
socket.destroyed || socket.connecting ||
socket.bytesRead > 0 || socket.bytesWritten > 0 ||
socket.readableLength > 0 || socket.writableLength > 0 ||
socket.readableEncoding != null) {
throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Socket');
}
}
Socket.prototype[kTransferList] = function() {
assertTransferableSocket(this);
return [this._handle];
};
Socket.prototype[kTransfer] = function() {
assertTransferableSocket(this);
const handle = this._handle;
const data = {
handle,
allowHalfOpen: this.allowHalfOpen,
};
// Detach the handle from this source socket; the messaging layer takes
// ownership of it via TCPWrap::TransferForMessaging(). Destroy the source so
// any further use on the sending side fails cleanly (the socket is now owned
// by the receiving thread) instead of silently dropping data.
this._handle = null;
this.destroy();
return {
data,
deserializeInfo: 'net:Socket',
};
};
Socket.prototype[kDeserialize] = function(data) {
const handle = data?.handle;
if (handle == null || !(handle instanceof TCP)) {
throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Socket');
}
this.allowHalfOpen = Boolean(data.allowHalfOpen);
this[kReinitializeHandle](handle);
this.readable = this.writable = true;
};
function socketToDnsFamily(family) {
switch (family) {
case 'IPv4':
return 4;
case 'IPv6':
return 6;
}
return family;
}
function lookupAndConnect(self, options) {
const { localAddress, localPort } = options;
const host = options.host || 'localhost';
let { port, autoSelectFamilyAttemptTimeout, autoSelectFamily } = options;
validateStringWithoutNullBytes(host, 'options.host');
// An adopted BoundSocket already owns the local endpoint and address family.
if (self[kBoundSource]) {
if (localAddress !== undefined || localPort !== undefined) {
throw new ERR_INVALID_ARG_VALUE(
'options',
options,
'localAddress and localPort cannot be used with an adopted bound socket');
}
autoSelectFamily = false;
}
if (localAddress && !isIP(localAddress)) {
throw new ERR_INVALID_IP_ADDRESS(localAddress);
}
if (localPort) {
validateNumber(localPort, 'options.localPort');
}
if (port !== undefined) {
if (typeof port !== 'number' && typeof port !== 'string') {
throw new ERR_INVALID_ARG_TYPE('options.port',
['number', 'string'], port);
}
validatePort(port);
}
port |= 0;
if (autoSelectFamily != null) {
validateBoolean(autoSelectFamily, 'options.autoSelectFamily');
} else {
autoSelectFamily = autoSelectFamilyDefault;
}
if (autoSelectFamilyAttemptTimeout != null) {
validateInt32(autoSelectFamilyAttemptTimeout, 'options.autoSelectFamilyAttemptTimeout', 1);
if (autoSelectFamilyAttemptTimeout < 10) {
autoSelectFamilyAttemptTimeout = 10;
}
} else {
autoSelectFamilyAttemptTimeout = autoSelectFamilyAttemptTimeoutDefault;
}
// If host is an IP, skip performing a lookup
const addressType = isIP(host);
// Retain the host before the IP fast path for TLS session identity binding.
self._host = host;
if (addressType) {
// An adopted BoundSocket needs no lookup: issue connect(2) in-tick so the
// kernel-resolved concrete source address is observable synchronously.
// Connection failures are still reported via a deferred 'error' event.
if (self[kBoundSource]) {
defaultTriggerAsyncIdScope(
self[async_id_symbol],
internalConnect,
self, host, port, addressType, localAddress, localPort,
);
return;
}
defaultTriggerAsyncIdScope(self[async_id_symbol], process.nextTick, () => {
if (self.connecting)
defaultTriggerAsyncIdScope(
self[async_id_symbol],
internalConnect,
self, host, port, addressType, localAddress, localPort,
);
});
return;
}
if (options.lookup != null)
validateFunction(options.lookup, 'options.lookup');
if (dns === undefined) dns = require('dns');
const dnsopts = {
family: socketToDnsFamily(options.family),
hints: options.hints || 0,
};
if (!isWindows &&
dnsopts.family !== 4 &&
dnsopts.family !== 6 &&
dnsopts.hints === 0) {
dnsopts.hints = dns.ADDRCONFIG;
}
debug('connect: find host', host);
debug('connect: dns options', dnsopts);
const lookup = options.lookup || dns.lookup;
if (dnsopts.family !== 4 && dnsopts.family !== 6 && !localAddress && autoSelectFamily) {
debug('connect: autodetecting');
dnsopts.all = true;
defaultTriggerAsyncIdScope(self[async_id_symbol], function() {
lookupAndConnectMultiple(
self,
async_id_symbol,
lookup,
host,
options,
dnsopts,
port,
localAddress,
localPort,
autoSelectFamilyAttemptTimeout,
);
});
return;
}
defaultTriggerAsyncIdScope(self[async_id_symbol], function() {
lookup(host, dnsopts, function emitLookup(err, ip, addressType) {
self.emit('lookup', err, ip, addressType, host);
// It's possible we were destroyed while looking this up.
// XXX it would be great if we could cancel the promise returned by
// the look up.
if (!self.connecting) return;
if (err) {
// net.createConnection() creates a net.Socket object and immediately
// calls net.Socket.connect() on it (that's us). There are no event
// listeners registered yet so defer the error event to the next tick.
process.nextTick(connectErrorNT, self, err);
} else if ((typeof ip !== 'string') || !isIP(ip)) {
err = new ERR_INVALID_IP_ADDRESS(ip);
process.nextTick(connectErrorNT, self, err);
} else if (addressType !== 4 && addressType !== 6) {
err = new ERR_INVALID_ADDRESS_FAMILY(addressType,
options.host,
options.port);
process.nextTick(connectErrorNT, self, err);
} else {
self._unrefTimer();
defaultTriggerAsyncIdScope(
self[async_id_symbol],
internalConnect,
self, ip, port, addressType, localAddress, localPort,
);
}
});
});
}
function lookupAndConnectMultiple(
self, async_id_symbol, lookup, host, options, dnsopts, port, localAddress, localPort, timeout,
) {
defaultTriggerAsyncIdScope(self[async_id_symbol], function emitLookup() {
lookup(host, dnsopts, function emitLookup(err, addresses) {
// It's possible we were destroyed while looking this up.
// XXX it would be great if we could cancel the promise returned by
// the look up.
if (!self.connecting) {
return;
} else if (err) {
self.emit('lookup', err, undefined, undefined, host);
// net.createConnection() creates a net.Socket object and immediately
// calls net.Socket.connect() on it (that's us). There are no event
// listeners registered yet so defer the error event to the next tick.
process.nextTick(connectErrorNT, self, err);
return;
}
// Filter addresses by only keeping the one which are either IPv4 or IPV6.
// The first valid address determines which group has preference on the
// alternate family sorting which happens later.
const validAddresses = [[], []];
const validIps = [[], []];
let destinations;
for (let i = 0, l = addresses.length; i < l; i++) {
const address = addresses[i];
const { address: ip, family: addressType } = address;
self.emit('lookup', err, ip, addressType, host);
// It's possible we were destroyed while looking this up.
if (!self.connecting) {
return;
}
if (isIP(ip) && (addressType === 4 || addressType === 6)) {
destinations ||= addressType === 6 ? { 6: 0, 4: 1 } : { 4: 0, 6: 1 };
const destination = destinations[addressType];
// Only try an address once
if (!ArrayPrototypeIncludes(validIps[destination], ip)) {
ArrayPrototypePush(validAddresses[destination], address);
ArrayPrototypePush(validIps[destination], ip);
}
}
}
// When no AAAA or A records are available, fail on the first one
if (!validAddresses[0].length && !validAddresses[1].length) {
const { address: firstIp, family: firstAddressType } = addresses[0];
if (!isIP(firstIp)) {
err = new ERR_INVALID_IP_ADDRESS(firstIp);
process.nextTick(connectErrorNT, self, err);
} else if (firstAddressType !== 4 && firstAddressType !== 6) {
err = new ERR_INVALID_ADDRESS_FAMILY(firstAddressType,
options.host,
options.port);
process.nextTick(connectErrorNT, self, err);
}
return;
}
// Sort addresses alternating families
const toAttempt = [];
for (let i = 0, l = MathMax(validAddresses[0].length, validAddresses[1].length); i < l; i++) {
if (i in validAddresses[0]) {
ArrayPrototypePush(toAttempt, validAddresses[0][i]);
}
if (i in validAddresses[1]) {
ArrayPrototypePush(toAttempt, validAddresses[1][i]);
}
}
if (toAttempt.length === 1) {
debug('connect/multiple: only one address found, switching back to single connection');
const { address: ip, family: addressType } = toAttempt[0];
self._unrefTimer();
defaultTriggerAsyncIdScope(
self[async_id_symbol],
internalConnect,
self,
ip,
port,
addressType,
localAddress,
localPort,
);
return;
}
self.autoSelectFamilyAttemptedAddresses = [];
debug('connect/multiple: will try the following addresses', toAttempt);
const context = {
socket: self,
addresses: toAttempt,
current: 0,
port,
localPort,
timeout,
[kTimeout]: null,
errors: [],
pending: [],
socketHandleState: kSocketHandleUnused,
done: false,
};
self[kAutoSelectFamilyContext] = context;
self._unrefTimer();
defaultTriggerAsyncIdScope(self[async_id_symbol], internalConnectMultiple, context);
});
});
}
function connectErrorNT(self, err) {
self.destroy(err);
}
Socket.prototype.ref = function() {
if (!this._handle) {
this.once('connect', this.ref);
return this;
}
if (typeof this._handle.ref === 'function') {
this._handle.ref();
}
const context = this[kAutoSelectFamilyContext];
if (context) {
context[kTimeout]?.ref();
for (let i = 0; i < context.pending.length; i++) {
context.pending[i].handle.ref();
}
}
return this;
};
Socket.prototype.unref = function() {
if (!this._handle) {
this.once('connect', this.unref);
return this;
}
if (typeof this._handle.unref === 'function') {
this._handle.unref();
}
const context = this[kAutoSelectFamilyContext];
if (context) {
context[kTimeout]?.unref();
for (let i = 0; i < context.pending.length; i++) {
context.pending[i].handle.unref();
}
}
return this;
};
function afterConnect(status, handle, req, readable, writable) {
const self = handle[owner_symbol];
// Callback may come after call to destroy
if (self.destroyed) {
return;
}
debug('afterConnect');
assert(self.connecting);
self.connecting = false;
self._sockname = null;
if (status === 0) {
if (self.readable && !readable) {
self.push(null);
self.read();
}
if (self.writable && !writable) {
self.end();
}
self._unrefTimer();
if (self[kSetNoDelay] && self._handle.setNoDelay) {
self._handle.setNoDelay(true);
}
if (self[kSetKeepAlive] && self._handle.setKeepAlive) {
self._handle.setKeepAlive(true, self[kSetKeepAliveInitialDelay],
self[kSetKeepAliveInterval],
self[kSetKeepAliveCount]);
}
if (self[kSetTOS] !== undefined && self._handle.setTypeOfService) {
const err = self._handle.setTypeOfService(self[kSetTOS]);
// On Windows, setting TOS is best-effort. If it fails, we shouldn't destroy
// the connection or emit an error, as the socket is otherwise healthy.
if (err && err !== UV_EBADF && !isWindows) {
self.emit('error', new ErrnoException(err, 'setTypeOfService'));
}
}
self.emit('connect');
self.emit('ready');
// Start the first read, or get an immediate EOF.
// this doesn't actually consume any bytes, because len=0.
if (readable && !self.isPaused())
self.read(0);
if (self[kPerfHooksNetConnectContext] && hasObserver('net')) {
stopPerf(self, kPerfHooksNetConnectContext);
}
} else {
let details;
if (req.localAddress && req.localPort) {
details = req.localAddress + ':' + req.localPort;
}
const ex = new ExceptionWithHostPort(status,
'connect',
req.address,
req.port,
details);
if (details) {
ex.localAddress = req.localAddress;
ex.localPort = req.localPort;
}
self.emit('connectionAttemptFailed', req.address, req.port, req.addressType, ex);
self.destroy(ex);
}
}
function addClientAbortSignalOption(self, options) {
validateAbortSignal(options.signal, 'options.signal');
const { signal } = options;
let disposable;
function onAbort() {
disposable?.[SymbolDispose]();
self._aborted = true;
}
if (signal.aborted) {
process.nextTick(onAbort);
} else {
process.nextTick(() => {
disposable = addAbortListener(signal, onAbort);
});
}
}
function createConnectionError(req, status) {
let details;
if (req.localAddress && req.localPort) {
details = req.localAddress + ':' + req.localPort;
}
const ex = new ExceptionWithHostPort(status,
'connect',
req.address,
req.port,
details);
if (details) {
ex.localAddress = req.localAddress;
ex.localPort = req.localPort;
}
return ex;
}
function afterConnectMultiple(context, status, handle, req, readable, writable) {
debug('connect/multiple: connection attempt to %s:%s completed with status %s', req.address, req.port, status);
const attempt = removePendingAttempt(context, req);
if (attempt === undefined) return;
const self = context.socket;
if (context.done || !self.connecting) {
if (!attempt.own) attempt.handle.close();
return;
}
if (status !== 0) {
releaseAttemptHandle(context, attempt.handle, attempt.own);
const ex = createConnectionError(req, status);
context.errors[attempt.index] = ex;
self.emit('connectionAttemptFailed', req.address, req.port, req.addressType, ex);
connectionAttemptFailed(context, attempt.index);
return;
}
closeConnectionAttempts(context);
self[kAutoSelectFamilyContext] = undefined;
if (!attempt.own) {
const unrefed = self._handle.hasRef() === false;
self[kReinitializeHandle](attempt.handle);
if (unrefed) self._handle.unref();
}
if (hasObserver('net')) {
startPerf(
self,
kPerfHooksNetConnectContext,
{ type: 'net', name: 'connect', detail: { host: req.address, port: req.port } },
);
}
afterConnect(status, self._handle, req, readable, writable);
}
function addServerAbortSignalOption(self, options) {
if (options?.signal === undefined) {
return;
}
validateAbortSignal(options.signal, 'options.signal');
const { signal } = options;
const onAborted = () => {
self.close();
};
if (signal.aborted) {
process.nextTick(onAborted);
} else {
const disposable = addAbortListener(signal, onAborted);
self.once('close', disposable[SymbolDispose]);
}
}
function Server(options, connectionListener) {
if (!(this instanceof Server))
return new Server(options, connectionListener);
EventEmitter.call(this);
// A listening TCP Server can be moved to another thread by listing it in the
// transferList of a worker_threads postMessage() call. See [kTransfer]().
markTransferMode(this, false, true);
if (typeof options === 'function') {
connectionListener = options;
options = kEmptyObject;
this.on('connection', connectionListener);
} else if (options == null || typeof options === 'object') {
options = { ...options };
if (typeof connectionListener === 'function') {
this.on('connection', connectionListener);
}
} else {
throw new ERR_INVALID_ARG_TYPE('options', 'Object', options);
}
if (options.keepAliveInitialDelay !== undefined) {
validateNumber(
options.keepAliveInitialDelay, 'options.keepAliveInitialDelay',
);
if (options.keepAliveInitialDelay < 0) {
options.keepAliveInitialDelay = 0;
}
}
if (options.highWaterMark !== undefined) {
validateNumber(
options.highWaterMark, 'options.highWaterMark',
);
if (options.highWaterMark < 0) {
options.highWaterMark = getDefaultHighWaterMark();
}
}
this._connections = 0;
this[async_id_symbol] = -1;
this._handle = null;
this._listening = false;
this._usingWorkers = false;
this._workers = [];
this._unref = false;
this._listeningId = 1;
this.allowHalfOpen = options.allowHalfOpen || false;
this.pauseOnConnect = !!options.pauseOnConnect;
this.noDelay = Boolean(options.noDelay);
this.keepAlive = Boolean(options.keepAlive);
this.keepAliveInitialDelay = ~~(options.keepAliveInitialDelay / 1000);
this.highWaterMark = options.highWaterMark ?? getDefaultHighWaterMark();
if (options.blockList) {
if (!module.exports.BlockList.isBlockList(options.blockList)) {
throw new ERR_INVALID_ARG_TYPE('options.blockList', 'net.BlockList', options.blockList);
}
this.blockList = options.blockList;
}
}
ObjectSetPrototypeOf(Server.prototype, EventEmitter.prototype);
ObjectSetPrototypeOf(Server, EventEmitter);
function toNumber(x) { return (x = Number(x)) >= 0 ? x : false; }
// Returns handle if it can be created, or error code if it can't
function createServerHandle(address, port, addressType, fd, flags) {
let err = 0;
// Assign handle in listen, and clean up if bind or listen fails
let handle;
let isTCP = false;
if (typeof fd === 'number' && fd >= 0) {
try {
handle = createHandle(fd, true);
} catch (e) {
// Not a fd we can listen on. This will trigger an error.
debug('listen invalid fd=%d:', fd, e.message);
return UV_EINVAL;
}
err = handle.open(fd);
if (err)
return err;
assert(!address && !port);
} else if (port === -1 && addressType === -1) {
handle = new Pipe(PipeConstants.SERVER);
if (isWindows) {
const instances = NumberParseInt(process.env.NODE_PENDING_PIPE_INSTANCES);
if (!NumberIsNaN(instances)) {
handle.setPendingInstances(instances);
}
}
} else {
handle = new TCP(TCPConstants.SERVER);
isTCP = true;
}
if (address || port || isTCP) {
debug('bind to', address || 'any');
if (!address) {
// Try binding to ipv6 first
err = handle.bind6(DEFAULT_IPV6_ADDR, port, flags);
if (err) {
handle.close();
// Fallback to ipv4
return createServerHandle(DEFAULT_IPV4_ADDR, port, undefined, undefined, flags);
}
} else if (addressType === 6) {
err = handle.bind6(address, port, flags);
} else {
err = handle.bind(address, port, flags);
}
}
if (err) {
handle.close();
return err;
}
return handle;
}
function setupListenHandle(address, port, addressType, backlog, fd, flags) {
debug('setupListenHandle', address, port, addressType, backlog, fd);
// If there is not yet a handle, we need to create one and bind.
// In the case of a server sent via IPC, we don't need to do this.
if (this._handle) {
debug('setupListenHandle: have a handle already');
} else {
debug('setupListenHandle: create a handle');
let rval = null;
// Try to bind to the unspecified IPv6 address, see if IPv6 is available
if (!address && typeof fd !== 'number') {
rval = createServerHandle(DEFAULT_IPV6_ADDR, port, 6, fd, flags);
if (typeof rval === 'number') {
rval = null;
address = DEFAULT_IPV4_ADDR;
addressType = 4;
} else {
address = DEFAULT_IPV6_ADDR;
addressType = 6;
}
}
if (rval === null)
rval = createServerHandle(address, port, addressType, fd, flags);
if (typeof rval === 'number') {
const error = new UVExceptionWithHostPort(rval, 'listen', address, port);
if (netServerListen.hasSubscribers) {
netServerListen.error.publish({ server: this, error });
}
process.nextTick(emitErrorNT, this, error);
return;
}
this._handle = rval;
}
this[async_id_symbol] = getNewAsyncId(this._handle);
this._handle.onconnection = onconnection;
this._handle[owner_symbol] = this;
// Use a backlog of 512 entries. We pass 511 to the listen() call because
// the kernel does: backlogsize = roundup_pow_of_two(backlogsize + 1);
// which will thus give us a backlog of 512 entries.
const err = this._handle.listen(backlog || 511);
if (err) {
const ex = new UVExceptionWithHostPort(err, 'listen', address, port);
this._handle.close();
this._handle = null;
if (netServerListen.hasSubscribers) {
netServerListen.error.publish({ server: this, error: ex });
}
defaultTriggerAsyncIdScope(this[async_id_symbol],
process.nextTick,
emitErrorNT,
this,
ex);
return;
}
if (netServerListen.hasSubscribers) {
netServerListen.asyncEnd.publish({ server: this });
}
// Generate connection key, this should be unique to the connection
this._connectionKey = addressType + ':' + address + ':' + port;
// Unref the handle if the server was unref'ed prior to listening
if (this._unref)
this.unref();
defaultTriggerAsyncIdScope(this[async_id_symbol],
process.nextTick,
emitListeningNT,
this);
}
// Legacy alias for `setupListenHandle`, kept around only because it is an
// undocumented monkeypatch point. Nothing in core calls it unless it has been
// overridden, see `callSetupListenHandle`.
const legacyListen2 = deprecate(
setupListenHandle,
'Server.prototype._listen2 is deprecated. Use Server.prototype.listen() instead.',
'DEP0208');
Server.prototype._listen2 = legacyListen2;
// Set up the listen handle, going through `_listen2` when userland replaced it
// so that the monkeypatch keeps taking effect (DEP0208). Servers that did not
// touch `_listen2` must not trigger the deprecation warning.
function callSetupListenHandle(server, address, port, addressType, backlog,
fd, flags) {
if (server._listen2 !== legacyListen2) {
server._listen2(address, port, addressType, backlog, fd, flags);
return;
}
FunctionPrototypeCall(setupListenHandle, server, address, port, addressType,
backlog, fd, flags);
}
// A listening TCP Server can be transferred to another thread, which moves the
// underlying listening socket (and its pending accept queue) to that thread's
// event loop. Only a server bound to a live TCP handle can be transferred.
function assertTransferableServer(server) {
if (server._handle == null || !(server._handle instanceof TCP)) {
throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Server');
}
}
Server.prototype[kTransferList] = function() {
assertTransferableServer(this);
return [this._handle];
};
Server.prototype[kTransfer] = function() {
assertTransferableServer(this);
const handle = this._handle;
const data = {
handle,
// Construction-time options that govern accepted sockets, so the receiving
// server reproduces the same behaviour.
allowHalfOpen: this.allowHalfOpen,
pauseOnConnect: this.pauseOnConnect,
noDelay: this.noDelay,
keepAlive: this.keepAlive,
keepAliveInitialDelay: this.keepAliveInitialDelay * 1000,
highWaterMark: this.highWaterMark,
};
// Detach so the source server no longer references the handle being moved.
this._handle = null;
this._listening = false;
return {
data,
deserializeInfo: 'net:Server',
};
};
Server.prototype[kDeserialize] = function(data) {
const { handle, ...options } = data ?? {};
if (handle == null || !(handle instanceof TCP)) {
throw new ERR_WORKER_HANDLE_NOT_TRANSFERABLE('net.Server');
}
this.allowHalfOpen = Boolean(options.allowHalfOpen);
this.pauseOnConnect = Boolean(options.pauseOnConnect);
this.noDelay = Boolean(options.noDelay);
this.keepAlive = Boolean(options.keepAlive);
this.keepAliveInitialDelay = ~~(options.keepAliveInitialDelay / 1000);
this.highWaterMark = options.highWaterMark ?? getDefaultHighWaterMark();
// Adopt the transferred listening handle (mirrors the child_process server
// hand-off path), which re-arms accept() on this thread's event loop.
this.listen(handle);
};
function emitErrorNT(self, err) {
self._listening = false;
self.emit('error', err);
}
function emitListeningNT(self) {
// Ensure handle hasn't closed
if (self._handle)
self.emit('listening');
}
function listenInCluster(server, address, port, addressType,
backlog, fd, exclusive, flags, options) {
exclusive = !!exclusive;
if (cluster === undefined) cluster = require('cluster');
if (cluster.isPrimary || exclusive) {
// Will create a new handle
callSetupListenHandle(server, address, port, addressType, backlog, fd,
flags);
return;
}
const serverQuery = {
address: address,
port: port,
addressType: addressType,
fd: fd,
flags,
backlog,
...options,
};
const listeningId = server._listeningId;
// Get the primary's server handle, and listen on it
cluster._getServer(server, serverQuery, listenOnPrimaryHandle);
function listenOnPrimaryHandle(err, handle) {
if (listeningId !== server._listeningId) {
handle.close();
return;
}
err = checkBindError(err, port, handle);
if (err) {
const ex = new ExceptionWithHostPort(err, 'bind', address, port);
server._listening = false;
return server.emit('error', ex);
}
// If there was a handle, just close it to avoid fd leak
// but it doesn't look like that's going to happen right now
if (server._handle) {
server._handle.close();
}
// Reuse primary's server handle
server._handle = handle;
callSetupListenHandle(server, address, port, addressType, backlog, fd,
flags);
}
}
Server.prototype.listen = function(...args) {
const normalized = normalizeArgs(args);
let options = normalized[0];
const cb = normalized[1];
if (this._handle || this._listening) {
throw new ERR_SERVER_ALREADY_LISTEN();
}
if (netServerListen.hasSubscribers) {
netServerListen.asyncStart.publish({ server: this, options });
}
if (cb !== null) {
this.once('listening', cb);
}
const backlogFromArgs =
// (handle, backlog) or (path, backlog) or (port, backlog)
toNumber(args.length > 1 && args[1]) ||
toNumber(args.length > 2 && args[2]); // (port, host, backlog)
// (boundSocket[, ...]) or ({ handle: boundSocket }[, ...]): adopt the bound
// handle, transferring ownership so the BoundSocket can no longer close it.
let boundSocket = null;
if (options instanceof BoundSocket) {
boundSocket = options;
} else if (options.handle instanceof BoundSocket) {
boundSocket = options.handle;
}
if (boundSocket !== null) {
const boundPath = boundSocket[kBoundSocketPath];
this._handle = boundSocket[kBoundSocketConsume]();
// A pipe-backed handle reports its path via Server.address().
if (boundPath !== undefined) {
this._pipeName = boundPath;
}
this[async_id_symbol] = this._handle.getAsyncId();
this._listening = true;
this._listeningId++;
listenInCluster(this, null, -1, -1, backlogFromArgs, undefined, true);
return this;
}
options = options._handle || options.handle || options;
const flags = getFlags(options);
// Refresh the id to make the previous call invalid
this._listeningId++;
// (handle[, backlog][, cb]) where handle is an object with a handle
if (options instanceof TCP) {
this._handle = options;
this[async_id_symbol] = this._handle.getAsyncId();
this._listening = true;
listenInCluster(this, null, -1, -1, backlogFromArgs, undefined, true);
return this;
}
addServerAbortSignalOption(this, options);
// (handle[, backlog][, cb]) where handle is an object with a fd
if (typeof options.fd === 'number' && options.fd >= 0) {
this._listening = true;
listenInCluster(this, null, null, null, backlogFromArgs, options.fd);
return this;
}
// ([port][, host][, backlog][, cb]) where port is omitted,
// that is, listen(), listen(null), listen(cb), or listen(null, cb)
// or (options[, cb]) where options.port is explicitly set as undefined or
// null, bind to an arbitrary unused port
if (args.length === 0 || typeof args[0] === 'function' ||
(options.port === undefined && 'port' in options) ||
options.port === null) {
options.port = 0;
}
// ([port][, host][, backlog][, cb]) where port is specified
// or (options[, cb]) where options.port is specified
// or if options.port is normalized as 0 before
let backlog;
if (typeof options.port === 'number' || typeof options.port === 'string') {
validatePort(options.port, 'options.port');
backlog = options.backlog || backlogFromArgs;
if (options.reusePort === true) {
options.exclusive = true;
}
// start TCP server listening on host:port
this._listening = true;
if (options.host) {
lookupAndListen(this, options.port | 0, options.host, backlog,
options.exclusive, flags);
} else { // Undefined host, listens on unspecified address
// Default addressType 4 will be used to search for primary server
listenInCluster(this, null, options.port | 0, 4,
backlog, undefined, options.exclusive, flags);
}
return this;
}
// (path[, backlog][, cb]) or (options[, cb])
// where path or options.path is a UNIX domain socket or Windows pipe
if (options.path && isPipeName(options.path)) {
// We can not call fchmod on abstract unix socket
if (options.path[0] === '\0' &&
(options.readableAll || options.writableAll)) {
const msg = 'can not set readableAll or writableAllt to true when path is abstract unix socket';
throw new ERR_INVALID_ARG_VALUE('options', options, msg);
}
const pipeName = this._pipeName = options.path;
backlog = options.backlog || backlogFromArgs;
this._listening = true;
listenInCluster(this,
pipeName,
-1,
-1,
backlog,
undefined,
options.exclusive,
undefined,
{
readableAll: options.readableAll,
writableAll: options.writableAll,
});
if (!this._handle) {
// Failed and an error shall be emitted in the next tick.
// Therefore, we directly return.
return this;
}
let mode = 0;
if (options.readableAll === true)
mode |= PipeConstants.UV_READABLE;
if (options.writableAll === true)
mode |= PipeConstants.UV_WRITABLE;
if (mode !== 0) {
const err = this._handle.fchmod(mode);
if (err) {
this._handle.close();
this._handle = null;
this._listening = false;
throw new ErrnoException(err, 'uv_pipe_chmod');
}
}
return this;
}
if (!(('port' in options) || ('path' in options))) {
throw new ERR_INVALID_ARG_VALUE('options', options,
'must have the property "port" or "path"');
}
throw new ERR_INVALID_ARG_VALUE('options', options);
};
function isIpv6LinkLocal(ip) {
if (!isIPv6(ip)) { return false; }
const ipv6Buffer = convertIpv6StringToBuffer(ip);
const firstByte = ipv6Buffer[0]; // The first 8 bits
const secondByte = ipv6Buffer[1]; // The next 8 bits
// The link-local prefix is `1111111010`, which in hexadecimal is `fe80`
// First 8 bits (firstByte) should be `11111110` (0xfe)
// The next 2 bits of the second byte should be `10` (0x80)
const isFirstByteCorrect = (firstByte === 0xfe); // 0b11111110 == 0xfe
const isSecondByteCorrect = (secondByte & 0xc0) === 0x80; // 0b10xxxxxx == 0x80
return isFirstByteCorrect && isSecondByteCorrect;
}
function filterOnlyValidAddress(addresses) {
// Return the first non IPV6 link-local address if present
for (const address of addresses) {
if (!isIpv6LinkLocal(address.address)) {
return address;
}
}
// Otherwise return the first address
return addresses[0];
}
function lookupAndListen(self, port, address, backlog,
exclusive, flags) {
if (dns === undefined) dns = require('dns');
const listeningId = self._listeningId;
dns.lookup(address, { all: true }, (err, addresses) => {
if (listeningId !== self._listeningId) {
return;
}
if (err) {
self._listening = false;
self.emit('error', err);
} else {
const validAddress = filterOnlyValidAddress(addresses);
const family = validAddress?.family || 4;
listenInCluster(self, validAddress.address, port, family,
backlog, undefined, exclusive, flags);
}
});
}
ObjectDefineProperty(Server.prototype, 'listening', {
__proto__: null,
get: function() {
return !!this._handle;
},
configurable: true,
enumerable: true,
});
Server.prototype.address = function() {
if (this._handle?.getsockname) {
const out = {};
const err = this._handle.getsockname(out);
if (err) {
throw new ErrnoException(err, 'address');
}
return out;
} else if (this._pipeName) {
return this._pipeName;
}
return null;
};
function onconnection(err, clientHandle) {
const handle = this;
const self = handle[owner_symbol];
debug('onconnection');
if (err) {
self.emit('error', new ErrnoException(err, 'accept'));
return;
}
if (self.maxConnections != null && self._connections >= self.maxConnections) {
if (clientHandle.getsockname || clientHandle.getpeername) {
const data = { __proto__: null };
if (clientHandle.getsockname) {
const localInfo = { __proto__: null };
clientHandle.getsockname(localInfo);
data.localAddress = localInfo.address;
data.localPort = localInfo.port;
data.localFamily = localInfo.family;
}
if (clientHandle.getpeername) {
const remoteInfo = { __proto__: null };
clientHandle.getpeername(remoteInfo);
data.remoteAddress = remoteInfo.address;
data.remotePort = remoteInfo.port;
data.remoteFamily = remoteInfo.family;
}
self.emit('drop', data);
} else {
self.emit('drop');
}
clientHandle.close();
return;
}
if (self.blockList && typeof clientHandle.getpeername === 'function') {
const remoteInfo = { __proto__: null };
clientHandle.getpeername(remoteInfo);
const addressType = isIP(remoteInfo.address);
// Fail closed: if a BlockList is configured but the peer address cannot
// be obtained (e.g. the connection was reset before getpeername()),
// refuse the connection instead of silently bypassing the access control.
if (!addressType || self.blockList.check(remoteInfo.address, `ipv${addressType}`)) {
clientHandle.close();
return;
}
}
const socket = new Socket({
handle: clientHandle,
allowHalfOpen: self.allowHalfOpen,
pauseOnCreate: self.pauseOnConnect,
readable: true,
writable: true,
readableHighWaterMark: self.highWaterMark,
writableHighWaterMark: self.highWaterMark,
});
if (self.noDelay && clientHandle.setNoDelay) {
socket[kSetNoDelay] = true;
clientHandle.setNoDelay(true);
}
if (self.keepAlive && clientHandle.setKeepAlive) {
socket[kSetKeepAlive] = true;
socket[kSetKeepAliveInitialDelay] = self.keepAliveInitialDelay;
clientHandle.setKeepAlive(true, self.keepAliveInitialDelay);
}
self._connections++;
socket.server = self;
socket._server = self;
self.emit('connection', socket);
if (netServerSocketChannel.hasSubscribers) {
netServerSocketChannel.publish({
socket,
});
}
}
/**
* Gets the number of concurrent connections on the server
* @param {Function} cb
* @returns {Server}
*/
Server.prototype.getConnections = function(cb) {
const self = this;
function end(err, connections) {
defaultTriggerAsyncIdScope(self[async_id_symbol],
process.nextTick,
cb,
err,
connections);
}
if (!this._usingWorkers) {
end(null, this._connections);
return this;
}
// Poll workers
let left = this._workers.length;
let total = this._connections;
function oncount(err, count) {
if (err) {
left = -1;
return end(err);
}
total += count;
if (--left === 0) return end(null, total);
}
for (let n = 0; n < this._workers.length; n++) {
this._workers[n].getConnections(oncount);
}
return this;
};
Server.prototype.close = function(cb) {
this._listening = false;
this._listeningId++;
if (typeof cb === 'function') {
if (!this._handle) {
this.once('close', function close() {
cb(new ERR_SERVER_NOT_RUNNING());
});
} else {
this.once('close', cb);
}
}
if (this._handle) {
this._handle.close();
this._handle = null;
}
if (this._usingWorkers) {
let left = this._workers.length;
const onWorkerClose = () => {
if (--left !== 0) return;
this._connections = 0;
this._emitCloseIfDrained();
};
// Increment connections to be sure that, even if all sockets will be closed
// during polling of workers, `close` event will be emitted only once.
this._connections++;
// Poll workers
for (let n = 0; n < this._workers.length; n++)
this._workers[n].close(onWorkerClose);
} else {
this._emitCloseIfDrained();
}
return this;
};
Server.prototype[SymbolAsyncDispose] = async function() {
if (!this._handle) {
return;
}
await FunctionPrototypeCall(promisify(this.close), this);
};
Server.prototype[SymbolAsyncIterator] = function() {
kFirstEventParam ??= require('internal/events/symbols').kFirstEventParam;
return EventEmitter.on(this, 'connection', {
close: ['close'],
[kFirstEventParam]: true,
});
};
Server.prototype._emitCloseIfDrained = function() {
debug('SERVER _emitCloseIfDrained');
if (this._handle || this._connections) {
debug('SERVER handle? %j connections? %d',
!!this._handle, this._connections);
return;
}
defaultTriggerAsyncIdScope(this[async_id_symbol],
process.nextTick,
emitCloseNT,
this);
};
function emitCloseNT(self) {
debug('SERVER: emit close');
self.emit('close');
}
Server.prototype[EventEmitter.captureRejectionSymbol] = function(
err, event, sock) {
switch (event) {
case 'connection':
sock.destroy(err);
break;
default:
this.emit('error', err);
}
};
// Legacy alias on the C++ wrapper object. This is not public API, so we may
// want to runtime-deprecate it at some point. There's no hurry, though.
ObjectDefineProperty(TCP.prototype, 'owner', {
__proto__: null,
get() { return this[owner_symbol]; },
set(v) { return this[owner_symbol] = v; },
});
ObjectDefineProperty(Socket.prototype, '_handle', {
__proto__: null,
get() { return this[kHandle]; },
set(v) { return this[kHandle] = v; },
});
Server.prototype._setupWorker = function(socketList) {
this._usingWorkers = true;
this._workers.push(socketList);
socketList.once('exit', (socketList) => {
const index = ArrayPrototypeIndexOf(this._workers, socketList);
this._workers.splice(index, 1);
});
};
Server.prototype.ref = function() {
this._unref = false;
if (this._handle)
this._handle.ref();
return this;
};
Server.prototype.unref = function() {
this._unref = true;
if (this._handle)
this._handle.unref();
return this;
};
module.exports = {
_createServerHandle: createServerHandle,
_normalizeArgs: normalizeArgs,
_TransferredBoundSocket,
get BlockList() {
BlockList ??= require('internal/blocklist').BlockList;
return BlockList;
},
get SocketAddress() {
SocketAddress ??= require('internal/socketaddress').SocketAddress;
return SocketAddress;
},
BoundSocket,
connect,
createConnection: connect,
createServer,
get promises() {
netPromises ??= require('internal/net/promises');
return netPromises;
},
isIP: isIP,
isIPv4: isIPv4,
isIPv6: isIPv6,
Server,
Socket,
Stream: Socket, // Legacy naming
getDefaultAutoSelectFamily,
setDefaultAutoSelectFamily,
getDefaultAutoSelectFamilyAttemptTimeout,
setDefaultAutoSelectFamilyAttemptTimeout,
};