Обработка ошибок с помощью потоков node.js

Ответ 1

преобразование

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

Для получения информации о том, как создать поток преобразования, смотрите здесь и здесь. Все, что вам нужно сделать, это:

1. включить модуль потока
2. создавать (или наследовать) класс Transform
3. реализовать метод _transform, который принимает (chunk, encoding, callback).

Кусок - это ваши данные. Большую часть времени вам не нужно беспокоиться о кодировании, если вы работаете в objectMode = true. Обратный вызов вызывается, когда вы закончите обработку чанка. Затем этот фрагмент передается следующему потоку.

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

Для обработки ошибок продолжайте читать.

pipe

В цепочке pipeопроводов обработка ошибок действительно нетривиальна. Согласно этот поток.pipe() не создан для пересылки ошибок. Так что-то вроде...

var a = createStream();
a.pipe(b).pipe(c).on('error', function(e){handleError(e)});

... будет прослушивать только ошибки в потоке c. Если на a было сгенерировано событие ошибки, оно не будет передано и, фактически, сгенерирует. Чтобы сделать это правильно:

var a = createStream();
a.on('error', function(e){handleError(e)})
.pipe(b)
.on('error', function(e){handleError(e)})
.pipe(c)
.on('error', function(e){handleError(e)});

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

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

конец

При возникновении события ошибки конечное событие не будет запущено (явно). Передача события ошибки завершит поток.

домены

По моему опыту, домены работают очень хорошо большую часть времени. Если у вас есть необработанное событие ошибки (то есть сообщение об ошибке в потоке без прослушивателя), сервер может дать сбой. Теперь, как указано в приведенной выше статье, вы можете заключить поток в домен, который должен правильно отлавливать все ошибки.

var d = domain.create();
 d.on('error', handleAllErrors);
 d.run(function() {
     fs.createReadStream(tarball)
       .pipe(gzip.Gunzip())
       .pipe(tar.Extract({ path: targetPath }))
       .on('close', cb);
 });

• приведенный выше пример кода взят из этого поста

Прелесть доменов в том, что они сохранят следы стека. Хотя Event-Stream также хорошо справляется с этой задачей.

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

Ответ 2

устарели. вам они не нужны.

для этого вопроса различия между трансформируемым или записываемым не так важны.

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

var a = createReadableStream()
var b = anotherTypeOfStream()
var c = createWriteStream()

a.on('error', handler)
b.on('error', handler)
c.on('error', handler)

a.pipe(b).pipe(c)

function handler (err) { console.log(err) }

это предотвращает печально известное неперехваченное исключение, если один из этих потоков запускает свое событие ошибки

Ответ 3

Если вы используете узел> = v10.0.0, вы можете использовать stream.pipeline и stream.finished.

Например:

const { pipeline, finished } = require('stream');

pipeline(
  input, 
  transformA, 
  transformB, 
  transformC, 
  (err) => {
    if (err) {
      console.error('Pipeline failed', err);
    } else {
      console.log('Pipeline succeeded');
    }
});


finished(input, (err) => {
  if (err) {
    console.error('Stream failed', err);
  } else {
    console.log('Stream is done reading');
  }
});

Смотрите этот github PR для дальнейшего обсуждения.

Ответ 4

Ошибки всей цепочки могут быть переданы в самый правый поток с помощью простой функции:

function safePipe (readable, transforms) {
    while (transforms.length > 0) {
        var new_readable = transforms.shift();
        readable.on("error", function(e) { new_readable.emit("error", e); });
        readable.pipe(new_readable);
        readable = new_readable;
    }
    return readable;
}

который можно использовать как:

safePipe(readable, [ transform1, transform2, ... ]);

Ответ 5

.on("error", handler) заботится только об ошибках потока, но если вы используете пользовательские потоки преобразования, .on("error", handler) не перехватывает ошибки, возникающие внутри функции _transform. Таким образом, можно сделать что-то вроде этого для управления потоком приложений:

this ключевое слово в функции _transform относится к самому Stream, который является EventEmitter. Таким образом, вы можете использовать try catch как try catch ниже, чтобы перехватить ошибки и затем передать их в пользовательские обработчики событий.

// CustomTransform.js
CustomTransformStream.prototype._transform = function (data, enc, done) {
  var stream = this
  try {
    // Do your transform code
  } catch (e) {
    // Now based on the error type, with an if or switch statement
    stream.emit("CTError1", e)
    stream.emit("CTError2", e)
  }
  done()
}

// StreamImplementation.js
someReadStream
  .pipe(CustomTransformStream)
  .on("CTError1", function (e) { console.log(e) })
  .on("CTError2", function (e) { /*Lets do something else*/ })
  .pipe(someWriteStream)

Таким образом, вы можете хранить свою логику и обработчики ошибок отдельно. Кроме того, вы можете обрабатывать только некоторые ошибки и игнорировать другие.

ОБНОВИТЬ
Альтернатива: наблюдаемый RXJS

Ответ 6

Используйте шаблон Node.js, создав механику потока Transform и вызывая ее обратный вызов done с аргументом для распространения ошибки:

var transformStream1 = new stream.Transform(/*{objectMode: true}*/);

transformStream1.prototype._transform = function (chunk, encoding, done) {
  //var stream = this;

  try {
    // Do your transform code
    /* ... */
  } catch (error) {
    // nodejs style for propagating an error
    return done(error);
  }

  // Here, everything went well
  done();
}

// Let use the transform stream, assuming `someReadStream`
// and `someWriteStream` have been defined before
someReadStream
  .pipe(transformStream1)
  .on('error', function (error) {
    console.error('Error in transformStream1:');
    console.error(error);
    process.exit(-1);
   })
  .pipe(someWriteStream)
  .on('close', function () {
    console.log('OK.');
    process.exit();
  })
  .on('error', function (error) {
    console.error(error);
    process.exit(-1);
   });

Ответ 7

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

https://nodejs.org/dist/latest-v10.x/docs/api/errors.html