Слияние потоков в nodejs

Question

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

var stream = new Readable({ objectMode: true });

var cursor = collection.find();
cursor.forEach(fetch);

function fetch(document) {
  stream.push(document);
  // Get all joined documents and run fetch() on them
}

return stream;

Я получаю некоторые ошибки, потому что он не реализует _read, Этот метод также усложняет поиск времени вызова stream.push(null),

Каково решение этой проблемы?

Answer 1

Метод _read необходим для реализации читаемого потока. Если вы предпочитаете более простой интерфейс, вы, вероятно, предпочтете использовать поток PassThrough:

var stream = new PassThrough({ objectMode: true });

var cursor = collection.find();
cursor.forEach(fetch);

function fetch(document) {
  stream.write(document);
  // Get all joined documents and run fetch() on them
}

return stream;

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

Кроме того, посмотрите на API mongodb, чтобы увидеть, как правильно проверить конец потока записей коллекции и соответственно вызвать метод stream.end().