@@ -56,6 +56,7 @@ const { |
56 | 56 | UV_EINVAL, |
57 | 57 | UV_ENOTCONN, |
58 | 58 | UV_ECANCELED, |
| 59 | + UV_ETIMEDOUT, |
59 | 60 | } = internalBinding('uv'); |
60 | 61 |
|
61 | 62 | const { Buffer } = require('buffer'); |
@@ -482,6 +483,10 @@ function Socket(options) { |
482 | 483 | } |
483 | 484 | } |
484 | 485 |
|
| 486 | + if (options.signal) { |
| 487 | + addClientAbortSignalOption(this, options); |
| 488 | + } |
| 489 | + |
485 | 490 | // Reserve properties |
486 | 491 | this.server = null; |
487 | 492 | this._server = null; |
@@ -1091,6 +1096,11 @@ function internalConnectMultiple(context, canceled) { |
1091 | 1096 | clearTimeout(context[kTimeout]); |
1092 | 1097 | const self = context.socket; |
1093 | 1098 |
|
| 1099 | + // We were requested to abort. Stop all operations |
| 1100 | + if (self._aborted) { |
| 1101 | + return; |
| 1102 | + } |
| 1103 | + |
1094 | 1104 | // All connections have been tried without success, destroy with error |
1095 | 1105 | if (canceled || context.current === context.addresses.length) { |
1096 | 1106 | if (context.errors.length === 0) { |
@@ -1105,7 +1115,11 @@ function internalConnectMultiple(context, canceled) { |
1105 | 1115 | assert(self.connecting); |
1106 | 1116 |
|
1107 | 1117 | const current = context.current++; |
1108 | | - const handle = current === 0 ? self._handle : new TCP(TCPConstants.SOCKET); |
| 1118 | + |
| 1119 | + if (current > 0) { |
| 1120 | + self[kReinitializeHandle](new TCP(TCPConstants.SOCKET)); |
| 1121 | + } |
| 1122 | + |
1109 | 1123 | const { localPort, port, flags } = context; |
1110 | 1124 | const { address, family: addressType } = context.addresses[current]; |
1111 | 1125 | let localAddress; |
@@ -1114,16 +1128,16 @@ function internalConnectMultiple(context, canceled) { |
1114 | 1128 | if (localPort) { |
1115 | 1129 | if (addressType === 4) { |
1116 | 1130 | localAddress = DEFAULT_IPV4_ADDR; |
1117 | | - err = handle.bind(localAddress, localPort); |
| 1131 | + err = self._handle.bind(localAddress, localPort); |
1118 | 1132 | } else { // addressType === 6 |
1119 | 1133 | localAddress = DEFAULT_IPV6_ADDR; |
1120 | | - err = handle.bind6(localAddress, localPort, flags); |
| 1134 | + err = self._handle.bind6(localAddress, localPort, flags); |
1121 | 1135 | } |
1122 | 1136 |
|
1123 | 1137 | debug('connect/multiple: binding to localAddress: %s and localPort: %d (addressType: %d)', |
1124 | 1138 | localAddress, localPort, addressType); |
1125 | 1139 |
|
1126 | | - err = checkBindError(err, localPort, handle); |
| 1140 | + err = checkBindError(err, localPort, self._handle); |
1127 | 1141 | if (err) { |
1128 | 1142 | ArrayPrototypePush(context.errors, exceptionWithHostPort(err, 'bind', localAddress, localPort)); |
1129 | 1143 | internalConnectMultiple(context); |
@@ -1143,9 +1157,9 @@ function internalConnectMultiple(context, canceled) { |
1143 | 1157 | ArrayPrototypePush(self.autoSelectFamilyAttemptedAddresses, `${address}:${port}`); |
1144 | 1158 |
|
1145 | 1159 | if (addressType === 4) { |
1146 | | - err = handle.connect(req, address, port); |
| 1160 | + err = self._handle.connect(req, address, port); |
1147 | 1161 | } else { |
1148 | | - err = handle.connect6(req, address, port); |
| 1162 | + err = self._handle.connect6(req, address, port); |
1149 | 1163 | } |
1150 | 1164 |
|
1151 | 1165 | if (err) { |
@@ -1165,7 +1179,7 @@ function internalConnectMultiple(context, canceled) { |
1165 | 1179 | debug('connect/multiple: setting the attempt timeout to %d ms', context.timeout); |
1166 | 1180 |
|
1167 | 1181 | // If the attempt has not returned an error, start the connection timer |
1168 | | - context[kTimeout] = setTimeout(internalConnectMultipleTimeout, context.timeout, context, req, handle); |
| 1182 | + context[kTimeout] = setTimeout(internalConnectMultipleTimeout, context.timeout, context, req, self._handle); |
1169 | 1183 | } |
1170 | 1184 | } |
1171 | 1185 |
|
@@ -1183,6 +1197,15 @@ Socket.prototype.connect = function(...args) { |
1183 | 1197 | const options = normalized[0]; |
1184 | 1198 | const cb = normalized[1]; |
1185 | 1199 |
|
| 1200 | + if (cb !== null) { |
| 1201 | + this.once('connect', cb); |
| 1202 | + } |
| 1203 | + |
| 1204 | + // If the parent is already connecting, do not attempt to connect again |
| 1205 | + if (this._parent && this._parent.connecting) { |
| 1206 | + return this; |
| 1207 | + } |
| 1208 | + |
1186 | 1209 | // options.port === null will be checked later. |
1187 | 1210 | if (options.port === undefined && options.path == null) |
1188 | 1211 | throw new ERR_MISSING_ARGS(['options', 'port', 'path']); |
@@ -1207,10 +1230,6 @@ Socket.prototype.connect = function(...args) { |
1207 | 1230 | initSocketHandle(this); |
1208 | 1231 | } |
1209 | 1232 |
|
1210 | | - if (cb !== null) { |
1211 | | - this.once('connect', cb); |
1212 | | - } |
1213 | | - |
1214 | 1233 | this._unrefTimer(); |
1215 | 1234 |
|
1216 | 1235 | this.connecting = true; |
@@ -1583,7 +1602,47 @@ function afterConnect(status, handle, req, readable, writable) { |
1583 | 1602 | } |
1584 | 1603 | } |
1585 | 1604 |
|
| 1605 | +function addClientAbortSignalOption(self, options) { |
| 1606 | + validateAbortSignal(options.signal, 'options.signal'); |
| 1607 | + const { signal } = options; |
| 1608 | + |
| 1609 | + function onAbort() { |
| 1610 | + signal.removeEventListener('abort', onAbort); |
| 1611 | + self._aborted = true; |
| 1612 | + } |
| 1613 | + |
| 1614 | + if (signal.aborted) { |
| 1615 | + process.nextTick(onAbort); |
| 1616 | + } else { |
| 1617 | + process.nextTick(() => { |
| 1618 | + signal.addEventListener('abort', onAbort); |
| 1619 | + }); |
| 1620 | + } |
| 1621 | +} |
| 1622 | + |
| 1623 | +function createConnectionError(req, status) { |
| 1624 | + let details; |
| 1625 | + |
| 1626 | + if (req.localAddress && req.localPort) { |
| 1627 | + details = req.localAddress + ':' + req.localPort; |
| 1628 | + } |
| 1629 | + |
| 1630 | + const ex = exceptionWithHostPort(status, |
| 1631 | + 'connect', |
| 1632 | + req.address, |
| 1633 | + req.port, |
| 1634 | + details); |
| 1635 | + if (details) { |
| 1636 | + ex.localAddress = req.localAddress; |
| 1637 | + ex.localPort = req.localPort; |
| 1638 | + } |
| 1639 | + |
| 1640 | + return ex; |
| 1641 | +} |
| 1642 | + |
1586 | 1643 | function afterConnectMultiple(context, current, status, handle, req, readable, writable) { |
| 1644 | + debug('connect/multiple: connection attempt to %s:%s completed with status %s', req.address, req.port, status); |
| 1645 | + |
1587 | 1646 | // Make sure another connection is not spawned |
1588 | 1647 | clearTimeout(context[kTimeout]); |
1589 | 1648 |
|
@@ -1596,35 +1655,15 @@ function afterConnectMultiple(context, current, status, handle, req, readable, w |
1596 | 1655 |
|
1597 | 1656 | const self = context.socket; |
1598 | 1657 |
|
1599 | | - |
1600 | 1658 | // Some error occurred, add to the list of exceptions |
1601 | 1659 | if (status !== 0) { |
1602 | | - let details; |
1603 | | - if (req.localAddress && req.localPort) { |
1604 | | - details = req.localAddress + ':' + req.localPort; |
1605 | | - } |
1606 | | - const ex = exceptionWithHostPort(status, |
1607 | | - 'connect', |
1608 | | - req.address, |
1609 | | - req.port, |
1610 | | - details); |
1611 | | - if (details) { |
1612 | | - ex.localAddress = req.localAddress; |
1613 | | - ex.localPort = req.localPort; |
1614 | | - } |
1615 | | - |
1616 | | - ArrayPrototypePush(context.errors, ex); |
| 1660 | + ArrayPrototypePush(context.errors, createConnectionError(req, status)); |
1617 | 1661 |
|
1618 | 1662 | // Try the next address |
1619 | 1663 | internalConnectMultiple(context, status === UV_ECANCELED); |
1620 | 1664 | return; |
1621 | 1665 | } |
1622 | 1666 |
|
1623 | | - if (context.current > 1 && self[kReinitializeHandle]) { |
1624 | | - self[kReinitializeHandle](handle); |
1625 | | - handle = self._handle; |
1626 | | - } |
1627 | | - |
1628 | 1667 | if (hasObserver('net')) { |
1629 | 1668 | startPerf( |
1630 | 1669 | self, |
@@ -1633,17 +1672,18 @@ function afterConnectMultiple(context, current, status, handle, req, readable, w |
1633 | 1672 | ); |
1634 | 1673 | } |
1635 | 1674 |
|
1636 | | - afterConnect(status, handle, req, readable, writable); |
| 1675 | + afterConnect(status, self._handle, req, readable, writable); |
1637 | 1676 | } |
1638 | 1677 |
|
1639 | 1678 | function internalConnectMultipleTimeout(context, req, handle) { |
1640 | 1679 | debug('connect/multiple: connection to %s:%s timed out', req.address, req.port); |
1641 | 1680 | req.oncomplete = undefined; |
| 1681 | + ArrayPrototypePush(context.errors, createConnectionError(req, UV_ETIMEDOUT)); |
1642 | 1682 | handle.close(); |
1643 | 1683 | internalConnectMultiple(context); |
1644 | 1684 | } |
1645 | 1685 |
|
1646 | | -function addAbortSignalOption(self, options) { |
| 1686 | +function addServerAbortSignalOption(self, options) { |
1647 | 1687 | if (options?.signal === undefined) { |
1648 | 1688 | return; |
1649 | 1689 | } |
@@ -1932,7 +1972,7 @@ Server.prototype.listen = function(...args) { |
1932 | 1972 | listenInCluster(this, null, -1, -1, backlogFromArgs); |
1933 | 1973 | return this; |
1934 | 1974 | } |
1935 | | - addAbortSignalOption(this, options); |
| 1975 | + addServerAbortSignalOption(this, options); |
1936 | 1976 | // (handle[, backlog][, cb]) where handle is an object with a fd |
1937 | 1977 | if (typeof options.fd === 'number' && options.fd >= 0) { |
1938 | 1978 | listenInCluster(this, null, null, null, backlogFromArgs, options.fd); |
|