async / await с потоком событий mapSync не работает

Question

Команды await, с которыми я комментировал // ****** Это ожидание не работает * / не похоже на работу. Не уверен, связано ли это с тем, что они находятся в потоке событий, или с проблемой с обещанием в импортированном модуле.

Когда я вызываю функцию run из сопоставленного массива для вставки данных из нескольких источников, функция run возвращается немедленно, а не в ожидании, пока knex завершит вставку данных.

 app.get("/api/pull/all_data", (req, res)=>{
    const dataSources = [
      {resource: 'DI_ORDER_TYPE', tableName: 'ln_order_type'},
      {resource: 'DI_DATES', tableName: 'ln_dates'},
      {resource: 'WHINR140_INVENTORY', tableName: 'ln_inventory'}
    ]

    dataSources.map(async (ds)=>{
      console.log(`Importing ${ds.resource} into table ${ds.tableName}`)
      await get_all_data(ds.tableName, ds.resource)
    })

    console.log("Import complete")
  })

Вот моя функция запуска, которая вызывается из кода выше.

const request = require('request')
const JSONStream = require('JSONStream')
const es = require('event-stream')
const knex_ln = require('../knex_ln')
const insertData = require('../insert_data')
const create_table = require('../create_table_from_json.js')
const clean_fieldnames = require('../clean_fieldnames')

function run(tableName, resourceName) {
    return new Promise(async (resolve, reject)=>{
        let tableData = []
        let recordCount = 0
        let maxRecords = 10000
        let totalRecords = 0
        // let tableName = 'LN_di_order_type'
        // let resourceName = 'DI_ORDER_TYPE'
        let rowData = {}
        //Delete Existing Data and wait for it to complete
        await knex_ln.schema.hasTable(tableName).then(async (exists)=>{
            if(exists){
                try {
                    await knex_ln(tableName).delete().then(()=>{})
                } catch (error) {

                }
            }
        })
        //Get LN replica data and pipe data into JSONStream
        request(`${process.env.API_BASE_URL}/${process.env.SECURITY_NAME}/${resourceName}`,
        {
            auth: {
                'user': process.env.API_USER,
                'pass': process.env.API_PASS
            }
        }
        )
        .pipe(JSONStream.parse([true, {recurse: true}, `${process.env.SECURITY_NAME}.row`, true]))
        .pipe(es.mapSync(async (row)=>{
            rowData = row
            let cleanData = await clean_fieldnames(row)
            tableData.push(cleanData)
            recordCount += 1
            totalRecords += 1
            if(recordCount >= maxRecords){
                try {
                    //******This await does not work */
                    await create_table(tableName, row)
                } catch (error) {
                    console.log("Unable to create table", error)
                }

                //Insert records
                try {
                    //******This await does not work */
                    await insertData(tableName, tableData)
                    console.log(`inserting ${recordCount} records into table ${tableName}`)
                } catch (error) {
                    console.log("Unable to insert data: ", error)
                }

                //Reset tracker variables
                recordCount = 0
                tableData = []
            }
        }))
        .on('end', async ()=>{
            await create_table(tableName, rowData)
            await insertData(tableName, tableData)
            console.log(`Inserted ${totalRecords} into table ${tableName}`)
            resolve('OK')
        })
        .on('error',(err)=>{
            reject(err)
        })
    })
}
module.exports = run

Вот мой файл модуля, который возвращает обещание

//insert_data.js
const knex_ln = require('./knex_ln')

module.exports = async (tableName, tableData) => 
    new Promise(async (resolve, reject) => {
      try {
        await knex_ln(tableName).insert(tableData)
        console.log("Inserting Data: ", tableData.length)
        resolve()
      } catch (error) {
        console.log("Error inserting data: ", err)
        reject(err)
      }
    })

Вот пример вывода

Importing DI_ORDER_TYPE into table ln_order_type
Importing DI_DATES into table ln_dates
Importing WHINR140_INVENTORY into table ln_inventory
Importing WHWMD210_WAREHOUSE_ITEM_DATA into table ln_warehouse_item_data
Importing TDIPU010_ITEM_BUY_FROM_BP_INFORMATION into table ln_item_buy_from_bp_information
Importing TDIPU001_ITEM_PURCHASE_DATA into table ln_item_purchase_data
Importing TDPCG031_PRICE_BOOKS into table ln_price_books
Importing TDPUR300_PURCHASE_CONTRACTS into table ln_purchase_contracts
Importing TDPUR301_PURCHASE_CONTRACT_LINES into table ln_purchase_contract_lines
Inserted 72 records into table ln_order_type
Inserted 217 records into table ln_purchase_contracts
inserting 10000 records into table ln_inventory
Inserted 4694 records into table ln_purchase_contract_lines
inserting 10000 records into table ln_item_buy_from_bp_information
inserting 10000 records into table ln_dates
inserting 10000 records into table ln_inventory
inserting 10000 records into table ln_price_books
inserting 10000 records into table ln_item_purchase_data
inserting 10000 records into table ln_inventory
inserting 10000 records into table ln_price_books
inserting 10000 records into table ln_dates
inserting 10000 records into table ln_inventory
inserting 10000 records into table ln_price_books
inserting 10000 records into table ln_item_purchase_data

Answer 1

Решением для меня было использование bluebird Promise.each

. Он будет обрабатывать каждый из элементов массива dataSources и ждать возврата обещания, прежде чем обрабатывать следующий элемент в списке.

Promise.each(dataSources, function(ds){
     ....
    }).then(()=>{
     ....
    })

http://bluebirdjs.com/docs/api/promise.each.html

Answer 2

Вы ожидаете только внутренних функций в map, а не функции верхнего уровня.

Вам необходимо добавить ожидание в функцию верхнего уровня:

await Promise.all(dataSources.map(async (ds)=> {
  console.log(`Importing ${ds.resource} into table ${ds.tableName}`)
  await get_all_data(ds.tableName, ds.resource)
}));

В противном случае вы ждете только во внутренней функции, а не в самом обработчике маршрута.