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
KafkaProducerCliente o devolver.
Exemplo de utilização
import { KafkaFactory } from '@develop-fapp/kafka-client'
async function bootstrap() {
const factory = new KafkaFactory({
clientId: 'test-app',
broker: 'localhost:9092'
})
const producer = factory.getProducer()
try {
// Producer a one message by time.
await producer.send({ topic: 'quickstart', message: 'Test kafka.' })
// Producer messages in batch by time.
await producer.sendBatch({
topic: 'quickstart',
messages: [
'Test kafka.',
'Other message,',
'One more message'
]
})
} catch (error) {
console.log({ error })
}
}
bootstrap()Consumidor
Também contamos com o método abaixo para consumidores:
-
getConsumer - Responsável por criar um
KafkaConsumerCliente o devolver.
Exemplo de utilização
import { KafkaFactory, KafkaConsumerConfig } from '@develop-fapp/kafka-client'
async function bootstrap() {
const factory = new KafkaFactory({
clientId: 'test-app',
broker: 'localhost:9092'
})
const consumerConfig: KafkaConsumerConfig = {
topic: 'quickstart',
fromBeginning: true,
config: {
groupId: 'quickstart-group'
}
}
const consumer = factory.getConsumer(consumerConfig)
try {
// Consumer a one message by time.
await consumer.consume({
eachMessage: async ({ topic, partition, message }) => {
console.log({
partition,
offset: message.offset,
value: message.value.toString(),
})
}
})
// Consumer messages in batch by time.
await consumer.consume({
eachBatch: async ({ batch }) => {
for (const message of batch.messages) {
const prefix = `${batch.topic}[${batch.partition} | ${message.offset}] / ${message.timestamp}`
console.log(`- ${prefix} # ${message.value}`)
}
}
})
} catch (error) {
console.log({ error })
}
}
bootstrap()