dgram: add support for UDP connected sockets

Added the `dgram.connect()` and `dgram.disconnect()` methods that
associate/disassociate a udp socket to/from a remote address.
It optimizes for cases where lots of packets are sent to the same
address.
Also added the `dgram.remoteAddress()` method to retrieve the associated
remote address.

PR-URL: https://github.com/nodejs/node/pull/26871
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

doc/api/dgram.md

@@ -49,6 +49,14 @@ added: v0.1.99
 The `'close'` event is emitted after a socket is closed with [`close()`][].
 Once triggered, no new `'message'` events will be emitted on this socket.
 
+### Event: 'connect'
+<!-- YAML
+added: REPLACEME
+-->
+
+The `'connect'` event is emitted after a socket is associated to a remote
+address as a result of a successful [`connect()`][] call.
+
 ### Event: 'error'
 <!-- YAML
 added: v0.1.99
@@ -237,6 +245,34 @@ added: v0.1.99
 Close the underlying socket and stop listening for data on it. If a callback is
 provided, it is added as a listener for the [`'close'`][] event.
 
+### socket.connect(port[, address][, callback])
+<!-- YAML
+added: REPLACEME
+-->
+
+* `port` {integer}
+* `address` {string}
+* `callback` {Function} Called when the connection is completed or on error.
+
+Associates the `dgram.Socket` to a remote address and port. Every
+message sent by this handle is automatically sent to that destination. Also,
+the socket will only receive messages from that remote peer.
+Trying to call `connect()` on an already connected socket will result
+in an [`ERR_SOCKET_DGRAM_IS_CONNECTED`][] exception. If `address` is not
+provided, `'127.0.0.1'` (for `udp4` sockets) or `'::1'` (for `udp6` sockets)
+will be used by default. Once the connection is complete, a `'connect'` event
+is emitted and the optional `callback` function is called. In case of failure,
+the `callback` is called or, failing this, an `'error'` event is emitted.
+
+### socket.disconnect()
+<!-- YAML
+added: REPLACEME
+-->
+
+A synchronous function that disassociates a connected `dgram.Socket` from
+its remote address. Trying to call `disconnect()` on an already disconnected
+socket will result in an [`ERR_SOCKET_DGRAM_NOT_CONNECTED`][] exception.
+
 ### socket.dropMembership(multicastAddress[, multicastInterface])
 <!-- YAML
 added: v0.6.9
@@ -283,7 +319,18 @@ Calling `socket.ref()` multiples times will have no additional effect.
 The `socket.ref()` method returns a reference to the socket so calls can be
 chained.
 
-### socket.send(msg[, offset, length], port[, address][, callback])
+### socket.remoteAddress()
+<!-- YAML
+added: REPLACEME
+-->
+
+* Returns: {Object}
+
+Returns an object containing the `address`, `family`, and `port` of the remote
+endpoint. It throws an [`ERR_SOCKET_DGRAM_NOT_CONNECTED`][] exception if the
+socket is not connected.
+
+### socket.send(msg[, offset, length][, port][, address][, callback])
 <!-- YAML
 added: v0.1.99
 changes:
@@ -301,6 +348,9 @@ changes:
     pr-url: https://github.com/nodejs/node/pull/4374
     description: The `msg` parameter can be an array now. Also, the `offset`
                  and `length` parameters are optional now.
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/26871
+    description: Added support for sending data on connected sockets.
 -->
 
 * `msg` {Buffer|Uint8Array|string|Array} Message to be sent.
@@ -310,8 +360,10 @@ changes:
 * `address` {string} Destination hostname or IP address.
 * `callback` {Function} Called when the message has been sent.
 
-Broadcasts a datagram on the socket. The destination `port` and `address` must
-be specified.
+Broadcasts a datagram on the socket.
+For connectionless sockets, the destination `port` and `address` must be
+specified. Connected sockets, on the other hand, will use their associated
+remote endpoint, so the `port` and `address` arguments must not be set.
 
 The `msg` argument contains the message to be sent.
 Depending on its type, different behavior can apply. If `msg` is a `Buffer`
@@ -375,6 +427,20 @@ application and operating system. It is important to run benchmarks to
 determine the optimal strategy on a case-by-case basis. Generally speaking,
 however, sending multiple buffers is faster.
 
+Example of sending a UDP packet using a socket connected to a port on
+`localhost`:
+
+```js
+const dgram = require('dgram');
+const message = Buffer.from('Some bytes');
+const client = dgram.createSocket('udp4');
+client.connect(41234, 'localhost', (err) => {
+  client.send(message, (err) => {
+    client.close();
+  });
+});
+```
+
 **A Note about UDP datagram size**
 
 The maximum size of an `IPv4/v6` datagram depends on the `MTU`
@@ -651,10 +717,13 @@ and `udp6` sockets). The bound address and port can be retrieved using
 
 [`'close'`]: #dgram_event_close
 [`Error`]: errors.html#errors_class_error
+[`ERR_SOCKET_DGRAM_IS_CONNECTED`]: errors.html#errors_err_socket_dgram_is_connected
+[`ERR_SOCKET_DGRAM_NOT_CONNECTED`]: errors.html#errors_err_socket_dgram_not_connected
 [`EventEmitter`]: events.html
 [`System Error`]: errors.html#errors_class_systemerror
 [`close()`]: #dgram_socket_close_callback
 [`cluster`]: cluster.html
+[`connect()`]: #dgram_socket_connect_port_address_callback
 [`dgram.Socket#bind()`]: #dgram_socket_bind_options_callback
 [`dgram.createSocket()`]: #dgram_dgram_createsocket_options_callback
 [`dns.lookup()`]: dns.html#dns_dns_lookup_hostname_options_callback

doc/api/errors.md

@@ -1645,6 +1645,17 @@ Data could be sent on a socket.
 
 An attempt was made to operate on an already closed socket.
 
+<a id="ERR_SOCKET_DGRAM_IS_CONNECTED"></a>
+### ERR_SOCKET_DGRAM_IS_CONNECTED
+
+A [`dgram.connect()`][] call was made on an already connected socket.
+
+<a id="ERR_SOCKET_DGRAM_NOT_CONNECTED"></a>
+### ERR_SOCKET_DGRAM_NOT_CONNECTED
+
+A [`dgram.disconnect()`][] or [`dgram.remoteAddress()`][] call was made on a
+disconnected socket.
+
 <a id="ERR_SOCKET_DGRAM_NOT_RUNNING"></a>
 ### ERR_SOCKET_DGRAM_NOT_RUNNING
 
@@ -2288,7 +2299,10 @@ such as `process.stdout.on('data')`.
 [`crypto.scrypt()`]: crypto.html#crypto_crypto_scrypt_password_salt_keylen_options_callback
 [`crypto.scryptSync()`]: crypto.html#crypto_crypto_scryptsync_password_salt_keylen_options
 [`crypto.timingSafeEqual()`]: crypto.html#crypto_crypto_timingsafeequal_a_b
+[`dgram.connect()`]: dgram.html#dgram_socket_connect_port_address_callback
 [`dgram.createSocket()`]: dgram.html#dgram_dgram_createsocket_options_callback
+[`dgram.disconnect()`]: dgram.html#dgram_socket_disconnect
+[`dgram.remoteAddress()`]: dgram.html#dgram_socket_remoteaddress
 [`errno`(3) man page]: http://man7.org/linux/man-pages/man3/errno.3.html
 [`fs.readFileSync`]: fs.html#fs_fs_readfilesync_path_options
 [`fs.readdir`]: fs.html#fs_fs_readdir_path_options_callback

lib/dgram.js

@@ -28,6 +28,9 @@ const {
   newHandle,
   guessHandleType,
 } = require('internal/dgram');
+const {
+  isLegalPort,
+} = require('internal/net');
 const {
   ERR_INVALID_ARG_TYPE,
   ERR_MISSING_ARGS,
@@ -36,6 +39,8 @@ const {
   ERR_SOCKET_BAD_PORT,
   ERR_SOCKET_BUFFER_SIZE,
   ERR_SOCKET_CANNOT_SEND,
+  ERR_SOCKET_DGRAM_IS_CONNECTED,
+  ERR_SOCKET_DGRAM_NOT_CONNECTED,
   ERR_SOCKET_DGRAM_NOT_RUNNING,
   ERR_INVALID_FD_TYPE
 } = errors.codes;
@@ -64,6 +69,10 @@ const BIND_STATE_UNBOUND = 0;
 const BIND_STATE_BINDING = 1;
 const BIND_STATE_BOUND = 2;
 
+const CONNECT_STATE_DISCONNECTED = 0;
+const CONNECT_STATE_CONNECTING = 1;
+const CONNECT_STATE_CONNECTED = 2;
+
 const RECV_BUFFER = true;
 const SEND_BUFFER = false;
 
@@ -101,6 +110,7 @@ function Socket(type, listener) {
     handle,
     receiving: false,
     bindState: BIND_STATE_UNBOUND,
+    connectState: CONNECT_STATE_DISCONNECTED,
     queue: undefined,
     reuseAddr: options && options.reuseAddr, // Use UV_UDP_REUSEADDR if true.
     ipv6Only: options && options.ipv6Only,
@@ -148,6 +158,9 @@ function replaceHandle(self, newHandle) {
   // Replace the existing handle by the handle we got from master.
   oldHandle.close();
   state.handle = newHandle;
+  // Check if the udp handle was connected and set the state accordingly
+  if (isConnected(self))
+    state.connectState = CONNECT_STATE_CONNECTED;
 }
 
 function bufferSize(self, size, buffer) {
@@ -238,6 +251,10 @@ Socket.prototype.bind = function(port_, address_ /* , callback */) {
     if (err)
       throw errnoException(err, 'open');
 
+    // Check if the udp handle was connected and set the state accordingly
+    if (isConnected(this))
+      state.connectState = CONNECT_STATE_CONNECTED;
+
     startListening(this);
     return this;
   }
@@ -313,6 +330,106 @@ Socket.prototype.bind = function(port_, address_ /* , callback */) {
 };
 
 
+function validatePort(port) {
+  const legal = isLegalPort(port);
+  if (legal)
+    port = port | 0;
+
+  if (!legal || port === 0)
+    throw new ERR_SOCKET_BAD_PORT(port);
+
+  return port;
+}
+
+
+Socket.prototype.connect = function(port, address, callback) {
+  port = validatePort(port);
+  if (typeof address === 'function') {
+    callback = address;
+    address = '';
+  } else if (address === undefined) {
+    address = '';
+  }
+
+  validateString(address, 'address');
+
+  const state = this[kStateSymbol];
+
+  if (state.connectState !== CONNECT_STATE_DISCONNECTED)
+    throw new ERR_SOCKET_DGRAM_IS_CONNECTED();
+
+  state.connectState = CONNECT_STATE_CONNECTING;
+  if (state.bindState === BIND_STATE_UNBOUND)
+    this.bind({ port: 0, exclusive: true }, null);
+
+  if (state.bindState !== BIND_STATE_BOUND) {
+    enqueue(this, _connect.bind(this, port, address, callback));
+    return;
+  }
+
+  _connect.call(this, port, address, callback);
+};
+
+
+function _connect(port, address, callback) {
+  const state = this[kStateSymbol];
+  if (callback)
+    this.once('connect', callback);
+
+  const afterDns = (ex, ip) => {
+    defaultTriggerAsyncIdScope(
+      this[async_id_symbol],
+      doConnect,
+      ex, this, ip, address, port, callback
+    );
+  };
+
+  state.handle.lookup(address, afterDns);
+}
+
+
+function doConnect(ex, self, ip, address, port, callback) {
+  const state = self[kStateSymbol];
+  if (!state.handle)
+    return;
+
+  if (!ex) {
+    const err = state.handle.connect(ip, port);
+    if (err) {
+      ex = exceptionWithHostPort(err, 'connect', address, port);
+    }
+  }
+
+  if (ex) {
+    state.connectState = CONNECT_STATE_DISCONNECTED;
+    return process.nextTick(() => {
+      if (callback) {
+        self.removeListener('connect', callback);
+        callback(ex);
+      } else {
+        self.emit('error', ex);
+      }
+    });
+  }
+
+  state.connectState = CONNECT_STATE_CONNECTED;
+  process.nextTick(() => self.emit('connect'));
+}
+
+
+Socket.prototype.disconnect = function() {
+  const state = this[kStateSymbol];
+  if (state.connectState !== CONNECT_STATE_CONNECTED)
+    throw new ERR_SOCKET_DGRAM_NOT_CONNECTED();
+
+  const err = state.handle.disconnect();
+  if (err)
+    throw errnoException(err, 'connect');
+  else
+    state.connectState = CONNECT_STATE_DISCONNECTED;
+};
+
+
 // Thin wrapper around `send`, here for compatibility with dgram_legacy.js
 Socket.prototype.sendto = function(buffer,
                                    offset,
@@ -398,8 +515,18 @@ function clearQueue() {
     queue[i]();
 }
 
+function isConnected(self) {
+  try {
+    this.remoteAddress();
+    return true;
+  } catch {
+    return false;
+  }
+}
+
 
 // valid combinations
+// For connectionless sockets
 // send(buffer, offset, length, port, address, callback)
 // send(buffer, offset, length, port, address)
 // send(buffer, offset, length, port, callback)
@@ -408,20 +535,39 @@ function clearQueue() {
 // send(bufferOrList, port, address)
 // send(bufferOrList, port, callback)
 // send(bufferOrList, port)
+// For connected sockets
+// send(buffer, offset, length, callback)
+// send(buffer, offset, length)
+// send(bufferOrList, callback)
+// send(bufferOrList)
 Socket.prototype.send = function(buffer,
                                  offset,
                                  length,
                                  port,
                                  address,
                                  callback) {
-  let list;
 
-  if (address || (port && typeof port !== 'function')) {
+  let list;
+  const state = this[kStateSymbol];
+  const connected = state.connectState === CONNECT_STATE_CONNECTED;
+  if (!connected) {
+    if (address || (port && typeof port !== 'function')) {
+      buffer = sliceBuffer(buffer, offset, length);
+    } else {
+      callback = port;
+      port = offset;
+      address = length;
+    }
+  } else if (typeof length === 'number') {
     buffer = sliceBuffer(buffer, offset, length);
+    if (typeof port === 'function') {
+      callback = port;
+      port = null;
+    } else if (port || address) {
+      throw new ERR_SOCKET_DGRAM_IS_CONNECTED();
+    }
   } else {
-    callback = port;
-    port = offset;
-    address = length;
+    callback = offset;
   }
 
   if (!Array.isArray(buffer)) {
@@ -439,9 +585,8 @@ Socket.prototype.send = function(buffer,
                                    ['Buffer', 'string'], buffer);
   }
 
-  port = port >>> 0;
-  if (port === 0 || port > 65535)
-    throw new ERR_SOCKET_BAD_PORT(port);
+  if (!connected)
+    port = validatePort(port);
 
   // Normalize callback so it's either a function or undefined but not anything
   // else.
@@ -457,8 +602,6 @@ Socket.prototype.send = function(buffer,
 
   healthCheck(this);
 
-  const state = this[kStateSymbol];
-
   if (state.bindState === BIND_STATE_UNBOUND)
     this.bind({ port: 0, exclusive: true }, null);
 
@@ -480,7 +623,11 @@ Socket.prototype.send = function(buffer,
     );
   };
 
-  state.handle.lookup(address, afterDns);
+  if (!connected) {
+    state.handle.lookup(address, afterDns);
+  } else {
+    afterDns(null, null);
+  }
 };
 
 function doSend(ex, self, ip, list, address, port, callback) {
@@ -507,12 +654,11 @@ function doSend(ex, self, ip, list, address, port, callback) {
     req.oncomplete = afterSend;
   }
 
-  const err = state.handle.send(req,
-                                list,
-                                list.length,
-                                port,
-                                ip,
-                                !!callback);
+  let err;
+  if (port)
+    err = state.handle.send(req, list, list.length, port, ip, !!callback);
+  else
+    err = state.handle.send(req, list, list.length, !!callback);
 
   if (err && callback) {
     // Don't emit as error, dgram_legacy.js compatibility
@@ -573,6 +719,21 @@ Socket.prototype.address = function() {
   return out;
 };
 
+Socket.prototype.remoteAddress = function() {
+  healthCheck(this);
+
+  const state = this[kStateSymbol];
+  if (state.connectState !== CONNECT_STATE_CONNECTED)
+    throw new ERR_SOCKET_DGRAM_NOT_CONNECTED();
+
+  var out = {};
+  var err = state.handle.getpeername(out);
+  if (err)
+    throw errnoException(err, 'getpeername');
+
+  return out;
+};
+
 
 Socket.prototype.setBroadcast = function(arg) {
   const err = this[kStateSymbol].handle.setBroadcast(arg ? 1 : 0);

lib/internal/dgram.js

@@ -45,6 +45,7 @@ function newHandle(type, lookup) {
 
     handle.lookup = lookup6.bind(handle, lookup);
     handle.bind = handle.bind6;
+    handle.connect = handle.connect6;
     handle.send = handle.send6;
     return handle;
   }

lib/internal/errors.js

@@ -1019,6 +1019,8 @@ E('ERR_SOCKET_BUFFER_SIZE',
   SystemError);
 E('ERR_SOCKET_CANNOT_SEND', 'Unable to send data', Error);
 E('ERR_SOCKET_CLOSED', 'Socket is closed', Error);
+E('ERR_SOCKET_DGRAM_IS_CONNECTED', 'Already connected', Error);
+E('ERR_SOCKET_DGRAM_NOT_CONNECTED', 'Not connected', Error);
 E('ERR_SOCKET_DGRAM_NOT_RUNNING', 'Not running', Error);
 E('ERR_SRI_PARSE',
   'Subresource Integrity string %s had an unexpected at %d',

src/udp_wrap.cc

@@ -114,11 +114,16 @@ void UDPWrap::Initialize(Local<Object> target,
 
   env->SetProtoMethod(t, "open", Open);
   env->SetProtoMethod(t, "bind", Bind);
+  env->SetProtoMethod(t, "connect", Connect);
   env->SetProtoMethod(t, "send", Send);
   env->SetProtoMethod(t, "bind6", Bind6);
+  env->SetProtoMethod(t, "connect6", Connect6);
   env->SetProtoMethod(t, "send6", Send6);
+  env->SetProtoMethod(t, "disconnect", Disconnect);
   env->SetProtoMethod(t, "recvStart", RecvStart);
   env->SetProtoMethod(t, "recvStop", RecvStop);
+  env->SetProtoMethod(t, "getpeername",
+                      GetSockOrPeerName<UDPWrap, uv_udp_getpeername>);
   env->SetProtoMethod(t, "getsockname",
                       GetSockOrPeerName<UDPWrap, uv_udp_getsockname>);
   env->SetProtoMethod(t, "addMembership", AddMembership);
@@ -215,6 +220,30 @@ void UDPWrap::DoBind(const FunctionCallbackInfo<Value>& args, int family) {
 }
 
 
+void UDPWrap::DoConnect(const FunctionCallbackInfo<Value>& args, int family) {
+  UDPWrap* wrap;
+  ASSIGN_OR_RETURN_UNWRAP(&wrap,
+                          args.Holder(),
+                          args.GetReturnValue().Set(UV_EBADF));
+
+  CHECK_EQ(args.Length(), 2);
+
+  node::Utf8Value address(args.GetIsolate(), args[0]);
+  Local<Context> ctx = args.GetIsolate()->GetCurrentContext();
+  uint32_t port;
+  if (!args[1]->Uint32Value(ctx).To(&port))
+    return;
+  struct sockaddr_storage addr_storage;
+  int err = sockaddr_for_family(family, address.out(), port, &addr_storage);
+  if (err == 0) {
+    err = uv_udp_connect(&wrap->handle_,
+                         reinterpret_cast<const sockaddr*>(&addr_storage));
+  }
+
+  args.GetReturnValue().Set(err);
+}
+
+
 void UDPWrap::Open(const FunctionCallbackInfo<Value>& args) {
   UDPWrap* wrap;
   ASSIGN_OR_RETURN_UNWRAP(&wrap,
@@ -273,6 +302,30 @@ void UDPWrap::BufferSize(const FunctionCallbackInfo<Value>& args) {
   args.GetReturnValue().Set(size);
 }
 
+
+void UDPWrap::Connect(const FunctionCallbackInfo<Value>& args) {
+  DoConnect(args, AF_INET);
+}
+
+
+void UDPWrap::Connect6(const FunctionCallbackInfo<Value>& args) {
+  DoConnect(args, AF_INET6);
+}
+
+
+void UDPWrap::Disconnect(const FunctionCallbackInfo<Value>& args) {
+  UDPWrap* wrap;
+  ASSIGN_OR_RETURN_UNWRAP(&wrap,
+                          args.Holder(),
+                          args.GetReturnValue().Set(UV_EBADF));
+
+  CHECK_EQ(args.Length(), 0);
+
+  int err = uv_udp_connect(&wrap->handle_, nullptr);
+
+  args.GetReturnValue().Set(err);
+}
+
 #define X(name, fn)                                                            \
   void UDPWrap::name(const FunctionCallbackInfo<Value>& args) {                \
     UDPWrap* wrap = Unwrap<UDPWrap>(args.Holder());                            \
@@ -353,22 +406,28 @@ void UDPWrap::DoSend(const FunctionCallbackInfo<Value>& args, int family) {
                           args.Holder(),
                           args.GetReturnValue().Set(UV_EBADF));
 
-  // send(req, list, list.length, port, address, hasCallback)
+  CHECK(args.Length() == 4 || args.Length() == 6);
   CHECK(args[0]->IsObject());
   CHECK(args[1]->IsArray());
   CHECK(args[2]->IsUint32());
-  CHECK(args[3]->IsUint32());
-  CHECK(args[4]->IsString());
-  CHECK(args[5]->IsBoolean());
+
+  bool sendto = args.Length() == 6;
+  if (sendto) {
+    // send(req, list, list.length, port, address, hasCallback)
+    CHECK(args[3]->IsUint32());
+    CHECK(args[4]->IsString());
+    CHECK(args[5]->IsBoolean());
+  } else {
+    // send(req, list, list.length, hasCallback)
+    CHECK(args[3]->IsBoolean());
+  }
 
   Local<Object> req_wrap_obj = args[0].As<Object>();
   Local<Array> chunks = args[1].As<Array>();
   // it is faster to fetch the length of the
   // array in js-land
   size_t count = args[2].As<Uint32>()->Value();
-  const unsigned short port = args[3].As<Uint32>()->Value();
-  node::Utf8Value address(env->isolate(), args[4]);
-  const bool have_callback = args[5]->IsTrue();
+  const bool have_callback = sendto ? args[5]->IsTrue() : args[3]->IsTrue();
 
   SendWrap* req_wrap;
   {
@@ -391,14 +450,24 @@ void UDPWrap::DoSend(const FunctionCallbackInfo<Value>& args, int family) {
 
   req_wrap->msg_size = msg_size;
 
-  struct sockaddr_storage addr_storage;
-  int err = sockaddr_for_family(family, address.out(), port, &addr_storage);
+  int err = 0;
+  sockaddr* addr = nullptr;
+  if (sendto) {
+    struct sockaddr_storage addr_storage;
+    const unsigned short port = args[3].As<Uint32>()->Value();
+    node::Utf8Value address(env->isolate(), args[4]);
+    err = sockaddr_for_family(family, address.out(), port, &addr_storage);
+    if (err == 0) {
+      addr = reinterpret_cast<sockaddr*>(&addr_storage);
+    }
+  }
+
   if (err == 0) {
     err = req_wrap->Dispatch(uv_udp_send,
                              &wrap->handle_,
                              *bufs,
                              count,
-                             reinterpret_cast<const sockaddr*>(&addr_storage),
+                             addr,
                              OnSend);
   }
 

src/udp_wrap.h

@@ -45,9 +45,12 @@ class UDPWrap: public HandleWrap {
   static void New(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Open(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Bind(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void Connect(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Send(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Bind6(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void Connect6(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void Send6(const v8::FunctionCallbackInfo<v8::Value>& args);
+  static void Disconnect(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void RecvStart(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void RecvStop(const v8::FunctionCallbackInfo<v8::Value>& args);
   static void AddMembership(const v8::FunctionCallbackInfo<v8::Value>& args);
@@ -79,6 +82,8 @@ class UDPWrap: public HandleWrap {
 
   static void DoBind(const v8::FunctionCallbackInfo<v8::Value>& args,
                      int family);
+  static void DoConnect(const v8::FunctionCallbackInfo<v8::Value>& args,
+                     int family);
   static void DoSend(const v8::FunctionCallbackInfo<v8::Value>& args,
                      int family);
   static void SetMembership(const v8::FunctionCallbackInfo<v8::Value>& args,

test/parallel/test-dgram-connect-send-callback-buffer-length.js

@@ -0,0 +1,24 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+
+const dgram = require('dgram');
+const client = dgram.createSocket('udp4');
+
+const buf = Buffer.allocUnsafe(256);
+const offset = 20;
+const len = buf.length - offset;
+
+const messageSent = common.mustCall(function messageSent(err, bytes) {
+  assert.ifError(err);
+  assert.notStrictEqual(bytes, buf.length);
+  assert.strictEqual(bytes, buf.length - offset);
+  client.close();
+});
+
+client.bind(0, () => {
+  client.connect(client.address().port, common.mustCall(() => {
+    client.send(buf, offset, len, messageSent);
+  }));
+});

test/parallel/test-dgram-connect-send-callback-buffer.js

@@ -0,0 +1,21 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const dgram = require('dgram');
+
+const client = dgram.createSocket('udp4');
+
+const buf = Buffer.allocUnsafe(256);
+
+const onMessage = common.mustCall(function(err, bytes) {
+  assert.ifError(err);
+  assert.strictEqual(bytes, buf.length);
+  client.close();
+});
+
+client.bind(0, () => {
+  client.connect(client.address().port, common.mustCall(() => {
+    client.send(buf, onMessage);
+  }));
+});

test/parallel/test-dgram-connect-send-callback-multi-buffer.js

@@ -0,0 +1,29 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const dgram = require('dgram');
+
+const client = dgram.createSocket('udp4');
+
+const messageSent = common.mustCall((err, bytes) => {
+  assert.strictEqual(bytes, buf1.length + buf2.length);
+});
+
+const buf1 = Buffer.alloc(256, 'x');
+const buf2 = Buffer.alloc(256, 'y');
+
+client.on('listening', common.mustCall(() => {
+  const port = client.address().port;
+  client.connect(port, common.mustCall(() => {
+    client.send([buf1, buf2], messageSent);
+  }));
+}));
+
+client.on('message', common.mustCall((buf, info) => {
+  const expected = Buffer.concat([buf1, buf2]);
+  assert.ok(buf.equals(expected), 'message was received correctly');
+  client.close();
+}));
+
+client.bind(0);

test/parallel/test-dgram-connect-send-default-host.js

@@ -0,0 +1,48 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const dgram = require('dgram');
+
+const client = dgram.createSocket('udp4');
+const server = dgram.createSocket('udp4');
+
+const toSend = [Buffer.alloc(256, 'x'),
+                Buffer.alloc(256, 'y'),
+                Buffer.alloc(256, 'z'),
+                'hello'];
+
+const received = [];
+
+server.on('listening', common.mustCall(() => {
+  const port = server.address().port;
+  client.connect(port, (err) => {
+    assert.ifError(err);
+    client.send(toSend[0], 0, toSend[0].length);
+    client.send(toSend[1]);
+    client.send([toSend[2]]);
+    client.send(toSend[3], 0, toSend[3].length);
+
+    client.send(new Uint8Array(toSend[0]), 0, toSend[0].length);
+    client.send(new Uint8Array(toSend[1]));
+    client.send([new Uint8Array(toSend[2])]);
+    client.send(new Uint8Array(Buffer.from(toSend[3])),
+                0, toSend[3].length);
+  });
+}));
+
+server.on('message', common.mustCall((buf, info) => {
+  received.push(buf.toString());
+
+  if (received.length === toSend.length * 2) {
+    // The replies may arrive out of order -> sort them before checking.
+    received.sort();
+
+    const expected = toSend.concat(toSend).map(String).sort();
+    assert.deepStrictEqual(received, expected);
+    client.close();
+    server.close();
+  }
+}, toSend.length * 2));
+
+server.bind(0);

test/parallel/test-dgram-connect-send-empty-array.js

@@ -0,0 +1,22 @@
+'use strict';
+
+const common = require('../common');
+
+const assert = require('assert');
+const dgram = require('dgram');
+
+const client = dgram.createSocket('udp4');
+
+client.on('message', common.mustCall((buf, info) => {
+  const expected = Buffer.alloc(0);
+  assert.ok(buf.equals(expected), `Expected empty message but got ${buf}`);
+  client.close();
+}));
+
+client.on('listening', common.mustCall(() => {
+  client.connect(client.address().port,
+                 common.localhostIPv4,
+                 common.mustCall(() => client.send([])));
+}));
+
+client.bind(0);

test/parallel/test-dgram-connect-send-empty-buffer.js

@@ -0,0 +1,20 @@
+'use strict';
+const common = require('../common');
+
+const assert = require('assert');
+const dgram = require('dgram');
+
+const client = dgram.createSocket('udp4');
+
+client.bind(0, common.mustCall(function() {
+  const port = this.address().port;
+  client.connect(port, common.mustCall(() => {
+    const buf = Buffer.alloc(0);
+    client.send(buf, 0, 0, common.mustCall());
+  }));
+
+  client.on('message', common.mustCall((buffer) => {
+    assert.strictEqual(buffer.length, 0);
+    client.close();
+  }));
+}));

test/parallel/test-dgram-connect-send-empty-packet.js

@@ -0,0 +1,28 @@
+'use strict';
+const common = require('../common');
+
+const assert = require('assert');
+const dgram = require('dgram');
+
+const client = dgram.createSocket('udp4');
+
+client.bind(0, common.mustCall(function() {
+  client.connect(client.address().port, common.mustCall(() => {
+    client.on('message', common.mustCall(callback));
+    const buf = Buffer.alloc(1);
+
+    const interval = setInterval(function() {
+      client.send(buf, 0, 0, common.mustCall(callback));
+    }, 10);
+
+    function callback(firstArg) {
+      // If client.send() callback, firstArg should be null.
+      // If client.on('message') listener, firstArg should be a 0-length buffer.
+      if (firstArg instanceof Buffer) {
+        assert.strictEqual(firstArg.length, 0);
+        clearInterval(interval);
+        client.close();
+      }
+    }
+  }));
+}));

test/parallel/test-dgram-connect-send-multi-buffer-copy.js

@@ -0,0 +1,30 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const dgram = require('dgram');
+
+const client = dgram.createSocket('udp4');
+
+const onMessage = common.mustCall(common.mustCall((err, bytes) => {
+  assert.ifError(err);
+  assert.strictEqual(bytes, buf1.length + buf2.length);
+}));
+
+const buf1 = Buffer.alloc(256, 'x');
+const buf2 = Buffer.alloc(256, 'y');
+
+client.on('listening', function() {
+  const toSend = [buf1, buf2];
+  client.connect(client.address().port, common.mustCall(() => {
+    client.send(toSend, onMessage);
+  }));
+});
+
+client.on('message', common.mustCall(function onMessage(buf, info) {
+  const expected = Buffer.concat([buf1, buf2]);
+  assert.ok(buf.equals(expected), 'message was received correctly');
+  client.close();
+}));
+
+client.bind(0);

test/parallel/test-dgram-connect-send-multi-string-array.js

@@ -0,0 +1,17 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const dgram = require('dgram');
+const socket = dgram.createSocket('udp4');
+const data = ['foo', 'bar', 'baz'];
+
+socket.on('message', common.mustCall((msg, rinfo) => {
+  socket.close();
+  assert.deepStrictEqual(msg.toString(), data.join(''));
+}));
+
+socket.bind(0, () => {
+  socket.connect(socket.address().port, common.mustCall(() => {
+    socket.send(data);
+  }));
+});

test/parallel/test-dgram-connect.js

@@ -0,0 +1,76 @@
+'use strict';
+
+const common = require('../common');
+const { addresses } = require('../common/internet');
+const assert = require('assert');
+const dgram = require('dgram');
+
+const PORT = 12345;
+
+const client = dgram.createSocket('udp4');
+client.connect(PORT, common.mustCall(() => {
+  const remoteAddr = client.remoteAddress();
+  assert.strictEqual(remoteAddr.port, PORT);
+  assert.throws(() => {
+    client.connect(PORT, common.mustNotCall());
+  }, {
+    name: 'Error',
+    message: 'Already connected',
+    code: 'ERR_SOCKET_DGRAM_IS_CONNECTED'
+  });
+
+  client.disconnect();
+  assert.throws(() => {
+    client.disconnect();
+  }, {
+    name: 'Error',
+    message: 'Not connected',
+    code: 'ERR_SOCKET_DGRAM_NOT_CONNECTED'
+  });
+
+  assert.throws(() => {
+    client.remoteAddress();
+  }, {
+    name: 'Error',
+    message: 'Not connected',
+    code: 'ERR_SOCKET_DGRAM_NOT_CONNECTED'
+  });
+
+  client.connect(PORT, addresses.INVALID_HOST, common.mustCall((err) => {
+    assert.ok(err.code === 'ENOTFOUND' || err.code === 'EAI_AGAIN');
+
+    client.once('error', common.mustCall((err) => {
+      assert.ok(err.code === 'ENOTFOUND' || err.code === 'EAI_AGAIN');
+      client.once('connect', common.mustCall(() => client.close()));
+      client.connect(PORT);
+    }));
+
+    client.connect(PORT, addresses.INVALID_HOST);
+  }));
+}));
+
+assert.throws(() => {
+  client.connect(PORT);
+}, {
+  name: 'Error',
+  message: 'Already connected',
+  code: 'ERR_SOCKET_DGRAM_IS_CONNECTED'
+});
+
+assert.throws(() => {
+  client.disconnect();
+}, {
+  name: 'Error',
+  message: 'Not connected',
+  code: 'ERR_SOCKET_DGRAM_NOT_CONNECTED'
+});
+
+[ 0, null, 78960, undefined ].forEach((port) => {
+  assert.throws(() => {
+    client.connect(port);
+  }, {
+    name: 'RangeError',
+    message: /^Port should be >= 0 and < 65536/,
+    code: 'ERR_SOCKET_BAD_PORT'
+  });
+});

test/parallel/test-dgram-send-bad-arguments.js

@@ -28,40 +28,77 @@ const buf = Buffer.from('test');
 const host = '127.0.0.1';
 const sock = dgram.createSocket('udp4');
 
-// First argument should be a buffer.
-common.expectsError(
-  () => { sock.send(); },
-  {
-    code: 'ERR_INVALID_ARG_TYPE',
-    type: TypeError,
-    message: 'The "buffer" argument must be one of type ' +
-    'Buffer, Uint8Array, or string. Received type undefined'
-  }
-);
+function checkArgs(connected) {
+  // First argument should be a buffer.
+  common.expectsError(
+    () => { sock.send(); },
+    {
+      code: 'ERR_INVALID_ARG_TYPE',
+      type: TypeError,
+      message: 'The "buffer" argument must be one of type ' +
+      'Buffer, Uint8Array, or string. Received type undefined'
+    }
+  );
 
-// send(buf, offset, length, port, host)
-assert.throws(() => { sock.send(buf, 1, 1, -1, host); }, RangeError);
-assert.throws(() => { sock.send(buf, 1, 1, 0, host); }, RangeError);
-assert.throws(() => { sock.send(buf, 1, 1, 65536, host); }, RangeError);
+  // send(buf, offset, length, port, host)
+  if (connected) {
+    common.expectsError(
+      () => { sock.send(buf, 1, 1, -1, host); },
+      {
+        code: 'ERR_SOCKET_DGRAM_IS_CONNECTED',
+        type: Error,
+        message: 'Already connected'
+      }
+    );
 
-// send(buf, port, host)
-common.expectsError(
-  () => { sock.send(23, 12345, host); },
-  {
-    code: 'ERR_INVALID_ARG_TYPE',
-    type: TypeError,
-    message: 'The "buffer" argument must be one of type ' +
-    'Buffer, Uint8Array, or string. Received type number'
-  }
-);
+    common.expectsError(
+      () => { sock.send(buf, 1, 1, 0, host); },
+      {
+        code: 'ERR_SOCKET_DGRAM_IS_CONNECTED',
+        type: Error,
+        message: 'Already connected'
+      }
+    );
 
-// send([buf1, ..], port, host)
-common.expectsError(
-  () => { sock.send([buf, 23], 12345, host); },
-  {
-    code: 'ERR_INVALID_ARG_TYPE',
-    type: TypeError,
-    message: 'The "buffer list arguments" argument must be one of type ' +
-    'Buffer or string. Received type object'
+    common.expectsError(
+      () => { sock.send(buf, 1, 1, 65536, host); },
+      {
+        code: 'ERR_SOCKET_DGRAM_IS_CONNECTED',
+        type: Error,
+        message: 'Already connected'
+      }
+    );
+  } else {
+    assert.throws(() => { sock.send(buf, 1, 1, -1, host); }, RangeError);
+    assert.throws(() => { sock.send(buf, 1, 1, 0, host); }, RangeError);
+    assert.throws(() => { sock.send(buf, 1, 1, 65536, host); }, RangeError);
   }
-);
+
+  // send(buf, port, host)
+  common.expectsError(
+    () => { sock.send(23, 12345, host); },
+    {
+      code: 'ERR_INVALID_ARG_TYPE',
+      type: TypeError,
+      message: 'The "buffer" argument must be one of type ' +
+      'Buffer, Uint8Array, or string. Received type number'
+    }
+  );
+
+  // send([buf1, ..], port, host)
+  common.expectsError(
+    () => { sock.send([buf, 23], 12345, host); },
+    {
+      code: 'ERR_INVALID_ARG_TYPE',
+      type: TypeError,
+      message: 'The "buffer list arguments" argument must be one of type ' +
+      'Buffer or string. Received type object'
+    }
+  );
+}
+
+checkArgs();
+sock.connect(12345, common.mustCall(() => {
+  checkArgs(true);
+  sock.close();
+}));