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

'use strict';

const debug = require('debug')('cluster-client#follower');
const is = require('is-type-of');
const Base = require('tcp-base');
const Packet = require('./protocol/packet');
const Request = require('./protocol/request');
const Response = require('./protocol/response');

class Follower extends Base {
  /**
   * "Fake" Client, forward request to leader
   *
   * @param {Object} options
   *  - {Number} port - the port
   *  - {Map} descriptors - interface descriptors
   *  - {Transcode} transcode - serialze / deserialze methods
   *  - {Number} responseTimeout - the timeout
   * @constructor
   */
  constructor(options) {
    // local address
    options.host = '127.0.0.1';
    super(options);
    this._publishMethodName = this._findMethodName('publish');
    this._subInfo = new Set();
    this._subData = new Map();
    this._transcode = options.transcode;
    this._closeByUser = false;

    this.on('request', req => this._handleRequest(req));
    // avoid warning message
    this.setMaxListeners(100);
  }
  get isLeader() {
    return false;
  }

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

  get heartBeatPacket() {
    const heartbeat = new Request({
      connObj: {
        type: 'heartbeat',
      },
      timeout: this.options.responseTimeout,
    });
    return heartbeat.encode();
  }

  getHeader() {
    return this.read(24);
  }

  getBodyLength(header) {
    return header.readInt32BE(16) + header.readInt32BE(20);
  }

  close(err) {
    this._closeByUser = true;
    return super.close(err);
  }

  decode(body, header) {
    const buf = Buffer.concat([ header, body ]);
    const packet = Packet.decode(buf);
    const connObj = packet.connObj;
    if (connObj && connObj.type === 'invoke_result') {
      let data;
      if (packet.data) {
        data = this.options.transcode.decode(packet.data);
      }
      if (connObj.success) {
        return {
          id: packet.id,
          isResponse: packet.isResponse,
          data,
        };
      }
      const error = new Error(data.message);
      Object.assign(error, data);
      return {
        id: packet.id,
        isResponse: packet.isResponse,
        error,
      };
    }
    return {
      id: packet.id,
      isResponse: packet.isResponse,
      connObj: packet.connObj,
      data: packet.data,
    };
  }

  send(...args) {
    // just ignore after close
    if (this._closeByUser) {
      return;
    }
    return super.send(...args);
  }

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

  subscribe(reg, listener) {
    const key = this.formatKey(reg);
    this.on(key, listener);

    // no need duplicate subscribe
    if (!this._subInfo.has(key)) {
      debug('[Follower:%s] subscribe %j for first time', this.options.name, reg);
      const req = new Request({
        connObj: { type: 'subscribe', key, reg },
        timeout: this.options.responseTimeout,
      });

      // send subscription
      this.send({
        id: req.id,
        oneway: true,
        data: req.encode(),
      });
      this._subInfo.add(key);
    } else if (this._subData.has(key)) {
      debug('[Follower:%s] subscribe %j', this.options.name, reg);
      process.nextTick(() => {
        listener(this._subData.get(key));
      });
    }
    return this;
  }

  unSubscribe(reg, listener) {
    const key = this.formatKey(reg);
    if (listener) {
      this.removeListener(key, listener);
    } else {
      this.removeAllListeners(key);
    }
    if (this.listeners(key).length === 0) {
      debug('[Follower:%s] no more subscriber for %j, send unSubscribe req to leader', this.options.name, reg);
      this._subInfo.delete(key);

      const req = new Request({
        connObj: { type: 'unSubscribe', key, reg },
        timeout: this.options.responseTimeout,
      });
      // send subscription
      this.send({
        id: req.id,
        oneway: true,
        data: req.encode(),
      });
    }
  }

  publish(reg) {
    this.invoke(this._publishMethodName, [ reg ]);
    return this;
  }

  invoke(method, args, callback) {
    const oneway = !is.function(callback); // if no callback, means oneway
    const argLength = args.length;
    let data;
    // data:
    // +-----+---------------+-----+---------------+
    // | len |   arg1 body   | len |   arg2 body   |  ...
    // +-----+---------------+-----+---------------+
    if (argLength > 0) {
      let argsBufLength = 0;
      const arr = [];
      for (const arg of args) {
        const argBuf = this._transcode.encode(arg);
        const len = argBuf.length;
        const buf = Buffer.alloc(4 + len);
        buf.writeInt32BE(len, 0);
        argBuf.copy(buf, 4, 0, len);
        arr.push(buf);
        argsBufLength += (len + 4);
      }
      data = Buffer.concat(arr, argsBufLength);
    }
    const req = new Request({
      connObj: {
        type: 'invoke',
        method,
        argLength,
        oneway,
      },
      data,
      timeout: this.options.responseTimeout,
    });
    // send invoke request
    this.send({
      id: req.id,
      oneway,
      data: req.encode(),
    }, callback);
  }

  _registerChannel() {
    const req = new Request({
      connObj: {
        type: 'register_channel',
        channelName: this.options.name,
      },
      timeout: this.options.responseTimeout,
    });
    // send invoke request
    this.send({
      id: req.id,
      oneway: false,
      data: req.encode(),
    }, (err, data) => {
      if (err) {
        // if socket alive, do retry
        if (this._socket) {
          err.message = `register to channel: ${this.options.name} failed, will retry after 3s, ${err.message}`;
          this.logger.warn(err);
          // if exception, retry after 3s
          setTimeout(() => this._registerChannel(), 3000);
        } else {
          this.ready(err);
        }
        return;
      }
      const res = this._transcode.decode(data);
      if (res.success) {
        debug('[Follower:%s] register to channel: %s success', this.options.name, this.options.name);
        this.ready(true);
      } else {
        const error = new Error(res.error.message);
        Object.assign(error, res.error);
        this.ready(error);
      }
    });
  }

  _findMethodName(type) {
    for (const method of this.options.descriptors.keys()) {
      const descriptor = this.options.descriptors.get(method);
      if (descriptor.type === 'delegate' && descriptor.to === type) {
        return method;
      }
    }
    return null;
  }

  _handleRequest(req) {
    debug('[Follower:%s] receive req: %j from leader', this.options.name, req);
    const connObj = req.connObj || {};
    if (connObj.type === 'subscribe_result') {
      const result = this._transcode.decode(req.data);
      this.emit(connObj.key, result);
      this._subData.set(connObj.key, result);
      // feedback
      const res = new Response({
        id: req.id,
        timeout: req.timeout,
        connObj: { type: 'subscribe_result_res' },
      });
      this.send({
        id: req.id,
        oneway: true,
        data: res.encode(),
      });
    }
  }

  _connect(done) {
    if (!done) {
      done = err => {
        if (err) {
          this.ready(err);
        } else {
          // register to proper channel, difference type of client into difference channel
          this._registerChannel();
        }
      };
    }
    super._connect(done);
  }
}

module.exports = Follower;