Code cleanups. Removing unneeded dependency.

Stefán J. Sigurðarson · lukebakken · commit b5d507facf54 · 2020-01-17T07:31:11.000-08:00
diff --git a/projects/client/RabbitMQ.Client/RabbitMQ.Client.csproj b/projects/client/RabbitMQ.Client/RabbitMQ.Client.csproj
@@ -50,9 +50,6 @@
   </PropertyGroup>
 
   <ItemGroup Condition="$('TargetFramework') == 'net462'">
-    <PackageReference Include="System.Threading.Channels" Version="4.7.0" />
-  </ItemGroup>
-
   <ItemGroup Condition=" '$(TargetFramework)' == 'net452' ">
     <PackageReference Include="Microsoft.NETFramework.ReferenceAssemblies" Version="1.0.0" PrivateAssets="All" />
   </ItemGroup>
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerWorkService.cs b/projects/client/RabbitMQ.Client/src/client/impl/AsyncConsumerWorkService.cs
@@ -1,21 +1,16 @@
 ﻿using RabbitMQ.Client.Impl;
 
-using System;
 using System.Collections.Concurrent;
 using System.Threading;
-#if CORECLR
-using System.Threading.Channels;
-#endif
 using System.Threading.Tasks;
 
 namespace RabbitMQ.Client
 {
     internal sealed class AsyncConsumerWorkService : ConsumerWorkService
     {
-        readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();
+        private readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();
 
-        public void Schedule<TWork>(ModelBase model, TWork work)
-            where TWork : Work
+        public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work
         {
             workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(work);
         }
@@ -29,78 +24,32 @@ private WorkPool StartNewWorkPool(IModel model)
 
         public void Stop(IModel model)
         {
-            WorkPool workPool;
-            if (workPools.TryRemove(model, out workPool))
+            if (workPools.TryRemove(model, out WorkPool workPool))
             {
                 workPool.Stop();
             }
         }
 
         public void Stop()
         {
-            foreach (var model in workPools.Keys)
+            foreach (IModel model in workPools.Keys)
             {
                 Stop(model);
             }
         }
 
-#if CORECLR
-        class WorkPool
-        {
-            readonly Channel<Work> actionChannel = Channel.CreateUnbounded<Work>(new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });
-            readonly ModelBase model;
-            Task worker;
-
-            public WorkPool(ModelBase model)
-            {
-                this.model = model;
-            }
-
-            public void Start()
-            {
-                worker = Task.Run(Loop);
-            }
-
-            public void Enqueue(Work action)
-            {
-                actionChannel.Writer.TryWrite(action);
-            }
-
-            async Task Loop()
-            {
-                while (await actionChannel.Reader.WaitToReadAsync().ConfigureAwait(false))
-                {
-                    while (actionChannel.Reader.TryRead(out Work action))
-                    {
-                        await action.Execute(model).ConfigureAwait(false);
-                    }
-                }
-            }
-
-            public void Stop()
-            {
-                actionChannel.Writer.Complete();
-                worker.Wait();
-            }
-        }
-    }
-}
-#else
         class WorkPool
         {
             readonly ConcurrentQueue<Work> workQueue;
-            readonly TimeSpan waitTime;
             readonly CancellationTokenSource tokenSource;
             readonly ModelBase model;
-            TaskCompletionSource<bool> messageArrived;
+            readonly SemaphoreSlim semaphore = new SemaphoreSlim(0);
             private Task task;
 
             public WorkPool(ModelBase model)
             {
                 this.model = model;
                 workQueue = new ConcurrentQueue<Work>();
-                messageArrived = new TaskCompletionSource<bool>();
-                waitTime = TimeSpan.FromMilliseconds(100);
                 tokenSource = new CancellationTokenSource();
             }
 
@@ -112,30 +61,26 @@ public void Start()
             public void Enqueue(Work work)
             {
                 workQueue.Enqueue(work);
-                messageArrived.TrySetResult(true);
+                semaphore.Release();
             }
 
             async Task Loop()
             {
                 while (tokenSource.IsCancellationRequested == false)
                 {
-                    while (workQueue.TryDequeue(out Work work))
+                    await semaphore.WaitAsync().ConfigureAwait(false);
+                    if (workQueue.TryDequeue(out Work work))
                     {
                         await work.Execute(model).ConfigureAwait(false);
                     }
-
-                    await Task.WhenAny(Task.Delay(waitTime, tokenSource.Token), messageArrived.Task).ConfigureAwait(false);
-                    messageArrived.TrySetResult(true);
-                    messageArrived = new TaskCompletionSource<bool>();
                 }
             }
 
-            public Task Stop()
+            public void Stop()
             {
                 tokenSource.Cancel();
-                return task;
+                task.Wait();
             }
         }
     }
 }
-#endif
diff --git a/projects/client/RabbitMQ.Client/src/client/impl/ConsumerWorkService.cs b/projects/client/RabbitMQ.Client/src/client/impl/ConsumerWorkService.cs
@@ -1,21 +1,16 @@
 ﻿using System;
 using System.Collections.Concurrent;
 using System.Threading;
-#if CORECLR
-using System.Threading.Channels;
-#endif
 using System.Threading.Tasks;
 
 namespace RabbitMQ.Client
 {
     public class ConsumerWorkService
     {
-        readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();
+        private readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();
 
         public void AddWork(IModel model, Action fn)
         {
-            // two step approach is taken, as TryGetValue does not aquire locks
-            // if this fails, GetOrAdd is called, which takes a lock
             workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(fn);
         }
 
@@ -28,80 +23,30 @@ private WorkPool StartNewWorkPool(IModel model)
 
         public void StopWork(IModel model)
         {
-            WorkPool workPool;
-            if (workPools.TryRemove(model, out workPool))
+            if (workPools.TryRemove(model, out WorkPool workPool))
             {
                 workPool.Stop();
             }
         }
 
         public void StopWork()
         {
-            foreach (var model in workPools.Keys)
+            foreach (IModel model in workPools.Keys)
             {
                 StopWork(model);
             }
         }
 
-#if CORECLR
-        class WorkPool
-        {
-            readonly Channel<Action> actionChannel = Channel.CreateUnbounded<Action>(new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });
-            Task worker;
-
-            public WorkPool(IModel model)
-            {
-            }
-
-            public void Start()
-            {
-                worker = Task.Run(Loop);
-            }
-
-            public void Enqueue(Action action)
-            {
-                actionChannel.Writer.TryWrite(action);
-            }
-
-            async Task Loop()
-            {
-                while (await actionChannel.Reader.WaitToReadAsync().ConfigureAwait(false))
-                {
-                    while (actionChannel.Reader.TryRead(out Action action))
-                    {
-                        try
-                        {
-                            action();
-                        }
-                        catch (Exception)
-                        {
-                        }
-                    }
-                }
-            }
-
-            public void Stop()
-            {
-                actionChannel.Writer.Complete();
-                worker.Wait();
-            }
-        }
-    }
-}
-#else
         class WorkPool
         {
             readonly ConcurrentQueue<Action> actions;
-            readonly AutoResetEvent messageArrived;
-            readonly TimeSpan waitTime;
+            readonly SemaphoreSlim semaphore = new SemaphoreSlim(0);
             readonly CancellationTokenSource tokenSource;
-            Task worker;
+            private Task worker;
 
             public WorkPool(IModel model)
             {
                 actions = new ConcurrentQueue<Action>();
-                messageArrived = new AutoResetEvent(false);
-                waitTime = TimeSpan.FromMilliseconds(100);
                 tokenSource = new CancellationTokenSource();
             }
 
@@ -113,14 +58,16 @@ public void Start()
             public void Enqueue(Action action)
             {
                 actions.Enqueue(action);
-                messageArrived.Set();
+                semaphore.Release();
             }
 
-            void Loop()
+            async Task Loop()
             {
                 while (tokenSource.IsCancellationRequested == false)
                 {
-                    while (actions.TryDequeue(out Action action))
+                    await semaphore.WaitAsync().ConfigureAwait(false);
+
+                    if (actions.TryDequeue(out Action action))
                     {
                         try
                         {
@@ -130,16 +77,14 @@ void Loop()
                         {
                         }
                     }
-
-                    messageArrived.WaitOne(waitTime);
                 }
             }
 
             public void Stop()
             {
                 tokenSource.Cancel();
+                worker.Wait();
             }
         }
     }
 }
-#endif

Original file line number	Diff line number	Diff line change
`@@ -1,21 +1,16 @@`
`1`	`1`	`using RabbitMQ.Client.Impl;`
`2`	`2`
`3`		`-using System;`
`4`	`3`	`using System.Collections.Concurrent;`
`5`	`4`	`using System.Threading;`
`6`		`-#if CORECLR`
`7`		`-using System.Threading.Channels;`
`8`		`-#endif`
`9`	`5`	`using System.Threading.Tasks;`
`10`	`6`
`11`	`7`	`namespace RabbitMQ.Client`
`12`	`8`	`{`
`13`	`9`	`internal sealed class AsyncConsumerWorkService : ConsumerWorkService`
`14`	`10`	`{`
`15`		`- readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();`
	`11`	`+ private readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();`
`16`	`12`
`17`		`- public void Schedule<TWork>(ModelBase model, TWork work)`
`18`		`- where TWork : Work`
	`13`	`+ public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work`
`19`	`14`	`{`
`20`	`15`	`workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(work);`
`21`	`16`	`}`
`@@ -29,78 +24,32 @@ private WorkPool StartNewWorkPool(IModel model)`
`29`	`24`
`30`	`25`	`public void Stop(IModel model)`
`31`	`26`	`{`
`32`		`- WorkPool workPool;`
`33`		`- if (workPools.TryRemove(model, out workPool))`
	`27`	`+ if (workPools.TryRemove(model, out WorkPool workPool))`
`34`	`28`	`{`
`35`	`29`	`workPool.Stop();`
`36`	`30`	`}`
`37`	`31`	`}`
`38`	`32`
`39`	`33`	`public void Stop()`
`40`	`34`	`{`
`41`		`- foreach (var model in workPools.Keys)`
	`35`	`+ foreach (IModel model in workPools.Keys)`
`42`	`36`	`{`
`43`	`37`	`Stop(model);`
`44`	`38`	`}`
`45`	`39`	`}`
`46`	`40`
`47`		`-#if CORECLR`
`48`		`- class WorkPool`
`49`		`- {`
`50`		`- readonly Channel<Work> actionChannel = Channel.CreateUnbounded<Work>(new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });`
`51`		`- readonly ModelBase model;`
`52`		`- Task worker;`
`53`		`-`
`54`		`- public WorkPool(ModelBase model)`
`55`		`- {`
`56`		`- this.model = model;`
`57`		`- }`
`58`		`-`
`59`		`- public void Start()`
`60`		`- {`
`61`		`- worker = Task.Run(Loop);`
`62`		`- }`
`63`		`-`
`64`		`- public void Enqueue(Work action)`
`65`		`- {`
`66`		`- actionChannel.Writer.TryWrite(action);`
`67`		`- }`
`68`		`-`
`69`		`- async Task Loop()`
`70`		`- {`
`71`		`- while (await actionChannel.Reader.WaitToReadAsync().ConfigureAwait(false))`
`72`		`- {`
`73`		`- while (actionChannel.Reader.TryRead(out Work action))`
`74`		`- {`
`75`		`- await action.Execute(model).ConfigureAwait(false);`
`76`		`- }`
`77`		`- }`
`78`		`- }`
`79`		`-`
`80`		`- public void Stop()`
`81`		`- {`
`82`		`- actionChannel.Writer.Complete();`
`83`		`- worker.Wait();`
`84`		`- }`
`85`		`- }`
`86`		`- }`
`87`		`-}`
`88`		`-#else`
`89`	`41`	`class WorkPool`
`90`	`42`	`{`
`91`	`43`	`readonly ConcurrentQueue<Work> workQueue;`
`92`		`- readonly TimeSpan waitTime;`
`93`	`44`	`readonly CancellationTokenSource tokenSource;`
`94`	`45`	`readonly ModelBase model;`
`95`		`- TaskCompletionSource<bool> messageArrived;`
	`46`	`+ readonly SemaphoreSlim semaphore = new SemaphoreSlim(0);`
`96`	`47`	`private Task task;`
`97`	`48`
`98`	`49`	`public WorkPool(ModelBase model)`
`99`	`50`	`{`
`100`	`51`	`this.model = model;`
`101`	`52`	`workQueue = new ConcurrentQueue<Work>();`
`102`		`- messageArrived = new TaskCompletionSource<bool>();`
`103`		`- waitTime = TimeSpan.FromMilliseconds(100);`
`104`	`53`	`tokenSource = new CancellationTokenSource();`
`105`	`54`	`}`
`106`	`55`
`@@ -112,30 +61,26 @@ public void Start()`
`112`	`61`	`public void Enqueue(Work work)`
`113`	`62`	`{`
`114`	`63`	`workQueue.Enqueue(work);`
`115`		`- messageArrived.TrySetResult(true);`
	`64`	`+ semaphore.Release();`
`116`	`65`	`}`
`117`	`66`
`118`	`67`	`async Task Loop()`
`119`	`68`	`{`
`120`	`69`	`while (tokenSource.IsCancellationRequested == false)`
`121`	`70`	`{`
`122`		`- while (workQueue.TryDequeue(out Work work))`
	`71`	`+ await semaphore.WaitAsync().ConfigureAwait(false);`
	`72`	`+ if (workQueue.TryDequeue(out Work work))`
`123`	`73`	`{`
`124`	`74`	`await work.Execute(model).ConfigureAwait(false);`
`125`	`75`	`}`
`126`		`-`
`127`		`- await Task.WhenAny(Task.Delay(waitTime, tokenSource.Token), messageArrived.Task).ConfigureAwait(false);`
`128`		`- messageArrived.TrySetResult(true);`
`129`		`- messageArrived = new TaskCompletionSource<bool>();`
`130`	`76`	`}`
`131`	`77`	`}`
`132`	`78`
`133`		`- public Task Stop()`
	`79`	`+ public void Stop()`
`134`	`80`	`{`
`135`	`81`	`tokenSource.Cancel();`
`136`		`- return task;`
	`82`	`+ task.Wait();`
`137`	`83`	`}`
`138`	`84`	`}`
`139`	`85`	`}`
`140`	`86`	`}`
`141`		`-#endif`
Original file line number	Diff line number	Diff line change
`@@ -1,21 +1,16 @@`
`1`	`1`	`using System;`
`2`	`2`	`using System.Collections.Concurrent;`
`3`	`3`	`using System.Threading;`
`4`		`-#if CORECLR`
`5`		`-using System.Threading.Channels;`
`6`		`-#endif`
`7`	`4`	`using System.Threading.Tasks;`
`8`	`5`
`9`	`6`	`namespace RabbitMQ.Client`
`10`	`7`	`{`
`11`	`8`	`public class ConsumerWorkService`
`12`	`9`	`{`
`13`		`- readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();`
	`10`	`+ private readonly ConcurrentDictionary<IModel, WorkPool> workPools = new ConcurrentDictionary<IModel, WorkPool>();`
`14`	`11`
`15`	`12`	`public void AddWork(IModel model, Action fn)`
`16`	`13`	`{`
`17`		`- // two step approach is taken, as TryGetValue does not aquire locks`
`18`		`- // if this fails, GetOrAdd is called, which takes a lock`
`19`	`14`	`workPools.GetOrAdd(model, StartNewWorkPool).Enqueue(fn);`
`20`	`15`	`}`
`21`	`16`
`@@ -28,80 +23,30 @@ private WorkPool StartNewWorkPool(IModel model)`
`28`	`23`
`29`	`24`	`public void StopWork(IModel model)`
`30`	`25`	`{`
`31`		`- WorkPool workPool;`
`32`		`- if (workPools.TryRemove(model, out workPool))`
	`26`	`+ if (workPools.TryRemove(model, out WorkPool workPool))`
`33`	`27`	`{`
`34`	`28`	`workPool.Stop();`
`35`	`29`	`}`
`36`	`30`	`}`
`37`	`31`
`38`	`32`	`public void StopWork()`
`39`	`33`	`{`
`40`		`- foreach (var model in workPools.Keys)`
	`34`	`+ foreach (IModel model in workPools.Keys)`
`41`	`35`	`{`
`42`	`36`	`StopWork(model);`
`43`	`37`	`}`
`44`	`38`	`}`
`45`	`39`
`46`		`-#if CORECLR`
`47`		`- class WorkPool`
`48`		`- {`
`49`		`- readonly Channel<Action> actionChannel = Channel.CreateUnbounded<Action>(new UnboundedChannelOptions { SingleReader = true, SingleWriter = false });`
`50`		`- Task worker;`
`51`		`-`
`52`		`- public WorkPool(IModel model)`
`53`		`- {`
`54`		`- }`
`55`		`-`
`56`		`- public void Start()`
`57`		`- {`
`58`		`- worker = Task.Run(Loop);`
`59`		`- }`
`60`		`-`
`61`		`- public void Enqueue(Action action)`
`62`		`- {`
`63`		`- actionChannel.Writer.TryWrite(action);`
`64`		`- }`
`65`		`-`
`66`		`- async Task Loop()`
`67`		`- {`
`68`		`- while (await actionChannel.Reader.WaitToReadAsync().ConfigureAwait(false))`
`69`		`- {`
`70`		`- while (actionChannel.Reader.TryRead(out Action action))`
`71`		`- {`
`72`		`- try`
`73`		`- {`
`74`		`- action();`
`75`		`- }`
`76`		`- catch (Exception)`
`77`		`- {`
`78`		`- }`
`79`		`- }`
`80`		`- }`
`81`		`- }`
`82`		`-`
`83`		`- public void Stop()`
`84`		`- {`
`85`		`- actionChannel.Writer.Complete();`
`86`		`- worker.Wait();`
`87`		`- }`
`88`		`- }`
`89`		`- }`
`90`		`-}`
`91`		`-#else`
`92`	`40`	`class WorkPool`
`93`	`41`	`{`
`94`	`42`	`readonly ConcurrentQueue<Action> actions;`
`95`		`- readonly AutoResetEvent messageArrived;`
`96`		`- readonly TimeSpan waitTime;`
	`43`	`+ readonly SemaphoreSlim semaphore = new SemaphoreSlim(0);`
`97`	`44`	`readonly CancellationTokenSource tokenSource;`
`98`		`- Task worker;`
	`45`	`+ private Task worker;`
`99`	`46`
`100`	`47`	`public WorkPool(IModel model)`
`101`	`48`	`{`
`102`	`49`	`actions = new ConcurrentQueue<Action>();`
`103`		`- messageArrived = new AutoResetEvent(false);`
`104`		`- waitTime = TimeSpan.FromMilliseconds(100);`
`105`	`50`	`tokenSource = new CancellationTokenSource();`
`106`	`51`	`}`
`107`	`52`
`@@ -113,14 +58,16 @@ public void Start()`
`113`	`58`	`public void Enqueue(Action action)`
`114`	`59`	`{`
`115`	`60`	`actions.Enqueue(action);`
`116`		`- messageArrived.Set();`
	`61`	`+ semaphore.Release();`
`117`	`62`	`}`
`118`	`63`
`119`		`- void Loop()`
	`64`	`+ async Task Loop()`
`120`	`65`	`{`
`121`	`66`	`while (tokenSource.IsCancellationRequested == false)`
`122`	`67`	`{`
`123`		`- while (actions.TryDequeue(out Action action))`
	`68`	`+ await semaphore.WaitAsync().ConfigureAwait(false);`
	`69`	`+`
	`70`	`+ if (actions.TryDequeue(out Action action))`
`124`	`71`	`{`
`125`	`72`	`try`
`126`	`73`	`{`
`@@ -130,16 +77,14 @@ void Loop()`
`130`	`77`	`{`
`131`	`78`	`}`
`132`	`79`	`}`
`133`		`-`
`134`		`- messageArrived.WaitOne(waitTime);`
`135`	`80`	`}`
`136`	`81`	`}`
`137`	`82`
`138`	`83`	`public void Stop()`
`139`	`84`	`{`
`140`	`85`	`tokenSource.Cancel();`
	`86`	`+ worker.Wait();`
`141`	`87`	`}`
`142`	`88`	`}`
`143`	`89`	`}`
`144`	`90`	`}`
`145`		`-#endif`