kafka client

Biblioteca para padronizar e facilitar a utilização do Apache Kafka como um broker de mensagens.

Estrutura

Como utilizar

Configuração

A classe KafkaFactory foi constuída sob o design pattern Factory que possui o comportamento de criação e devolução de objetos para determinado fim.

A classe KafkaFactory conta com um objeto em seu construtor e mais dois métodos após a configuração de sua instância:

Em seu construtor, fica a seu critério aplicar as configurações que achar pertinente com referência nas configurações do KafkaJs, porém esta abstração, possui a particularidade de que só é possível configurar um broker por instância de KafkaFactory. Sendo assim, o uníco campo que sofre alteração no objeto de configuração é o brokers, ficando da seguinte maneira:

antes

• brokers: ['kafka1:9092', 'kafka2:9092']

agora

• broker: 'kafka:9092'

Produtor

Após a configuração inicial da instância de factory, para o produtor, contamos com o método:

• getProducer - Responsável por criar um KafkaProducerClient e o devolver.

Exemplo de utilização

1import { KafkaFactory } from '@develop-fapp/kafka-client'
2
3async function bootstrap() {
4  const factory = new KafkaFactory({
5    clientId: 'test-app',
6    broker: 'localhost:9092'
7  })
8
9  const producer = factory.getProducer()
10
11  try {
12    // Producer a one message by time.
13    await producer.send({ topic: 'quickstart', message: 'Test kafka.' })
14
15    // Producer messages in batch by time.
16    await producer.sendBatch({
17      topic: 'quickstart',
18      messages: [
19        'Test kafka.',
20        'Other message,',
21        'One more message'
22      ]
23    })
24  } catch (error) {
25    console.log({ error })
26  }
27}
28
29bootstrap()

Consumidor

Também contamos com o método abaixo para consumidores:

• getConsumer - Responsável por criar um KafkaConsumerClient e o devolver.

Exemplo de utilização

1import { KafkaFactory, KafkaConsumerConfig } from '@develop-fapp/kafka-client'
2
3async function bootstrap() {
4  const factory = new KafkaFactory({
5    clientId: 'test-app',
6    broker: 'localhost:9092'
7  })
8
9  const consumerConfig: KafkaConsumerConfig = {
10    topic: 'quickstart',
11    fromBeginning: true,
12    config: {
13      groupId: 'quickstart-group'
14    }
15  }
16
17  const consumer = factory.getConsumer(consumerConfig)
18
19  try {
20    // Consumer a one message by time.
21    await consumer.consume({
22      eachMessage: async ({ topic, partition, message }) => {
23        console.log({
24          partition,
25          offset: message.offset,
26          value: message.value.toString(),
27        })
28      }
29    })
30
31    // Consumer messages in batch by time.
32    await consumer.consume({
33      eachBatch: async ({ batch }) => {
34        for (const message of batch.messages) {
35          const prefix = `${batch.topic}[${batch.partition} | ${message.offset}] / ${message.timestamp}`
36          console.log(`- ${prefix} # ${message.value}`)
37        }
38      }
39    })
40  } catch (error) {
41    console.log({ error })
42  }
43}
44
45bootstrap()