Предпосылка

Я пытаюсь найти правильный способ преждевременного прекращения серии потоковых потоков (конвейер) в Node.js: иногда я хочу изящно прервать поток до его завершения. В частности, я имею дело с главным образом objectMode: true и не-родными параллельными потоками, но это не имеет большого значения.

Проблема

Проблема заключается в том, когда я unpipe конвейер, данные остаются в каждом буфере потока и drain ed. Это может быть хорошо для большинства промежуточных потоков (например, Readable/Transform), но последний Writable по-прежнему истощается до его цели записи (например, файла или базы данных или сокета или w/e). Это может быть проблематично, если буфер содержит сотни или тысячи кусков, которые требуют значительного количества времени для слива. Я хочу, чтобы он немедленно остановился, то есть не слил; зачем тратить циклы и память на данные, которые не имеют значения?

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

Вопрос

Каков правильный способ изящного уничтожения конвейера потоков в форме a.pipe(b).pipe(c).pipe(z)?

Решение?

Решение, которое я придумал, состоит из трех шагов:

unpipe каждый поток в конвейере в обратном порядке
Пустой буфер потока, который реализует Writable
end каждый поток, реализующий Writable

Некоторые псевдокоды, иллюстрирующие весь процесс:

var pipeline = [ // define the pipeline
  readStream,
  transformStream0,
  transformStream1,
  writeStream
];

// build and start the pipeline
var tmpBuildStream;
pipeline.forEach(function(stream) {
    if ( !tmpBuildStream ) {
        tmpBuildStream = stream;
        continue;
    }
    tmpBuildStream = lastStream.pipe(stream);
});

// sleep, timeout, event, etc...

// tear down the pipeline
var tmpTearStream;
pipeline.slice(0).reverse().forEach(function(stream) {
    if ( !tmpTearStream ) {
        tmpTearStream = stream;
        continue;
    }
    tmpTearStream = stream.unpipe(tmpTearStream);
});

// empty and end the pipeline
pipeline.forEach(function(stream) {
  if ( typeof stream._writableState === 'object' ) { // empty
    stream._writableState.length -= stream._writableState.buffer.length;
    stream._writableState.buffer = [];
  }
  if ( typeof stream.end === 'function' ) { // kill
    stream.end();
  }
});

Меня действительно беспокоит использование stream._writableState и изменение внутренних свойств buffer и length (_ означает частное свойство). Это похоже на хак. Также обратите внимание на то, что, поскольку я пишу, такие вещи, как pause и resume, наш вопрос не может быть (на основании предложения, полученного мной от IRC).

Я также собрал исполняемую версию (довольно неряшливую), которую вы можете захватить из github: https://github.com/zamnuts/multipipe-proto (git clone, npm install, просмотр readme, запуск npm)

Ответ 1

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

Я хотел бы сосредоточиться на вашей главной цели здесь:

 INPUT >----[read] → [transform0] → [transform1] → [write]-----> OUTPUT
               |          |              |            |
 KILL_ALL------o----------o--------------o------------o--------[nothing to drain]

Я считаю, что вышеуказанная структура может быть достигнута путем комбинирования пользовательских:

duplex stream - для собственной реализации _write(chunk, encoding, cb) и _read(bytes) с
transform stream - для собственной реализации _transform(chunk, encoding, cb).

Поскольку вы используете пакет writable-stream-parallel, вы также можете просмотреть их библиотеки, так как их реализация duplex может быть найдена здесь: https://github.com/Clever/writable-stream-parallel/blob/master/lib/duplex.js. И их реализация transform stream находится здесь: https://github.com/Clever/writable-stream-parallel/blob/master/lib/transform.js. Здесь они обрабатывают highWaterMark.

Возможное решение

Их поток записи: https://github.com/Clever/writable-stream-parallel/blob/master/lib/writable.js#L189 имеет интересную функцию writeOrBuffer, я думаю, вы можете немного ее настроить для прерывания запись данных из буфера.

Примечание. Эти 3 флажки управляют очисткой буфера:

( !finished && !state.bufferProcessing && state.buffer.length )

Литература: