From 03987ae802a072aca2af6de7ffe3a68de5044793 Mon Sep 17 00:00:00 2001
From: delvedor <tommydelved@gmail.com>
Date: Tue, 15 Dec 2020 10:23:06 +0100
Subject: [PATCH 1/2] Fix #1374

---
 lib/Connection.js | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/lib/Connection.js b/lib/Connection.js
index a9084ed77..011644380 100644
--- a/lib/Connection.js
+++ b/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
     }
   }
 

From 9ce30d130489485465eaefafc52317d38996711c Mon Sep 17 00:00:00 2001
From: delvedor <tommydelved@gmail.com>
Date: Tue, 15 Dec 2020 10:23:16 +0100
Subject: [PATCH 2/2] Updated test

---
 test/unit/connection.test.js | 29 +++++++++++++++++++++++++++++
 1 file changed, 29 insertions(+)

diff --git a/test/unit/connection.test.js b/test/unit/connection.test.js
index 95c5e5faf..87b14af9b 100644
--- a/test/unit/connection.test.js
+++ b/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())
+})