joe/logAggregator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Only send next chunk after the LogAggregator feedback arrived.

Browse Source
This commit is contained in:
Joe
2019-01-14 07:32:58 -06:00
parent fcd56db730
commit ea1884f819
2 changed files with 10 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
casRpsLogExtractor/casRpsLogExtractor.config.json.example
View File

@@ -2,7 +2,6 @@
"sourceTag": "CCERpsLog", "sourceTag": "CCERpsLog",
"serverURI": "http://localhost:2222", "serverURI": "http://localhost:2222",
"pollIntervalIdle": 10000, "pollIntervalIdle": 10000,
"pollIntervalBusy": 1000,
"UP_sqlConfig": { "UP_sqlConfig": {
"user": "casdbuser", "user": "casdbuser",
"password": "4KcLao2016!", "password": "4KcLao2016!",

21
casRpsLogExtractor/casRpsLogExtractor.js
View File

@@ -45,8 +45,6 @@ let maxPKey = "";
let maxPKeyUnsent = ""; let maxPKeyUnsent = "";
let allRows = []; let allRows = [];
let currPollInterval = config.pollIntervalBusy;
let lastPollInterval;
let theInterval; let theInterval;
@@ -72,7 +70,7 @@ sql.connect(config.sqlConfig, err => {
sqlRequest.query("select top " + maxResultsPerMessage + " * from rpslog where pkey > '" + maxPKey + "' order by pkey"); sqlRequest.query("select top " + maxResultsPerMessage + " * from rpslog where pkey > '" + maxPKey + "' order by pkey");
} }
theInterval = setInterval(runNextPoll, currPollInterval); runNextPoll();
sqlRequest.on('row', row => { sqlRequest.on('row', row => {
outData = { sourceTag: config.sourceTag, processingType: processingType, data: row }; outData = { sourceTag: config.sourceTag, processingType: processingType, data: row };
@@ -89,6 +87,7 @@ sql.connect(config.sqlConfig, err => {
}); });
sqlRequest.on('done', result => { sqlRequest.on('done', result => {
clearInterval(theInterval);
if (allRows.length>0){ if (allRows.length>0){
request({ method: 'POST', uri: config.serverURI, body: JSON.stringify(allRows), json: true }).then(_ => { request({ method: 'POST', uri: config.serverURI, body: JSON.stringify(allRows), json: true }).then(_ => {
// count up the maxPKey only when the request was successful... // count up the maxPKey only when the request was successful...
@@ -96,21 +95,17 @@ sql.connect(config.sqlConfig, err => {
fs.writeFileSync("casRpsLogExtractor.maxPKey.dat", maxPKey); fs.writeFileSync("casRpsLogExtractor.maxPKey.dat", maxPKey);
}).catch(err=>{ }).catch(err=>{
logger.log("error","Sending Request to Service: " + err) logger.log("error","Sending Request to Service: " + err)
}).finally(_=>{
runNextPoll();
}); });
process.stdout.write(">" + allRows.length + ">"); process.stdout.write(">" + allRows.length + ">");
if (allRows.length === maxResultsPerMessage) {
currPollInterval = config.pollIntervalBusy;
} else { } else {
currPollInterval = config.pollIntervalIdle;
}
if (currPollInterval !== lastPollInterval) { if (currPollInterval !== lastPollInterval) {
clearInterval(theInterval); theInterval = setInterval(runNextPoll, config.pollIntervalIdle);
theInterval = setInterval(runNextPoll, currPollInterval);
} }
allRows=[];
}
process.stdout.write("."); process.stdout.write(".");
}
allRows=[];
}); });
}) })