Harden stream error handling for cases of a non-Error instance emit.

We have received an error report that suggests a stream registered
with a Pipeline emitted a non-Error (specifically undefined). This
really should not happen, but since in principle it was intended
that arbitrary streams could be registered, we have always opted
to be extremely thorough around error handling. Opt to make sure
such an occurrence results in gracefully tearing down the stream
and attempt to provide some breadcrumbs around what occurred.

src/Pipeline.js

@@ -225,11 +225,18 @@ module.exports = class Pipeline extends Stream.Duplex {
 
       // protect against filters emitting errors more than once
       stream.once('error', (err) => {
+        const engineName = this.usedEngines[i].name;
         let commandArgs;
-        if (
-          (commandArgs = err.commandArgs || this.usedEngines[i].commandArgs)
-        ) {
-          const engineName = this.usedEngines[i].name;
+        if (!(err instanceof Error)) {
+          const messageForNonError = `${engineName} emitted non-Error (stream index ${i})`;
+          err = new Error(messageForNonError);
+          commandArgs = null;
+        } else if (err.commandArgs) {
+          commandArgs = err.commandArgs;
+        } else {
+          commandArgs = this.usedEngines[i].commandArgs;
+        }
+        if (commandArgs) {
           err.commandLine = `${engineName} ${commandArgs.join(' ')}`;
         }
         this._fail(err, true);
@@ -355,7 +362,7 @@ module.exports = class Pipeline extends Stream.Duplex {
   addStream(stream) {
     this._preflush = true;
     this._attach(stream);
-    this.usedEngines.push({ name: '_stream' });
+    this.usedEngines.push({ name: '_stream', commandArgs: null });
     return this;
   }
 

test/impro.spec.js

@@ -1955,6 +1955,27 @@ describe('impro', () => {
       );
     });
 
+    it('should not break on error from an arbitrary stream that is not an Error', () => {
+      const erroringStream = new stream.Transform();
+      erroringStream._transform = () => {
+        setImmediate(() => {
+          erroringStream.emit('error', undefined);
+        });
+      };
+
+      const pipeline = impro.createPipeline().addStream(erroringStream);
+
+      return expect(
+        'bulb.gif',
+        'when piped through',
+        pipeline,
+        'to error with',
+        expect
+          .it('to have message', '_stream emitted non-Error (stream index 0)')
+          .and('not to have property', 'commandLine')
+      );
+    });
+
     it('should not break on error occurring on the internal passthrough', () => {
       const error = new Error('arranged error');
       class UnchangedStream extends stream.Transform {