Egg 多進(jìn)程研發(fā)模式增強(qiáng)

+--------+   +--------+
| Client |   | Client |   ... n
+--------+   +--------+
    |  \     /   |
    |    \ /     |        n * m 個(gè)鏈接
    |    / \     |
    |  /     \   |
+--------+   +--------+
| Server |   | Server |   ... m
+--------+   +--------+

             +-------+
             | start |
             +---+---+
                 |
        +--------+---------+
      __| port competition |__
win /   +------------------+  \ lose
   /                           \
+---------------+     tcp conn     +-------------------+
| Leader(Agent) |<---------------->| Follower(Worker1) |
+---------------+                  +-------------------+
    |            \ tcp conn
    |             \
+--------+         +-------------------+
| Client |         | Follower(Worker2) |
+--------+         +-------------------+

const Base = require('sdk-base');

class Client extends Base {
  constructor(options) {
    super(options);
    // 在初始化成功以后記得 ready
    this.ready(true);
  }

  /**
   * 訂閱
   *
   * @param {Object} info - 訂閱的信息（一個(gè) JSON 對象，注意盡量不要包含 Function, Buffer, Date 這類屬性）
   * @param {Function} listener - 監(jiān)聽的回調(diào)函數(shù)，接收一個(gè)參數(shù)就是監(jiān)聽到的結(jié)果對象
   */
  subscribe(info, listener) {
    // ...
  }

  /**
   * 發(fā)布
   *
   * @param {Object} info - 發(fā)布的信息，和上面 subscribe 的 info 類似
   */
  publish(info) {
    // ...
  }

  /**
   * 獲取數(shù)據(jù) (invoke)
   *
   * @param {String} id - id
   * @return {Object} result
   */
  async getData(id) {
    // ...
  }
}

0       1       2               4                                                              12
+-------+-------+---------------+---------------------------------------------------------------+
|version|req/res|    reserved   |                          request id                           |
+-------------------------------+-------------------------------+-------------------------------+
|           timeout             |   connection object length    |   application object length   |
+-------------------------------+---------------------------------------------------------------+
|         conn object (JSON format)  ...                    |            app object             |
+-----------------------------------------------------------+                                   |
|                                          ...                                                  |
+-----------------------------------------------------------------------------------------------+

+----------+             +---------------+          +---------+
| Follower |             |  Local Server |          |  Leader |
+----------+             +---------------+          +---------+
     |     register channel     |       assign to        |
     + -----------------------> |  --------------------> |
     |                          |                        |
     |                                subscribe          |
     + ------------------------------------------------> |
     |                                 publish           |
     + ------------------------------------------------> |
     |                                                   |
     |       subscribe result                            |
     | <------------------------------------------------ +
     |                                                   |
     |                                 invoke            |
     + ------------------------------------------------> |
     |          invoke result                            |
     | <------------------------------------------------ +
     |                                                   |

// registry_client.js
const URL = require('url');
const Base = require('sdk-base');

class RegistryClient extends Base {
  constructor(options) {
    super({
      // 指定異步啟動的方法
      initMethod: 'init',
    });
    this._options = options;
    this._registered = new Map();
  }

  /**
   * 啟動邏輯
   */
  async init() {
    this.ready(true);
  }

  /**
   * 獲取配置
   * @param {String} dataId - the dataId
   * @return {Object} 配置
   */
  async getConfig(dataId) {
    return this._registered.get(dataId);
  }

  /**
   * 訂閱
   * @param {Object} reg
   *   - {String} dataId - the dataId
   * @param {Function}  listener - the listener
   */
  subscribe(reg, listener) {
    const key = reg.dataId;
    this.on(key, listener);

    const data = this._registered.get(key);
    if (data) {
      process.nextTick(() => listener(data));
    }
  }

  /**
   * 發(fā)布
   * @param {Object} reg
   *   - {String} dataId - the dataId
   *   - {String} publishData - the publish data
   */
  publish(reg) {
    const key = reg.dataId;
    let changed = false;

    if (this._registered.has(key)) {
      const arr = this._registered.get(key);
      if (arr.indexOf(reg.publishData) === -1) {
        changed = true;
        arr.push(reg.publishData);
      }
    } else {
      changed = true;
      this._registered.set(key, [reg.publishData]);
    }
    if (changed) {
      this.emit(key, this._registered.get(key).map(url => URL.parse(url, true)));
    }
  }
}

module.exports = RegistryClient;

// agent.js
const RegistryClient = require('registry_client');

module.exports = agent => {
  // 對 RegistryClient 進(jìn)行封裝和實(shí)例化
  agent.registryClient = agent.cluster(RegistryClient)
    // create 方法的參數(shù)就是 RegistryClient 構(gòu)造函數(shù)的參數(shù)
    .create({});

  agent.beforeStart(async () => {
    await agent.registryClient.ready();
    agent.coreLogger.info('registry client is ready');
  });
};

// app.js
const RegistryClient = require('registry_client');

module.exports = app => {
  app.registryClient = app.cluster(RegistryClient).create({});
  app.beforeStart(async () => {
    await app.registryClient.ready();
    app.coreLogger.info('registry client is ready');

    // 調(diào)用 subscribe 進(jìn)行訂閱
    app.registryClient.subscribe({
      dataId: 'demo.DemoService',
    }, val => {
      // ...
    });

    // 調(diào)用 publish 發(fā)布數(shù)據(jù)
    app.registryClient.publish({
      dataId: 'demo.DemoService',
      publishData: 'xxx',
    });

    // 調(diào)用 getConfig 接口
    const res = await app.registryClient.getConfig('demo.DemoService');
    console.log(res);
  });
};

class MockClient extends Base {
  constructor(options) {
    super({
      initMethod: 'init',
    });
    this._options = options;
    this._registered = new Map();
  }

  async init() {
    this.ready(true);
  }

  sub(info, listener) {
    const key = reg.dataId;
    this.on(key, listener);

    const data = this._registered.get(key);
    if (data) {
      process.nextTick(() => listener(data));
    }
  }

  ...
}

// agent.js
module.exports = agent => {
  agent.mockClient = agent.cluster(MockClient)
    // 將 sub 代理到 subscribe 邏輯上
    .delegate('sub', 'subscribe')
    .create();

  agent.beforeStart(async () => {
    await agent.mockClient.ready();
  });
};

// app.js
module.exports = app => {
  app.mockClient = app.cluster(MockClient)
    // 將 sub 代理到 subscribe 邏輯上
    .delegate('sub', 'subscribe')
    .create();

  app.beforeStart(async () => {
    await app.mockClient.ready();

    app.sub({ id: 'test-id' }, val => {
      // put your code here
    });
  });
};

// some-client/index.js
const cluster = require('cluster-client');
const RegistryClient = require('./registry_client');

class APIClient extends Base {
  constructor(options) {
    super(options);

    // options.cluster 用于給 Egg 的插件傳遞 app.cluster 進(jìn)來
    this._client = (options.cluster || cluster)(RegistryClient).create(options);
    this._client.ready(() => this.ready(true));

    this._cache = {};

    // subMap:
    // {
    //   foo: reg1,
    //   bar: reg2,
    // }
    const subMap = options.subMap;

    for (const key in subMap) {
      this.subscribe(subMap[key], value => {
        this._cache[key] = value;
      });
    }
  }

  subscribe(reg, listener) {
    this._client.subscribe(reg, listener);
  }

  publish(reg) {
    this._client.publish(reg);
  }

  get(key) {
    return this._cache[key];
  }
}

// 最終模塊向外暴露這個(gè) APIClient
module.exports = APIClient;

// app.js || agent.js
const APIClient = require('some-client'); // 上面那個(gè)模塊
module.exports = app => {
  const config = app.config.apiClient;
  app.apiClient = new APIClient(Object.assign({}, config, { cluster: app.cluster });
  app.beforeStart(async () => {
    await app.apiClient.ready();
  });
};

// config.${env}.js
exports.apiClient = {
  subMap: {
    foo: {
      id: '',
    },
    // bar...
  }
};

const APIClientBase = require('cluster-client').APIClientBase;
const RegistryClient = require('./registry_client');

class APIClient extends APIClientBase {
  // 返回原始的客戶端類
  get DataClient() {
    return RegistryClient;
  }

  // 用于設(shè)置 cluster-client 相關(guān)參數(shù)，等同于 cluster 方法的第二個(gè)參數(shù)
  get clusterOptions() {
    return {
      responseTimeout: 120 * 1000,
    };
  }

  subscribe(reg, listener) {
    this._client.subscribe(reg, listener);
  }

  publish(reg) {
    this._client.publish(reg);
  }

  get(key) {
    return this._cache[key];
  }
}

+------------------------------------------------+
| APIClient                                      |
|       +----------------------------------------|
|       | ClusterClient                          |
|       |      +---------------------------------|
|       |      | RegistryClient                  |
+------------------------------------------------+

/**
 * @property {Number} responseTimeout - response timeout, default is 60000
 * @property {Transcode} [transcode]
 *   - {Function} encode - custom serialize method
 *   - {Function} decode - custom deserialize method
 */
config.clusterClient = {
  responseTimeout: 60000,
};

app.registryClient = app.cluster(RegistryClient, {
  responseTimeout: 120 * 1000, // 這里傳入的是和 cluster-client 相關(guān)的參數(shù)
}).create({
  // 這里傳入的是 RegistryClient 需要的參數(shù)
});

const APIClientBase = require('cluster-client').APIClientBase;
const RegistryClient = require('./registry_client');

class APIClient extends APIClientBase {
  get DataClient() {
    return RegistryClient;
  }

  get clusterOptions() {
    return {
      responseTimeout: 120 * 1000,
    };
  }

  // ...
}

module.exports = APIClient;