lib/dgram.js

   1 // Copyright Joyent, Inc. and other Node contributors.
   2 //
   3 // Permission is hereby granted, free of charge, to any person obtaining a
   4 // copy of this software and associated documentation files (the
   5 // "Software"), to deal in the Software without restriction, including
   6 // without limitation the rights to use, copy, modify, merge, publish,
   7 // distribute, sublicense, and/or sell copies of the Software, and to permit
   8 // persons to whom the Software is furnished to do so, subject to the
   9 // following conditions:
  10 //
  11 // The above copyright notice and this permission notice shall be included
  12 // in all copies or substantial portions of the Software.
  13 //
  14 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
  15 // OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  16 // MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
  17 // NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
  18 // DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
  19 // OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
  20 // USE OR OTHER DEALINGS IN THE SOFTWARE.
  21
  22 var assert = require('assert');
  23 var util = require('util');
  24 var events = require('events');
  25
  26 var UDP = process.binding('udp_wrap').UDP;
  27
  28 var BIND_STATE_UNBOUND = 0;
  29 var BIND_STATE_BINDING = 1;
  30 var BIND_STATE_BOUND = 2;
  31
  32 // lazily loaded
  33 var cluster = null;
  34 var dns = null;
  35 var net = null;
  36
  37 var errnoException = util._errnoException;
  38
  39 // no-op callback
  40 function noop() {
  41 }
  42
  43
  44 function isIP(address) {
  45   if (!net)
  46     net = require('net');
  47
  48   return net.isIP(address);
  49 }
  50
  51
  52 function lookup(address, family, callback) {
  53   if (!dns)
  54     dns = require('dns');
  55
  56   return dns.lookup(address, family, callback);
  57 }
  58
  59
  60 function lookup4(address, callback) {
  61   return lookup(address || '0.0.0.0', 4, callback);
  62 }
  63
  64
  65 function lookup6(address, callback) {
  66   return lookup(address || '::0', 6, callback);
  67 }
  68
  69
  70 function newHandle(type) {
  71   if (type == 'udp4') {
  72     var handle = new UDP;
  73     handle.lookup = lookup4;
  74     return handle;
  75   }
  76
  77   if (type == 'udp6') {
  78     var handle = new UDP;
  79     handle.lookup = lookup6;
  80     handle.bind = handle.bind6;
  81     handle.send = handle.send6;
  82     return handle;
  83   }
  84
  85   if (type == 'unix_dgram')
  86     throw new Error('unix_dgram sockets are not supported any more.');
  87
  88   throw new Error('Bad socket type specified. Valid types are: udp4, udp6');
  89 }
  90
  91
  92 exports._createSocketHandle = function(address, port, addressType, fd) {
  93   // Opening an existing fd is not supported for UDP handles.
  94   assert(!util.isNumber(fd) || fd < 0);
  95
  96   var handle = newHandle(addressType);
  97
  98   if (port || address) {
  99     var err = handle.bind(address, port || 0, 0);
 100     if (err) {
 101       handle.close();
 102       return err;
 103     }
 104   }
 105
 106   return handle;
 107 };
 108
 109
 110 function Socket(type, listener) {
 111   events.EventEmitter.call(this);
 112
 113   var handle = newHandle(type);
 114   handle.owner = this;
 115
 116   this._handle = handle;
 117   this._receiving = false;
 118   this._bindState = BIND_STATE_UNBOUND;
 119   this.type = type;
 120   this.fd = null; // compatibility hack
 121
 122   if (util.isFunction(listener))
 123     this.on('message', listener);
 124 }
 125 util.inherits(Socket, events.EventEmitter);
 126 exports.Socket = Socket;
 127
 128
 129 exports.createSocket = function(type, listener) {
 130   return new Socket(type, listener);
 131 };
 132
 133
 134 function startListening(socket) {
 135   socket._handle.onmessage = onMessage;
 136   // Todo: handle errors
 137   socket._handle.recvStart();
 138   socket._receiving = true;
 139   socket._bindState = BIND_STATE_BOUND;
 140   socket.fd = -42; // compatibility hack
 141
 142   socket.emit('listening');
 143 }
 144
 145 function replaceHandle(self, newHandle) {
 146
 147   // Set up the handle that we got from master.
 148   newHandle.lookup = self._handle.lookup;
 149   newHandle.bind = self._handle.bind;
 150   newHandle.send = self._handle.send;
 151   newHandle.owner = self;
 152
 153   // Replace the existing handle by the handle we got from master.
 154   self._handle.close();
 155   self._handle = newHandle;
 156 }
 157
 158 Socket.prototype.bind = function(/*port, address, callback*/) {
 159   var self = this;
 160
 161   self._healthCheck();
 162
 163   if (this._bindState != BIND_STATE_UNBOUND)
 164     throw new Error('Socket is already bound');
 165
 166   this._bindState = BIND_STATE_BINDING;
 167
 168   if (util.isFunction(arguments[arguments.length - 1]))
 169     self.once('listening', arguments[arguments.length - 1]);
 170
 171   var UDP = process.binding('udp_wrap').UDP;
 172   if (arguments[0] instanceof UDP) {
 173     replaceHandle(self, arguments[0]);
 174     startListening(self);
 175     return;
 176   }
 177
 178   var port = arguments[0];
 179   var address = arguments[1];
 180   if (util.isFunction(address)) address = '';  // a.k.a. "any address"
 181
 182   // resolve address first
 183   self._handle.lookup(address, function(err, ip) {
 184     if (err) {
 185       self._bindState = BIND_STATE_UNBOUND;
 186       self.emit('error', err);
 187       return;
 188     }
 189
 190     if (!cluster)
 191       cluster = require('cluster');
 192
 193     if (cluster.isWorker) {
 194       cluster._getServer(self, ip, port, self.type, -1, function(err, handle) {
 195         if (err) {
 196           self.emit('error', errnoException(err, 'bind'));
 197           self._bindState = BIND_STATE_UNBOUND;
 198           return;
 199         }
 200
 201         if (!self._handle)
 202           // handle has been closed in the mean time.
 203           return handle.close();
 204
 205         replaceHandle(self, handle);
 206         startListening(self);
 207       });
 208
 209     } else {
 210       if (!self._handle)
 211         return; // handle has been closed in the mean time
 212
 213       var err = self._handle.bind(ip, port || 0, /*flags=*/ 0);
 214       if (err) {
 215         self.emit('error', errnoException(err, 'bind'));
 216         self._bindState = BIND_STATE_UNBOUND;
 217         // Todo: close?
 218         return;
 219       }
 220
 221       startListening(self);
 222     }
 223   });
 224 };
 225
 226
 227 // thin wrapper around `send`, here for compatibility with dgram_legacy.js
 228 Socket.prototype.sendto = function(buffer,
 229                                    offset,
 230                                    length,
 231                                    port,
 232                                    address,
 233                                    callback) {
 234   if (!util.isNumber(offset) || !util.isNumber(length))
 235     throw new Error('send takes offset and length as args 2 and 3');
 236
 237   if (!util.isString(address))
 238     throw new Error(this.type + ' sockets must send to port, address');
 239
 240   this.send(buffer, offset, length, port, address, callback);
 241 };
 242
 243
 244 Socket.prototype.send = function(buffer,
 245                                  offset,
 246                                  length,
 247                                  port,
 248                                  address,
 249                                  callback) {
 250   var self = this;
 251
 252   if (!util.isBuffer(buffer))
 253     throw new TypeError('First argument must be a buffer object.');
 254
 255   offset = offset | 0;
 256   if (offset < 0)
 257     throw new RangeError('Offset should be >= 0');
 258
 259   if (offset >= buffer.length)
 260     throw new RangeError('Offset into buffer too large');
 261
 262   // Sending a zero-length datagram is kind of pointless but it _is_
 263   // allowed, hence check that length >= 0 rather than > 0.
 264   length = length | 0;
 265   if (length < 0)
 266     throw new RangeError('Length should be >= 0');
 267
 268   if (offset + length > buffer.length)
 269     throw new RangeError('Offset + length beyond buffer length');
 270
 271   port = port | 0;
 272   if (port <= 0 || port > 65535)
 273     throw new RangeError('Port should be > 0 and < 65536');
 274
 275   callback = callback || noop;
 276
 277   self._healthCheck();
 278
 279   if (self._bindState == BIND_STATE_UNBOUND)
 280     self.bind(0, null);
 281
 282   // If the socket hasn't been bound yet, push the outbound packet onto the
 283   // send queue and send after binding is complete.
 284   if (self._bindState != BIND_STATE_BOUND) {
 285     // If the send queue hasn't been initialized yet, do it, and install an
 286     // event handler that flushes the send queue after binding is done.
 287     if (!self._sendQueue) {
 288       self._sendQueue = [];
 289       self.once('listening', function() {
 290         // Flush the send queue.
 291         for (var i = 0; i < self._sendQueue.length; i++)
 292           self.send.apply(self, self._sendQueue[i]);
 293         self._sendQueue = undefined;
 294       });
 295     }
 296     self._sendQueue.push([buffer, offset, length, port, address, callback]);
 297     return;
 298   }
 299
 300   self._handle.lookup(address, function(ex, ip) {
 301     if (ex) {
 302       if (callback) callback(ex);
 303       self.emit('error', ex);
 304     }
 305     else if (self._handle) {
 306       var req = { cb: callback, oncomplete: afterSend };
 307       var err = self._handle.send(req, buffer, offset, length, port, ip);
 308       if (err) {
 309         // don't emit as error, dgram_legacy.js compatibility
 310         process.nextTick(function() {
 311           callback(errnoException(err, 'send'));
 312         });
 313       }
 314     }
 315   });
 316 };
 317
 318
 319 function afterSend(err) {
 320   if (this.cb)
 321     this.cb(err ? errnoException(err, 'send') : null);
 322 }
 323
 324
 325 Socket.prototype.close = function() {
 326   this._healthCheck();
 327   this._stopReceiving();
 328   this._handle.close();
 329   this._handle = null;
 330   this.emit('close');
 331 };
 332
 333
 334 Socket.prototype.address = function() {
 335   this._healthCheck();
 336
 337   var out = {};
 338   var err = this._handle.getsockname(out);
 339   if (err) {
 340     throw errnoException(err, 'getsockname');
 341   }
 342
 343   return out;
 344 };
 345
 346
 347 Socket.prototype.setBroadcast = function(arg) {
 348   var err = this._handle.setBroadcast(arg ? 1 : 0);
 349   if (err) {
 350     throw errnoException(err, 'setBroadcast');
 351   }
 352 };
 353
 354
 355 Socket.prototype.setTTL = function(arg) {
 356   if (!util.isNumber(arg)) {
 357     throw new TypeError('Argument must be a number');
 358   }
 359
 360   var err = this._handle.setTTL(arg);
 361   if (err) {
 362     throw errnoException(err, 'setTTL');
 363   }
 364
 365   return arg;
 366 };
 367
 368
 369 Socket.prototype.setMulticastTTL = function(arg) {
 370   if (!util.isNumber(arg)) {
 371     throw new TypeError('Argument must be a number');
 372   }
 373
 374   var err = this._handle.setMulticastTTL(arg);
 375   if (err) {
 376     throw errnoException(err, 'setMulticastTTL');
 377   }
 378
 379   return arg;
 380 };
 381
 382
 383 Socket.prototype.setMulticastLoopback = function(arg) {
 384   var err = this._handle.setMulticastLoopback(arg ? 1 : 0);
 385   if (err) {
 386     throw errnoException(err, 'setMulticastLoopback');
 387   }
 388
 389   return arg; // 0.4 compatibility
 390 };
 391
 392
 393 Socket.prototype.addMembership = function(multicastAddress,
 394                                           interfaceAddress) {
 395   this._healthCheck();
 396
 397   if (!multicastAddress) {
 398     throw new Error('multicast address must be specified');
 399   }
 400
 401   var err = this._handle.addMembership(multicastAddress, interfaceAddress);
 402   if (err) {
 403     throw new errnoException(err, 'addMembership');
 404   }
 405 };
 406
 407
 408 Socket.prototype.dropMembership = function(multicastAddress,
 409                                            interfaceAddress) {
 410   this._healthCheck();
 411
 412   if (!multicastAddress) {
 413     throw new Error('multicast address must be specified');
 414   }
 415
 416   var err = this._handle.dropMembership(multicastAddress, interfaceAddress);
 417   if (err) {
 418     throw new errnoException(err, 'dropMembership');
 419   }
 420 };
 421
 422
 423 Socket.prototype._healthCheck = function() {
 424   if (!this._handle)
 425     throw new Error('Not running'); // error message from dgram_legacy.js
 426 };
 427
 428
 429 Socket.prototype._stopReceiving = function() {
 430   if (!this._receiving)
 431     return;
 432
 433   this._handle.recvStop();
 434   this._receiving = false;
 435   this.fd = null; // compatibility hack
 436 };
 437
 438
 439 function onMessage(nread, handle, buf, rinfo) {
 440   var self = handle.owner;
 441   if (nread < 0) {
 442     return self.emit('error', errnoException(nread, 'recvmsg'));
 443   }
 444   rinfo.size = buf.length; // compatibility
 445   self.emit('message', buf, rinfo);
 446 }
 447
 448
 449 Socket.prototype.ref = function() {
 450   if (this._handle)
 451     this._handle.ref();
 452 };
 453
 454
 455 Socket.prototype.unref = function() {
 456   if (this._handle)
 457     this._handle.unref();
 458 };