Как в NodeJS сделать EventEmitter для нескольких одновременных запросов?

Question

[Daniel Newman]

Запрашиваю два API, возвращающих данные источника и хранилища данных. Заполняю ими БД моего приложения, сравнивая данные источника и хранимые, для последующего обновления:

var dataStoredLoad = function( dataListOffset ) {
    eventEmitter.emit('dataStoredFetched');
}
var dataSourceLoad = function( dataListOffset ) {
    eventEmitter.emit('dataSourceFetched');
}

var middlewareDataToUpdate = function ( dataStoredLoad , dataSourceLoad ) {
    dataStoredLoad.forEach( function (dataStored, index) {
        dataSourceLoad.forEach( function (dataStored, index) {
            if ( dataStored.id == dataSource ) {
                 // key.value comparsion procedure and .pop() / .push()
            }
        }
    }
}

Оба API работаю async, а сравнение нужно запускать лишь когда оба массива данных загружены в БД.
Неизвестно, какой API вернет значения первым.

Что тут можно сделать и как это красиво пишут другие?

UPD:
Внутри функций - запросы к API, выполненные в неблокирующем async потоке, а забор данных выполняется в серии обращений через API, т.е. решение с async.parallel, предложенное также ниже, всегда валится в "колбэк(?)" function(err, results) где results is equal to: { one: ' ', two: ' ' }

Видимо, я как-то должен вопрос переформулировать, но не понимаю, как именно.

UPD2:

var answerAPI = crunchbase.request(arg1) ;

не null, с момента запуска, а сразу какая-то '', в ожидании ответа сервера API. что запрокидывает нас дальше по этому водопаду асинхронных функций к функции результата, функция отрабатывается с пустыми значениями, а потом (в моем случае) в консоль валятся результаты запросов к API.

async.series({
  one: function(callback) {
      var answerAPIone = crunchbase.request(arg1);
      callback(null, answerAPIone );
  },
  two: function(callback) {
      var answerAPItwo = airtable.request(arg1);
      callback(null, answerAPItwo);
  }
},
function(err, results) {
  console.log( 'ИТОГО: <', results.one, '><', results.two, '>' );
  // ИТОГО: <><>
});

Все уперлось в правильное написание функции с колбеком и использование async по его прямому назначению. Остальные варианты - замечательно расширили горизонты. Всем спасибо.

Answer 1

[Тимур Шемсединов]

https://github.com/caolan/async

async.series({
  one: function(callback) {
    setTimeout(function() {
      callback(null, 1);
    }, 200);
  },
  two: function(callback) {
    setTimeout(function() {
      callback(null, 2);
    }, 100);
  }
},
function(err, results) {
  // results is now equal to: { one: 1, two: 2 }
});

Comments

[fr33zy]

https://github.com/caolan/async#seriestasks-callback
> each one running once the previous function has completed.

Полагаю, автор хотел вот это: https://github.com/caolan/async#paralleltasks-callback

[Daniel Newman]

Есть "но": внутри функций - запросы к API, выполненные в том же неблокирующем формате async, а забор данных выполняется в серии обращений через API, т.е. у меня всегда все сразу валится в function(err, results) где results is equal to: { one: '', two: '' }

Я, видимо, как-то должен вопрос переформулировать, но не понимаю, как именно.

[Тимур Шемсединов]

fr33zy, согласен, или async.each но уж точно, что eventEmitter тут ни при чем. Daniel Newman, опишите задачу в общем, без своей реализации, возможно, ее нужно решать вообще иначе.

[Daniel Newman]

Нужно сгрести данных через два API:
https://github.com/coding-house/crunchbase2 - источник
https://github.com/Airtable/airtable.js - хранилище

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

[Тимур Шемсединов]

А, это нужно сделать вложенный async.parallel. Наружный запускает две функции, каждая обращается к одному API, но обращений несколько, когда заказчиваются вложенные распараллеленные обращения, то две ветки сливаются во внешней parallel.

[Daniel Newman]

Тимур Шемсединов: upd

[fr33zy]

Для сложной логики лучше auto: https://github.com/caolan/async#auto

[Тимур Шемсединов]

UPD прочитал, но понятнее не стало, мы запускаем много функций, которые параллельно что-то добывают, конечно они не сами добывают, а вызывают разные асинхронные API, передавая по эстафете им колбеки, и когда все вызовы вернут значения, то вызывается наконец и завершающая функция, что нужно если не это?

[Daniel Newman]

Тимур Шемсединов: Мне как-то неудобно вывалить всю проблему сюда (комментарии для кода никак не годятся). Если вас не затруднит - я выслал почтой доступ к репозиторию и среде ide.c9.io

[Тимур Шемсединов]

Я не работаю с Ц9, мне там неудобно. Какой файл смотреть - не ясно, в index.js там не во всех случаях вызываются колбеки из функций внутри async.parallel, проследите, чтобы в лббом случае был вызван колбек.

[Тимур Шемсединов]

В общем, нужно разобраться с принципом колбеков сначала и потом с принципом асинхронности. Вкратце: колбек - это функция, которая передается параметром в другую функцию и будет вызвана 1 раз из нее, когда будет готов результат. Есть негласное соглашение, что к колбека первый параметр это ошибка или null (если все удачно), а второй параметр - это результат (случается, что результатов несколько, тогда их или группируют в объект или просто у колбека становится больше двух параметров, четких правил тут нет). Колбек нужно вызвать 1 и только 1 раз. Вызывать его больше - это уже не колбек, а некий аналог события. Теперь про асинхронность, нужно понимать, что это просто способ возврата результата не через значение функции, а через колбек, что приводит к созданию цепочек вызовов и даже к ветвлению вызовов. Для упрощения и управления последовательностью операций, а глевное - для создания удобных синтаксических структур в коде - есть много путей, один и них - использование async.js. Его доку можно и нужно прочитать полностью, она вполне вменяемама и лежит на одной странице прямо с примерами https://github.com/caolan/async Еще полезно пройти соответствующий раздел nodeschool.io

Answer 2

[Юрий Пузыня]

Я советую вам посмотреть в сторону обещаний.
Начните с канонической библиотека promise
Когда разберетесь с ней, вам возможно потребуется комбайн посложнее, смотрите в сторону Q

// npm install promise

var fs = require('fs');
var Promise = require('promise');

// В качестве источника данных в данном случае файловая система,
// в вашем случае это видимо будет net.Connection
function dataStoredLoad( dataName ) {
    return new Promise(function(resolve, reject) {
        fs.readFile(dataName, 'utf8', function(err, data) {
            if (err) return reject(err);

            resolve(data);
        });
    })
}

// Это другой способ записать то, что написано в dataStoredLoad
function dataSourceLoad( dataName ) {
    return Promise.denodeify(fs.readFile)(dataName, 'utf8');
}

Promise.all(dataStoredLoad, dataSourceLoad).then(function(res) {
    var stored = res[0],
          sourse = res[1];

    stored.forEach( function (dataStored, index) {
        sourse.forEach( function (dataStored, index) {
            if ( dataStored.id == dataSource ) {
                 // key.value comparsion procedure and .pop() / .push()
            }
        }
    }
});

Comments

[Daniel Newman]

Я уже почти понял, в чем у меня проблема. В вашем варианте с промисами, функция fs.read возвращает, а в моем - не существует EOF. И я свой вариант EOF для запроса к API пока написать не осилил головой.

Answer 3

[dtestyk]

однажды для подобной проблемы коллега предложил считать события.

также, можно использовать Promise.all(db_promise, src_promise), при некотором старании и Promise.race даже получится обработать таймаут.

Answer 4

[Алексей Комаров]

Во-первых, в случае async вам больше подходит parallel, а не series. async.parallel не ждет вызова колбека из первой функции и сразу запустит вторую после возврата первой.
Во-вторых, API, которые вы запрашиваете, должны быть (и наверняка являются) асинхронными, а потому сразу возвращают вам ничего. Вы должны передавать этим API колбек, внутри которого вызывать колбек асинка.

Как-то так:

async.parallel({
  one: function(callback) {
      crunchbase.request(arg1, function(error, answerAPIone){
        callback(null, answerAPIone);
      });
  },
  two: function(callback) {
      airtable.request(arg1, function(error, answerAPItwo){
        callback(null, answerAPItwo);
      });
  }
},
function(err, results) {
  console.log( 'ИТОГО: <', results.one, '><', results.two, '>' );
});

Answer 5

[Alex]

https://github.com/dfilatov/vow

Comments

[Daniel Newman]

Ничего про промисы не знаю, чуть выше откомментировал про async свою проблему. Как-то проблема перекладывается на решение с промисами?