# enn-egg-kafka

[![NPM version][npm-image]][npm-url]
[![build status][travis-image]][travis-url]
[![Test coverage][codecov-image]][codecov-url]
[![David deps][david-image]][david-url]
[![Known Vulnerabilities][snyk-image]][snyk-url]
[![npm download][download-image]][download-url]

[npm-image]: https://img.shields.io/npm/v/egg-enn-egg-kafka.svg?style=flat-square
[npm-url]: https://npmjs.org/package/enn-egg-kafka
[travis-image]: https://img.shields.io/travis/eggjs/egg-enn-egg-kafka.svg?style=flat-square
[travis-url]: https://travis-ci.org/eggjs/egg-enn-egg-kafka
[codecov-image]: https://img.shields.io/codecov/c/github/eggjs/egg-enn-egg-kafka.svg?style=flat-square
[codecov-url]: https://codecov.io/github/eggjs/egg-enn-egg-kafka?branch=master
[david-image]: https://img.shields.io/david/eggjs/egg-enn-egg-kafka.svg?style=flat-square
[david-url]: https://david-dm.org/eggjs/egg-enn-egg-kafka
[snyk-image]: https://snyk.io/test/npm/egg-enn-egg-kafka/badge.svg?style=flat-square
[snyk-url]: https://snyk.io/test/npm/egg-enn-egg-kafka
[download-image]: https://img.shields.io/npm/dm/egg-enn-egg-kafka.svg?style=flat-square
[download-url]: https://npmjs.org/package/enn-egg-kafka

<!--
Description here.
-->

## 依赖说明

### 依赖的 egg 版本

enn-egg-kafka 版本 | egg 1.x
--- | ---
1.x | 😁
0.x | ❌

### 依赖的插件
<!--

如果有依赖其它插件，请在这里特别说明。如

- security
- multipart

-->

## 开启插件

```js
// config/plugin.js
exports.kafka = {
  enable: true,
  package: 'enn-egg-kafka',
};
```

## 使用场景

- 对[kafkajs](https://www.npmjs.com/package/kafkajs)模块封装的eggjs插件


## 配置

- 配置 `config/config.default.js`
```js
  config.kafka = {
    client: {
      clientId: 'broker-data-sync',
      brokers: ['10.0.3.42:9092'],
      consumer: {
        groupId: 'group-test',
      },
      producer: {
        allowAutoTopicCreation: true,
      },
    },
    afterReady: false,
  };
```
其中afterReady为新增字段，设置为true后，等待端口启动完成，再启动插件，其他配置字段详见 https://kafka.js.org/docs/getting-started
## 使用

```js
/*
* 生产消息
*/
  await app.kafka.producerSend('topic-enn-rd', [{ key: 'key2111', value: 'hey hey!1111' }])
  /*
committed offset when starting to fetch messages. If the offset is invalid or not defined, fromBeginning defines the behavior of the consumer group. This can be configured when subscribing to a topic.
  //When fromBeginning is true, the group will use the earliest offset. If set to false, it will use the latest offset. The default is false.
/*
* 消费消息-单条 fromBeginning默认为true，从commit后的offset开始消费
*/
  await app.kafka.consumerInit('test_topic');
  app.kafka.consumer.run({
    eachMessage: async ({
      topic,
      partition,
      message,
      heartbeat
    }) => {
      const data = {
        partition,
        topic,
        key: message.key.toString(),
        value: message.value.toString(),
        headers: message.headers,
        timestamp: message.timestamp,
        offset: message.offset,
      };
      if (message.extra) {
        data.extra = message.extra.toString();
      }
      console.log(message)
    }
  });
  
/*
* 消费消息-批量
*/
  await app.kafka.consumerInit('test_topic');
  app.kafka.consumer.run({
    eachBatchAutoResolve: true,
    eachBatch: async ({
      batch,
      resolveOffset,
      heartbeat,
      commitOffsetsIfNecessary,
      uncommittedOffsets,
      isRunning,
      isStale,
    }) => {
      const datas = [];
      for (let message of batch.messages) {
        console.log(message)
        const data = {
          topic: batch.topic,
          partition: batch.partition,
          highWatermark: batch.highWatermark,
          message: {
            offset: message.offset,
            key: message.key.toString(),
            value: message.value.toString(),
            headers: message.headers,
          }
        };
        if (message.extra) {
          data.extra = message.extra.toString();
        }
        resolveOffset(message.offset);
        await heartbeat();
      }
    },
  })
```

## License

[MIT](LICENSE)