Fehlerbehandlung mit node.js-Streams

Was ist der richtige Weg, um Fehler mit Streams zu behandeln? Ich weiß bereits, dass es ein "Fehler" -Ereignis gibt, das Sie abhören können, aber ich möchte mehr Details über willkürlich komplizierte Situationen erfahren.

Was machen Sie für den Anfang, wenn Sie eine einfache Rohrkette machen möchten:

input.pipe(transformA).pipe(transformB).pipe(transformC)...

Und wie erstellen Sie eine dieser Transformationen richtig, damit Fehler korrekt behandelt werden?

Weitere verwandte Fragen:

• Was passiert mit dem Endereignis, wenn ein Fehler auftritt? Wird es nie gefeuert? Wird es manchmal gefeuert? Kommt es auf die Transformation / den Stream an? Was sind die Standards hier?
• Gibt es Mechanismen, um Fehler durch die Rohre zu verbreiten?
• Lösen Domains dieses Problem effektiv? Beispiele wären schön.
• Haben Fehler, die aus "Fehler" -Ereignissen resultieren, Stapelspuren? Manchmal? Noch nie? Gibt es eine Möglichkeit, einen von ihnen zu bekommen?

verwandeln

Transformations-Streams sind sowohl lesbar als auch beschreibbar und daher wirklich gute "mittlere" Streams. Aus diesem Grund werden sie manchmal als throughStreams bezeichnet. Sie ähneln auf diese Weise einem Duplex-Stream, bieten jedoch eine schöne Schnittstelle zum Bearbeiten der Daten, anstatt sie nur zu senden. Der Zweck eines Transformationsstroms besteht darin, die Daten zu manipulieren, während sie durch den Strom geleitet werden. Möglicherweise möchten Sie beispielsweise einige asynchrone Aufrufe ausführen oder einige Felder ableiten, einige Dinge neu zuordnen usw.

Informationen zum Erstellen eines Transformations-Streams finden Sie hier und hier . Alles was du tun musst, ist :

1. Fügen Sie das Stream-Modul hinzu
2. Instanziieren (oder erben) der Transform-Klasse
3. Implementieren Sie eine _transformMethode, die a (chunk, encoding, callback).

Der Block sind Ihre Daten. Die meiste Zeit müssen Sie sich keine Gedanken über die Codierung machen, wenn Sie arbeiten objectMode = true. Der Rückruf wird aufgerufen, wenn Sie mit der Verarbeitung des Blocks fertig sind. Dieser Block wird dann zum nächsten Stream weitergeleitet.

Wenn Sie ein nettes Hilfsmodul möchten, mit dem Sie ganz einfach Streams durchführen können, empfehle ich through2 .

Lesen Sie zur Fehlerbehandlung weiter.

Rohr

In einer Rohrkette ist die Behandlung von Fehlern in der Tat nicht trivial. Laut diesem Thread ist .pipe () nicht dafür ausgelegt, Fehler weiterzuleiten. Also so etwas wie ...

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

... würde nur auf Fehler im Stream warten c. Wenn ein Fehlerereignis ausgegeben awürde, würde dieses nicht weitergegeben und würde tatsächlich werfen. Um dies richtig zu machen:

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)});

Obwohl der zweite Weg ausführlicher ist, können Sie zumindest den Kontext beibehalten, in dem Ihre Fehler auftreten. Dies ist normalerweise eine gute Sache.

Eine Bibliothek, die ich jedoch hilfreich finde, wenn Sie einen Fall haben, in dem Sie nur die Fehler am Ziel erfassen möchten und sich nicht so sehr darum kümmern, wo es passiert ist, ist der Ereignisstrom .

Ende

Wenn ein Fehlerereignis ausgelöst wird, wird das Endereignis nicht (explizit) ausgelöst. Das Ausgeben eines Fehlerereignisses beendet den Stream.

Domänen

Nach meiner Erfahrung funktionieren Domains die meiste Zeit sehr gut. Wenn Sie ein nicht behandeltes Fehlerereignis haben (dh einen Fehler in einem Stream ohne Listener ausgeben), kann der Server abstürzen. Wie im obigen Artikel ausgeführt, können Sie den Stream jetzt in eine Domäne einschließen, die alle Fehler ordnungsgemäß abfangen sollte.

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);
 });

• Das obige Codebeispiel stammt aus diesem Beitrag

Das Schöne an Domänen ist, dass sie die Stapelspuren beibehalten. Auch wenn Event-Stream gute Arbeit leistet.

Weitere Informationen finden Sie im Stream-Handbuch . Ziemlich ausführlich, aber sehr nützlich und bietet einige großartige Links zu vielen hilfreichen Modulen.

Wenn Sie node> = v10.0.0 verwenden, können Sie stream.pipeline und stream.finished verwenden .

Beispielsweise:

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');
  }
});

Weitere Informationen finden Sie in diesem Github PR .

Domains sind veraltet. du brauchst sie nicht.

Für diese Frage sind Unterscheidungen zwischen transformiert oder beschreibbar nicht so wichtig.

Die Antwort von mshell_lauren ist großartig, aber alternativ können Sie auch explizit auf das Fehlerereignis in jedem Stream warten, von dem Sie glauben, dass er fehlerhaft ist. und verwenden Sie die Handlerfunktion erneut, wenn Sie dies bevorzugen.

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) }

Dies verhindert die berüchtigte nicht erfasste Ausnahme, falls einer dieser Streams sein Fehlerereignis auslöst

Fehler aus der gesamten Kette können mit einer einfachen Funktion an den Stream ganz rechts weitergegeben werden:

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;
}

welches verwendet werden kann wie:

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

.on("error", handler)kümmert sich nur um Stream-Fehler, aber wenn Sie benutzerdefinierte Transformations-Streams verwenden, sollten .on("error", handler)Sie die in der _transformFunktion auftretenden Fehler nicht abfangen . So etwas kann man zur Steuerung des Anwendungsflusses tun: -

thisSchlüsselwort in _transformFunktion bezieht sich auf sich Streamselbst, was ein ist EventEmitter. Sie können also try catchwie unten beschrieben die Fehler abfangen und später an die benutzerdefinierten Ereignishandler übergeben.

// 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)

Auf diese Weise können Sie Ihre Logik- und Fehlerbehandlungsroutinen getrennt halten. Sie können auch festlegen, dass nur einige Fehler behandelt und andere ignoriert werden.

UPDATE
Alternative: RXJS Observable

Verwenden Sie das Multipipe- Paket, um mehrere Streams zu einem Duplex-Stream zu kombinieren. Und behandeln Sie Fehler an einem Ort.

const pipe = require('multipipe')

// pipe streams
const stream = pipe(streamA, streamB, streamC) 


// centralized error handling
stream.on('error', fn)

Verwenden Sie das Node.js-Muster, indem Sie eine Transform-Stream-Mechanik erstellen und ihren Rückruf donemit einem Argument aufrufen , um den Fehler zu verbreiten:

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's 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);
   });

Try catch erfasst nicht die Fehler, die im Stream aufgetreten sind, da sie ausgelöst werden, nachdem der aufrufende Code bereits beendet wurde. Sie können sich auf die Dokumentation beziehen:

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