Fix Query stream to emit close after ending

fixes #1349
closes #1350

Changes.md

@@ -7,6 +7,7 @@ you spot any mistakes.
 ## HEAD
 
 * Add `POOL_CONNLIMIT` code to "No connections available." error #1332
+* Fix Query stream to emit close after ending #1349 #1350
 * Performance improvements for array/object escaping in SqlString #1331
 
 ## v2.10.2 (2016-01-12)

lib/protocol/sequences/Query.js

@@ -197,6 +197,12 @@ Query.prototype.stream = function(options) {
     self._connection && self._connection.resume();
   };
 
+  stream.once('end', function() {
+    process.nextTick(function () {
+      stream.emit('close');
+    });
+  });
+
   this.on('result',function(row,i) {
     if (!stream.push(row)) self._connection.pause();
     stream.emit('result',row,i);  // replicate old emitter
@@ -207,7 +213,6 @@ Query.prototype.stream = function(options) {
   });
 
   this.on('end', function() {
-    stream.emit('close');  // notify readers that query has completed
     stream.push(null);  // pushing null, indicating EOF
   });
 

test/unit/query/test-stream-emits-close-after-end.js

@@ -0,0 +1,39 @@
+var assert     = require('assert');
+var common     = require('../../common');
+var connection = common.createConnection({port: common.fakeServerPort});
+
+var server = common.createFakeServer();
+
+server.listen(common.fakeServerPort, function (err) {
+  assert.ifError(err);
+
+  var closed = false;
+  var ended  = false;
+  var query  = connection.query('SELECT * FROM stream LIMIT 2');
+  var stream = query.stream();
+  var wait   = 3;
+
+  function done() {
+    if (--wait) return;
+    server.destroy();
+  }
+
+  stream.once('close', function () {
+    assert.ok(ended);
+    closed = true;
+    done();
+  });
+
+  stream.once('end', function () {
+    assert.ok(!closed);
+    ended = true;
+    done();
+  });
+
+  stream.on('data', function noop() {});
+
+  connection.end(function (err) {
+    assert.ifError(err);
+    done();
+  });
+});

test/unit/query/test-stream-emits-close.js

@@ -18,6 +18,8 @@ server.listen(common.fakeServerPort, function (err) {
 
   stream.once('close', done);
 
+  stream.on('data', function noop() {});
+
   connection.end(function (err) {
     assert.ifError(err);
     done();