Merge pull request #1165 from rabbitmq/rabbitmq-dotnet-client-938-950-main

Port rabbitmq/rabbitmq-dotnet-client #950 to main

Author: michaelklishin

projects/RabbitMQ.Client/client/api/IConnectionExtensions.cs

@@ -19,7 +19,7 @@ public static class IConnectionExtensions
         /// </remarks>
         public static void Close(this IConnection connection)
         {
-            connection.Close(Constants.ReplySuccess, "Goodbye", Timeout.InfiniteTimeSpan, false);
+            connection.Close(Constants.ReplySuccess, "Goodbye", TimeSpan.FromSeconds(30), false);
         }
 
         /// <summary>
@@ -37,7 +37,7 @@ public static void Close(this IConnection connection)
         /// </remarks>
         public static void Close(this IConnection connection, ushort reasonCode, string reasonText)
         {
-            connection.Close(reasonCode, reasonText, Timeout.InfiniteTimeSpan, false);
+            connection.Close(reasonCode, reasonText, TimeSpan.FromSeconds(30), false);
         }
 
         /// <summary>
@@ -93,7 +93,7 @@ public static void Close(this IConnection connection, ushort reasonCode, string
         /// </remarks>
         public static void Abort(this IConnection connection)
         {
-            connection.Close(Constants.ReplySuccess, "Connection close forced", Timeout.InfiniteTimeSpan, true);
+            connection.Close(Constants.ReplySuccess, "Connection close forced", TimeSpan.FromSeconds(5), true);
         }
 
         /// <summary>
@@ -111,7 +111,7 @@ public static void Abort(this IConnection connection)
         /// </remarks>
         public static void Abort(this IConnection connection, ushort reasonCode, string reasonText)
         {
-            connection.Close(reasonCode, reasonText, Timeout.InfiniteTimeSpan, true);
+            connection.Close(reasonCode, reasonText, TimeSpan.FromSeconds(5), true);
         }
 
         /// <summary>

projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs

@@ -163,6 +163,8 @@ public RecoveryAwareModel CreateNonRecoveringModel()
         public override string ToString()
             => $"AutorecoveringConnection({InnerConnection.Id},{Endpoint},{GetHashCode()})";
 
+        internal IFrameHandler FrameHandler => InnerConnection.FrameHandler;
+
         internal void Init()
         {
             Init(_factory.EndpointResolverFactory(new List<AmqpTcpEndpoint> { _factory.Endpoint }));

projects/RabbitMQ.Client/client/impl/Connection.cs

@@ -108,11 +108,6 @@ public Connection(IConnectionFactory factory, IFrameHandler frameHandler, string
         public bool IsOpen => CloseReason is null;
 
         public int LocalPort => _frameHandler.LocalPort;
-
-        ///<summary>Another overload of a Protocol property, useful
-        ///for exposing a tighter type.</summary>
-        internal ProtocolBase Protocol => (ProtocolBase)Endpoint.Protocol;
-
         public int RemotePort => _frameHandler.RemotePort;
 
         public IDictionary<string, object?>? ServerProperties { get; private set; }
@@ -123,6 +118,16 @@ public Connection(IConnectionFactory factory, IFrameHandler frameHandler, string
         ///<summary>Explicit implementation of IConnection.Protocol.</summary>
         IProtocol IConnection.Protocol => Endpoint.Protocol;
 
+        ///<summary>Another overload of a Protocol property, useful
+        ///for exposing a tighter type.</summary>
+        internal ProtocolBase Protocol => (ProtocolBase)Endpoint.Protocol;
+
+        ///<summary>Used for testing only.</summary>
+        internal IFrameHandler FrameHandler
+        {
+            get { return _frameHandler; }
+        }
+
         public event EventHandler<CallbackExceptionEventArgs> CallbackException
         {
             add => _callbackExceptionWrapper.AddHandler(value);
@@ -259,7 +264,7 @@ public void Close(ushort reasonCode, string reasonText, TimeSpan timeout, bool a
         ///</para>
         ///<para>
         ///Timeout determines how much time internal close operations should be given
-        ///to complete. System.Threading.Timeout.InfiniteTimeSpan value means infinity.
+        ///to complete.
         ///</para>
         ///</remarks>
         internal void Close(ShutdownEventArgs reason, bool abort, TimeSpan timeout)
@@ -279,8 +284,11 @@ internal void Close(ShutdownEventArgs reason, bool abort, TimeSpan timeout)
                 try
                 {
                     // Try to send connection.close wait for CloseOk in the MainLoop
-                    var cmd = new ConnectionClose(reason.ReplyCode, reason.ReplyText, 0, 0);
-                    _session0.Transmit(ref cmd);
+                    if (!_closed)
+                    {
+                        var cmd = new ConnectionClose(reason.ReplyCode, reason.ReplyText, 0, 0);
+                        _session0.Transmit(ref cmd);
+                    }
                 }
                 catch (AlreadyClosedException)
                 {

projects/RabbitMQ.Client/client/impl/ModelBase.cs

@@ -742,7 +742,7 @@ protected void HandleConnectionStart(in IncomingCommand cmd)
             if (m_connectionStartCell is null)
             {
                 var reason = new ShutdownEventArgs(ShutdownInitiator.Library, Constants.CommandInvalid, "Unexpected Connection.Start");
-                Session.Connection.Close(reason, false, Timeout.InfiniteTimeSpan);
+                Session.Connection.Close(reason, false, TimeSpan.FromSeconds(30));
             }
 
             var method = new ConnectionStart(cmd.MethodBytes.Span);

projects/RabbitMQ.Client/client/impl/SocketFrameHandler.cs

@@ -198,7 +198,11 @@ public void Close()
         {
             lock (_semaphore)
             {
-                if (!_closed)
+                if (_closed || _socket == null)
+                {
+                    return;
+                }
+                else
                 {
                     try
                     {

projects/Unit/TestConnectionShutdown.cs

@@ -33,6 +33,7 @@
 using System.Threading;
 
 using RabbitMQ.Client.Impl;
+using RabbitMQ.Client.Framing.Impl;
 
 using Xunit;
 
@@ -41,6 +42,61 @@ namespace RabbitMQ.Client.Unit
 
     public class TestConnectionShutdown : IntegrationFixture
     {
+        [Fact]
+        public void TestCleanClosureWithSocketClosedOutOfBand()
+        {
+            _conn = CreateAutorecoveringConnection();
+            _model = _conn.CreateModel();
+
+            var latch = new ManualResetEventSlim(false);
+            _model.ModelShutdown += (model, args) => {
+                latch.Set();
+            };
+
+            var c = (AutorecoveringConnection)_conn;
+            c.FrameHandler.Close();
+
+            _conn.Close(TimeSpan.FromSeconds(4));
+            Wait(latch, TimeSpan.FromSeconds(5));
+        }
+
+        [Fact]
+        public void TestAbortWithSocketClosedOutOfBand()
+        {
+            _conn = CreateAutorecoveringConnection();
+            _model = _conn.CreateModel();
+
+            var latch = new ManualResetEventSlim(false);
+            _model.ModelShutdown += (model, args) => {
+                latch.Set();
+            };
+
+            var c = (AutorecoveringConnection)_conn;
+            c.FrameHandler.Close();
+
+            _conn.Abort();
+            // default Connection.Abort() timeout and then some
+            Wait(latch, TimeSpan.FromSeconds(6));
+        }
+
+        [Fact]
+        public void TestDisposedWithSocketClosedOutOfBand()
+        {
+            _conn = CreateAutorecoveringConnection();
+            _model = _conn.CreateModel();
+
+            var latch = new ManualResetEventSlim(false);
+            _model.ModelShutdown += (model, args) => {
+                latch.Set();
+            };
+
+            var c = (AutorecoveringConnection)_conn;
+            c.FrameHandler.Close();
+
+            _conn.Dispose();
+            Wait(latch, TimeSpan.FromSeconds(3));
+        }
+
         [Fact]
         public void TestShutdownSignalPropagationToChannels()
         {