Merge pull request #225 from ggoodman/fix-stream-error-window

Handle stream payload errors before socket connection

lib/index.js

@@ -336,10 +336,17 @@ internals.deferPipeUntilSocketConnects = function (req, stream) {
     };
     const onSocketConnect = () => {
 
+        stream.removeListener('error', onStreamError);
         stream.pipe(req);
     };
+    const onStreamError = (err) => {
+
+        req.emit('error', err);
+    };
+
 
     req.once('socket', onSocket);
+    stream.on('error', onStreamError);
 };
 
 

test/index.js

@@ -1127,25 +1127,19 @@ describe('read()', () => {
         const stream = new Stream.Readable({
             read() {
 
-                piped = true;
+                read = true;
                 this.push(null);
             }
         });
-        const onPiped = () => {
-
-            piped = true;
-        };
-        let piped = false;
-
-        stream.on('pipe', onPiped);
+        let read = false;
 
         const promiseA = Wreck.request('post', 'http://localhost:0', {
             agent,
             payload: stream
         });
 
         await expect(promiseA).to.reject(Error, /Unable to obtain socket/);
-        expect(piped).to.equal(false);
+        expect(read).to.equal(false);
 
         const handler = (req, res) => {
 
@@ -1158,10 +1152,27 @@ describe('read()', () => {
             payload: stream
         });
         expect(res.statusCode).to.equal(200);
-        expect(piped).to.equal(true);
+        expect(read).to.equal(true);
         server.close();
     });
 
+    it('will handle stream payload errors between request creation and connection establishment', async () => {
+
+        const agent = new internals.SlowAgent();
+        const stream = new Stream.Readable();
+        const promiseA = Wreck.request('post', 'http://localhost:0', {
+            agent,
+            payload: stream
+        });
+
+        process.nextTick(() => {
+
+            stream.emit('error', new Error('Asynchronous stream error'));
+        });
+
+        await expect(promiseA).to.reject(Error, /Asynchronous stream error/);
+    });
+
     it('times out when stream read takes too long', async () => {
 
         const TestStream = function () {