before(async function setup() {
    cfg = sconfig(process.cwd() + '/test');
    const logger = new chassis.Logger(cfg.get('logger'));

    events = new Events(cfg.get('events:kafka'), logger);
    await events.start();

    const topics = cfg.get('events:kafka:topics');
    testTopic = events.topic(cfg.get('events:kafka:topics:test.resource:topic'));
    commandTopic = events.topic(cfg.get('events:kafka:topics:command:topic'));
    // subscribe all response events
    for (let eventName of cfg.get('events:kafka:topics:command:events')) {
      await commandTopic.on(eventName, eventListener);
    }

    server = new Server(cfg.get('server'), logger);
    db = await database.get(cfg.get('database:arango'), logger);
    await db.truncate();

    const config = cfg.get();
    delete config.database.nedb;  // not supported in default implementation

    const cis = new CommandInterface(server, config, logger, events);
    await server.bind('commandinterface', cis);
    await server.start();

    const client = new Client(cfg.get('client:commandinterface'), logger);
    service = await client.connect();
  });

 /**
 * updates the topic offset in redis periodically
 *
 */
 updateTopicOffsets(): any {
   // Iterate through the topics and updateOffsets periodically for each topic
   // events.topic(TopicName) - gives the topic object
   const kafkaCfg = this.config.get('events:kafka');
   const topicTypes = _.keys(kafkaCfg.topics);
   for (let i = 0; i < topicTypes.length; i += 1) {
     const topicType = topicTypes[i];
     const topicName = kafkaCfg.topics[topicType].topic;
     this.topics[topicType] = this.kafkaEvents.topic(topicName);
     this.timerID[i] = setInterval(this.storeOffset.bind(this),
       this.config.get('redis:offsetStoreInterval'), this.topics[topicType], topicName);
   }
 }

describe('CommandInterfaceService', () => {
  let db: any;
  let server: Server;
  let events: Events;
  const testEvent = {
    value: 'a test event',
    count: 0,
  };
  let service;
  let cfg;
  let testTopic;
  let commandTopic;
  let validate;
  const eventListener = async function (msg: any,
    context: any, config: any, eventName: string): Promise<any> {
    await validate(msg, eventName);
  };
  before(async function setup() {
    cfg = sconfig(process.cwd() + '/test');
    const logger = new chassis.Logger(cfg.get('logger'));

    events = new Events(cfg.get('events:kafka'), logger);
    await events.start();

    const topics = cfg.get('events:kafka:topics');
    testTopic = events.topic(cfg.get('events:kafka:topics:test.resource:topic'));
    commandTopic = events.topic(cfg.get('events:kafka:topics:command:topic'));
    // subscribe all response events
    for (let eventName of cfg.get('events:kafka:topics:command:events')) {
      await commandTopic.on(eventName, eventListener);
    }

    server = new Server(cfg.get('server'), logger);
    db = await database.get(cfg.get('database:arango'), logger);
    await db.truncate();

    const config = cfg.get();
    delete config.database.nedb;  // not supported in default implementation

    const cis = new CommandInterface(server, config, logger, events);
    await server.bind('commandinterface', cis);
    await server.start();

    const client = new Client(cfg.get('client:commandinterface'), logger);
    service = await client.connect();
  });
  after(async function teardown() {
    await server.stop();
    await events.stop();
  });
  describe('check', () => {
    it('should return the status', async function checkHealth() {
      let cmdPayload = encodeMsg({
        service: 'commandinterface'
      });

      const msg = {
        name: 'health_check',
        payload: cmdPayload
      };

      // validator called by the event listener
      validate = function (msg: any, eventName: string): void {
        eventName.should.equal('healthCheckResponse');
        should.exist(msg.services);
        msg.services.should.containEql('commandinterface');
        should.exist(msg.payload);
        const payload = decodeMsg(msg.payload);
        should.not.exist(payload.error);
        should.exist(payload.status);
        payload.status.should.equal('SERVING');
      };

      let offset = await commandTopic.$offset(-1);
      // check commandinterface service, should serve
      let resp = await service.command(msg);
      await commandTopic.$wait(offset); // wait for response on both Kafka & gRPC

      should.not.exist(resp.error);
      should.exist(resp.data);
      let data = decodeMsg(resp.data);
      should.exist(data.status);
      data.status.should.equal('SERVING');

      // should not serve if service does not exist
      cmdPayload = encodeMsg({
        service: 'does_not_exist'
      });
      // check none existing service, should throw error
      resp = await service.command({
        name: 'health_check',
        payload: cmdPayload
      });
      should.exist(resp.data);
      data = decodeMsg(resp.data);
      should.not.exist(resp.error); // no exception thrown
      should.exist(data.error);  // tolerant error handling
      data.error.should.equal('Service does_not_exist does not exist');
      // should check all binded services if no service is specified
      cmdPayload = encodeMsg({
//.........这里部分代码省略.........

 beforeEach(async function start() {
   events = new Events(cfg.get('events:kafka'), logger);
   await events.start();
 });

describe('offsetStore', () => {
  let events: Events;
  const topicName = 'test';
  let topic: Topic;
  let offsetStore: OffsetStore;
  const eventName = 'testCreated';
  const testMessage = { value: 'testValue', count: 1 };

  const cfg = sconfig(process.cwd() + '/test');
  const logger = new Logger(cfg.get('logger'));

  beforeEach(async function start() {
    events = new Events(cfg.get('events:kafka'), logger);
    await events.start();
  });
  afterEach(async function stop() {
    await offsetStore.stop();
    await events.stop();
  });

  it('should emit an event and verify the stored offset value from redis',
    async function testStoredOffsetValue() {
      this.timeout(10000);
      offsetStore = new OffsetStore(events, cfg, logger);
      topic = await (events.topic(topicName));

      const listener = function listener(message, context) {
        testMessage.value.should.equal(message.value);
        testMessage.count.should.equal(message.count);
      };
      // get the current offsetValue for 'test' topic before emitting message
      const currentOffset = await topic.$offset(-1);
      // emit message to kafka
      await topic.on(eventName, listener);
      await topic.emit(eventName, testMessage);
      const newOffset = await new Promise((resolve, reject) => {
        setTimeout(async () => {
          const offsetValue = await offsetStore.getOffset(topicName);
          resolve(offsetValue);
        }, 8000);
      });
      should.exist(newOffset);
      Number(newOffset).should.equal(currentOffset + 1);
    });
  it('should consume a previously emitted message from Kafka',
    async function testConsumeListener() {
      this.timeout(4000);
      // emit testMessage to kafka
      topic = await events.topic(topicName);
      await topic.emit(eventName, testMessage);

      // start offsetTracker subscribing to previous offset value read
      // from redis and consume the above message
      offsetStore = new OffsetStore(events, cfg, logger);
      const listener = async function listener(message, context) {
        testMessage.value.should.equal(message.value);
        testMessage.count.should.equal(message.count);
      };

      // get the current offsetValue for 'test' topic before emitting message
      let startingOffset = await offsetStore.getOffset(topicName);
      await topic.on(eventName, listener, { startingOffset });

      // wait for 2sec so that message is consumed and
      // test is not ended immediately
      return new Promise((resolve, reject) => {
        setTimeout(() => {
          resolve();
        }, 2000);
      });
    });
});

  constructor(server: Server, config: any, logger: Logger, events: Events) {
    if (_.isNil(events)) {
      if (logger.error) {
        logger.error('No Kafka client was provided. Disabling all commands.');
        return;
      }
    }
    if (!_.has(config, 'server.services')) {
      throw new Error('missing config server.services');
    }

    this.config = config;
    this.logger = logger;

    if (!_.has(this.config, 'events')
      || !_.has(this.config.events, 'kafka')
      || !_.has(this.config.events.kafka, 'topics')
      || !_.has(this.config.events.kafka.topics, 'command')) {
      throw new Error('Commands topic configuration was not provided.');
    }

    this.kafkaEvents = events;

    // Health
    this.health = {
      status: ServingStatus.UNKNOWN,
    };
    this.service = {};
    const service = this.service;
    const health = this.health;
    _.forEach(config.server.services, (serviceCfg, serviceName) => {
      service[serviceName] = {
        bound: false,
        transport: {},
      };
    });
    server.on('bound', (serviceName) => {
      service[serviceName].bound = true;
      health.status = ServingStatus.NOT_SERVING;
    });
    server.on('serving', (transports) => {
      health.status = ServingStatus.SERVING;
      _.forEach(transports, (transport, transportName) => {
        _.forEach(service, (srv, serviceName) => {
          service[serviceName].transport[transportName] = ServingStatus.SERVING;
        });
      });
    });
    server.on('stopped', (transports) => {
      health.status = ServingStatus.NOT_SERVING;
      _.forEach(transports, (transport, transportName) => {
        _.forEach(service, (srv, serviceName) => {
          service[serviceName].transport[transportName] = ServingStatus.NOT_SERVING;
        });
      });
    });

    // list of available commands
    this.commands = {
      reset: this.reset,
      restore: this.restore,
      reconfigure: this.reconfigure,
      health_check: this.check,
      version: this.version
    };
    const topicCfg = config.events.kafka.topics.command;
    this.commandTopic = events.topic(topicCfg.topic);
    // check for buffer fields
    this.bufferedCollection = new Map<string, string>();
    if (this.config.fieldHandlers && this.config.fieldHandlers.bufferFields) {
      for (let bufferedCollection in this.config.fieldHandlers.bufferFields) {
        this.bufferedCollection.set(bufferedCollection,
          this.config.fieldHandlers.bufferFields[bufferedCollection]);
      }
      this.logger.info('Buffered collections are:', this.bufferedCollection);
    }
  }

    async function testConsumeListener() {
      this.timeout(4000);
      // emit testMessage to kafka
      topic = await events.topic(topicName);
      await topic.emit(eventName, testMessage);

      // start offsetTracker subscribing to previous offset value read
      // from redis and consume the above message
      offsetStore = new OffsetStore(events, cfg, logger);
      const listener = async function listener(message, context) {
        testMessage.value.should.equal(message.value);
        testMessage.count.should.equal(message.count);
      };

      // get the current offsetValue for 'test' topic before emitting message
      let startingOffset = await offsetStore.getOffset(topicName);
      await topic.on(eventName, listener, { startingOffset });

      // wait for 2sec so that message is consumed and
      // test is not ended immediately
      return new Promise((resolve, reject) => {
        setTimeout(() => {
          resolve();
        }, 2000);
      });
    });

    async function testStoredOffsetValue() {
      this.timeout(10000);
      offsetStore = new OffsetStore(events, cfg, logger);
      topic = await (events.topic(topicName));

      const listener = function listener(message, context) {
        testMessage.value.should.equal(message.value);
        testMessage.count.should.equal(message.count);
      };
      // get the current offsetValue for 'test' topic before emitting message
      const currentOffset = await topic.$offset(-1);
      // emit message to kafka
      await topic.on(eventName, listener);
      await topic.emit(eventName, testMessage);
      const newOffset = await new Promise((resolve, reject) => {
        setTimeout(async () => {
          const offsetValue = await offsetStore.getOffset(topicName);
          resolve(offsetValue);
        }, 8000);
      });
      should.exist(newOffset);
      Number(newOffset).should.equal(currentOffset + 1);
    });

 after(async function teardown() {
   await server.stop();
   await events.stop();
 });

 afterEach(async function stop() {
   await offsetStore.stop();
   await events.stop();
 });