Node.js Streams: есть ли способ преобразовать или обернуть поток записи fs в поток Transform?

Question

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

Конвейер выглядит следующим образом: для простоты удален несоответствующий код:

function handler (req, res) {
  req.pipe(jsonParse())
     .pipe(addTimeStamp())
     .pipe(jsonStringify())
     .pipe(saveToFs('saved.json'))
     .pipe(res);
}

Пользовательские потоки Transform довольно просты, но у меня нет элегантного способа написания saveToFs. Это выглядит так:

function saveToFs (filename) {
  const write$ = fs.createWriteStream(filename);
  write$.on('open', () => console.log('opened'));
  write$.on('close', () => console.log('closed'));

  const T = new Transform();
  T._transform = function (chunk, encoding, cb) {
    write$.write(chunk);
    cb(null, chunk);
  }
  return T;
}

Идея состоит в том, чтобы просто направить данные в поток записи, а затем через поток ответов, но fs.createWriteStream(<file.name>) является только записываемым потоком, поэтому он затрудняет этот подход.

Прямо сейчас у этого кода есть две проблемы, которые я вижу: поток записи никогда не запускает событие закрытия (утечка памяти?), И я хотел бы, чтобы данные проходили через запись файловой системы перед возвратом данных в поток ответов. по существу многоадресной передачи на две раковины.

Буду очень признателен за любые предложения или указания на основные вещи, которые я пропустил.

Answer 1

Что вам нужно сделать, это сохранить stream, возвращаемый .pipe до saveToFs, и затем передать его в файл и res.

function handler(req, res) {
    const transformed = req.pipe(jsonParse())
        .pipe(addTimeStamp())
        .pipe(jsonStringify());

    transformed.pipe(fs.createWriteStream('saved.json'));
    transformed.pipe(res);
}

Подводя итог, вы можете направить один и тот же читаемый поток (transformed) в несколько записываемых потоков.

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

Использовать опцию { end: false } при передаче по трубопроводу на res.

transformed.pipe(res, { end: false });

А затем позвоните res.end(), когда файл будет записан или когда вы захотите.