lib/dgram.js

   1 'use strict';
   2
   3 const assert = require('assert');
   4 const Buffer = require('buffer').Buffer;
   5 const util = require('util');
   6 const EventEmitter = require('events');
   7 const constants = require('constants');
   8
   9 const UDP = process.binding('udp_wrap').UDP;
  10 const SendWrap = process.binding('udp_wrap').SendWrap;
  11
  12 const BIND_STATE_UNBOUND = 0;
  13 const BIND_STATE_BINDING = 1;
  14 const BIND_STATE_BOUND = 2;
  15
  16 // lazily loaded
  17 var cluster = null;
  18 var dns = null;
  19
  20 const errnoException = util._errnoException;
  21 const exceptionWithHostPort = util._exceptionWithHostPort;
  22
  23 function lookup(address, family, callback) {
  24   if (!dns)
  25     dns = require('dns');
  26
  27   return dns.lookup(address, family, callback);
  28 }
  29
  30
  31 function lookup4(address, callback) {
  32   return lookup(address || '0.0.0.0', 4, callback);
  33 }
  34
  35
  36 function lookup6(address, callback) {
  37   return lookup(address || '::0', 6, callback);
  38 }
  39
  40
  41 function newHandle(type) {
  42   if (type == 'udp4') {
  43     var handle = new UDP();
  44     handle.lookup = lookup4;
  45     return handle;
  46   }
  47
  48   if (type == 'udp6') {
  49     var handle = new UDP();
  50     handle.lookup = lookup6;
  51     handle.bind = handle.bind6;
  52     handle.send = handle.send6;
  53     return handle;
  54   }
  55
  56   if (type == 'unix_dgram')
  57     throw new Error('unix_dgram sockets are not supported any more.');
  58
  59   throw new Error('Bad socket type specified. Valid types are: udp4, udp6');
  60 }
  61
  62
  63 exports._createSocketHandle = function(address, port, addressType, fd, flags) {
  64   // Opening an existing fd is not supported for UDP handles.
  65   assert(typeof fd !== 'number' || fd < 0);
  66
  67   var handle = newHandle(addressType);
  68
  69   if (port || address) {
  70     var err = handle.bind(address, port || 0, flags);
  71     if (err) {
  72       handle.close();
  73       return err;
  74     }
  75   }
  76
  77   return handle;
  78 };
  79
  80
  81 function Socket(type, listener) {
  82   EventEmitter.call(this);
  83
  84   if (typeof type === 'object') {
  85     var options = type;
  86     type = options.type;
  87   }
  88
  89   var handle = newHandle(type);
  90   handle.owner = this;
  91
  92   this._handle = handle;
  93   this._receiving = false;
  94   this._bindState = BIND_STATE_UNBOUND;
  95   this.type = type;
  96   this.fd = null; // compatibility hack
  97
  98   // If true - UV_UDP_REUSEADDR flag will be set
  99   this._reuseAddr = options && options.reuseAddr;
 100
 101   if (typeof listener === 'function')
 102     this.on('message', listener);
 103 }
 104 util.inherits(Socket, EventEmitter);
 105 exports.Socket = Socket;
 106
 107
 108 exports.createSocket = function(type, listener) {
 109   return new Socket(type, listener);
 110 };
 111
 112
 113 function startListening(socket) {
 114   socket._handle.onmessage = onMessage;
 115   // Todo: handle errors
 116   socket._handle.recvStart();
 117   socket._receiving = true;
 118   socket._bindState = BIND_STATE_BOUND;
 119   socket.fd = -42; // compatibility hack
 120
 121   socket.emit('listening');
 122 }
 123
 124 function replaceHandle(self, newHandle) {
 125
 126   // Set up the handle that we got from master.
 127   newHandle.lookup = self._handle.lookup;
 128   newHandle.bind = self._handle.bind;
 129   newHandle.send = self._handle.send;
 130   newHandle.owner = self;
 131
 132   // Replace the existing handle by the handle we got from master.
 133   self._handle.close();
 134   self._handle = newHandle;
 135 }
 136
 137 Socket.prototype.bind = function(port /*, address, callback*/) {
 138   var self = this;
 139
 140   self._healthCheck();
 141
 142   if (this._bindState != BIND_STATE_UNBOUND)
 143     throw new Error('Socket is already bound');
 144
 145   this._bindState = BIND_STATE_BINDING;
 146
 147   if (typeof arguments[arguments.length - 1] === 'function')
 148     self.once('listening', arguments[arguments.length - 1]);
 149
 150   if (port instanceof UDP) {
 151     replaceHandle(self, port);
 152     startListening(self);
 153     return self;
 154   }
 155
 156   var address;
 157   var exclusive;
 158
 159   if (port !== null && typeof port === 'object') {
 160     address = port.address || '';
 161     exclusive = !!port.exclusive;
 162     port = port.port;
 163   } else {
 164     address = typeof arguments[1] === 'function' ? '' : arguments[1];
 165     exclusive = false;
 166   }
 167
 168   // resolve address first
 169   self._handle.lookup(address, function(err, ip) {
 170     if (err) {
 171       self._bindState = BIND_STATE_UNBOUND;
 172       self.emit('error', err);
 173       return;
 174     }
 175
 176     if (!cluster)
 177       cluster = require('cluster');
 178
 179     var flags = 0;
 180     if (self._reuseAddr)
 181       flags |= constants.UV_UDP_REUSEADDR;
 182
 183     if (cluster.isWorker && !exclusive) {
 184       function onHandle(err, handle) {
 185         if (err) {
 186           var ex = exceptionWithHostPort(err, 'bind', ip, port);
 187           self.emit('error', ex);
 188           self._bindState = BIND_STATE_UNBOUND;
 189           return;
 190         }
 191
 192         if (!self._handle)
 193           // handle has been closed in the mean time.
 194           return handle.close();
 195
 196         replaceHandle(self, handle);
 197         startListening(self);
 198       }
 199       cluster._getServer(self, {
 200         address: ip,
 201         port: port,
 202         addressType: self.type,
 203         fd: -1,
 204         flags: flags
 205       }, onHandle);
 206
 207     } else {
 208       if (!self._handle)
 209         return; // handle has been closed in the mean time
 210
 211       var err = self._handle.bind(ip, port || 0, flags);
 212       if (err) {
 213         var ex = exceptionWithHostPort(err, 'bind', ip, port);
 214         self.emit('error', ex);
 215         self._bindState = BIND_STATE_UNBOUND;
 216         // Todo: close?
 217         return;
 218       }
 219
 220       startListening(self);
 221     }
 222   });
 223
 224   return self;
 225 };
 226
 227
 228 // thin wrapper around `send`, here for compatibility with dgram_legacy.js
 229 Socket.prototype.sendto = function(buffer,
 230                                    offset,
 231                                    length,
 232                                    port,
 233                                    address,
 234                                    callback) {
 235   if (typeof offset !== 'number' || typeof length !== 'number')
 236     throw new Error('send takes offset and length as args 2 and 3');
 237
 238   if (typeof address !== 'string')
 239     throw new Error(this.type + ' sockets must send to port, address');
 240
 241   this.send(buffer, offset, length, port, address, callback);
 242 };
 243
 244
 245 Socket.prototype.send = function(buffer,
 246                                  offset,
 247                                  length,
 248                                  port,
 249                                  address,
 250                                  callback) {
 251   var self = this;
 252
 253   if (typeof buffer === 'string')
 254     buffer = new Buffer(buffer);
 255   else if (!(buffer instanceof Buffer))
 256     throw new TypeError('First argument must be a buffer or string.');
 257
 258   offset = offset | 0;
 259   if (offset < 0)
 260     throw new RangeError('Offset should be >= 0');
 261
 262   if ((length == 0 && offset > buffer.length) ||
 263       (length > 0 && offset >= buffer.length))
 264     throw new RangeError('Offset into buffer too large');
 265
 266   // Sending a zero-length datagram is kind of pointless but it _is_
 267   // allowed, hence check that length >= 0 rather than > 0.
 268   length = length | 0;
 269   if (length < 0)
 270     throw new RangeError('Length should be >= 0');
 271
 272   if (offset + length > buffer.length)
 273     throw new RangeError('Offset + length beyond buffer length');
 274
 275   port = port | 0;
 276   if (port <= 0 || port > 65535)
 277     throw new RangeError('Port should be > 0 and < 65536');
 278
 279   // Normalize callback so it's either a function or undefined but not anything
 280   // else.
 281   if (typeof callback !== 'function')
 282     callback = undefined;
 283
 284   self._healthCheck();
 285
 286   if (self._bindState == BIND_STATE_UNBOUND)
 287     self.bind({port: 0, exclusive: true}, null);
 288
 289   // If the socket hasn't been bound yet, push the outbound packet onto the
 290   // send queue and send after binding is complete.
 291   if (self._bindState != BIND_STATE_BOUND) {
 292     // If the send queue hasn't been initialized yet, do it, and install an
 293     // event handler that flushes the send queue after binding is done.
 294     if (!self._sendQueue) {
 295       self._sendQueue = [];
 296       self.once('listening', function() {
 297         // Flush the send queue.
 298         for (var i = 0; i < self._sendQueue.length; i++)
 299           self.send.apply(self, self._sendQueue[i]);
 300         self._sendQueue = undefined;
 301       });
 302     }
 303     self._sendQueue.push([buffer, offset, length, port, address, callback]);
 304     return;
 305   }
 306
 307   self._handle.lookup(address, function(ex, ip) {
 308     if (ex) {
 309       if (typeof callback === 'function') {
 310         callback(ex);
 311         return;
 312       }
 313
 314       self.emit('error', ex);
 315     } else if (self._handle) {
 316       var req = new SendWrap();
 317       req.buffer = buffer;  // Keep reference alive.
 318       req.length = length;
 319       req.address = address;
 320       req.port = port;
 321       if (callback) {
 322         req.callback = callback;
 323         req.oncomplete = afterSend;
 324       }
 325       var err = self._handle.send(req,
 326                                   buffer,
 327                                   offset,
 328                                   length,
 329                                   port,
 330                                   ip,
 331                                   !!callback);
 332       if (err && callback) {
 333         // don't emit as error, dgram_legacy.js compatibility
 334         var ex = exceptionWithHostPort(err, 'send', address, port);
 335         process.nextTick(callback, ex);
 336       }
 337     }
 338   });
 339 };
 340
 341
 342 function afterSend(err) {
 343   if (err) {
 344     err = exceptionWithHostPort(err, 'send', this.address, this.port);
 345   }
 346   this.callback(err, this.length);
 347 }
 348
 349
 350 Socket.prototype.close = function(callback) {
 351   if (typeof callback === 'function')
 352     this.on('close', callback);
 353   this._healthCheck();
 354   this._stopReceiving();
 355   this._handle.close();
 356   this._handle = null;
 357   var self = this;
 358   process.nextTick(socketCloseNT, self);
 359
 360   return this;
 361 };
 362
 363
 364 function socketCloseNT(self) {
 365   self.emit('close');
 366 }
 367
 368
 369 Socket.prototype.address = function() {
 370   this._healthCheck();
 371
 372   var out = {};
 373   var err = this._handle.getsockname(out);
 374   if (err) {
 375     throw errnoException(err, 'getsockname');
 376   }
 377
 378   return out;
 379 };
 380
 381
 382 Socket.prototype.setBroadcast = function(arg) {
 383   var err = this._handle.setBroadcast(arg ? 1 : 0);
 384   if (err) {
 385     throw errnoException(err, 'setBroadcast');
 386   }
 387 };
 388
 389
 390 Socket.prototype.setTTL = function(arg) {
 391   if (typeof arg !== 'number') {
 392     throw new TypeError('Argument must be a number');
 393   }
 394
 395   var err = this._handle.setTTL(arg);
 396   if (err) {
 397     throw errnoException(err, 'setTTL');
 398   }
 399
 400   return arg;
 401 };
 402
 403
 404 Socket.prototype.setMulticastTTL = function(arg) {
 405   if (typeof arg !== 'number') {
 406     throw new TypeError('Argument must be a number');
 407   }
 408
 409   var err = this._handle.setMulticastTTL(arg);
 410   if (err) {
 411     throw errnoException(err, 'setMulticastTTL');
 412   }
 413
 414   return arg;
 415 };
 416
 417
 418 Socket.prototype.setMulticastLoopback = function(arg) {
 419   var err = this._handle.setMulticastLoopback(arg ? 1 : 0);
 420   if (err) {
 421     throw errnoException(err, 'setMulticastLoopback');
 422   }
 423
 424   return arg; // 0.4 compatibility
 425 };
 426
 427
 428 Socket.prototype.addMembership = function(multicastAddress,
 429                                           interfaceAddress) {
 430   this._healthCheck();
 431
 432   if (!multicastAddress) {
 433     throw new Error('multicast address must be specified');
 434   }
 435
 436   var err = this._handle.addMembership(multicastAddress, interfaceAddress);
 437   if (err) {
 438     throw errnoException(err, 'addMembership');
 439   }
 440 };
 441
 442
 443 Socket.prototype.dropMembership = function(multicastAddress,
 444                                            interfaceAddress) {
 445   this._healthCheck();
 446
 447   if (!multicastAddress) {
 448     throw new Error('multicast address must be specified');
 449   }
 450
 451   var err = this._handle.dropMembership(multicastAddress, interfaceAddress);
 452   if (err) {
 453     throw errnoException(err, 'dropMembership');
 454   }
 455 };
 456
 457
 458 Socket.prototype._healthCheck = function() {
 459   if (!this._handle)
 460     throw new Error('Not running'); // error message from dgram_legacy.js
 461 };
 462
 463
 464 Socket.prototype._stopReceiving = function() {
 465   if (!this._receiving)
 466     return;
 467
 468   this._handle.recvStop();
 469   this._receiving = false;
 470   this.fd = null; // compatibility hack
 471 };
 472
 473
 474 function onMessage(nread, handle, buf, rinfo) {
 475   var self = handle.owner;
 476   if (nread < 0) {
 477     return self.emit('error', errnoException(nread, 'recvmsg'));
 478   }
 479   rinfo.size = buf.length; // compatibility
 480   self.emit('message', buf, rinfo);
 481 }
 482
 483
 484 Socket.prototype.ref = function() {
 485   if (this._handle)
 486     this._handle.ref();
 487
 488   return this;
 489 };
 490
 491
 492 Socket.prototype.unref = function() {
 493   if (this._handle)
 494     this._handle.unref();
 495
 496   return this;
 497 };