I am trying to run a number of promises in parallels in a batch but it starts the next batch before I have finished processing the first batch the code works when I use simple timeout promises. Can any one help give me a quick bit of help ?
await asyncFunctionsInBatches.reduce(async (previousBatch, currentBatch, index) => {
await previousBatch;
console.time(`batch ${index}`);
console.log(`Processing batch ${index}...`);
const currentBatchPromises = currentBatch.map(async (row) =>
await sqlFetch(queryDate, queryDate1, username, password, row)
);
await Promise.all(currentBatchPromises);
console.timeEnd(`batch ${index}`);
}, Promise.resolve());
}
This is the promise I am using
const sqlFetch = async (
queryDate,
queryDate1,
username,
password,
rowNumber
) => {
( await createUnixSocketPool(username, password)).query(
fetchQuery.sql,
[queryDate, queryDate1, rowNumber],
async function (err, result) {
if (err) {
console.log(err);
}
console.log(result)
return publishMessage(result).catch(console.error);
}
);
};
In sqlFetch, you're awaiting the result of createUnixSocketPool but not the result of query:
const sqlFetch = async (
queryDate,
queryDate1,
username,
password,
rowNumber
) => {
( await createUnixSocketPool(username, password)).query(
// −−−−− −−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−−− −−−−−−
// ↑ ↑ ↑
// \−−− only applies to / not to −−−−−−−−−−−−−−−/
fetchQuery.sql,
[queryDate, queryDate1, rowNumber],
async function (err, result) {
if (err) {
console.log(err);
}
console.log(result)
return publishMessage(result).catch(console.error);
}
);
};
As a result, the promise from sqlFetch is immediately fulfilled with undefined, while the query is still running.
Without knowing what query expects from its callback (you're giving it a promise; does it really expect one?) or whether it returns a promise it's hard to help, other than to say that if it returns a promise, await that, and if it doesn't, handle that (e.g., as described in the answers to this question).
Here's an update based on information gleaned from the answer you posted:
const sqlFetch = (
queryDate,
queryDate1,
username,
password,
rowNumber
) => new Promise((resolve, reject) => {
createUnixSocketPool(username, password)
.then(pool => {
pool.query(
fetchQuery.sql,
[queryDate, queryDate1, rowNumber],
(err, result) => {
if (err) {
reject(err);
} else {
resolve(publishMessage(result));
}
}
);
})
.catch(reject);
});
Or better yet, write a promise-enabled wrapper for query as described in the link above:
function asyncQuery(pool, ...args) {
return new Promise((resolve, reject) => {
pool.query(...args, (err, result) => {
if (err) {
reject(err);
} else {
resolve(result);
}
});
});
}
and then it's really nice and simple:
const sqlFetch = async (
queryDate,
queryDate1,
username,
password,
rowNumber
) => {
const pool = await createUnixSocketPool(username, password)
const result = await asyncQuery(pool, fetchQuery.sql, [queryDate, queryDate1, rowNumber]);
return publishMessage(result)
});
I added a promise
const sqlFetch = (
queryDate,
queryDate1,
username,
password,
rowNumber
) => new Promise(async (resolve, reject) => {
(await createUnixSocketPool(username, password)
.catch(reject)).query(
fetchQuery.sql,
[queryDate, queryDate1, rowNumber],
function (err, result) {
if (err) reject(err);
else publishMessage(result).catch(reject).then(resolve);
}
);
});