Hello, everyone,
I still have a problem with my MySQL data recovery system at MongoDB.
I have a loop that allows me to recover MySQL data by stop ([bus] station) name.
Inside this loop, I have my SQL query with a function that will store in a table the data obtained through a loop, to finally insert them in Mongodb.
The problem is that my curls start asynchronously.
For example :
array =[1,2,3];
array.forEach(item => {
console.log(item);
sql.query(query, function(error, results, fields){
results.forEach(data => {}){
console.log(data);
});
});
});
result :
-1
-2
-3
-data
-data
-data…
The worst part is that I’ve data that multiplies x times!
do you know how I can fix this problem?
I want the loop to be locked until all the following actions have been performed.
I tried to use async and await, wrapasync, setTimeout, for… but without success!
Here is my code:
let Future = require("fibers/future");
let mysql = require('mysql');
let sql = mysql.createConnection({
host : '127.0.0.1',
user : 'root',
password : '',
database : 'bdsi'
});
/**
* Suppression des anciennes données MongoDB (Date < NOW())
*/
Depart.remove({"jourExploitation":{$lte:new Date()}, "heureTheoriqueDepart":{$lte:new Date()}});
/**
* Récupération des arrets depuis MongoDB sans doublon
*/
let rawArrets = Arret.rawCollection();
let arretsDistinct = Meteor.wrapAsync(rawArrets.distinct, rawArrets);
let arrets = arretsDistinct("nomArret");
/**
* @collection Depart
* Utilisé pour effectuer des opérations d'agrégations
*/
manyUpSert = Depart.rawCollection();
sql.connect();
arrets.forEach(arret => {
let dataJsonUpdate = [];
let dataJsonInsert = [];
sql.query('SELECT /*+ MAX_EXECUTION_TIME(60000) */ DISTINCT LIGNE.mnemoCom AS numeroVehicule, ARRET.mnemol AS numeroArret,HORAIRE.htd AS heureTheoriqueDepart, ARRET.nom AS nomArret, DEST.sens AS sens, DEST.nom AS nomDestination, HORAIRE.jour AS jourExploitation, VEHICULE.avance AS avance, VEHICULE.retard AS retard FROM DEST INNER JOIN (HORAIRE INNER JOIN (VEHICULE INNER JOIN (LIGNE INNER JOIN (CHAINAGE INNER JOIN (ARRET INNER JOIN ARRETCHN ON ARRET.ref = ARRETCHN.arret) ON CHAINAGE.ref = ARRETCHN.chainage) ON LIGNE.ref = CHAINAGE.ligne) ON VEHICULE.ligne = LIGNE.ref) ON HORAIRE.arretchn = arretchn.ref) ON (DEST.ligne = LIGNE.ref) AND (DEST.ref = CHAINAGE.dest) WHERE (((ARRET.nom)="' + arret + '" AND (HORAIRE.jour BETWEEN CURDATE() AND CURDATE()+1) AND ((CHAINAGE.type)="C") AND (ARRET.nom)<>DEST.nom)) GROUP BY LIGNE.mnemoCom, ARRET.mnemol, HORAIRE.htd, DEST.sens, DEST.nom, HORAIRE.jour, ARRET.nom ORDER BY HORAIRE.htd ASC, LIGNE.mnemoCom ASC, DEST.nom ASC', function (error, results, fields)
{
if (error)
console.log("Exception : " + error);
else {
let upSert = function () {
results.forEach(result => {
/**
* transforme l'heure en date complète année-mois-jour heure:minute:seconde
* @type Date(}
*/
let getDate = new Date(result.jourExploitation);
let hours = result.heureTheoriqueDepart.replace(/:/g, ',').split(',');
getDate.setHours(hours[0], hours[1], hours[2]);
// Nous verifions que les données n'existent pas dans notre base de données MongoDB
let dataExist = Depart.find({
'numeroArret': parseInt(result.numeroArret),
'jourExploitation': result.jourExploitation,
'heureTheoriqueDepart': getDate,
"numeroVehicule": parseInt(result.numeroVehicule)
}, {limit: 1, fields: {"heureTheoriqueDepart": 1}, transform: null}).fetch();
if (dataExist.length === 0) {
dataJsonInsert.push({
'jourExploitation': result.jourExploitation,
'nomArret': result.nomArret,
'numeroArret': parseInt(result.numeroArret),
'heureTheoriqueDepart': getDate,
'nomDestination': result.nomDestination,
'numeroVehicule': parseInt(result.numeroVehicule),
'sens': result.sens,
'retard': result.retard,
'avance': result.avance,
'debutPerturbation': new Date('1970-01-01 01:00:00.000'),
'finPerturbation': new Date('1970-01-01 01:00:00.000'),
'estParti': false
});
}
});
}.future();
// On exécute notre fonction upSert
try {
Future.task(function () {
upSert().wait();
if (dataJsonInsert.length > 0)
manyUpSert.insertMany(dataJsonInsert, { forceServerObjectId: true });
}).detach();
}
catch (e) {
console.log(e);
}
}
});
});
sql.end();
