|
|
@ -286,22 +286,22 @@ exports.kafkaProducerAsync = function (brokers, topic, message, options = {}, sa
|
|
|
|
|
|
|
|
|
|
|
|
producer.connect().then(
|
|
|
|
producer.connect().then(
|
|
|
|
() => {
|
|
|
|
() => {
|
|
|
|
try {
|
|
|
|
|
|
|
|
producer.send({
|
|
|
|
producer.send({
|
|
|
|
topic: topic,
|
|
|
|
topic: topic,
|
|
|
|
messages: [{
|
|
|
|
messages: [{
|
|
|
|
value: message,
|
|
|
|
value: message,
|
|
|
|
}],
|
|
|
|
}],
|
|
|
|
});
|
|
|
|
}).then((_) => {
|
|
|
|
connectedToKafka = true;
|
|
|
|
|
|
|
|
clearTimeout(timeoutID);
|
|
|
|
|
|
|
|
resolve("Message sent successfully");
|
|
|
|
resolve("Message sent successfully");
|
|
|
|
} catch (e) {
|
|
|
|
}).catch((e) => {
|
|
|
|
connectedToKafka = true;
|
|
|
|
connectedToKafka = true;
|
|
|
|
producer.disconnect();
|
|
|
|
producer.disconnect();
|
|
|
|
clearTimeout(timeoutID);
|
|
|
|
clearTimeout(timeoutID);
|
|
|
|
reject(new Error("Error sending message: " + e.message));
|
|
|
|
reject(new Error("Error sending message: " + e.message));
|
|
|
|
}
|
|
|
|
}).finally(() => {
|
|
|
|
|
|
|
|
connectedToKafka = true;
|
|
|
|
|
|
|
|
clearTimeout(timeoutID);
|
|
|
|
|
|
|
|
});
|
|
|
|
}
|
|
|
|
}
|
|
|
|
).catch(
|
|
|
|
).catch(
|
|
|
|
(e) => {
|
|
|
|
(e) => {
|
|
|
@ -313,8 +313,10 @@ exports.kafkaProducerAsync = function (brokers, topic, message, options = {}, sa
|
|
|
|
);
|
|
|
|
);
|
|
|
|
|
|
|
|
|
|
|
|
producer.on("producer.network.request_timeout", (_) => {
|
|
|
|
producer.on("producer.network.request_timeout", (_) => {
|
|
|
|
|
|
|
|
if (!connectedToKafka) {
|
|
|
|
clearTimeout(timeoutID);
|
|
|
|
clearTimeout(timeoutID);
|
|
|
|
reject(new Error("producer.network.request_timeout"));
|
|
|
|
reject(new Error("producer.network.request_timeout"));
|
|
|
|
|
|
|
|
}
|
|
|
|
});
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
|
|
producer.on("producer.disconnect", (_) => {
|
|
|
|
producer.on("producer.disconnect", (_) => {
|
|
|
|