fix: remove stream.emit('close') on stream end

lib/commands/query.js

@@ -227,7 +227,7 @@ class Query extends Command {
   }
 
   /* eslint no-unused-vars: ["error", { "argsIgnorePattern": "^_" }] */
-  row(packet, _connection) { 
+  row(packet, _connection) {
     if (packet.isEOF()) {
       const status = packet.eofStatusFlags();
       const moreResults = status & ServerStatus.SERVER_MORE_RESULTS_EXISTS;
@@ -279,10 +279,13 @@ class Query extends Command {
     });
     this.on('end', () => {
       stream.push(null); // pushing null, indicating EOF
-      stream.emit('close'); // notify readers that query has completed
     });
     this.on('fields', fields => {
       stream.emit('fields', fields); // replicate old emitter
+      // Node >= 18 Error: Premature close
+      if (process.versions.node.split(".", 1)[0] <= 14) {
+        stream.emit('close'); // notify readers that query has completed
+      }
     });
     return stream;
   }
@@ -302,7 +305,7 @@ class Query extends Command {
       Timers.clearTimeout(this.queryTimeout);
       this.queryTimeout = null;
     }
-    
+
     const err = new Error('Query inactivity timeout');
     err.errorno = 'PROTOCOL_SEQUENCE_TIMEOUT';
     err.code = 'PROTOCOL_SEQUENCE_TIMEOUT';

test/integration/connection/test-stream.js

@@ -3,6 +3,7 @@
 const common = require('../../common');
 const connection = common.createConnection();
 const assert = require('assert');
+const { finished } = require("stream/promises");
 
 let rows;
 const rows1 = [];
@@ -62,8 +63,31 @@ connection.execute('SELECT * FROM announcements', (err, _rows) => {
   });
 });
 
+
+const rowsStreamForAwait = []
+async function testStreamForAwait() {
+  const stream = connection.execute('SELECT * FROM announcements').stream()
+  for await (const row of stream) {
+    rowsStreamForAwait.push(row)
+  }
+}
+testStreamForAwait().then()
+
+
+const rowsStreamFinished = []
+async function testStreamFinished() {
+  const stream = connection.execute('SELECT * FROM announcements').stream()
+  stream.on('result', row => {
+    rowsStreamFinished.push(row)
+  })
+  await finished(stream)
+}
+testStreamFinished().then()
+
 process.on('exit', () => {
   assert.deepEqual(rows.length, 2);
   assert.deepEqual(rows, rows1);
   assert.deepEqual(rows, rows2);
+  assert.deepEqual(rows, rowsStreamForAwait);
+  assert.deepEqual(rows, rowsStreamFinished);
 });