Harden stream error handling for the case of a non-Error emit.

src/Pipeline.js

@@ -225,11 +225,17 @@ 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)) {
+          err = new Error(`${engineName} with index ${i} emitted non-Error`);
+          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 +361,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 with index 0 emitted non-Error')
+          .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 {