Emit 'done' event on writable streams Needle's response is being piped to

README.md

@@ -27,11 +27,13 @@ var data = {
 needle
   .post('https://my.server.com/foo', data, { multipart: true })
   .on('readable', function() { /* eat your chunks */ })
-  .on('done', function(err, resp) {
+  .on('done', function(err) {
     console.log('Ready-o!');
   })
 ```
 
+Pipe
+
 From version 2.0.x up, Promises are also supported. Just call `needle()` directly and you'll get a native Promise object.
 
 ```js
@@ -72,7 +74,7 @@ Usage
 
 ```js
 // using promises
-needle('get', 'https://server.com/posts/12')
+needle('get', 'https://server.com/posts/123')
   .then(function(resp) {
     // ...
   })
@@ -90,18 +92,11 @@ needle.get('ifconfig.me/all.json', function(error, response, body) {
 });
 
 // no callback, using streams
-// without pipelines
-needle.get('/api.json', {json: true})
-  .on('done', function(err, resp) {
-    console.log(JSON.stringify(resp.body, null, 4));
-  })
-// with pipelines, here the event is triggered by fs which has a 'close' event
-var out = fs.createWriteStream('logo.png');
 needle.get('https://google.com/images/logo.png')
-  .pipe(out)
-  .on('close', function() {
+  .pipe(fs.createWriteStream('logo.png'))
+  .on('done', function(err) {
     console.log('Pipe finished!');
-   });
+  });
 ```
 
 As you can see, you can use Needle with Promises or without them. When using Promises or when a callback is passed, the response's body will be buffered and written to `response.body`, and the callback will be fired when all of the data has been collected and processed (e.g. decompressed, decoded and/or parsed).

lib/needle.js

@@ -495,6 +495,11 @@ Needle.prototype.send_request = function(count, method, uri, config, post_data,
     // NOTE: this event used to be called 'end', but the behaviour was confusing
     // when errors ocurred, because the stream would still emit an 'end' event.
     out.emit('done', err);
+
+    // trigger the 'done' event on streams we're being piped to, if any
+    var pipes = out._readableState.pipes || [];
+    if (!pipes.forEach) pipes = [pipes];
+    pipes.forEach(function(st) { st.emit('done', err); })
   }
 
   function had_error(err) {
@@ -745,10 +750,10 @@ Needle.prototype.send_request = function(count, method, uri, config, post_data,
       // is triggered after all data has been written to it.
       if (out.file) {
         out.file.on('close', function() {
-          done(null, resp, resp.body);
+          done(null, resp);
         })
       } else { // elvis has left the building.
-        done(null, resp, resp.body);
+        done(null, resp);
       }
 
     });

test/errors_spec.js

@@ -81,6 +81,7 @@ describe('errors', function() {
             stream = needle.get(url);
 
         stream.on('done', function(err) {
+          err.code.should.match(/ENOTFOUND|EADDRINFO|EAI_AGAIN/)
           callcount++;
         })
 
@@ -90,21 +91,6 @@ describe('errors', function() {
         }, 200)
       })
 
-      it('error should be ENOTFOUND or EADDRINFO or EAI_AGAIN', function(done) {
-        var errorific,
-            stream = needle.get(url);
-
-        stream.on('done', function(err) {
-          errorific = err;
-        })
-
-        setTimeout(function() {
-          should.exist(errorific);
-          errorific.code.should.match(/ENOTFOUND|EADDRINFO|EAI_AGAIN/)
-          done();
-        }, 200)
-      })
-
       it('does not emit a readable event', function(done) {
         var called = false,
             stream = needle.get(url);
@@ -113,24 +99,24 @@ describe('errors', function() {
           called = true;
         })
 
-        setTimeout(function() {
+        stream.on('done', function(err) {
           called.should.be.false;
           done();
-        }, 50)
+        })
       })
 
       it('does not emit an error event', function(done) {
         var emitted = false,
-            req = needle.get(url);
+            stream = needle.get(url);
 
-        req.on('error', function() {
+        stream.on('error', function() {
           emitted = true;
         })
 
-        setTimeout(function() {
+        stream.on('done', function(err) {
           emitted.should.eql(false);
           done();
-        }, 100);
+        })
       })
 
     })
@@ -211,10 +197,12 @@ describe('errors', function() {
     describe('without callback', function() {
 
       it('emits done event once, with error', function(done) {
-        var called = 0,
+        var error,
+            called = 0,
             stream = send_request();
 
         stream.on('done', function(err) {
+          err.code.should.equal('ECONNRESET');
           called++;
         })
 
@@ -242,13 +230,9 @@ describe('errors', function() {
             stream = send_request();
 
         stream.on('done', function(err) {
-          error = err;
-        })
-
-        setTimeout(function() {
-          error.code.should.equal('ECONNRESET')
+          err.code.should.equal('ECONNRESET')
           done();
-        }, 250)
+        })
       })
 
       it('does not emit a readable event', function(done) {
@@ -259,24 +243,26 @@ describe('errors', function() {
           called = true;
         })
 
-        setTimeout(function() {
+        stream.on('done', function(err) {
           called.should.be.false;
           done();
-        }, 250)
+        })
       })
 
       it('does not emit an error event', function(done) {
         var emitted = false;
-        var req = send_request();
+        var stream = send_request();
 
-        req.on('error', function() {
+        stream.on('error', function() {
           emitted = true;
         })
 
-        setTimeout(function() {
+        stream.on('done', function(err) {
+          err.should.be.a.Error;
+          err.code.should.equal('ECONNRESET')
           emitted.should.eql(false);
           done();
-        }, 100);
+        })
       })
 
     })

test/helpers.js

@@ -37,6 +37,8 @@ helpers.server = function(opts, cb) {
   }
 
   var finish = function(req, res) {
+    if (opts.handler) return opts.handler(req, res);
+
     res.writeHead(get('code'), get('headers'));
     res.end(opts.response || mirror_response(req));
   }

test/stream_events_spec.js

@@ -0,0 +1,144 @@
+var needle  = require('../'),
+    fs = require('fs'),
+    should  = require('should'),
+    helpers = require('./helpers');
+
+describe('stream events', function() {
+
+  var server,
+      port = 3456,
+      responseData,
+      serverOpts = {},
+      requestHandler = function(req, res) { res.end('OK') }
+
+  before(function() {
+    var opts = {
+      port: port,
+      handler: function(req, res) { requestHandler(req, res) }
+    }
+    server = helpers.server(opts);
+  })
+
+  after(function() {
+    server.close();
+  })
+
+  beforeEach(function() {
+    responseData = '';
+  })
+
+  describe('when consuming data directly', function() {
+
+    function send_request(opts, cb) {
+      return needle
+              .get('http://localhost:' + port, opts)
+              .on('data', function(data) { responseData += data })
+    }
+
+    describe('and request stream fails', function() {
+
+      it('emits done event with error', function(done) {
+        requestHandler = function(req, res) { req.socket.destroy() }
+
+        send_request({}).on('done', function(err) {
+          err.code.should.eql('ECONNRESET');
+          responseData.should.eql('');
+          done()
+        })
+      })
+
+    })
+
+    describe('and request succeeds but decoding fails', function() {
+
+      it('emits done event without error', function(done) {
+        requestHandler = function(req, res) {
+          res.setHeader('Content-Type', 'application/json')
+          res.end('invalid:json')
+        }
+
+        send_request({ json: true }).on('done', function(err) {
+          should.not.exist(err);
+          responseData.should.eql('invalid:json');
+          done()
+        })
+      })
+
+    })
+
+    describe('and request succeeds and pipeline works ok', function() {
+
+      it('emits done event without error', function(done) {
+        requestHandler = function(req, res) { res.end('{"ok":1}') }
+
+        send_request({ json: true }).on('done', function(err) {
+          should.not.exist(err);
+          responseData.should.eql('{"ok":1}');
+          done()
+        })
+      })
+
+    })
+
+  })
+
+  describe('when piping to a fs writableStream', function() {
+
+    var outFile = 'test/tmp.dat';
+
+    function send_request(opts, cb) {
+      return needle
+              .get('http://localhost:' + port, opts)
+              .pipe(fs.createWriteStream(outFile))
+              .on('data', function(data) { responseData += data })
+    }
+
+    after(function(done) {
+      fs.unlink(outFile, done)
+    })
+
+    describe('and request stream fails', function() {
+
+      it('final stream emits done event with error', function(done) {
+        requestHandler = function(req, res) { req.socket.destroy() }
+
+        send_request({}).on('done', function(err) {
+          err.code.should.eql('ECONNRESET');
+          done()
+        })
+      })
+
+    })
+
+    describe('and request succeeds but decoding fails', function() {
+
+      it('final stream emits done event without error', function(done) {
+        requestHandler = function(req, res) {
+          res.setHeader('Content-Type', 'application/json')
+          res.end('invalid:json')
+        }
+
+        send_request({ json: true }).on('done', function(err) {
+          should.not.exist(err);
+          done()
+        })
+      })
+
+    })
+
+    describe('and request succeeds and pipeline works ok', function() {
+
+      it('final stream emits done event without error', function(done) {
+        requestHandler = function(req, res) { res.end('{"ok":1}') }
+
+        send_request({ json: true }).on('done', function(err) {
+          should.not.exist(err);
+          done()
+        })
+      })
+
+    })
+
+  })
+
+})