Merge pull request #8 from FlowzPlatform/develop

data not uploaded start again job queue

Author: chiragob

uploader.js

@@ -48,6 +48,7 @@ let ESClient = new elasticsearch.Client({
   requestTimeout: 100000
 //  ,log: 'trace'
 })
+let uploadedRecord = 0
 
 let optionsES = {
   tls: 'https://',
@@ -118,6 +119,10 @@ let doJob = async function (objWorkJob, next) {
 
 function updateImportTrackerStatus (trackerId) {
   return new Promise(async (resolve, reject) => {
+    if (uploadedRecord <=0) {
+      reject({"message" :"data not uploaded, record count is zero"})
+    }
+
     rethinkDbConnectionObj = await connectRethinkDB(rethinkDBConnection)
     rethink.db(rethinkDBConnection.db).table(rethinkDBConnection.table)
     .filter({'id': trackerId})
@@ -506,7 +511,7 @@ let delayPromise = (delay) => {
 
 let perPageDataUpload = 100
 let batchPromise = []
-let uploadedRecord = 0
+
 // to make batch for data upload
 async function makeBatch (objWorkJob, listObjects, currentProductsData, makeProductUpdateJsonObj) {
   return new Promise(async (resolve, reject) => {

uploaderworker.js

@@ -44,12 +44,18 @@ let queueOption = {
 }
 
 process.on('unhandledRejection', (reason, p) => {
-  console.log('Unhandled Rejection reason:', reason);
+  console.log('Unhandled Rejection reason:', reason, p);
   // application specific logging, throwing an error, or other logic here
 });
 
 const objQ = new rfqQueue(connctionOption, queueOption)
 
+objQ.on('error', (err) => {
+  console.log('Queue Id: ' + err.queueId)
+  console.error(err)
+})
+
+
 function getJobQueue () {
   objQ.process(async (job, next) => {
     try {