ali-ons/lib/remoting_client.js

'use strict';

const assert = require('assert');
const Base = require('sdk-base');
const logger = require('./logger');
const Channel = require('./channel');

const defaultOptions = {
  logger,
  responseTimeout: 30000,
};

class RemotingClient extends Base {

  /**
   * rocketmq remoting client
   * @param {Object} options
   *   - {HttpClient} httpclient - http request client
   *   - {Object} [logger] - log module
   *   - {Number} [responseTimeout] - tcp response timeout
   * @class
   */
  constructor(options) {
    assert(options.onsAddr || options.nameSrv, '[RemotingClient] options.onsAddr or options.nameSrv one of them is required');
    assert(options.httpclient, '[RemotingClient] options.httpclient is required');
    super(Object.assign({ initMethod: 'init' }, defaultOptions, options));
    this._index = 0;
    this._inited = false;
    this._namesrvAddrList = [];
    this._channels = new Map();
  }

  /**
   * @property {HttpClient} RemotingClient#httpclient
   */
  get httpclient() {
    return this.options.httpclient;
  }

  /**
   * @property {Object} RemotingClient#logger
   */
  get logger() {
    return this.options.logger;
  }

  /**
   * @property {Number} RemotingClient#responseTimeout
   */
  get responseTimeout() {
    return this.options.responseTimeout;
  }

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

  /**
   * start the client
   * @return {void}
   */
  async init() {
    // get name server address at first
    await this.updateNameServerAddressList();
    this._inited = true;
    this.logger.info('[mq:remoting_client] remoting client started');
  }

  /**
   * close the client
   * @return {void}
   */
  async close() {
    if (!this._inited) {
      return;
    }

    // wait all channel close
    await Promise.all(Array.from(this._channels.keys()).map(addr => {
      return new Promise(resolve => {
        const channel = this._channels.get(addr);
        if (channel && channel.isOK) {
          channel.once('close', resolve);
          channel.close();
        } else {
          resolve();
        }
        this._channels.delete(addr);
      });
    }));

    this._inited = false;
    this.emit('close');
    this.removeAllListeners();

    this.logger.info('[mq:remoting_client] remoting client is closed');
  }

  /**
   * default error handler
   * @param {Error} err - error object
   * @return {void}
   */
  error(err) {
    this.emit('error', err);
  }

  async handleClose(addr, channel) {
    if (this._channels.get(addr) && this._channels.get(addr).clientId === channel.clientId) {
      this._channels.delete(addr);
    }
    await this.updateNameServerAddressListThrottle();
  }

  async updateNameServerAddressListThrottle() {
    if (this._updatingNameServerAddressList) {
      // if it is in the process of updating and there is a new update task
      // it needs to be recorded at this time and update again at the end to avoid missing the last update.
      this._needToUpdatingNameServerAddressList = true;
      return;
    }
    this._updatingNameServerAddressList = true;
    let updateError;
    try {
      await this.updateNameServerAddressList();
    } catch(e) {
      updateError = e;
    }
    this._updatingNameServerAddressList = false;
    if (this._needToUpdatingNameServerAddressList) {
      this._needToUpdatingNameServerAddressList = false;
      this.updateNameServerAddressListThrottle();
    }
    if (updateError) {
      throw updateError;
    }
  }

  /**
   * fetch name server address list
   * @return {void}
   */
  async updateNameServerAddressList() {
    if (this.options.nameSrv) {
      this._namesrvAddrList = Array.isArray(this.options.nameSrv) ? this.options.nameSrv : [ this.options.nameSrv ];
      return;
    }
    let url = this.options.onsAddr;
    if (this.unitName) {
      url = url + '-' + this.unitName + '?nofix=1';
    }
    const ret = await this.httpclient.request(url, {
      timeout: this.options.connectTimeout || 10000,
    });
    if (ret.status === 200) {
      const addrs = ret.data.toString().trim();
      const newList = addrs.split(';');
      if (newList.length) {
        this._namesrvAddrList = newList;
      }
      this.logger.info('[mq:remoting_client] fetch name server addresses successfully, address: %s', addrs);
    } else {
      throw new Error('[mq:remoting_client] fetch name server addresses failed, ret.statusCode: ' + ret.status);
    }
  }

  /**
   * invoke command
   * @param {String} addr - server address
   * @param {RemotingCommand} command - remoting command
   * @param {Number} [timeout] - response timeout
   * @return {Object} response
   */
  async invoke(addr, command, timeout) {
    if (!this._inited) {
      await this.ready();
    }
    return await this.getAndCreateChannel(addr)
      .invoke(command, timeout || this.responseTimeout);
  }

  /**
   * invoke command for namesrv, return success at least once success
   * @param {RemotingCommand} command - remoting command
   * @param {Number} [timeout] - response timeout
   * @return {Object} response
   */
  async invokeForNameSrvAtLeastOnce(command, timeout) {
    let response = {};
    let count = this._namesrvAddrList.length;
    let err = null;
    while (count--) {
      try {
        response = await this.invoke(null, command, timeout);
        // 有一个成功即无需抛 invoke 错误
        err = null;
        break;
      } catch (e) {
        err = e;
        this.logger.warn(err);
      }
    }
    if (err) {
      throw err;
    }
    return response;
  }

  /**
   * invoke command without response
   * @param {String} addr - server address
   * @param {RemotingCommand} command - remoting command
   * @return {void}
   */
  async invokeOneway(addr, command) {
    if (!this._inited) {
      await this.ready();
    }
    await this.getAndCreateChannel(addr).invokeOneway(command);
  }

  /**
   * get request channel from address
   * @param {String} [addr] - server address
   * @return {Channel} request channel
   */
  getAndCreateChannel(addr) {
    if (!addr) {
      this._index = ++this._index % this._namesrvAddrList.length;
      addr = this._namesrvAddrList[this._index];
    }
    let channel = this._channels.get(addr);
    if (channel && channel.isOK) {
      return channel;
    }
    channel = new Channel(addr, this.options);
    this._channels.set(addr, channel);
    channel.once('close', this.handleClose.bind(this, addr, channel));
    channel.on('error', err => this.error(err));
    channel.on('request', (request, address) => this.emit('request', request, address));
    return channel;
  }
}

module.exports = RemotingClient;