Передайте данные, как только они будут готовы в Node.js

Question

У меня есть большой файл JSON, как это:

[
 {
  "id": 2000,
  "city": "New York",
  "lat": "",
  "lon": "",
 },
...
]

и я ищу лат и лон в каждом объекте. Я использую модуль JSONStream для передачи данных:

var writeStream = fs.createWriteStream('data/stream.json');
var readStream = fs.createReadStream('data/dummy/institucion_1_2000.json', {encoding: 'utf8'})

// Search for lat and lon
var search = es.mapSync(function(data){
  if(data.lat == ""){
    // Search for the lat and lon
    geocoder.geocode({address: data.city, country: "US"}) 
      .then(function(res) {
        console.log("Searched for " + res[0].formattedAddress);
        data.lat = res[0].latitude;
        data.lon = res[0].longitude;
      })
      .catch(function(err) {
        console.log("There was an error with element with id = " + data.id);
        console.log("Here is the error: " + err);
        if(err == 'Error: Status is OVER_QUERY_LIMIT. You have exceeded your rate-limit for this API.') {
          process.exit();
        }
      });
    return data;
  }
})

// Pipe
readStream
  .pipe(JSONStream.parse('*'))
  .pipe(search) 
  .pipe(JSONStream.stringify()) // This doesent wait until the search is finish
  .pipe(writeStream)

Геокодирование детали работает.

Моя проблема в том, что JSONStream.stringify читает и затем передает данные перед завершением функции поиска. Поэтому я получаю тот же файл JSON без необходимых мне модификаций. Если я просто попробую это:

if(data.lat == ""){
    lat = 1;
}

вместо геокодирования, которое занимает больше времени, работает. Я предполагаю, что моя проблема заключается во времени, необходимом для изменения потоковых данных. Итак, есть ли способ передать данные после того, как они были изменены?

РЕДАКТИРОВАТЬ У меня был беспорядок между синхронизацией и асинхронностью. Благодаря djones

var search = es.map(function (data, callback) {
  if(data.lat == ""){
    geocoder.geocode({address: data.city, country: "USA"}) // Choose between Comuna_Empresa and Comuna_Institucion
    .then(function(res) {
      console.log("Searched for " + res[0].formattedAddress);
      data.lat = res[0].latitude;
      data.lon = res[0].longitude;
      callback(null,data);
    })
    .catch(function(err) {
      console.log("There was an error with element at index = " + index);
      console.log("Here is the error: " + err);
    });
  }
})