Close and destroy underlying socket after any connection error

amqp-node · Aug 21, 2020 · dac924f · dac924f
1 parent e3e1016
commit dac924f
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 8 deletions.
diff --git a/lib/connect.js b/lib/connect.js
@@ -150,8 +150,11 @@ function connect(url, socketOptions, openCallback) {
       if (timeout) sock.setTimeout(0);
       if (err === null) {
         openCallback(null, c);
+      } else {
+        sock.end();
+        sock.destroy();
+        openCallback(err);
       }
-      else openCallback(err);
     });
   }
 
@@ -174,7 +177,11 @@ function connect(url, socketOptions, openCallback) {
   }
 
   sock.once('error', function(err) {
-    if (!sockok) openCallback(err);
+    if (!sockok) {
+      sock.end();
+      sock.destroy();
+      openCallback(err);
+    }
   });
 
 }

diff --git a/lib/connection.js b/lib/connection.js
@@ -179,8 +179,15 @@ C.open = function(allFields, openCallback0) {
   function send(Method) {
     // This can throw an exception if there's some problem with the
     // options; e.g., something is a string instead of a number.
-    try { self.sendMethod(0, Method, tunedOptions); }
-    catch (err) { bail(err); }
+    try {
+      self.sendMethod(0, Method, tunedOptions);
+    } catch (err) {
+      bail(err);
+
+      // We are in an inconsistent state so we can't continue,
+      // rethrowing the exception will stop further execution.
+      throw err;
+    }
   }
 
   function negotiate(server, desired) {
@@ -206,8 +213,13 @@ C.open = function(allFields, openCallback0) {
       return;
     }
     self.serverProperties = start.fields.serverProperties;
-    send(defs.ConnectionStartOk);
-    wait(afterStartOk);
+    try {
+      send(defs.ConnectionStartOk);
+      wait(afterStartOk);
+    } catch {
+      // Exit, callback with error already called,
+      return;
+    }
   }
 
   function afterStartOk(reply) {
@@ -228,8 +240,13 @@ C.open = function(allFields, openCallback0) {
         negotiate(fields.channelMax, allFields.channelMax);
       tunedOptions.heartbeat =
         negotiate(fields.heartbeat, allFields.heartbeat);
-      send(defs.ConnectionTuneOk);
-      send(defs.ConnectionOpen);
+      try {
+        send(defs.ConnectionTuneOk);
+        send(defs.ConnectionOpen);
+      } catch {
+        // Exit, callback with error already called,
+        return;
+      }
       expect(defs.ConnectionOpenOk, onOpenOk);
       break;
     default:
@@ -257,6 +274,10 @@ C.open = function(allFields, openCallback0) {
   // If the server closes the connection, it's probably because of
   // something we did
   function endWhileOpening(err) {
+    self.stream.removeListener('end', endWhileOpening);
+    self.stream.removeListener('error', endWhileOpening);
+    self.stream.end();
+
     bail(err || new Error('Socket closed abruptly ' +
                           'during opening handshake'));
   }