Есть ли способ работать с потоком узла как итеративный?

Question

Я пишу инструмент, который обрабатывает кучу текста, переданного в stdin, каждая строка является «записью».Я хотел бы сделать свой код более функциональным, поэтому я хотел бы рассматривать набор строк как «последовательность» или «итеративный» и выполнять итерацию по нему, используя reduce.

. Сейчас я нахожусьиспользуя модуль Node LineStream для обработки stdin как набора строк, но он работает, отправляя событие data для каждой строки - это нормально, он реализует Интерфейс Readable Stream .

Так что в настоящее время я делаю своего рода очень «ручное» уменьшение, передавая временное значение моей функции каждый раз, когда происходит событие data:

var windows = [];

linestream.on('data', function(line) {
  return windows = rollup(windows, extractDate(line), argv.w);
});

linestream.on('end', function() {
  return process.stdout.write(toCsv(windows));
});

process.stdin.resume();

Но было бы более функционально сделать что-то вроде:

linestream.lines.reduce(rollup, []);

function rollup(windows, line) {
    // would return a new interim or final value
}

Конечно, я мог бы «собрать» все строки в обычный массив и затем уменьшить его, но я попыталсяэто и использует слишком много памяти, когда я запускаю свой инструмент на большом наборе данных - так что нечто вроде итерации по потоку действительно то, что нужно.

Я думаю, что я спрашиваю, возможно линаписать Node-функцию / модуль wкоторый бы сделал это, или уже существует.

Спасибо!

Answer 1

Я не верю, что есть какой-то способ сделать это более функциональным, потому что вы имеете дело с асинхронностью.

Чтобы linestream.lines существовал, я думаю, что должно произойти одно из двух:

• Буфер в памяти каждой строки, который, как вы уже сказали, занимает слишком много памяти
• новая языковая конструкция, которая позволила бы асинхронному потоку управления выглядеть как синхронныйone.

Полагаю, вы могли бы сделать это (при условии использования jquery или другого обещающего API):

var op = (function(){

    var windows = []
        ,done = $.deferred();

    linestream.on('data', function(line) {
      return windows = rollup(windows, extractDate(line), argv.w);
    });

    linestream.on('end', function() {
      process.stdout.write(toCsv(windows));
      return done.resolve(windows);
    });

    process.stdin.resume();

    return done.promise();

})();

Но на самом деле это просто скрывает вещи.

Или вы можете использовать что-то вроде Rx , или подождать, пока не появятся генераторы .

Answer 2

Я не совсем понимаю, что должен делать rollup, но, как отмечали другие, у вас не может быть функции сокращения, которая рассчитывает получить все данные одновременно, не имея сначала всех данных одновременно и в памяти.

Что вы можете сделать, так это просто выполнить логику сокращения в обратном вызове события данных. Если ему требуется больше состояний, таких как последнее значение или общее количество значений, вы можете хранить эти данные в замкнутом круге обратного вызова.

Например, вот скользящее среднее для асинхронного потока чисел.

var total = 0;
var items = 0;
var average;

stream.on('data', function (line) {
  var num = parseInt(line, 10);
  total += line;
  items++;
  average = total / items;
});

stream.on('end', function () {
  console.log("The average is %s", average);
});

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

Answer 3

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

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

Я бы просто остался с узлом по умолчанию, если бы я был тобой.