src/js/net.js

   1 /* Copyright 2015-present Samsung Electronics Co., Ltd. and other contributors
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16
  17 var EventEmitter = require('events').EventEmitter;
  18 var stream = require('stream');
  19 var util = require('util');
  20 var assert = require('assert');
  21
  22 var TCP = process.binding(process.binding.tcp);
  23
  24
  25 function createTCP() {
  26   var tcp = new TCP();
  27   return tcp;
  28 }
  29
  30
  31 function SocketState(options) {
  32   // 'true' during connection handshaking.
  33   this.connecting = false;
  34
  35   // become 'true' when connection established.
  36   this.connected = false;
  37
  38   this.writable = true;
  39   this.readable = true;
  40
  41   this.destroyed = false;
  42
  43   this.allowHalfOpen = options && options.allowHalfOpen || false;
  44 }
  45
  46
  47 function Socket(options) {
  48   if (!(this instanceof Socket)) {
  49     return new Socket(options);
  50   }
  51
  52   if (util.isUndefined(options)) {
  53     options = {};
  54   }
  55
  56   stream.Duplex.call(this, options);
  57
  58   this._timer = null;
  59   this._timeout = 0;
  60
  61   this._socketState = new SocketState(options);
  62
  63   if (options.handle) {
  64     this._handle = options.handle;
  65     this._handle.owner = this;
  66   }
  67
  68   this.on('finish', onSocketFinish);
  69   this.on('end', onSocketEnd);
  70 }
  71
  72
  73 // Socket inherits Duplex.
  74 util.inherits(Socket, stream.Duplex);
  75
  76
  77 Socket.prototype.connect = function() {
  78   var self = this;
  79   var state = self._socketState;
  80
  81   var args = normalizeConnectArgs(arguments);
  82   var options = args[0];
  83   var callback = args[1];
  84
  85   if (state.connecting || state.connected) {
  86     return self;
  87   }
  88
  89   if (!self._handle) {
  90     self._handle = createTCP();
  91     self._handle.owner = self;
  92   }
  93
  94   if (util.isFunction(callback)) {
  95     self.once('connect', callback);
  96   }
  97
  98   resetSocketTimeout(self);
  99
 100   state.connecting = true;
 101
 102   var dns = require('dns');
 103   var host = options.host ? options.host : 'localhost';
 104   var port = options.port;
 105   var dnsopts = {
 106     family: options.family >>> 0,
 107     hints: 0
 108   };
 109
 110   if (!util.isNumber(port) || port < 0 || port > 65535)
 111     throw new RangeError('port should be >= 0 and < 65536: ' + options.port);
 112
 113   if (dnsopts.family !== 0 && dnsopts.family !== 4 && dnsopts.family !== 6)
 114     throw new RangeError('port should be 4 or 6: ' + dnsopts.family);
 115
 116   self._host = host;
 117   dns.lookup(host, dnsopts, function(err, ip, family) {
 118     self.emit('lookup', err, ip, family);
 119
 120     if (err) {
 121       process.nextTick(function() {
 122         self.emit('error', err);
 123         self._destroy();
 124       });
 125     } else {
 126       resetSocketTimeout(self);
 127       connect(self, ip, port);
 128     }
 129   });
 130
 131   return self;
 132 };
 133
 134
 135 Socket.prototype.write = function(data, callback) {
 136   if (!util.isString(data) && !util.isBuffer(data)) {
 137     throw new TypeError('invalid argument');
 138   }
 139
 140   return stream.Duplex.prototype.write.call(this, data, callback);
 141 };
 142
 143
 144 Socket.prototype._write = function(chunk, callback, afterWrite) {
 145   assert(util.isFunction(afterWrite));
 146
 147   var self = this;
 148
 149   resetSocketTimeout(self);
 150
 151   self._handle.owner = self;
 152
 153   self._handle.write(chunk, function(status) {
 154     afterWrite(status);
 155     if (util.isFunction(callback)) {
 156       callback.call(self, status);
 157     }
 158   });
 159 };
 160
 161
 162 Socket.prototype.end = function(data, callback) {
 163   var self = this;
 164   var state = self._socketState;
 165
 166   // end of writable stream.
 167   stream.Writable.prototype.end.call(self, data, callback);
 168
 169   // this socket is no longer writable.
 170   state.writable = false;
 171 };
 172
 173
 174 // Destroy this socket as fast as possible.
 175 Socket.prototype.destroy = function() {
 176   var self = this;
 177   var state = self._socketState;
 178
 179   if (state.destroyed) {
 180     return;
 181   }
 182
 183   if (state.writable) {
 184     self.end();
 185   }
 186
 187   // unset timeout
 188   clearSocketTimeout(self);
 189
 190   if (self._writableState.ended) {
 191     close(self);
 192     state.destroyed = true;
 193   } else {
 194     self.once('finish', function() {
 195       self.destroy();
 196     });
 197   }
 198 };
 199
 200
 201 // Destroy this socket as fast as possible if this socket is no longer readable.
 202 Socket.prototype.destroySoon = function() {
 203   var self = this;
 204   var state = self._socketState;
 205
 206   if (state.writable) {
 207     self.end();
 208   }
 209
 210   if (self._writableState.finished) {
 211     self.destroy();
 212   } else {
 213     self.once('finish', self.destroy);
 214   }
 215 }
 216
 217
 218 Socket.prototype.setKeepAlive = function(enable, delay) {
 219   var self = this;
 220   enable = +Boolean(enable);
 221   if (self._handle && self._handle.setKeepAlive) {
 222     self._handle.setKeepAlive(enable, ~~(delay / 1000));
 223   }
 224 };
 225
 226
 227 Socket.prototype.address = function() {
 228   if (!this._handle || !this._handle.getsockname) {
 229     return {};
 230   }
 231   if (!this._sockname) {
 232     var out = {};
 233     var err = this._handle.getsockname(out);
 234     if (err) return {};  // FIXME(bnoordhuis) Throw?
 235     this._sockname = out;
 236   }
 237   return this._sockname;
 238 };
 239
 240
 241 Socket.prototype.setTimeout = function(msecs, callback) {
 242   var self = this;
 243
 244   self._timeout = msecs;
 245   clearSocketTimeout(self);
 246
 247   if (msecs === 0) {
 248     if (callback) {
 249       self.removeListener('timeout', callback);
 250     }
 251   } else {
 252     self._timer = setTimeout(function() {
 253       self.emit('timeout');
 254       clearSocketTimeout(self);
 255     }, msecs);
 256     if (callback) {
 257       self.once('timeout', callback);
 258     }
 259   }
 260 };
 261
 262
 263 function connect(socket, ip, port) {
 264   var afterConnect = function(status) {
 265     var state = socket._socketState;
 266     state.connecting = false;
 267
 268     if (state.destroyed) {
 269       return;
 270     }
 271
 272     if (status == 0) {
 273       onSocketConnect(socket);
 274       socket.emit('connect');
 275     } else {
 276       socket.destroy();
 277       emitError(socket, new Error('connect failed - status: ' + status));
 278     }
 279   };
 280
 281   socket._handle.connect(ip, port, afterConnect);
 282 }
 283
 284
 285 function close(socket) {
 286   socket._handle.owner = socket;
 287   socket._handle.onclose = function() {
 288     socket.emit('close');
 289   };
 290
 291   socket._handle.close();
 292
 293   if (socket._server) {
 294     var server = socket._server;
 295     server._socketCount--;
 296     server._emitCloseIfDrained();
 297     socket._server = null;
 298   }
 299 }
 300
 301
 302 function resetSocketTimeout(socket) {
 303   var state = socket._socketState;
 304
 305   if (!state.destroyed) {
 306     // start timeout over again
 307     clearSocketTimeout(socket);
 308     socket._timer = setTimeout(function() {
 309       socket.emit('timeout');
 310       clearSocketTimeout(socket);
 311     }, socket._timeout);
 312   }
 313 };
 314
 315
 316 function clearSocketTimeout(socket) {
 317   if (socket._timer) {
 318     clearTimeout(socket._timer);
 319     socket._timer = null;
 320   }
 321 };
 322
 323
 324 function emitError(socket, err) {
 325   socket.emit('error', err);
 326 }
 327
 328
 329 function maybeDestroy(socket) {
 330   var state = socket._socketState;
 331
 332   if (!state.connecting &&
 333       !state.writable &&
 334       !state.readable) {
 335     socket.destroy();
 336   }
 337 }
 338
 339
 340 function onSocketConnect(socket) {
 341   var state = socket._socketState;
 342
 343   state.connecting = false;
 344   state.connected = true;
 345
 346   resetSocketTimeout(socket);
 347
 348   socket._readyToWrite();
 349
 350   // `readStart` on next tick, after connection event handled.
 351   process.nextTick(function() {
 352     socket._handle.owner = socket;
 353     socket._handle.onread = onread;
 354     socket._handle.readStart();
 355   });
 356 }
 357
 358
 359 function onread(socket, nread, isEOF, buffer) {
 360   var state = socket._socketState;
 361
 362   resetSocketTimeout(socket);
 363
 364   if (isEOF) {
 365     // pushing readable stream null means EOF.
 366     stream.Readable.prototype.push.call(socket, null);
 367
 368     if (socket._readableState.length == 0) {
 369       // this socket is no longer readable.
 370       state.readable = false;
 371       // destroy if this socket is not writable.
 372       maybeDestroy(socket);
 373     }
 374   } else if (nread < 0) {
 375     var err = new Error('read error: ' + nread);
 376     stream.Readable.prototype.error.call(socket, err);
 377   } else if (nread > 0) {
 378     if (process.platform  != 'nuttx') {
 379       stream.Readable.prototype.push.call(socket, buffer);
 380       return;
 381     }
 382
 383     var str = buffer.toString();
 384     var eofNeeded = false;
 385     if (str.length >= 6
 386       && str.substr(str.length - 6, str.length) == '\\e\\n\\d') {
 387       eofNeeded  = true;
 388       buffer = buffer.slice(0, str.length - 6);
 389     }
 390
 391     if (str.length == 6 && eofNeeded) {
 392       // Socket.prototype.end with no argument
 393     } else {
 394       stream.Readable.prototype.push.call(socket, buffer);
 395     }
 396
 397     if (eofNeeded) {
 398       onread(socket, 0, true, null);
 399     }
 400   }
 401 }
 402
 403
 404 // Writable stream finished.
 405 function onSocketFinish() {
 406   var self = this;
 407   var state = self._socketState;
 408
 409   if (!state.readable || self._readableState.ended) {
 410     // no readable stream or ended, destroy(close) socket.
 411     return self.destroy();
 412   } else {
 413     // Readable stream alive, shutdown only outgoing stream.
 414     var err = self._handle.shutdown(function() {
 415       if (self._readableState.ended) {
 416         self.destroy();
 417       }
 418     });
 419   }
 420 }
 421
 422
 423 // Readable stream ended.
 424 function onSocketEnd() {
 425   var state = this._socketState;
 426
 427   maybeDestroy(this);
 428
 429   if (!state.allowHalfOpen) {
 430     this.destroySoon();
 431   }
 432 }
 433
 434
 435
 436 function Server(options, connectionListener) {
 437   if (!(this instanceof Server)) {
 438     return new Server(options, connectionListener);
 439   }
 440
 441   EventEmitter.call(this);
 442
 443   if (util.isFunction(options)) {
 444     connectionListener = options;
 445     options = {};
 446   } else {
 447     options = options || {};
 448   }
 449
 450   if (util.isFunction(connectionListener)) {
 451     this.on('connection', connectionListener);
 452   }
 453
 454   this._handle = null;
 455   this._socketCount = 0;
 456
 457   this.allowHalfOpen = options.allowHalfOpen || false;
 458 }
 459
 460 // Server inherits EventEmitter.
 461 util.inherits(Server, EventEmitter);
 462
 463
 464 // listen
 465 Server.prototype.listen = function() {
 466   var self = this;
 467
 468   var args = normalizeListenArgs(arguments);
 469
 470   var options = args[0];
 471   var callback = args[1];
 472
 473   var port = options.port;
 474   var host = util.isString(options.host) ? options.host : '0.0.0.0';
 475   var backlog = util.isNumber(options.backlog) ? options.backlog : 511;
 476
 477   if (!util.isNumber(port)) {
 478     throw new Error('invalid argument - need port number');
 479   }
 480
 481   // register listening event listener.
 482   if (util.isFunction(callback)) {
 483     self.once('listening', callback);
 484   }
 485
 486   // Create server handle.
 487   if (!self._handle) {
 488     self._handle = createTCP();
 489   }
 490
 491   // bind port
 492   var err = self._handle.bind(host, port);
 493   if (err) {
 494     self._handle.close();
 495     return err;
 496   }
 497
 498   // listen
 499   self._handle.onconnection = onconnection;
 500   self._handle.createTCP = createTCP;
 501   self._handle.owner = self;
 502
 503   var err = self._handle.listen(backlog);
 504
 505   if (err) {
 506     self._handle.close();
 507     return err;
 508   }
 509
 510   process.nextTick(function() {
 511     if (self._handle) {
 512       self.emit('listening');
 513     }
 514   });
 515 };
 516
 517
 518 Server.prototype.address = function() {
 519   if (this._handle && this._handle.getsockname) {
 520     var out = {};
 521     this._handle.getsockname(out);
 522     // TODO(bnoordhuis) Check err and throw?
 523     return out;
 524   }
 525
 526   return null;
 527 };
 528
 529
 530 Server.prototype.close = function(callback) {
 531   if (util.isFunction(callback)) {
 532     if (!this._handle) {
 533       this.once('close', function() {
 534         callback(new Error('Not running'));
 535       });
 536     } else {
 537       this.once('close', callback);
 538     }
 539   }
 540   if (this._handle) {
 541     this._handle.close();
 542     this._handle = null;
 543   }
 544   this._emitCloseIfDrained();
 545   return this;
 546 };
 547
 548
 549 Server.prototype._emitCloseIfDrained = function() {
 550   var self = this;
 551
 552   if (self._handle || self._socketCount > 0) {
 553     return;
 554   }
 555
 556   process.nextTick(function() {
 557     self.emit('close');
 558   });
 559 };
 560
 561
 562 // This function is called after server accepted connection request
 563 // from a client.
 564 //  This binding
 565 //   * server tcp handle
 566 //  Parameters
 567 //   * status - status code
 568 //   * clientHandle - client socket handle (tcp).
 569 function onconnection(status, clientHandle) {
 570   var server = this.owner;
 571
 572   if (status) {
 573     server.emit('error', new Error('accept error: ' + status));
 574     return;
 575   }
 576
 577   // Create socket object for connecting client.
 578   var socket = new Socket({
 579     handle: clientHandle,
 580     allowHalfOpen: server.allowHalfOpen
 581   });
 582   socket._server = server;
 583
 584   onSocketConnect(socket);
 585
 586   server._socketCount++;
 587
 588   server.emit('connection', socket);
 589 }
 590
 591
 592 function normalizeListenArgs(args) {
 593   var options = {};
 594
 595   if (util.isObject(args[0])) {
 596     options = args[0];
 597   } else {
 598     var idx = 0;
 599     options.port = args[idx++];
 600     if (util.isString(args[idx])) {
 601       options.host = args[idx++];
 602     }
 603     if (util.isNumber(args[idx])) {
 604       options.backlog = args[idx++];
 605     }
 606   }
 607
 608   var cb = args[args.length - 1];
 609
 610   return util.isFunction(cb) ? [options, cb] : [options];
 611 }
 612
 613
 614 function normalizeConnectArgs(args) {
 615   var options = {};
 616
 617   if (util.isObject(args[0])) {
 618     options = args[0];
 619   } else {
 620     options.port = args[0];
 621     if (util.isString(args[1])) {
 622       options.host = args[1];
 623     }
 624   }
 625
 626   var cb = args[args.length - 1];
 627
 628   return util.isFunction(cb) ? [options, cb] : [options];
 629 }
 630
 631
 632 exports.createServer = function(options, callback) {
 633   return new Server(options, callback);
 634 };
 635
 636
 637 // net.connect(options[, connectListenr])
 638 // net.connect(port[, host][, connectListener])
 639 exports.connect = exports.createConnection = function() {
 640   var args = normalizeConnectArgs(arguments);
 641   var socket = new Socket(args[0]);
 642   return Socket.prototype.connect.apply(socket, args);
 643 };
 644
 645
 646 module.exports.Socket = Socket;
 647 module.exports.Server = Server;