Kafka-node-reply NPM

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 version 4.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": "example@email.com"
  },
  "headers": {
    "ContentType": "json/application"
  }
}

kafkaReqRes.requestSync(message).then((data)=> {
  console.log("Response success data", data)
}).error((err)=>{
  console.log("Response error message", err)
})