text: var logger = require('@sex-pomelo/sex-pomelo-logger').getLogger('pomelo-admin', 'MasterAgent');
  var MqttServer = require('../protocol/mqtt/mqttServer');
  var EventEmitter = require('events').EventEmitter;
  var MasterSocket = require('./masterSocket');
  var protocol = require('../util/protocol');
  var utils = require('../util/utils');
  var Util = require('util');
  
  var ST_INITED = 1;
  var ST_STARTED = 2;
  var ST_CLOSED = 3;
  
  /**
   * MasterAgent Constructor
   *
   * @class MasterAgent
   * @constructor
   * @param {Object} opts construct parameter
   *                 opts.consoleService {Object} consoleService
   *                 opts.id             {String} server id
   *                 opts.type           {String} server type, 'master', 'connector', etc.
   *                 opts.socket         {Object} socket-io object
   *                 opts.reqId          {Number} reqId add by 1
   *                 opts.callbacks      {Object} callbacks
   *                 opts.state          {Number} MasterAgent state
   * @api public
   */
  var MasterAgent = function(consoleService, opts) {
    EventEmitter.call(this);
    this.reqId = 1;
    this.idMap = {};
    this.msgMap = {};
    this.typeMap = {};
    this.clients = {};
    this.sockets = {};
    this.slaveMap = {};
    this.server = null;
    this.callbacks = {};
    this.state = ST_INITED;
    this.whitelist = opts.whitelist;
    this.consoleService = consoleService;
  };
  
  Util.inherits(MasterAgent, EventEmitter);
  
  /**
   * master listen to a port and handle register and request
   *
   * @param {String} port
   * @api public
   */
  MasterAgent.prototype.listen = function(port, cb) {
    if (this.state > ST_INITED) {
      logger.error('master agent has started or closed.');
      return;
    }
  
    this.state = ST_STARTED;
    this.server = new MqttServer();
    this.server.listen(port);
    // this.server = sio.listen(port);
    // this.server.set('log level', 0);
  
    cb = cb || function() {}
  
    var self = this;
    this.server.on('error', function(err) {
      self.emit('error', err);
      cb(err);
    });
  
    this.server.once('listening', function() {
      setImmediate(function() {
        cb();
      });
    });
  
    this.server.on('connection', function(socket) {
      // var id, type, info, registered, username;
      var masterSocket = new MasterSocket();
      masterSocket['agent'] = self;
      masterSocket['socket'] = socket;
  
      self.sockets[socket.id] = socket;
  
      socket.on('register', function(msg) {
        // register a new connection
        masterSocket.onRegister(msg);
      }); // end of on 'register'
  
      // message from monitor
      socket.on('monitor', function(msg) {
        masterSocket.onMonitor(msg);
      }); // end of on 'monitor'
  
      // message from client
      socket.on('client', function(msg) {
        masterSocket.onClient(msg);
      }); // end of on 'client'
  
      socket.on('reconnect', function(msg) {
        masterSocket.onReconnect(msg);
      });
  
      socket.on('disconnect', function() {
        masterSocket.onDisconnect();
      });
  
      socket.on('close', function() {
        masterSocket.onDisconnect();
      });
  
      socket.on('error', function(err) {
        masterSocket.onError(err);
      });
    }); // end of on 'connection'
  }; // end of listen
  
  /**
   * close master agent
   *
   * @api public
   */
  MasterAgent.prototype.close = function() {
    if (this.state > ST_STARTED) {
      return;
    }
    this.state = ST_CLOSED;
    this.server.close();
  };
  
  /**
   * set module
   *
   * @param {String} moduleId module id/name
   * @param {Object} value module object
   * @api public
   */
  MasterAgent.prototype.set = function(moduleId, value) {
    this.consoleService.set(moduleId, value);
  };
  
  /**
   * get module
   *
   * @param {String} moduleId module id/name
   * @api public
   */
  MasterAgent.prototype.get = function(moduleId) {
    return this.consoleService.get(moduleId);
  };
  
  /**
   * getClientById
   *
   * @param {String} clientId
   * @api public
   */
  MasterAgent.prototype.getClientById = function(clientId) {
    return this.clients[clientId];
  };
  
  /**
   * request monitor{master node} data from monitor
   *
   * @param {String} serverId
   * @param {String} moduleId module id/name
   * @param {Object} msg
   * @param {Function} callback function
   * @api public
   */
  MasterAgent.prototype.request = function(serverId, moduleId, msg, cb) {
    if (this.state > ST_STARTED) {
      return false;
    }
  
    cb = cb || function() {}
  
    var curId = this.reqId++;
    this.callbacks[curId] = cb;
  
    if (!this.msgMap[serverId]) {
      this.msgMap[serverId] = {};
    }
  
    this.msgMap[serverId][curId] = {
      moduleId: moduleId,
      msg: msg
    }
  
    var record = this.idMap[serverId];
    if (!record) {
      cb(new Error('unknown server id:' + serverId));
      return false;
    }
  
    sendToMonitor(record.socket, curId, moduleId, msg);
  
    return true;
  };
  
  /**
   * request server data from monitor by serverInfo{host:port}
   *
   * @param {String} serverId
   * @param {Object} serverInfo
   * @param {String} moduleId module id/name
   * @param {Object} msg
   * @param {Function} callback function
   * @api public
   */
  MasterAgent.prototype.requestServer = function(serverId, serverInfo, moduleId, msg, cb) {
    if (this.state > ST_STARTED) {
      return false;
    }
  
    var record = this.idMap[serverId];
    if (!record) {
      utils.invokeCallback(cb, new Error('unknown server id:' + serverId));
      return false;
    }
  
    var curId = this.reqId++;
    this.callbacks[curId] = cb;
  
    if (utils.compareServer(record, serverInfo)) {
      sendToMonitor(record.socket, curId, moduleId, msg);
    } else {
      var slaves = this.slaveMap[serverId];
      for (var i = 0, l = slaves.length; i < l; i++) {
        if (utils.compareServer(slaves[i], serverInfo)) {
          sendToMonitor(slaves[i].socket, curId, moduleId, msg);
          break;
        }
      }
    }
  
    return true;
  };
  
  /**
   * notify a monitor{master node} by id without callback
   *
   * @param {String} serverId
   * @param {String} moduleId module id/name
   * @param {Object} msg
   * @api public
   */
  MasterAgent.prototype.notifyById = function(serverId, moduleId, msg) {
    if (this.state > ST_STARTED) {
      return false;
    }
  
    var record = this.idMap[serverId];
    if (!record) {
      logger.error('fail to notifyById for unknown server id:' + serverId);
      return false;
    }
  
    sendToMonitor(record.socket, null, moduleId, msg);
  
    return true;
  };
  
  /**
   * notify a monitor by server{host:port} without callback
   *
   * @param {String} serverId
   * @param {Object} serverInfo{host:port}
   * @param {String} moduleId module id/name
   * @param {Object} msg
   * @api public
   */
  MasterAgent.prototype.notifyByServer = function(serverId, serverInfo, moduleId, msg) {
    if (this.state > ST_STARTED) {
      return false;
    }
  
    var record = this.idMap[serverId];
    if (!record) {
      logger.error('fail to notifyByServer for unknown server id:' + serverId);
      return false;
    }
  
    if (utils.compareServer(record, serverInfo)) {
      sendToMonitor(record.socket, null, moduleId, msg);
    } else {
      var slaves = this.slaveMap[serverId];
      for (var i = 0, l = slaves.length; i < l; i++) {
        if (utils.compareServer(slaves[i], serverInfo)) {
          sendToMonitor(slaves[i].socket, null, moduleId, msg);
          break;
        }
      }
    }
    return true;
  };
  
  /**
   * notify slaves by id without callback
   *
   * @param {String} serverId
   * @param {String} moduleId module id/name
   * @param {Object} msg
   * @api public
   */
  MasterAgent.prototype.notifySlavesById = function(serverId, moduleId, msg) {
    if (this.state > ST_STARTED) {
      return false;
    }
  
    var slaves = this.slaveMap[serverId];
    if (!slaves || slaves.length === 0) {
      logger.error('fail to notifySlavesById for unknown server id:' + serverId);
      return false;
    }
  
    broadcastMonitors(slaves, moduleId, msg);
    return true;
  };
  
  /**
   * notify monitors by type without callback
   *
   * @param {String} type serverType
   * @param {String} moduleId module id/name
   * @param {Object} msg
   * @api public
   */
  MasterAgent.prototype.notifyByType = function(type, moduleId, msg) {
    if (this.state > ST_STARTED) {
      return false;
    }
  
    var list = this.typeMap[type];
    if (!list || list.length === 0) {
      logger.error('fail to notifyByType for unknown server type:' + type);
      return false;
    }
    broadcastMonitors(list, moduleId, msg);
    return true;
  };
  
  /**
   * notify all the monitors without callback
   *
   * @param {String} moduleId module id/name
   * @param {Object} msg
   * @api public
   */
  MasterAgent.prototype.notifyAll = function(moduleId, msg) {
    if (this.state > ST_STARTED) {
      return false;
    }
    broadcastMonitors(this.idMap, moduleId, msg);
    return true;
  };
  
  /**
   * notify a client by id without callback
   *
   * @param {String} clientId
   * @param {String} moduleId module id/name
   * @param {Object} msg
   * @api public
   */
  MasterAgent.prototype.notifyClient = function(clientId, moduleId, msg) {
    if (this.state > ST_STARTED) {
      return false;
    }
  
    var record = this.clients[clientId];
    if (!record) {
      logger.error('fail to notifyClient for unknown client id:' + clientId);
      return false;
    }
    sendToClient(record.socket, null, moduleId, msg);
  };
  
  MasterAgent.prototype.notifyCommand = function(command, moduleId, msg) {
    if (this.state > ST_STARTED) {
      return false;
    }
    broadcastCommand(this.idMap, command, moduleId, msg);
    return true;
  };
  
  /**
   * add monitor,client to connection -- idMap
   *
   * @param {Object} agent agent object
   * @param {String} id
   * @param {String} type serverType
   * @param {Object} socket socket-io object
   * @api private
   */
  var addConnection = function(agent, id, type, pid, info, socket) {
    var record = {
      id: id,
      type: type,
      pid: pid,
      info: info,
      socket: socket
    };
    if (type === 'client') {
      agent.clients[id] = record;
    } else {
      if (!agent.idMap[id]) {
        agent.idMap[id] = record;
        var list = agent.typeMap[type] = agent.typeMap[type] || [];
        list.push(record);
      } else {
        var slaves = agent.slaveMap[id] = agent.slaveMap[id] || [];
        slaves.push(record);
      }
    }
    return record;
  };
  
  /**
   * remove monitor,client connection -- idMap
   *
   * @param {Object} agent agent object
   * @param {String} id
   * @param {String} type serverType
   * @api private
   */
  var removeConnection = function(agent, id, type, info) {
    if (type === 'client') {
      delete agent.clients[id];
    } else {
      // remove master node in idMap and typeMap
      var record = agent.idMap[id];
      if (!record) {
        return;
      }
      var _info = record['info']; // info {host, port}
      if (utils.compareServer(_info, info)) {
        delete agent.idMap[id];
        var list = agent.typeMap[type];
        if (list) {
          for (var i = 0, l = list.length; i < l; i++) {
            if (list[i].id === id) {
              list.splice(i, 1);
              break;
            }
          }
          if (list.length === 0) {
            delete agent.typeMap[type];
          }
        }
      } else {
        // remove slave node in slaveMap
        var slaves = agent.slaveMap[id];
        if (slaves) {
          for (var i = 0, l = slaves.length; i < l; i++) {
            if (utils.compareServer(slaves[i]['info'], info)) {
              slaves.splice(i, 1);
              break;
            }
          }
          if (slaves.length === 0) {
            delete agent.slaveMap[id];
          }
        }
      }
    }
  };
  
  /**
   * send msg to monitor
   *
   * @param {Object} socket socket-io object
   * @param {Number} reqId request id
   * @param {String} moduleId module id/name
   * @param {Object} msg message
   * @api private
   */
  var sendToMonitor = function(socket, reqId, moduleId, msg) {
    doSend(socket, 'monitor', protocol.composeRequest(reqId, moduleId, msg));
  };
  
  /**
   * send msg to client
   *
   * @param {Object} socket socket-io object
   * @param {Number} reqId request id
   * @param {String} moduleId module id/name
   * @param {Object} msg message
   * @api private
   */
  var sendToClient = function(socket, reqId, moduleId, msg) {
    doSend(socket, 'client', protocol.composeRequest(reqId, moduleId, msg));
  };
  
  var doSend = function(socket, topic, msg) {
    socket.send(topic, msg);
  }
  
  /**
   * broadcast msg to monitor
   *
   * @param {Object} record registered modules
   * @param {String} moduleId module id/name
   * @param {Object} msg message
   * @api private
   */
  var broadcastMonitors = function(records, moduleId, msg) {
    msg = protocol.composeRequest(null, moduleId, msg);
  
    if (records instanceof Array) {
      for (var i = 0, l = records.length; i < l; i++) {
        var socket = records[i].socket;
        doSend(socket, 'monitor', msg);
      }
    } else {
      for (var id in records) {
        var socket = records[id].socket;
        doSend(socket, 'monitor', msg);
      }
    }
  };
  
  var broadcastCommand = function(records, command, moduleId, msg) {
    msg = protocol.composeCommand(null, command, moduleId, msg);
  
    if (records instanceof Array) {
      for (var i = 0, l = records.length; i < l; i++) {
        var socket = records[i].socket;
        doSend(socket, 'monitor', msg);
      }
    } else {
      for (var id in records) {
        var socket = records[id].socket;
        doSend(socket, 'monitor', msg);
      }
    }
  };
  
  MasterAgent.prototype.doAuthUser = function(msg, socket, cb) {
    if (!msg.id) {
      // client should has a client id
      return cb(new Error('client should has a client id'));
    }
  
    var self = this;
    var username = msg.username;
    if (!username) {
      // client should auth with username
      doSend(socket, 'register', {
        code: protocol.PRO_FAIL,
        msg: 'client should auth with username'
      });
      return cb(new Error('client should auth with username'));
    }
  
    var authUser = self.consoleService.authUser;
    var env = self.consoleService.env;
    authUser(msg, env, function(user) {
      if (!user) {
        // client should auth with username
        doSend(socket, 'register', {
          code: protocol.PRO_FAIL,
          msg: 'client auth failed with username or password error'
        });
        return cb(new Error('client auth failed with username or password error'));
      }
  
      if (self.clients[msg.id]) {
        doSend(socket, 'register', {
          code: protocol.PRO_FAIL,
          msg: 'id has been registered. id:' + msg.id
        });
        return cb(new Error('id has been registered. id:' + msg.id));
      }
  
      logger.info('client user : ' + username + ' login to master');
      addConnection(self, msg.id, msg.type, null, user, socket);
      doSend(socket, 'register', {
        code: protocol.PRO_OK,
        msg: 'ok'
      });
  
      cb();
    });
  };
  
  MasterAgent.prototype.doAuthServer = function(msg, socket, cb) {
    var self = this;
    var authServer = self.consoleService.authServer;
    var env = self.consoleService.env;
    authServer(msg, env, function(status) {
      if (status !== 'ok') {
        doSend(socket, 'register', {
          code: protocol.PRO_FAIL,
          msg: 'server auth failed'
        });
        cb(new Error('server auth failed'));
        return;
      }
  
      var record = addConnection(self, msg.id, msg.serverType, msg.pid, msg.info, socket);
  
      doSend(socket, 'register', {
        code: protocol.PRO_OK,
        msg: 'ok'
      });
      msg.info = msg.info || {}
      msg.info.pid = msg.pid;
      self.emit('register', msg.info);
      cb(null);
    });
  };
  
  MasterAgent.prototype.doSend = doSend;
  
  MasterAgent.prototype.sendToMonitor = sendToMonitor;
  
  MasterAgent.prototype.addConnection = addConnection;
  
  MasterAgent.prototype.removeConnection = removeConnection;
  
  module.exports = MasterAgent;