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

error handling in dbExtractor

Browse Source
This commit is contained in:
Joe
2019-01-31 08:32:12 -06:00
parent b0d5fca2a4
commit bce3458160
1 changed files with 46 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
DBExtractor/DBExtractor.js
View File

@@ -58,51 +58,55 @@ try {
}
}
sql.connect(config.sqlConfig, err => {
if (err) {
logger.log("error", "Connecting to DB: " + err);
}
const sqlRequest = new sql.Request();
sqlRequest.stream = true;
try {
sql.connect(config.sqlConfig, err => {
if (err) {
logger.log("error", "Connecting to DB: " + err);
}
const sqlRequest = new sql.Request();
sqlRequest.stream = true;
function runNextPoll(){
sqlRequest.query("select top " + maxResultsPerMessage + " * from rpslog where pkey > '" + maxPKey + "' order by pkey");
}
runNextPoll();
function runNextPoll(){
sqlRequest.query("select top " + maxResultsPerMessage + " * from rpslog where pkey > '" + maxPKey + "' order by pkey");
}
runNextPoll();
sqlRequest.on('row', row => {
outData = { sourceTag: config.sourceTag, processingType: config.processingType, data: row };
process.stdout.write("#");
request.json = true;
sqlRequest.on('row', row => {
outData = { sourceTag: config.sourceTag, processingType: config.processingType, data: row };
process.stdout.write("#");
request.json = true;
logger.log('debug', "adding row to array:" + JSON.stringify(outData));
allRows.push(outData);
logger.log('debug', "adding row to array:" + JSON.stringify(outData));
allRows.push(outData);
if (row.PKey > maxPKeyUnsent) {
maxPKeyUnsent = row.PKey;
logger.log("debug", "MaxPKeyUnsent Written: " + maxPKeyUnsent);
}
});
if (row.PKey > maxPKeyUnsent) {
maxPKeyUnsent = row.PKey;
logger.log("debug", "MaxPKeyUnsent Written: " + maxPKeyUnsent);
}
});
sqlRequest.on('done', result => {
clearInterval(theInterval);
if (allRows.length>0){
request({ method: 'POST', uri: config.serverURI, body: JSON.stringify(allRows), json: true }).then(_ => {
// count up the maxPKey only when the request was successful...
maxPKey=maxPKeyUnsent;
fs.writeFileSync("DBExtractor.maxPKey.dat", maxPKey);
}).catch(err=>{
logger.log("error","Sending Request to Service: " + err)
}).finally(_=>{
runNextPoll();
});
process.stdout.write(">" + allRows.length + ">");
} else {
theInterval = setInterval(runNextPoll, config.pollIntervalIdle);
process.stdout.write(".");
}
allRows=[];
sqlRequest.on('done', result => {
clearInterval(theInterval);
if (allRows.length>0){
request({ method: 'POST', uri: config.serverURI, body: JSON.stringify(allRows), json: true }).then(_ => {
// count up the maxPKey only when the request was successful...
maxPKey=maxPKeyUnsent;
fs.writeFileSync("DBExtractor.maxPKey.dat", maxPKey);
}).catch(err=>{
logger.log("error","Sending Request to Service: " + err)
}).finally(_=>{
runNextPoll();
});
process.stdout.write(">" + allRows.length + ">");
} else {
theInterval = setInterval(runNextPoll, config.pollIntervalIdle);
process.stdout.write(".");
}
allRows=[];
});
})
});
})
} catch (ex) {
logger.log("error", `Fatal error occured: ${ex}`);
}