Асинхронное программирование в Node.js: Ожидание нескольких событий сразу

5Фев/105

Асинхронное программирование в Node.js: Ожидание нескольких событий сразу

Постановка задачи

Представим себе такую ситуацию. Нам надо сделать поиск в базе Tokyo Tyrant (или любой другой) и получить найденные значения в виде JavaScript-объектов. Проблема состоит в том, что получение объекта из Tokyo Tyrant происходит с коллбеком. Т.е., у нас на руках после поиска оказывается несколько коллбеков - по числу ожидаемых объектов:

var tyrant = require('./tyrant/tyrant');
var sys = require('sys');
 
tyrant.connect();
 
tyrant.addListener('connect', function() {
    tyrant.search(tyrant.is('type', 'blog'), tyrant.sort('time', 'desc')).addCallback(function(value) {
 
        var posts = [];
        for (item in value) {
            var page_id = value[item];
            tyrant.get(page_id).addCallback(function(raw_item) {
                sys.puts('Got record #' + page_id + ': ' + JSON.stringify(raw_item));
            });
        }
    });
});

Изначальное моё решение было простым (и неправильным): я объявил массив posts и наполнял его прямо из promises, ожидая их завершения с помощью wait(). Это работало, но, во-первых, противоречило идее асинхронной работы программы, и во-вторых, сломалось при первой же проверке на прочность с помощью ab: я получал кучу сообщений WARNING: promise.wait() is being called too often и итоговый сегфолт. Посмотрев в группе по этому поводу, нашёл только призыв Райана не пользоваться wait() вообще. Что ж, нужен был другой подход.

Callback-Комбо

Нам нужно каким то образом собрать полученные объекты и вызвать с ними функцию только когда все записи будут получены. Сама Node.js не предоставляет стандартных средств для такой задачи, но Tim Caswell написал маленькую функцию, позволяющую довольно просто собирать несколько событий в один event.

// combo library
    function Combo(callback) {
      this.callback = callback;
      this.items = 0;
      this.results = [];
    }
    Combo.prototype = {
      add: function () {
        var self = this;
        this.items++;
        return function () {
          self.check(self.items - 1, arguments);
        };
      },
      check: function (id, arguments) {
        this.results[id] = arguments;
        this.items--;
        if (this.items == 0) {
          this.callback.apply(this, this.results);
        }
      }
    }; 
 
    // Пример использования
    var both = new Combo(function () {
      // Сперва объявляем callback. Он просто выведет переданные аргументы
      puts(inspect(arguments));
    });
    // Ожидаем первое событие
    setTimeout(both.add(), 100);
    // ...и одновременно ожидаем второе
    setTimeout(both.add(), 50);

Я использовал её для сбора записей, получаемых из Tokyo Tyrant, но для этого мне пришлось её слегка изменить. Тим использовал в методе check функцию callback.apply. Т.к. у нас к моменту запуска коллбека будет массив значений, apply этот массив разделит на отдельные аргументы. То есть, если мы создадим Combo с функцией gather(), и получим пять результатов, callback.apply вызовет gather с пятью параметрами, которые нам придётся перебирать в коде. Поэтому я заменил callback.apply на callback.call:

function Combo(callback) {
  this.callback = callback;
  this.items = 0;
  this.results = [];
}
Combo.prototype = {
  add: function () {
    sys.puts('Adding promise to combo');
    var self = this;
    this.items++;
    return function () {
      sys.puts('Combo part fired with ' + (self.items - 1) + ' id, arguments is ' + JSON.stringify(arguments));
      self.check(self.items - 1, arguments);
    };
  },
  check: function (id, arguments_in) {
    this.results[id] = arguments_in;
    this.items--;
    if (this.items == 0) {
      sys.puts('Combo is approaching the end, resultset count is ' + this.results.length);
      this.callback.call(this, this.results);
    }
  }
};
 
var tyrant = require('./tyrant/tyrant');
var sys = require('sys');
 
tyrant.connect();
 
tyrant.addListener('connect', function() {
    tyrant.search(tyrant.is('type', 'blog'), tyrant.sort('time', 'desc')).addCallback(function(value) {
 
        var gatherer = new Combo(function(posts) {
            sys.puts('Got all records :' + JSON.stringify(posts));
        });
 
        for (item in value) {
            var page_id = value[item];
            tyrant.get(page_id).addCallback(gatherer.add());
        }
    });
});

После этой модификации всё отлично работает, проверка с помощью ab больше не убивает сайт и время выполнения запроса довольно стабильно.

Источник: Механический мир

Метки записи: node.js Оставить комментарий

Комментарии (5) Пинги (0) (подписаться на новые комментарии в этой ветке)

moro
Февраль 5th, 2010 - 18:23

Ну это стандартное решение, увеличивать счетчик при добавлении события, когда событие происходит – уменьшение, если счетчик равен нулю значит все закончилось.
Kuroki Kaze
Февраль 6th, 2010 - 13:52

В принципе да. Разве что возврат функции в коллбек для отлова события красиво сделан
admin
Февраль 11th, 2010 - 00:16

Было бы круто иметь некий сладкий синтаксис.. например добавить некий метод ко всем функциям типо:
Function.prototype.combo=function() { … };

чтобы в итоге иметь нечто вроде:
var gatherer = function(posts) {
sys.puts(‘Got all records :’ + JSON.stringify(posts));
}.combo(2);

после чего нужно дважды выполнить фунцию, чтобы она сработала
На Node такое написать можно!
Kuroki Kaze
Февраль 11th, 2010 - 09:35

Нам всё равно придётся передавать в обработчики событий функции-счётчики (как здесь gatherer.add() действует). Можно их считать при выдаче, незачем заранее указывать сколько их должно быть.

А присобачить к прототипу функции как нибудь можно конечно.
admin
Февраль 11th, 2010 - 20:04

неа.. функции счётчики не нужны.. Ведь мы можем при вызове Function.prototype.combo заменить нашу функцию собственно на счётчик, а саму функцию куда-нибудь подвесить