/**
   * Reset system data related to a service. Default implementation truncates
   * a set of ArangoDB instances, using the chassis-srv database provider.
   */
  async reset(): Promise<any> {
    this.logger.info('reset process started');
    if (this.health.status !== ServingStatus.NOT_SERVING) {
      this.logger.warn('reset process starting while server is serving');
    }

    let errorMsg = null;
    try {
      const dbCfgs = this.config.database;
      const dbCfgNames = _.keys(dbCfgs);
      for (let i = 0; i < dbCfgNames.length; i += 1) {
        const dbCfgName = dbCfgNames[i];
        const dbCfg = dbCfgs[dbCfgName];
        const db = await database.get(dbCfg, this.logger);
        switch (dbCfg.provider) {
          case 'arango':
            await db.truncate();
            this.logger.info(`arangodb ${dbCfg.database} truncated`);
            break;
          default:
            this.logger.error(
              `unsupported database provider ${dbCfg.provider} in database config ${dbCfgName}`);
            break;
        }
      }
    } catch (err) {
      this.logger.error('Unexpected error while resetting the system', err.message);
      errorMsg = err.message;
    }

    const eventObject = {
      services: _.keys(this.service),
      payload: null
    };

    if (errorMsg) {
      eventObject.payload = this.encodeMsg({
        error: errorMsg
      });
    } else {
      eventObject.payload = this.encodeMsg({
        status: 'Reset concluded successfully'
      });
    }
    await this.commandTopic.emit('resetResponse', eventObject);

    this.logger.info('reset process ended');

    if (errorMsg) {
      return {
        error: errorMsg
      };
    }
    return {};
  }

  private startToReceiveRestoreMessages(restoreTopic: Topic,
    drainEvent: Function): any {
    const asyncQueue = async.queue((msg, done) => {
      setImmediate(() => drainEvent(msg, err => {
        if (err) {
          done(err);
        } else {
          done();
        }
      }));
    }, 1);

    asyncQueue.drain = () => {
      // commit state first, before resuming
      this.logger.verbose('Committing offsets upon async queue drain');
      new Promise((resolve, reject) => {
        restoreTopic.consumer.commit((err, data) => {
          if (err) {
            return reject(err);
          }
          resolve(data);
        });
      }).then(() => {
        this.logger.info('Offset committed successfully');
      }).catch((err) => { });
    };
    this.logger.info('Async queue draining started.');
    return asyncQueue;
  }

 /**
  * get the offset value for the topic from redis
  * @param  {string} topic Topic name
  * @return {object}
  */
 async getOffset(topicName: string): Promise<number> {
   const redisKey = this.config.get('events:kafka:clientId') + ':' + topicName;
   const offsetValue = await new Promise<number>((resolve, reject) => {
     this.redisClient.get(redisKey, (err, response) => {
       if (err) {
         reject(err);
       }
       resolve(response);
     });
   });
   this.logger.info('The offset value retreived from redis for topic is:',
     { topicName, offsetValue });
   return offsetValue;
 }

  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);
    }
  }

 }).then(() => {
   this.logger.info('Offset committed successfully');
 }).catch((err) => { });

 /**
  * Reconfigure service
  * @param call
  * @param context
  */
 reconfigure(): any {
   this.logger.info('reconfigure is not implemented');
   throw new errors.Unimplemented('reconfigure is not implemented');
 }