dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixes to Kafka JS headers issue

Browse Source
This commit is contained in:
Andrii Shvaika 2020-04-30 15:17:39 +03:00
parent beb5d1d0ba
commit 30a2d19d2d
2 changed files with 4 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
msa/js-executor/api/jsInvokeMessageProcessor.js
View File

@ -59,7 +59,7 @@ JsInvokeMessageProcessor.prototype.onJsInvokeMessage = function(message) {
} else if (request.releaseRequest) { } else if (request.releaseRequest) {
this.processReleaseRequest(requestId, responseTopic, headers, request.releaseRequest); this.processReleaseRequest(requestId, responseTopic, headers, request.releaseRequest);
} else { } else {
logger.error('[%s] Unknown request recevied!', requestId); logger.error('[%s] Unknown request received!', requestId);
} }
} catch (err) { } catch (err) {

13
msa/js-executor/queue/kafkaTemplate.js
View File

@ -41,8 +41,6 @@ function KafkaProducer() {
} }
} }
let headersData = headers.data;
headersData = Object.fromEntries(Object.entries(headersData).map(([key, value]) => [key, Buffer.from(value)]));
return producer.send( return producer.send(
{ {
topic: responseTopic, topic: responseTopic,
@ -50,7 +48,7 @@ function KafkaProducer() {
{ {
key: scriptId, key: scriptId,
value: rawResponse, value: rawResponse,
headers: headersData headers: headers.data
} }
] ]
}); });
@ -96,15 +94,10 @@ function KafkaProducer() {
eachMessage: async ({topic, partition, message}) => { eachMessage: async ({topic, partition, message}) => {
let headers = message.headers; let headers = message.headers;
let key = message.key; let key = message.key;
let data = message.value;
let msg = {}; let msg = {};
headers = Object.fromEntries(
Object.entries(headers).map(([key, value]) => [key, [...value]]));
msg.key = key.toString('utf8'); msg.key = key.toString('utf8');
msg.data = [...data]; msg.data = message.value;
msg.headers = {data: headers} msg.headers = {data: headers};
messageProcessor.onJsInvokeMessage(msg); messageProcessor.onJsInvokeMessage(msg);
}, },
}); });