pseudo_process of mysql to es syncer

pseudo_process.js

function format_result(results,datas,callback){
    json = syncCall(foo);
    datas.push(json);
    callback();
    return;
}

function async_push_data_to_x(datas){
    ...
}

function fetch_data(connection,idx_b,last_data_set,fetch_callback){

    connection.query(sql, [], function(err, results) {
        if(idx_b === nb_batch-1 ) {  
            connection.release();  // always put connection back in pool after last query
            //es_singleton.getPool.close();
            fetch_callback(err, 'done'); //end loop
        } 
        else // if (idx_b < 3) 
        { //start other query in meantime gogogo !
            fetch_data(connection,++idx_b,last_data_set,fetch_callback); // looping to next batch
        }
        
        if(err) { 
            logger.error(err); 
            return; 
        }
        if(results.length <= 0){ 
            return; 
        }
        
        var datas = [];
        async.each(results,
            function(res,callback){ //treat all result in a async manner
                format_result(results,datas,callback); //writte into data
            },
             function(err){  // 3rd param is the function to call when everything's done
                async_push_data_to_x(datas);
                //do not fetch_callback !
                return; //end async.each final callback
            }
        );
        return;
    }
    return;
}

function main(){
    // ... some stuff init, get config and such
    var sql = "select count(1) from record where id > "+last_data;
    connection.query(sql, [], function(err, results) {
        if(err) { logger.error(err); reschedule(); return; }
         
        async.waterfall([
            function(callback) { //calc_nbToAdd
                const count_new=results[0]['count(1)'];
                calc_nbToAdd(count_set,callback);
                return;
            },
            function(count_set, callback) { //calc_nbbatch
                const batch_set = calc_nbbatch(count_toAdd,batch_opt,callback);
                return;
            },
            function(batch_set, callback) { //fetch_data
                // last_data,batch_set,'test_npm','test_npm'
                fetch_data(connection,idx_b,last_data_set,callback);
                return;
            }
        ], function (err, result) { //final
            var TimeStop = moment();
            const minDiff = TimeStop.diff(TimeStart, 'minutes');
            logger.info('All waterfall is done, took minDiff='+minDiff);
            reschedule(); //call us again
            return;
        });
    return;
    });
}