Merge pull request #295 from rabbitmq/rabbitmq-dotnet-client-294

michaelklishin · web-flow · commit 3fe6446cd5d7 · 2016-12-08T16:17:25.000+03:00
Ensures connection recovery does not keep going after the connection …
diff --git a/projects/client/RabbitMQ.Client/src/client/api/IConnection.cs b/projects/client/RabbitMQ.Client/src/client/api/IConnection.cs
@@ -194,7 +194,7 @@ public interface IConnection : NetworkConnection, IDisposable
         /// <remarks>
         /// Note that all active channels, sessions, and models will be closed if this method is called.
         /// In comparison to normal <see cref="Close()"/> method, <see cref="Abort()"/> will not throw
-        /// <see cref="AlreadyClosedException"/> or <see cref="IOException"/> during closing connection.
+        /// <see cref="IOException"/> during closing connection.
         ///This method waits infinitely for the in-progress close operation to complete.
         /// </remarks>
         void Abort();
@@ -206,7 +206,7 @@ public interface IConnection : NetworkConnection, IDisposable
         /// The method behaves in the same way as <see cref="Abort()"/>, with the only
         /// difference that the connection is closed with the given connection close code and message.
         /// <para>
-        /// The close code (See under "Reply Codes" in the AMQP specification)
+        /// The close code (See under "Reply Codes" in the AMQP 0-9-1 specification)
         /// </para>
         /// <para>
         /// A message indicating the reason for closing the connection
@@ -253,7 +253,7 @@ public interface IConnection : NetworkConnection, IDisposable
         /// closed if this method is called. It will wait for the in-progress
         /// close operation to complete. This method will not return to the caller
         /// until the shutdown is complete. If the connection is already closed
-        /// (or closing), then this method will throw <see cref="AlreadyClosedException"/>.
+        /// (or closing), then this method will do nothing.
         /// It can also throw <see cref="IOException"/> when socket was closed unexpectedly.
         /// </remarks>
         void Close();
@@ -281,7 +281,7 @@ public interface IConnection : NetworkConnection, IDisposable
         /// Note that all active channels, sessions, and models will be
         /// closed if this method is called. It will wait for the in-progress
         /// close operation to complete with a timeout. If the connection is
-        /// already closed (or closing), then this method will throw <see cref="AlreadyClosedException"/>.
+        /// already closed (or closing), then this method will do nothing.
         /// It can also throw <see cref="IOException"/> when socket was closed unexpectedly.
         /// If timeout is reached and the close operations haven't finished, then socket is forced to close.
         /// <para>
@@ -298,7 +298,7 @@ public interface IConnection : NetworkConnection, IDisposable
         /// The method behaves in the same way as <see cref="Close(int)"/>, with the only
         /// difference that the connection is closed with the given connection close code and message.
         /// <para>
-        /// The close code (See under "Reply Codes" in the AMQP specification).
+        /// The close code (See under "Reply Codes" in the AMQP 0-9-1 specification).
         /// </para>
         /// <para>
         /// A message indicating the reason for closing the connection.
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringConnection.cs b/projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringConnection.cs
@@ -397,14 +397,14 @@ public void BeginAutomaticRecovery()
 
                     recoveryTaskFactory.StartNew(() =>
                     {
-                        if(!self.ManuallyClosed)
+                        if (!self.ManuallyClosed)
                         {
                             try
                             {
 #if NETFX_CORE
-                            System.Threading.Tasks.Task.Delay(m_factory.NetworkRecoveryInterval).Wait();
+                                System.Threading.Tasks.Task.Delay(m_factory.NetworkRecoveryInterval).Wait();
 #else
-                            Thread.Sleep(m_factory.NetworkRecoveryInterval);
+                                Thread.Sleep(m_factory.NetworkRecoveryInterval);
 #endif
                                 self.PerformAutomaticRecovery();
                             }
@@ -422,19 +422,21 @@ protected void PerformAutomaticRecovery()
         {
             lock (recoveryLockTarget)
             {
-                RecoverConnectionDelegate();
-                RecoverConnectionShutdownHandlers();
-                RecoverConnectionBlockedHandlers();
-                RecoverConnectionUnblockedHandlers();
-
-                RecoverModels();
-                if (m_factory.TopologyRecoveryEnabled)
+                if (RecoverConnectionDelegate())
                 {
-                    RecoverEntities();
-                    RecoverConsumers();
-                }
+                    RecoverConnectionShutdownHandlers();
+                    RecoverConnectionBlockedHandlers();
+                    RecoverConnectionUnblockedHandlers();
+
+                    RecoverModels();
+                    if (m_factory.TopologyRecoveryEnabled)
+                    {
+                        RecoverEntities();
+                        RecoverConsumers();
+                    }
 
-                RunRecoveryEventHandlers();
+                    RunRecoveryEventHandlers();
+                }
             }
         }
 
@@ -659,56 +661,64 @@ private void Init(IFrameHandler fh)
         public void Abort()
         {
             this.ManuallyClosed = true;
-            m_delegate.Abort();
+            if(m_delegate.IsOpen)
+                m_delegate.Abort();
         }
 
         ///<summary>API-side invocation of connection abort.</summary>
         public void Abort(ushort reasonCode, string reasonText)
         {
             this.ManuallyClosed = true;
-            m_delegate.Abort(reasonCode, reasonText);
+            if (m_delegate.IsOpen)
+                m_delegate.Abort(reasonCode, reasonText);
         }
 
         ///<summary>API-side invocation of connection abort with timeout.</summary>
         public void Abort(int timeout)
         {
             this.ManuallyClosed = true;
-            m_delegate.Abort(timeout);
+            if (m_delegate.IsOpen)
+                m_delegate.Abort(timeout);
         }
 
         ///<summary>API-side invocation of connection abort with timeout.</summary>
         public void Abort(ushort reasonCode, string reasonText, int timeout)
         {
             this.ManuallyClosed = true;
-            m_delegate.Abort(reasonCode, reasonText, timeout);
+            if (m_delegate.IsOpen)
+                m_delegate.Abort(reasonCode, reasonText, timeout);
         }
 
         ///<summary>API-side invocation of connection.close.</summary>
         public void Close()
         {
             this.ManuallyClosed = true;
-            m_delegate.Close();
+            if (m_delegate.IsOpen)
+                m_delegate.Close();
         }
 
         ///<summary>API-side invocation of connection.close.</summary>
         public void Close(ushort reasonCode, string reasonText)
         {
             this.ManuallyClosed = true;
-            m_delegate.Close(reasonCode, reasonText);
+            if (m_delegate.IsOpen)
+                m_delegate.Close(reasonCode, reasonText);
         }
 
         ///<summary>API-side invocation of connection.close with timeout.</summary>
         public void Close(int timeout)
         {
             this.ManuallyClosed = true;
-            m_delegate.Close(timeout);
+            if (m_delegate.IsOpen)
+                m_delegate.Close(timeout);
         }
 
         ///<summary>API-side invocation of connection.close with timeout.</summary>
         public void Close(ushort reasonCode, string reasonText, int timeout)
         {
             this.ManuallyClosed = true;
-            m_delegate.Close(reasonCode, reasonText, timeout);
+            if (m_delegate.IsOpen)
+                m_delegate.Close(reasonCode, reasonText, timeout);
         }
 
         public IModel CreateModel()
@@ -736,23 +746,17 @@ public void HandleConnectionUnblocked()
 
         void IDisposable.Dispose()
         {
-            try {
+            try
+            {
                 Abort();
             }
-            finally
+            catch(Exception)
             {
-                m_models.Clear();
+                // TODO: log
             }
-            if (ShutdownReport.Count > 0)
+            finally
             {
-                foreach (ShutdownReportEntry entry in ShutdownReport)
-                {
-                    if (entry.Exception != null)
-                    {
-                        throw entry.Exception;
-                    }
-                }
-                throw new OperationInterruptedException(null);
+                m_models.Clear();
             }
         }
 
@@ -826,16 +830,15 @@ protected void RecoverConnectionBlockedHandlers()
             }
         }
 
-        protected void RecoverConnectionDelegate()
+        protected bool RecoverConnectionDelegate()
         {
-            bool recovering = true;
-            while (recovering)
+            while (!ManuallyClosed)
             {
                 try
                 {
                     var fh = endpoints.SelectOne(m_factory.CreateFrameHandler);
                     m_delegate = new Connection(m_factory, false, fh, this.ClientProvidedName);
-                    recovering = false;
+                    return true;
                 }
                 catch (Exception e)
                 {
@@ -866,6 +869,8 @@ protected void RecoverConnectionDelegate()
                     // TODO: provide a way to handle these exceptions
                 }
             }
+
+            return false;
         }
 
         protected void RecoverConnectionShutdownHandlers()
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/Connection.cs b/projects/client/RabbitMQ.Client/src/client/impl/Connection.cs
@@ -1240,9 +1240,9 @@ public void HandleConnectionUnblocked()
 
         void IDisposable.Dispose()
         {
-            MaybeStopHeartbeatTimers();
             try
             {
+                MaybeStopHeartbeatTimers();
                 Abort();
             }
             catch (OperationInterruptedException)
diff --git a/projects/client/Unit/src/unit/TestConnectionRecovery.cs b/projects/client/Unit/src/unit/TestConnectionRecovery.cs
@@ -151,7 +151,7 @@ public void TestBasicConnectionRecoveryWithHostnameList()
         public void TestBasicConnectionRecoveryWithHostnameListAndUnreachableHosts()
         {
             using(var c = CreateAutorecoveringConnection(new List<string> { "191.72.44.22", "127.0.0.1", "localhost" }))
-            {        
+            {
                 Assert.IsTrue(c.IsOpen);
                 CloseAndWaitForRecovery(c);
                 Assert.IsTrue(c.IsOpen);
@@ -189,6 +189,24 @@ public void TestBasicConnectionRecoveryErrorEvent()
             }
         }
 
+        [Test]
+        public void TestBasicConnectionRecoveryStopsAfterManualClose()
+        {
+            Assert.IsTrue(Conn.IsOpen);
+            var c = CreateAutorecoveringConnection();
+            var latch = new AutoResetEvent(false);
+            c.ConnectionRecoveryError += (o, args) => latch.Set();
+            StopRabbitMQ();
+            latch.WaitOne(30000); // we got the failed reconnection event.
+            var triedRecoveryAfterClose = false;
+            c.Close();
+            Thread.Sleep(5000);
+            c.ConnectionRecoveryError += (o, args) => triedRecoveryAfterClose = true;
+            Thread.Sleep(10000);
+            Assert.IsFalse(triedRecoveryAfterClose);
+            StartRabbitMQ();
+        }
+
         [Test]
         public void TestBasicConnectionRecoveryWithEndpointListAndUnreachableHosts()
         {

Original file line number	Diff line number	Diff line change
`@@ -397,14 +397,14 @@ public void BeginAutomaticRecovery()`
`397`	`397`
`398`	`398`	`recoveryTaskFactory.StartNew(() =>`
`399`	`399`	`{`
`400`		`- if(!self.ManuallyClosed)`
	`400`	`+ if (!self.ManuallyClosed)`
`401`	`401`	`{`
`402`	`402`	`try`
`403`	`403`	`{`
`404`	`404`	`#if NETFX_CORE`
`405`		`- System.Threading.Tasks.Task.Delay(m_factory.NetworkRecoveryInterval).Wait();`
	`405`	`+ System.Threading.Tasks.Task.Delay(m_factory.NetworkRecoveryInterval).Wait();`
`406`	`406`	`#else`
`407`		`- Thread.Sleep(m_factory.NetworkRecoveryInterval);`
	`407`	`+ Thread.Sleep(m_factory.NetworkRecoveryInterval);`
`408`	`408`	`#endif`
`409`	`409`	`self.PerformAutomaticRecovery();`
`410`	`410`	`}`
`@@ -422,19 +422,21 @@ protected void PerformAutomaticRecovery()`
`422`	`422`	`{`
`423`	`423`	`lock (recoveryLockTarget)`
`424`	`424`	`{`
`425`		`- RecoverConnectionDelegate();`
`426`		`- RecoverConnectionShutdownHandlers();`
`427`		`- RecoverConnectionBlockedHandlers();`
`428`		`- RecoverConnectionUnblockedHandlers();`
`429`		`-`
`430`		`- RecoverModels();`
`431`		`- if (m_factory.TopologyRecoveryEnabled)`
	`425`	`+ if (RecoverConnectionDelegate())`
`432`	`426`	`{`
`433`		`- RecoverEntities();`
`434`		`- RecoverConsumers();`
`435`		`- }`
	`427`	`+ RecoverConnectionShutdownHandlers();`
	`428`	`+ RecoverConnectionBlockedHandlers();`
	`429`	`+ RecoverConnectionUnblockedHandlers();`
	`430`	`+`
	`431`	`+ RecoverModels();`
	`432`	`+ if (m_factory.TopologyRecoveryEnabled)`
	`433`	`+ {`
	`434`	`+ RecoverEntities();`
	`435`	`+ RecoverConsumers();`
	`436`	`+ }`
`436`	`437`
`437`		`- RunRecoveryEventHandlers();`
	`438`	`+ RunRecoveryEventHandlers();`
	`439`	`+ }`
`438`	`440`	`}`
`439`	`441`	`}`
`440`	`442`
`@@ -659,56 +661,64 @@ private void Init(IFrameHandler fh)`
`659`	`661`	`public void Abort()`
`660`	`662`	`{`
`661`	`663`	`this.ManuallyClosed = true;`
`662`		`- m_delegate.Abort();`
	`664`	`+ if(m_delegate.IsOpen)`
	`665`	`+ m_delegate.Abort();`
`663`	`666`	`}`
`664`	`667`
`665`	`668`	`///<summary>API-side invocation of connection abort.</summary>`
`666`	`669`	`public void Abort(ushort reasonCode, string reasonText)`
`667`	`670`	`{`
`668`	`671`	`this.ManuallyClosed = true;`
`669`		`- m_delegate.Abort(reasonCode, reasonText);`
	`672`	`+ if (m_delegate.IsOpen)`
	`673`	`+ m_delegate.Abort(reasonCode, reasonText);`
`670`	`674`	`}`
`671`	`675`
`672`	`676`	`///<summary>API-side invocation of connection abort with timeout.</summary>`
`673`	`677`	`public void Abort(int timeout)`
`674`	`678`	`{`
`675`	`679`	`this.ManuallyClosed = true;`
`676`		`- m_delegate.Abort(timeout);`
	`680`	`+ if (m_delegate.IsOpen)`
	`681`	`+ m_delegate.Abort(timeout);`
`677`	`682`	`}`
`678`	`683`
`679`	`684`	`///<summary>API-side invocation of connection abort with timeout.</summary>`
`680`	`685`	`public void Abort(ushort reasonCode, string reasonText, int timeout)`
`681`	`686`	`{`
`682`	`687`	`this.ManuallyClosed = true;`
`683`		`- m_delegate.Abort(reasonCode, reasonText, timeout);`
	`688`	`+ if (m_delegate.IsOpen)`
	`689`	`+ m_delegate.Abort(reasonCode, reasonText, timeout);`
`684`	`690`	`}`
`685`	`691`
`686`	`692`	`///<summary>API-side invocation of connection.close.</summary>`
`687`	`693`	`public void Close()`
`688`	`694`	`{`
`689`	`695`	`this.ManuallyClosed = true;`
`690`		`- m_delegate.Close();`
	`696`	`+ if (m_delegate.IsOpen)`
	`697`	`+ m_delegate.Close();`
`691`	`698`	`}`
`692`	`699`
`693`	`700`	`///<summary>API-side invocation of connection.close.</summary>`
`694`	`701`	`public void Close(ushort reasonCode, string reasonText)`
`695`	`702`	`{`
`696`	`703`	`this.ManuallyClosed = true;`
`697`		`- m_delegate.Close(reasonCode, reasonText);`
	`704`	`+ if (m_delegate.IsOpen)`
	`705`	`+ m_delegate.Close(reasonCode, reasonText);`
`698`	`706`	`}`
`699`	`707`
`700`	`708`	`///<summary>API-side invocation of connection.close with timeout.</summary>`
`701`	`709`	`public void Close(int timeout)`
`702`	`710`	`{`
`703`	`711`	`this.ManuallyClosed = true;`
`704`		`- m_delegate.Close(timeout);`
	`712`	`+ if (m_delegate.IsOpen)`
	`713`	`+ m_delegate.Close(timeout);`
`705`	`714`	`}`
`706`	`715`
`707`	`716`	`///<summary>API-side invocation of connection.close with timeout.</summary>`
`708`	`717`	`public void Close(ushort reasonCode, string reasonText, int timeout)`
`709`	`718`	`{`
`710`	`719`	`this.ManuallyClosed = true;`
`711`		`- m_delegate.Close(reasonCode, reasonText, timeout);`
	`720`	`+ if (m_delegate.IsOpen)`
	`721`	`+ m_delegate.Close(reasonCode, reasonText, timeout);`
`712`	`722`	`}`
`713`	`723`
`714`	`724`	`public IModel CreateModel()`
`@@ -736,23 +746,17 @@ public void HandleConnectionUnblocked()`
`736`	`746`
`737`	`747`	`void IDisposable.Dispose()`
`738`	`748`	`{`
`739`		`- try {`
	`749`	`+ try`
	`750`	`+ {`
`740`	`751`	`Abort();`
`741`	`752`	`}`
`742`		`- finally`
	`753`	`+ catch(Exception)`
`743`	`754`	`{`
`744`		`- m_models.Clear();`
	`755`	`+ // TODO: log`
`745`	`756`	`}`
`746`		`- if (ShutdownReport.Count > 0)`
	`757`	`+ finally`
`747`	`758`	`{`
`748`		`- foreach (ShutdownReportEntry entry in ShutdownReport)`
`749`		`- {`
`750`		`- if (entry.Exception != null)`
`751`		`- {`
`752`		`- throw entry.Exception;`
`753`		`- }`
`754`		`- }`
`755`		`- throw new OperationInterruptedException(null);`
	`759`	`+ m_models.Clear();`
`756`	`760`	`}`
`757`	`761`	`}`
`758`	`762`
`@@ -826,16 +830,15 @@ protected void RecoverConnectionBlockedHandlers()`
`826`	`830`	`}`
`827`	`831`	`}`
`828`	`832`
`829`		`- protected void RecoverConnectionDelegate()`
	`833`	`+ protected bool RecoverConnectionDelegate()`
`830`	`834`	`{`
`831`		`- bool recovering = true;`
`832`		`- while (recovering)`
	`835`	`+ while (!ManuallyClosed)`
`833`	`836`	`{`
`834`	`837`	`try`
`835`	`838`	`{`
`836`	`839`	`var fh = endpoints.SelectOne(m_factory.CreateFrameHandler);`
`837`	`840`	`m_delegate = new Connection(m_factory, false, fh, this.ClientProvidedName);`
`838`		`- recovering = false;`
	`841`	`+ return true;`
`839`	`842`	`}`
`840`	`843`	`catch (Exception e)`
`841`	`844`	`{`
`@@ -866,6 +869,8 @@ protected void RecoverConnectionDelegate()`
`866`	`869`	`// TODO: provide a way to handle these exceptions`
`867`	`870`	`}`
`868`	`871`	`}`
	`872`	`+`
	`873`	`+ return false;`
`869`	`874`	`}`
`870`	`875`
`871`	`876`	`protected void RecoverConnectionShutdownHandlers()`
Original file line number	Diff line number	Diff line change
`@@ -1240,9 +1240,9 @@ public void HandleConnectionUnblocked()`
`1240`	`1240`
`1241`	`1241`	`void IDisposable.Dispose()`
`1242`	`1242`	`{`
`1243`		`- MaybeStopHeartbeatTimers();`
`1244`	`1243`	`try`
`1245`	`1244`	`{`
	`1245`	`+ MaybeStopHeartbeatTimers();`
`1246`	`1246`	`Abort();`
`1247`	`1247`	`}`
`1248`	`1248`	`catch (OperationInterruptedException)`