kafka-node-reply
v1.0.4
Published
Kafka node reply is a function support service can send message as request and receive response from other consumer to complete request. Write base package kafka-node https://www.npmjs.com/package/kafka-node
Downloads
99
Maintainers
Readme
kafka-node-reply
A kafka-node does not support synchronouse send message and wait for response message from another topic. So, this package may can help you do that and make it simpler. You can use it instead HTTP protocol
Let's getting started! Just install this package and use it as a library
Requirement
- Kafka node
kafka-node
version4.0.0
Kafka-node is a Node.js client for Apache Kafka 0.9 and later.
Installation
- Install package
kafka-node-reply
is super easy to install.
npm install kafka-node-reply
API
Concept
Each instance of kafkaNodeReply will have two part. The first is producer will send a message request to topic requestTopic
with a formatted message was defined (formatMessage
). Other consumers consume this topic will receive and process base on message request. After the process is done will send response into replyTopic
that kafkaNodeReply current consume to receive message response and complete this request
KafkaNodeReply
Create new instance of kafkaNodeReply.
client
The kafka-node client. Can create new client or re-use from kafka-node
admin
The kafka-node admin. Can create new admin or re-use from kafka-node client
requestTopicOptions
The options for producer
{
"topic": "TopicRequest", // Topic request name that producer will send request to
"requestTimeout": 30000, // Timeout of request.
"options": { // This options is kafka-node HighLevelProducer options
"partitionerType": 2,
"requireAcks": 0
}
}
responseTopicOptions
The options for response topic that kafkaNodeReply will consumer to receive response message
{
"topic": "TopicReply", // Topic reply that consumer will receive response message
"options": { // This options is kafka-node ConsumerGroup options
"groupId": "group-A-reply"
}
}
options
The options you are self define generic. It can merge from requestTopicOptions.options, responseTopicOptions.options, kafka-node client options...
requestSync
Sent a request in synchronuos
message
The message request to communicate. All of kafkaNodeRpely instances must use this format message. You can imagine this message similar http request, contain info such as: body, headers, params...
{
"action": "String", // The action name which consumer detect and process
"body": {}, // The body of request
"params": {}, // The parameters of request
"callback": { // Callback info that consumer will send response message. Default kafkaNodeReply are set for request. You can unset this value to use default. Recommend using default
"kafka": { // Callback via kafka. You can implement other callback like http, nitification...
"topic": "String", // Kafka topic reply name.
"partition": "Number", // Partition use for consumer in group
"key": "String" // Key determine request
}
}
}
How to use
Example
This package using base on kafka-node
. So you can re-use theirs client to connect this package. Such as: kafka-node.KafkaClient
, kafka-node.Admin
const kafka = require("kafka-node");
const kafkaReply = require("kafka-node-reply");
let client = new kafka.KafkaClient(options);
let admin = new kafka.Admin(client);
let requestTopicTops = {
"topic": "TopicRequest",
"requestTimeout": 30000,
"options": {
"partitionerType": 2,
"requireAcks": 0
}
}
let responseTopicOptions = {
"topic": "TopicReply",
"options": {
"groupId": "group-A-reply"
}
}
// Create new kafka request response instance
let kafkaReqRes = new kafkaReply.KafkaNodeReply(client, admin, requestTopicTops, responseTopicOptions, options)
let message = {
"ation": "getUser",
"body": {
"userId": 1,
"email": "[email protected]"
},
"headers": {
"ContentType": "json/application"
}
}
kafkaReqRes.requestSync(message).then((data)=> {
console.log("Response success data", data)
}).error((err)=>{
console.log("Response error message", err)
})
KafkaNodeReply
TODO
- [ ] Solution for kafka consumer group rebalancing make consumer cannot receive message
- [ ] Support request-reply stream
- [ ] Control error crash consumer
Change log
1.0.1
- Catch error when client disconnect
1.0.0
Release 1.0.0
Author
Quyen Hung - [email protected]