[Backport 7.x] Fix #1374 (#1378)

Co-authored-by: Tomas Della Vedova <delvedor@users.noreply.github.com>

lib/Connection.js

@@ -82,6 +82,7 @@ class Connection {
 
   request (params, callback) {
     this._openRequests++
+    let cleanedListeners = false
 
     const requestParams = this.buildRequestObject(params)
     // https://github.com/nodejs/node/commit/b961d9fd83
@@ -134,7 +135,7 @@ class Connection {
     if (isStream(params.body) === true) {
       pump(params.body, request, err => {
         /* istanbul ignore if  */
-        if (err != null) {
+        if (err != null && cleanedListeners === false) {
           cleanListeners()
           this._openRequests--
           callback(err, null)
@@ -151,6 +152,7 @@ class Connection {
       request.removeListener('timeout', onTimeout)
       request.removeListener('error', onError)
       request.removeListener('abort', onAbort)
+      cleanedListeners = true
     }
   }
 

test/unit/connection.test.js

@@ -23,6 +23,7 @@ const { test } = require('tap')
 const { inspect } = require('util')
 const { URL } = require('url')
 const { Agent } = require('http')
+const { Readable } = require('stream')
 const hpagent = require('hpagent')
 const intoStream = require('into-stream')
 const { buildServer } = require('../utils')
@@ -918,3 +919,31 @@ test('Proxy agent (https)', t => {
 
   t.true(connection.agent instanceof hpagent.HttpsProxyAgent)
 })
+
+test('Abort with a slow body', t => {
+  t.plan(1)
+
+  const connection = new Connection({
+    url: new URL('https://localhost:9200'),
+    proxy: 'http://localhost:8080'
+  })
+
+  const slowBody = new Readable({
+    read (size) {
+      setTimeout(() => {
+        this.push('{"size":1, "query":{"match_all":{}}}')
+        this.push(null) // EOF
+      }, 1000).unref()
+    }
+  })
+
+  const request = connection.request({
+    method: 'GET',
+    path: '/',
+    body: slowBody
+  }, (err, response) => {
+    t.ok(err instanceof RequestAbortedError)
+  })
+
+  setImmediate(() => request.abort())
+})