/** * Copyright (c) 2015-present, Peel Technologies, Inc. * All rights reserved. * * @providesModule TcpSocket * @flow */ 'use strict'; var inherits = require('inherits'); var EventEmitter = require('events').EventEmitter; var ipRegex = require('ip-regex'); var { DeviceEventEmitter, NativeModules } = require('react-native'); var Sockets = NativeModules.TcpSockets; var base64 = require('base64-js'); var Base64Str = require('./base64-str'); var noop = function () {}; var instances = 0; var STATE = { DISCONNECTED: 0, CONNECTING: 1, CONNECTED: 2 }; function TcpSocket(options: ?{ id: ?number }) { if (!(this instanceof TcpSocket)) { return new TcpSocket(options); } if (EventEmitter instanceof Function) { EventEmitter.call(this); } if (options && options.id) { // e.g. incoming server connections this._id = Number(options.id); if (this._id <= instances) { throw new Error('Socket id ' + this._id + 'already in use'); } } else { // javascript generated sockets range from 1-1000 this._id = instances++; } // ensure compatibility with node's EventEmitter if (!this.on) { this.on = this.addListener.bind(this); } // these will be set once there is a connection this.writable = this.readable = false; this._state = STATE.DISCONNECTED; } inherits(TcpSocket, EventEmitter); TcpSocket.prototype._debug = function() { if (__DEV__) { var args = [].slice.call(arguments); args.unshift('socket-' + this._id); console.log.apply(console, args); } }; // TODO : determine how to properly overload this with flow TcpSocket.prototype.connect = function(options, callback) : TcpSocket { this._registerEvents(); if (options === null || typeof options !== 'object') { // Old API: // connect(port, [host], [cb]) var args = this._normalizeConnectArgs(arguments); return TcpSocket.prototype.connect.apply(this, args); } if (typeof callback === 'function') { this.once('connect', callback); } var host = options.host || 'localhost'; var port = options.port || 0; var localAddress = options.localAddress; var localPort = options.localPort; if (localAddress && !ipRegex({exact: true}).test(localAddress)) { throw new TypeError('"localAddress" option must be a valid IP: ' + localAddress); } if (localPort && typeof localPort !== 'number') { throw new TypeError('"localPort" option should be a number: ' + localPort); } if (typeof port !== 'undefined') { if (typeof port !== 'number' && typeof port !== 'string') { throw new TypeError('"port" option should be a number or string: ' + port); } port = +port; if (!isLegalPort(port)) { throw new RangeError('"port" option should be >= 0 and < 65536: ' + port); } } this._state = STATE.CONNECTING; this._debug('connecting, host:', host, 'port:', port); this._destroyed = false; Sockets.connect(this._id, host, Number(port), options); return this; }; // Check that the port number is not NaN when coerced to a number, // is an integer and that it falls within the legal range of port numbers. function isLegalPort(port: number) : boolean { if (typeof port === 'string' && port.trim() === '') { return false; } return +port === (port >>> 0) && port >= 0 && port <= 0xFFFF; } TcpSocket.prototype.setTimeout = function(msecs: number, callback: () => void) { var self = this; if (this._timeout) { clearTimeout(this._timeout); this._timeout = null; } if (msecs > 0) { if (callback) { this.once('timeout', callback); } var self = this; this._timeout = setTimeout(function() { self.emit('timeout'); self._timeout = null; self.destroy(); }, msecs); } }; TcpSocket.prototype.address = function() : { port: number, address: string, family: string } { return this._address; }; TcpSocket.prototype.end = function(data, encoding) { if (this._destroyed) { return; } if (data) { this.write(data, encoding); } this._destroyed = true; this._debug('ending'); Sockets.end(this._id); }; TcpSocket.prototype.destroy = function() { if (!this._destroyed) { this._destroyed = true; this._debug('destroying'); Sockets.destroy(this._id); } }; TcpSocket.prototype._registerEvents = function(): void { if (this._subs && this._subs.length > 0) { return; } this._subs = [ DeviceEventEmitter.addListener( 'tcp-' + this._id + '-connect', this._onConnect.bind(this) ), DeviceEventEmitter.addListener( 'tcp-' + this._id + '-connection', this._onConnection.bind(this) ), DeviceEventEmitter.addListener( 'tcp-' + this._id + '-data', this._onData.bind(this) ), DeviceEventEmitter.addListener( 'tcp-' + this._id + '-close', this._onClose.bind(this) ), DeviceEventEmitter.addListener( 'tcp-' + this._id + '-error', this._onError.bind(this) ) ]; }; TcpSocket.prototype._unregisterEvents = function(): void { this._subs.forEach(function(listener) { listener.remove(); }); this._subs = []; }; TcpSocket.prototype._onConnect = function(address: { port: number, address: string, family: string }): void { this._debug('received', 'connect'); setConnected(this, address); this.emit('connect'); }; TcpSocket.prototype._onConnection = function(info: { id: number, address: { port: number, address: string, family: string } }): void { this._debug('received', 'connection'); var socket = new TcpSocket({ id: info.id }); socket._registerEvents(); setConnected(socket, info.address); this.emit('connection', socket); }; TcpSocket.prototype._onData = function(data: string): void { this._debug('received', 'data'); if (this._timeout) { clearTimeout(this._timeout); this._timeout = null; } // from base64 string var buffer = typeof Buffer === 'undefined' ? base64.toByteArray(data) : new global.Buffer(data, 'base64'); this.emit('data', buffer); }; TcpSocket.prototype._onClose = function(hadError: boolean): void { this._debug('received', 'close'); setDisconnected(this, hadError); }; TcpSocket.prototype._onError = function(error: string): void { this._debug('received', 'error'); this.emit('error', normalizeError(error)); this.destroy(); }; TcpSocket.prototype.write = function(buffer: any, callback: ?(err: ?Error) => void) : boolean { var self = this; if (this._state === STATE.DISCONNECTED) { throw new Error('Socket is not connected.'); } else if (this._state === STATE.CONNECTING) { // we're ok, GCDAsyncSocket handles queueing internally } var cb = callback || noop; var str; if (typeof buffer === 'string') { self._debug('socket.WRITE(): encoding as base64'); str = Base64Str.encode(buffer); } else if (typeof Buffer !== 'undefined' && global.Buffer.isBuffer(buffer)) { str = buffer.toString('base64'); } else if (buffer instanceof Uint8Array || Array.isArray(buffer)) { str = base64.fromByteArray(buffer); } else { throw new Error('invalid message format'); } Sockets.write(this._id, str, function(err) { if (self._timeout) { clearTimeout(self._timeout); self._timeout = null; } err = normalizeError(err); if (err) { self._debug('write failed', err); return cb(err); } cb(); }); return true; }; function setConnected(socket: TcpSocket, address: { port: number, address: string, family: string } ) { socket.writable = socket.readable = true; socket._state = STATE.CONNECTED; socket._address = address; } function setDisconnected(socket: TcpSocket, hadError: boolean): void { if (socket._state === STATE.DISCONNECTED) { return; } socket._unregisterEvents(); socket._state = STATE.DISCONNECTED; socket.emit('close', hadError); } function normalizeError(err) { if (err) { if (typeof err === 'string') { err = new Error(err); } return err; } } // Returns an array [options] or [options, cb] // It is the same as the argument of Socket.prototype.connect(). TcpSocket.prototype._normalizeConnectArgs = function(args) { var options = {}; if (args[0] !== null && typeof args[0] === 'object') { // connect(options, [cb]) options = args[0]; } else { // connect(port, [host], [cb]) options.port = args[0]; if (typeof args[1] === 'string') { options.host = args[1]; } } var cb = args[args.length - 1]; return typeof cb === 'function' ? [options, cb] : [options]; }; // unimplemented net.Socket apis TcpSocket.prototype.pause = TcpSocket.prototype.resume = TcpSocket.prototype.ref = TcpSocket.prototype.unref = TcpSocket.prototype.setNoDelay = TcpSocket.prototype.setKeepAlive = TcpSocket.prototype.setEncoding = function() { /* nop */ }; module.exports = TcpSocket;