Merge pull request #781 from danielmarbach/graceful

lukebakken · web-flow · commit d0f0e6cc087c · 2020-03-26T15:12:10.000-07:00
Graceful stop of worker services
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerDispatcher.cs b/projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerDispatcher.cs
@@ -1,4 +1,5 @@
 ﻿using System;
+using System.Threading.Tasks;
 
 namespace RabbitMQ.Client.Impl
 {
@@ -19,9 +20,9 @@ public void Quiesce()
             IsShutdown = true;
         }
 
-        public void Shutdown(IModel model)
+        public Task Shutdown(IModel model)
         {
-            _workService.Stop(model);
+            return _workService.Stop(model);
         }
 
         public bool IsShutdown
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerWorkService.cs b/projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerWorkService.cs
@@ -20,20 +20,14 @@ private WorkPool StartNewWorkPool(IModel model)
             return newWorkPool;
         }
 
-        public void Stop(IModel model)
+        public Task Stop(IModel model)
         {
             if (_workPools.TryRemove(model, out WorkPool workPool))
             {
-                workPool.Stop();
+                return workPool.Stop();
             }
-        }
 
-        public void Stop()
-        {
-            foreach (IModel model in _workPools.Keys)
-            {
-                Stop(model);
-            }
+            return Task.CompletedTask;
         }
 
         class WorkPool
@@ -78,17 +72,18 @@ async Task Loop()
                         // Swallowing the task cancellation in case we are stopping work.
                     }
 
-                    while (_tokenSource.IsCancellationRequested == false && _workQueue.TryDequeue(out Work work))
+                    while (_workQueue.TryDequeue(out Work work))
                     {
                         await work.Execute(_model).ConfigureAwait(false);
                     }
                 }
             }
 
-            public void Stop()
+            public Task Stop()
             {
                 _tokenSource.Cancel();
                 _tokenRegistration.Dispose();
+                return _worker;
             }
         }
     }
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringModel.cs b/projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringModel.cs
@@ -279,7 +279,7 @@ public void Close(ShutdownEventArgs reason, bool abort)
         {
             try
             {
-                _delegate.Close(reason, abort);
+                _delegate.Close(reason, abort).GetAwaiter().GetResult();;
             }
             finally
             {
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/ConcurrentConsumerDispatcher.cs b/projects/client/RabbitMQ.Client/src/client/impl/ConcurrentConsumerDispatcher.cs
@@ -1,6 +1,6 @@
 ﻿using System;
 using System.Collections.Generic;
-
+using System.Threading.Tasks;
 using RabbitMQ.Client.Events;
 
 namespace RabbitMQ.Client.Impl
@@ -22,9 +22,9 @@ public void Quiesce()
             IsShutdown = true;
         }
 
-        public void Shutdown(IModel model)
+        public Task Shutdown(IModel model)
         {
-            _workService.StopWork(model);
+            return _workService.StopWorkAsync(model);
         }
 
         public bool IsShutdown
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/ConsumerWorkService.cs b/projects/client/RabbitMQ.Client/src/client/impl/ConsumerWorkService.cs
@@ -21,20 +21,27 @@ private WorkPool StartNewWorkPool(IModel model)
             return newWorkPool;
         }
 
-        public void StopWork(IModel model)
+        public void StopWork()
         {
-            if (_workPools.TryRemove(model, out WorkPool workPool))
+            foreach (IModel model in _workPools.Keys)
             {
-                workPool.Stop();
+                StopWork(model);
             }
         }
 
-        public void StopWork()
+        public void StopWork(IModel model)
         {
-            foreach (IModel model in _workPools.Keys)
+            StopWorkAsync(model).GetAwaiter().GetResult();
+        }
+
+        internal Task StopWorkAsync(IModel model)
+        {
+            if (_workPools.TryRemove(model, out WorkPool workPool))
             {
-                StopWork(model);
+                return workPool.Stop();
             }
+
+            return Task.CompletedTask;
         }
 
         class WorkPool
@@ -77,7 +84,7 @@ async Task Loop()
                         // Swallowing the task cancellation exception for the semaphore in case we are stopping.
                     }
 
-                    while (_tokenSource.IsCancellationRequested == false && _actions.TryDequeue(out Action action))
+                    while (_actions.TryDequeue(out Action action))
                     {
                         try
                         {
@@ -91,10 +98,11 @@ async Task Loop()
                 }
             }
 
-            public void Stop()
+            public Task Stop()
             {
                 _tokenSource.Cancel();
                 _tokenRegistration.Dispose();
+                return _worker;
             }
         }
     }
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/IConsumerDispatcher.cs b/projects/client/RabbitMQ.Client/src/client/impl/IConsumerDispatcher.cs
@@ -39,6 +39,7 @@
 //---------------------------------------------------------------------------
 
 using System;
+using System.Threading.Tasks;
 
 namespace RabbitMQ.Client.Impl
 {
@@ -69,6 +70,6 @@ void HandleModelShutdown(IBasicConsumer consumer,
 
         void Quiesce();
 
-        void Shutdown(IModel model);
+        Task Shutdown(IModel model);
     }
 }
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs b/projects/client/RabbitMQ.Client/src/client/impl/ModelBase.cs
@@ -45,7 +45,7 @@
 using System.Runtime.CompilerServices;
 using System.Text;
 using System.Threading;
-
+using System.Threading.Tasks;
 using RabbitMQ.Client.Events;
 using RabbitMQ.Client.Exceptions;
 using RabbitMQ.Client.Framing;
@@ -184,14 +184,14 @@ public bool IsOpen
 
         public ISession Session { get; private set; }
 
-        public void Close(ushort replyCode, string replyText, bool abort)
+        public Task Close(ushort replyCode, string replyText, bool abort)
         {
-            Close(new ShutdownEventArgs(ShutdownInitiator.Application,
+            return Close(new ShutdownEventArgs(ShutdownInitiator.Application,
                 replyCode, replyText),
                 abort);
         }
 
-        public void Close(ShutdownEventArgs reason, bool abort)
+        public async Task Close(ShutdownEventArgs reason, bool abort)
         {
             var k = new ShutdownContinuation();
             ModelShutdown += k.OnConnectionShutdown;
@@ -204,7 +204,7 @@ public void Close(ShutdownEventArgs reason, bool abort)
                     _Private_ChannelClose(reason.ReplyCode, reason.ReplyText, 0, 0);
                 }
                 k.Wait(TimeSpan.FromMilliseconds(10000));
-                ConsumerDispatcher.Shutdown(this);
+                await ConsumerDispatcher.Shutdown(this).ConfigureAwait(false);
             }
             catch (AlreadyClosedException)
             {
@@ -508,7 +508,7 @@ public void OnSessionShutdown(object sender, ShutdownEventArgs reason)
             SetCloseReason(reason);
             OnModelShutdown(reason);
             BroadcastShutdownToConsumers(m_consumers, reason);
-            ConsumerDispatcher.Shutdown(this);
+            ConsumerDispatcher.Shutdown(this).GetAwaiter().GetResult();;
         }
 
         protected void BroadcastShutdownToConsumers(IDictionary<string, IBasicConsumer> cs, ShutdownEventArgs reason)
@@ -1394,7 +1394,7 @@ public void WaitForConfirmsOrDie(TimeSpan timeout)
                 Close(new ShutdownEventArgs(ShutdownInitiator.Application,
                     Constants.ReplySuccess,
                     "Nacks Received", new IOException("nack received")),
-                    false);
+                    false).GetAwaiter().GetResult();
                 throw new IOException("Nacks Received");
             }
             if (timedOut)
@@ -1403,7 +1403,7 @@ public void WaitForConfirmsOrDie(TimeSpan timeout)
                     Constants.ReplySuccess,
                     "Timed out waiting for acks",
                     new IOException("timed out waiting for acks")),
-                    false);
+                    false).GetAwaiter().GetResult();
                 throw new IOException("Timed out waiting for acks");
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`using System;`
	`2`	`+using System.Threading.Tasks;`
`2`	`3`
`3`	`4`	`namespace RabbitMQ.Client.Impl`
`4`	`5`	`{`
`@@ -19,9 +20,9 @@ public void Quiesce()`
`19`	`20`	`IsShutdown = true;`
`20`	`21`	`}`
`21`	`22`
`22`		`- public void Shutdown(IModel model)`
	`23`	`+ public Task Shutdown(IModel model)`
`23`	`24`	`{`
`24`		`- _workService.Stop(model);`
	`25`	`+ return _workService.Stop(model);`
`25`	`26`	`}`
`26`	`27`
`27`	`28`	`public bool IsShutdown`
Original file line number	Diff line number	Diff line change
`@@ -20,20 +20,14 @@ private WorkPool StartNewWorkPool(IModel model)`
`20`	`20`	`return newWorkPool;`
`21`	`21`	`}`
`22`	`22`
`23`		`- public void Stop(IModel model)`
	`23`	`+ public Task Stop(IModel model)`
`24`	`24`	`{`
`25`	`25`	`if (_workPools.TryRemove(model, out WorkPool workPool))`
`26`	`26`	`{`
`27`		`- workPool.Stop();`
	`27`	`+ return workPool.Stop();`
`28`	`28`	`}`
`29`		`- }`
`30`	`29`
`31`		`- public void Stop()`
`32`		`- {`
`33`		`- foreach (IModel model in _workPools.Keys)`
`34`		`- {`
`35`		`- Stop(model);`
`36`		`- }`
	`30`	`+ return Task.CompletedTask;`
`37`	`31`	`}`
`38`	`32`
`39`	`33`	`class WorkPool`
`@@ -78,17 +72,18 @@ async Task Loop()`
`78`	`72`	`// Swallowing the task cancellation in case we are stopping work.`
`79`	`73`	`}`
`80`	`74`
`81`		`- while (_tokenSource.IsCancellationRequested == false && _workQueue.TryDequeue(out Work work))`
	`75`	`+ while (_workQueue.TryDequeue(out Work work))`
`82`	`76`	`{`
`83`	`77`	`await work.Execute(_model).ConfigureAwait(false);`
`84`	`78`	`}`
`85`	`79`	`}`
`86`	`80`	`}`
`87`	`81`
`88`		`- public void Stop()`
	`82`	`+ public Task Stop()`
`89`	`83`	`{`
`90`	`84`	`_tokenSource.Cancel();`
`91`	`85`	`_tokenRegistration.Dispose();`
	`86`	`+ return _worker;`
`92`	`87`	`}`
`93`	`88`	`}`
`94`	`89`	`}`
Original file line number	Diff line number	Diff line change
`@@ -279,7 +279,7 @@ public void Close(ShutdownEventArgs reason, bool abort)`
`279`	`279`	`{`
`280`	`280`	`try`
`281`	`281`	`{`
`282`		`- _delegate.Close(reason, abort);`
	`282`	`+ _delegate.Close(reason, abort).GetAwaiter().GetResult();;`
`283`	`283`	`}`
`284`	`284`	`finally`
`285`	`285`	`{`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`using System;`
`2`	`2`	`using System.Collections.Generic;`
`3`		`-`
	`3`	`+using System.Threading.Tasks;`
`4`	`4`	`using RabbitMQ.Client.Events;`
`5`	`5`
`6`	`6`	`namespace RabbitMQ.Client.Impl`
`@@ -22,9 +22,9 @@ public void Quiesce()`
`22`	`22`	`IsShutdown = true;`
`23`	`23`	`}`
`24`	`24`
`25`		`- public void Shutdown(IModel model)`
	`25`	`+ public Task Shutdown(IModel model)`
`26`	`26`	`{`
`27`		`- _workService.StopWork(model);`
	`27`	`+ return _workService.StopWorkAsync(model);`
`28`	`28`	`}`
`29`	`29`
`30`	`30`	`public bool IsShutdown`
Original file line number	Diff line number	Diff line change
`@@ -21,20 +21,27 @@ private WorkPool StartNewWorkPool(IModel model)`
`21`	`21`	`return newWorkPool;`
`22`	`22`	`}`
`23`	`23`
`24`		`- public void StopWork(IModel model)`
	`24`	`+ public void StopWork()`
`25`	`25`	`{`
`26`		`- if (_workPools.TryRemove(model, out WorkPool workPool))`
	`26`	`+ foreach (IModel model in _workPools.Keys)`
`27`	`27`	`{`
`28`		`- workPool.Stop();`
	`28`	`+ StopWork(model);`
`29`	`29`	`}`
`30`	`30`	`}`
`31`	`31`
`32`		`- public void StopWork()`
	`32`	`+ public void StopWork(IModel model)`
`33`	`33`	`{`
`34`		`- foreach (IModel model in _workPools.Keys)`
	`34`	`+ StopWorkAsync(model).GetAwaiter().GetResult();`
	`35`	`+ }`
	`36`	`+`
	`37`	`+ internal Task StopWorkAsync(IModel model)`
	`38`	`+ {`
	`39`	`+ if (_workPools.TryRemove(model, out WorkPool workPool))`
`35`	`40`	`{`
`36`		`- StopWork(model);`
	`41`	`+ return workPool.Stop();`
`37`	`42`	`}`
	`43`	`+`
	`44`	`+ return Task.CompletedTask;`
`38`	`45`	`}`
`39`	`46`
`40`	`47`	`class WorkPool`
`@@ -77,7 +84,7 @@ async Task Loop()`
`77`	`84`	`// Swallowing the task cancellation exception for the semaphore in case we are stopping.`
`78`	`85`	`}`
`79`	`86`
`80`		`- while (_tokenSource.IsCancellationRequested == false && _actions.TryDequeue(out Action action))`
	`87`	`+ while (_actions.TryDequeue(out Action action))`
`81`	`88`	`{`
`82`	`89`	`try`
`83`	`90`	`{`
`@@ -91,10 +98,11 @@ async Task Loop()`
`91`	`98`	`}`
`92`	`99`	`}`
`93`	`100`
`94`		`- public void Stop()`
	`101`	`+ public Task Stop()`
`95`	`102`	`{`
`96`	`103`	`_tokenSource.Cancel();`
`97`	`104`	`_tokenRegistration.Dispose();`
	`105`	`+ return _worker;`
`98`	`106`	`}`
`99`	`107`	`}`
`100`	`108`	`}`
Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,7 @@`
`39`	`39`	`//---------------------------------------------------------------------------`
`40`	`40`
`41`	`41`	`using System;`
	`42`	`+using System.Threading.Tasks;`
`42`	`43`
`43`	`44`	`namespace RabbitMQ.Client.Impl`
`44`	`45`	`{`
`@@ -69,6 +70,6 @@ void HandleModelShutdown(IBasicConsumer consumer,`
`69`	`70`
`70`	`71`	`void Quiesce();`
`71`	`72`
`72`		`- void Shutdown(IModel model);`
	`73`	`+ Task Shutdown(IModel model);`
`73`	`74`	`}`
`74`	`75`	`}`
Original file line number	Diff line number	Diff line change
`@@ -45,7 +45,7 @@`
`45`	`45`	`using System.Runtime.CompilerServices;`
`46`	`46`	`using System.Text;`
`47`	`47`	`using System.Threading;`
`48`		`-`
	`48`	`+using System.Threading.Tasks;`
`49`	`49`	`using RabbitMQ.Client.Events;`
`50`	`50`	`using RabbitMQ.Client.Exceptions;`
`51`	`51`	`using RabbitMQ.Client.Framing;`
`@@ -184,14 +184,14 @@ public bool IsOpen`
`184`	`184`
`185`	`185`	`public ISession Session { get; private set; }`
`186`	`186`
`187`		`- public void Close(ushort replyCode, string replyText, bool abort)`
	`187`	`+ public Task Close(ushort replyCode, string replyText, bool abort)`
`188`	`188`	`{`
`189`		`- Close(new ShutdownEventArgs(ShutdownInitiator.Application,`
	`189`	`+ return Close(new ShutdownEventArgs(ShutdownInitiator.Application,`
`190`	`190`	`replyCode, replyText),`
`191`	`191`	`abort);`
`192`	`192`	`}`
`193`	`193`
`194`		`- public void Close(ShutdownEventArgs reason, bool abort)`
	`194`	`+ public async Task Close(ShutdownEventArgs reason, bool abort)`
`195`	`195`	`{`
`196`	`196`	`var k = new ShutdownContinuation();`
`197`	`197`	`ModelShutdown += k.OnConnectionShutdown;`
`@@ -204,7 +204,7 @@ public void Close(ShutdownEventArgs reason, bool abort)`
`204`	`204`	`_Private_ChannelClose(reason.ReplyCode, reason.ReplyText, 0, 0);`
`205`	`205`	`}`
`206`	`206`	`k.Wait(TimeSpan.FromMilliseconds(10000));`
`207`		`- ConsumerDispatcher.Shutdown(this);`
	`207`	`+ await ConsumerDispatcher.Shutdown(this).ConfigureAwait(false);`
`208`	`208`	`}`
`209`	`209`	`catch (AlreadyClosedException)`
`210`	`210`	`{`
`@@ -508,7 +508,7 @@ public void OnSessionShutdown(object sender, ShutdownEventArgs reason)`
`508`	`508`	`SetCloseReason(reason);`
`509`	`509`	`OnModelShutdown(reason);`
`510`	`510`	`BroadcastShutdownToConsumers(m_consumers, reason);`
`511`		`- ConsumerDispatcher.Shutdown(this);`
	`511`	`+ ConsumerDispatcher.Shutdown(this).GetAwaiter().GetResult();;`
`512`	`512`	`}`
`513`	`513`
`514`	`514`	`protected void BroadcastShutdownToConsumers(IDictionary<string, IBasicConsumer> cs, ShutdownEventArgs reason)`
`@@ -1394,7 +1394,7 @@ public void WaitForConfirmsOrDie(TimeSpan timeout)`
`1394`	`1394`	`Close(new ShutdownEventArgs(ShutdownInitiator.Application,`
`1395`	`1395`	`Constants.ReplySuccess,`
`1396`	`1396`	`"Nacks Received", new IOException("nack received")),`
`1397`		`- false);`
	`1397`	`+ false).GetAwaiter().GetResult();`
`1398`	`1398`	`throw new IOException("Nacks Received");`
`1399`	`1399`	`}`
`1400`	`1400`	`if (timedOut)`
`@@ -1403,7 +1403,7 @@ public void WaitForConfirmsOrDie(TimeSpan timeout)`
`1403`	`1403`	`Constants.ReplySuccess,`
`1404`	`1404`	`"Timed out waiting for acks",`
`1405`	`1405`	`new IOException("timed out waiting for acks")),`
`1406`		`- false);`
	`1406`	`+ false).GetAwaiter().GetResult();`
`1407`	`1407`	`throw new IOException("Timed out waiting for acks");`
`1408`	`1408`	`}`
`1409`	`1409`	`}`