ysjj-system/node_modules/cluster-client/lib/leader.js

'use strict';

const debug = require('debug')('cluster-client#leader');
const co = require('co');
const is = require('is-type-of');
const Base = require('sdk-base');
const utils = require('./utils');
const random = require('utility').random;
const ClusterServer = require('./server');
const Connection = require('./connection');
const Request = require('./protocol/request');
const Response = require('./protocol/response');

class Leader extends Base {
  /**
   * The Leader hold the real client
   *
   * @param {Object} options
   *  - {String} name - client name, default is the class name
   *  - {ClusterServer} server - the cluster server
   *  - {Boolean} isBroadcast - whether broadcast subscrption result to all followers or just one, default is true
   *  - {Number} heartbeatInterval - the heartbeat interval
   *  - {Function} createRealClient - to create the real client
   * @constructor
   */
  constructor(options) {
    super(options);
    this._connections = new Map();
    this._subListeners = new Map(); // subscribe key => listener
    this._subConnMap = new Map(); // subscribe key => conn key
    this._subData = new Map();
    // local socket server
    this._server = this.options.server;
    this._transcode = this.options.transcode;
    this._isReady = false;
    this._closeByUser = false;
    // the real client
    this._realClient = this.options.createRealClient();
    this._subscribeMethodName = this._findMethodName('subscribe');
    this._publishMethodName = this._findMethodName('publish');

    // event delegate
    utils.delegateEvents(this._realClient, this);

    if (is.function(this._realClient.ready)) {
      this._realClient.ready(err => {
        if (err) {
          this.ready(err);
        } else {
          this._isReady = true;
          this.ready(true);
        }
      });
    } else {
      this._isReady = true;
      this.ready(true);
    }

    this._handleConnection = this._handleConnection.bind(this);

    // subscribe its own channel
    this._server.on(`${this.options.name}_connection`, this._handleConnection);
    this._server.once('close', () => { this.emit('server_closed'); });
    this.on('server_closed', () => {
      this._handleClose().catch(err => { this.emit('error', err); });
    });

    // maxIdleTime is 3 times of heartbeatInterval
    const heartbeatInterval = this.options.heartbeatInterval;
    const maxIdleTime = this.options.heartbeatInterval * 3;

    this._heartbeatTimer = setInterval(() => {
      const now = Date.now();
      for (const conn of this._connections.values()) {
        const dur = now - conn.lastActiveTime;
        if (dur > maxIdleTime) {
          const err = new Error(`client no response in ${dur}ms exceeding maxIdleTime ${maxIdleTime}ms, maybe the connection is close on other side.`);
          err.name = 'ClusterClientNoResponseError';
          conn.close(err);
        }
      }
    }, heartbeatInterval);
  }

  get isLeader() {
    return true;
  }

  get logger() {
    return this.options.logger;
  }

  formatKey(reg) {
    return '$$inner$$__' + this.options.formatKey(reg);
  }

  subscribe(reg, listener) {
    const transcode = this._transcode;
    const conn = Object.create(Base.prototype, {
      isMock: { value: true },
      key: { value: `${this.options.name}_mock_conn_${utils.nextId()}` },
      lastActiveTime: {
        get() {
          return Date.now();
        },
      },
      listener: {
        get() {
          return listener;
        },
      },
      send: {
        value(req) {
          const result = transcode.decode(req.data);
          process.nextTick(() => {
            listener(result);
          });
        },
      },
      close: { value() {} },
    });
    conn.once('close', () => {
      this._connections.delete(conn.key);
      for (const connKeySet of this._subConnMap.values()) {
        connKeySet.delete(conn.key);
      }
    });

    this._connections.set(conn.key, conn);
    this._doSubscribe(reg, conn);
  }

  unSubscribe(reg, listener) {
    const key = this.formatKey(reg);
    const connKeySet = this._subConnMap.get(key) || new Set();
    const newConnKeySet = new Set();
    for (const connKey of connKeySet.values()) {
      const conn = this._connections.get(connKey);
      if (!conn) {
        continue;
      }
      if (conn.isMock && (!listener || conn.listener === listener)) {
        this._connections.delete(connKey);
        continue;
      }
      newConnKeySet.add(connKey);
    }
    this._subConnMap.set(key, newConnKeySet);
  }

  publish(reg) {
    this._realClient[this._publishMethodName](reg);
  }

  invoke(methodName, args, callback) {
    let method = this._realClient[methodName];
    // compatible with generatorFunction
    if (is.generatorFunction(method)) {
      method = co.wrap(method);
    }
    args.push(callback);
    const ret = method.apply(this._realClient, args);
    if (callback && is.promise(ret)) {
      ret.then(result => callback(null, result), err => callback(err))
        // to avoid uncaught exception in callback function, then cause unhandledRejection
        .catch(err => { this._errorHandler(err); });
    }
  }

  _doSubscribe(reg, conn) {
    const key = this.formatKey(reg);
    const callback = err => {
      if (err) {
        this._errorHandler(err);
      }
    };
    const isBroadcast = this.options.isBroadcast;
    const timeout = this.options.responseTimeout;

    const connKeySet = this._subConnMap.get(key) || new Set();
    connKeySet.add(conn.key);
    this._subConnMap.set(key, connKeySet);

    // only subscribe once in cluster mode, and broadcast to all followers
    if (!this._subListeners.has(key)) {
      const listener = result => {
        const data = this._transcode.encode(result);
        this._subData.set(key, data);

        const connKeySet = this._subConnMap.get(key);
        if (!connKeySet) {
          return;
        }
        let keys = Array.from(connKeySet.values());
        // if isBroadcast equal to false, random pick one to notify
        if (!isBroadcast) {
          keys = [ keys[random(keys.length)] ];
        }

        for (const connKey of keys) {
          const conn = this._connections.get(connKey);
          if (conn) {
            debug('[Leader:%s] push subscribe data to cluster client#%s', this.options.name, connKey);
            conn.send(new Request({
              timeout,
              connObj: {
                type: 'subscribe_result',
                key,
              },
              data,
            }), callback);
          }
        }
      };
      this._subListeners.set(key, listener);
      this._realClient[this._subscribeMethodName](reg, listener);
    } else if (this._subData.has(key) && isBroadcast) {
      conn.send(new Request({
        timeout,
        connObj: {
          type: 'subscribe_result',
          key,
        },
        data: this._subData.get(key),
      }), callback);
    }
  }

  _findMethodName(type) {
    return utils.findMethodName(this.options.descriptors, type);
  }

  // handle new socket connect
  _handleConnection(socket, req) {
    debug('[Leader:%s] socket connected, port: %d', this.options.name, socket.remotePort);

    const conn = new Connection({
      socket,
      name: this.options.name,
      logger: this.logger,
      transcode: this.options.transcode,
      requestTimeout: this.options.requestTimeout,
    });
    this._connections.set(conn.key, conn);
    conn.once('close', () => {
      this._connections.delete(conn.key);
      for (const connKeySet of this._subConnMap.values()) {
        connKeySet.delete(conn.key);
      }
    });
    conn.on('error', err => this._errorHandler(err));
    conn.on('request', (req, res) => this._handleRequest(req, res, conn));

    // handle register channel request
    const res = new Response({
      id: req.id,
      timeout: req.timeout,
    });
    this._handleRequest(req, res, conn);
  }

  _handleSubscribe(req, conn) {
    const connObj = req.connObj || {};
    this._doSubscribe(connObj.reg, conn);
  }

  _handleUnSubscribe(req, conn) {
    const connObj = req.connObj || {};
    const key = this.formatKey(connObj.reg);
    const connKeySet = this._subConnMap.get(key) || new Set();
    connKeySet.delete(conn.key);
    this._subConnMap.set(key, connKeySet);
  }

  // handle request from followers
  _handleRequest(req, res, conn) {
    const connObj = req.connObj || {};
    // update last active time to make sure not kick out by leader
    conn.lastActiveTime = Date.now();

    switch (connObj.type) {
      case 'subscribe':
        debug('[Leader:%s] received subscribe request from follower, req: %j, conn: %s', this.options.name, req, conn.key);
        this._handleSubscribe(req, conn);
        break;
      case 'unSubscribe':
        debug('[Leader:%s] received unSubscribe request from follower, req: %j, conn: %s', this.options.name, req, conn.key);
        this._handleUnSubscribe(req, conn);
        break;
      case 'invoke':
      {
        debug('[Leader:%s] received invoke request from follower, req: %j, conn: %s', this.options.name, req, conn.key);
        const argLength = connObj.argLength;
        const args = [];
        if (argLength > 0) {
          const data = req.data;
          for (let i = 0, offset = 0; i < argLength; ++i) {
            const len = data.readUInt32BE(offset);
            const arg = this._transcode.decode(data.slice(offset + 4, offset + 4 + len));
            args.push(arg);
            offset += (4 + len);
          }
        }

        if (connObj.oneway) {
          this.invoke(connObj.method, args);
        } else {
          const startTime = Date.now();
          this.invoke(connObj.method, args, (err, result) => {
            // no response if processing timeout, just record error
            if (req.timeout && Date.now() - startTime > req.timeout) {
              const err = new Error(`[Leader:${this.options.name}] invoke method:${connObj.method} timeout for req#${req.id}`);
              err.name = 'ClusterLeaderTimeoutError';
              err.method = connObj.method;
              err.args = args;
              this._errorHandler(err);
              return;
            }

            if (err) {
              const data = Object.assign({
                stack: err.stack,
                name: err.name,
                message: err.message,
              }, err);
              err.method = connObj.method;
              err.args = connObj.args;
              this._errorHandler(err);

              res.connObj = {
                type: 'invoke_result',
                success: false,
              };
              res.data = this._transcode.encode(data);
            } else {
              debug('[Leader:%s] send method:%s result to follower, result: %j', this.options.name, connObj.method, result);
              const data = this._transcode.encode(result);
              res.connObj = {
                type: 'invoke_result',
                success: true,
              };
              res.data = data;
            }
            conn.send(res);
          });
        }
        break;
      }
      case 'heartbeat':
        debug('[Leader:%s] received heartbeat request from follower, req: %j, conn: %s', this.options.name, req, conn.key);
        res.connObj = { type: 'heartbeat_res' };
        conn.send(res);
        break;
      case 'register_channel':
        // make sure response after leader is ready
        this.ready(err => {
          if (err) {
            res.connObj = { type: 'register_channel_res' };
            const data = Object.assign({
              message: err.message,
              stack: err.stack,
              name: err.name,
            }, err);
            res.data = this._transcode.encode({
              success: false,
              error: data,
            });
          } else {
            res.connObj = { type: 'register_channel_res' };
            res.data = this._transcode.encode({ success: true });
          }
          conn.send(res);
        });
        break;
      default:
      {
        const err = new Error(`unsupport data type: ${connObj.type}`);
        err.name = 'ClusterRequestTypeError';
        this._errorHandler(err);
        break;
      }
    }
  }

  // emit error asynchronously
  _errorHandler(err) {
    setImmediate(() => {
      if (!this._closeByUser) {
        this.emit('error', err);
      }
    });
  }

  async _handleClose() {
    debug('[Leader:%s] leader server is closed', this.options.name);
    // close the real client
    if (this._realClient) {
      const originClose = this._findMethodName('close');
      if (originClose) {
        // support common function, generatorFunction, and function returning a promise
        await utils.callFn(this._realClient[originClose].bind(this._realClient));
      }
    }
    clearInterval(this._heartbeatTimer);
    this._heartbeatTimer = null;
    this.emit('close');
  }

  async close() {
    this._closeByUser = true;
    debug('[Leader:%s] try to close leader', this.options.name);
    // 1. stop listening to server channel
    this._server.removeListener(`${this.options.name}_connection`, this._handleConnection);

    // 2. close all mock connections
    for (const conn of this._connections.values()) {
      if (conn.isMock) {
        conn.emit('close');
      }
    }

    // 3. close server
    //    CANNOT close server directly by server.close(), other cluster clients may be using it
    this.removeAllListeners('server_closed');
    await ClusterServer.close(this.options.name, this._server);

    // 5. close real client
    await this._handleClose();
  }
}

module.exports = Leader;