Merge pull request #858 from danielmarbach/unify-onmodel

michaelklishin · web-flow · commit 16560f2a8097 · 2020-07-07T20:42:41.000+03:00
Unify on IModel
diff --git a/projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs b/projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs
@@ -16,7 +16,7 @@ public AsyncConsumerWorkService(int concurrency) : base(concurrency)
             _startNewWorkPoolFunc = model => StartNewWorkPool(model);
         }
 
-        public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work
+        public void Schedule<TWork>(IModel model, TWork work) where TWork : Work
         {
             /*
              * rabbitmq/rabbitmq-dotnet-client#841
@@ -29,7 +29,7 @@ public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work
 
         private WorkPool StartNewWorkPool(IModel model)
         {
-            var newWorkPool = new WorkPool(model as ModelBase, _concurrency);
+            var newWorkPool = new WorkPool(model, _concurrency);
             newWorkPool.Start();
             return newWorkPool;
         }
@@ -47,13 +47,13 @@ public Task Stop(IModel model)
         class WorkPool
         {
             readonly Channel<Work> _channel;
-            readonly ModelBase _model;
+            readonly IModel _model;
             private Task _worker;
             private readonly int _concurrency;
             private SemaphoreSlim _limiter;
             private CancellationTokenSource _tokenSource;
 
-            public WorkPool(ModelBase model, int concurrency)
+            public WorkPool(IModel model, int concurrency)
             {
                 _concurrency = concurrency;
                 _model = model;
@@ -125,7 +125,7 @@ async Task LoopWithConcurrency(CancellationToken cancellationToken)
                 }
             }
 
-            static async Task HandleConcurrent(Work work, ModelBase model, SemaphoreSlim limiter)
+            static async Task HandleConcurrent(Work work, IModel model, SemaphoreSlim limiter)
             {
                 try
                 {
diff --git a/projects/RabbitMQ.Client/client/impl/BasicCancel.cs b/projects/RabbitMQ.Client/client/impl/BasicCancel.cs
@@ -15,20 +15,25 @@ public BasicCancel(IBasicConsumer consumer, string consumerTag) : base(consumer)
             _consumerTag = consumerTag;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
                 await consumer.HandleBasicCancel(_consumerTag).ConfigureAwait(false);
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>
                 {
                     {"consumer", consumer},
                     {"context",  "HandleBasicCancel"}
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
         }
     }
diff --git a/projects/RabbitMQ.Client/client/impl/BasicCancelOk.cs b/projects/RabbitMQ.Client/client/impl/BasicCancelOk.cs
@@ -15,20 +15,25 @@ public BasicCancelOk(IBasicConsumer consumer, string consumerTag) : base(consume
             _consumerTag = consumerTag;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
                 await consumer.HandleBasicCancelOk(_consumerTag).ConfigureAwait(false);
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>()
                 {
                     {"consumer", consumer},
                     {"context",  "HandleBasicCancelOk"}
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
         }
     }
diff --git a/projects/RabbitMQ.Client/client/impl/BasicConsumeOk.cs b/projects/RabbitMQ.Client/client/impl/BasicConsumeOk.cs
@@ -15,20 +15,25 @@ public BasicConsumeOk(IBasicConsumer consumer, string consumerTag) : base(consum
             _consumerTag = consumerTag;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
                 await consumer.HandleBasicConsumeOk(_consumerTag).ConfigureAwait(false);
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>()
                 {
                     {"consumer", consumer},
                     {"context",  "HandleBasicConsumeOk"}
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
         }
     }
diff --git a/projects/RabbitMQ.Client/client/impl/BasicDeliver.cs b/projects/RabbitMQ.Client/client/impl/BasicDeliver.cs
@@ -18,12 +18,12 @@ sealed class BasicDeliver : Work
         readonly IBasicProperties _basicProperties;
         readonly ReadOnlyMemory<byte> _body;
 
-        public BasicDeliver(IBasicConsumer consumer, 
-            string consumerTag, 
-            ulong deliveryTag, 
-            bool redelivered, 
-            string exchange, 
-            string routingKey, 
+        public BasicDeliver(IBasicConsumer consumer,
+            string consumerTag,
+            ulong deliveryTag,
+            bool redelivered,
+            string exchange,
+            string routingKey,
             IBasicProperties basicProperties,
             ReadOnlyMemory<byte> body) : base(consumer)
         {
@@ -36,7 +36,7 @@ public BasicDeliver(IBasicConsumer consumer,
             _body = body;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
@@ -50,12 +50,17 @@ await consumer.HandleBasicDeliver(_consumerTag,
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>()
                 {
                     {"consumer", consumer},
                     {"context",  "HandleBasicDeliver"}
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
             finally
             {
diff --git a/projects/RabbitMQ.Client/client/impl/ModelShutdown.cs b/projects/RabbitMQ.Client/client/impl/ModelShutdown.cs
@@ -15,20 +15,25 @@ public ModelShutdown(IBasicConsumer consumer, ShutdownEventArgs reason) : base(c
             _reason = reason;
         }
 
-        protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)
+        protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)
         {
             try
             {
                 await consumer.HandleModelShutdown(model, _reason).ConfigureAwait(false);
             }
             catch (Exception e)
             {
+                if (!(model is ModelBase modelBase))
+                {
+                    return;
+                }
+
                 var details = new Dictionary<string, object>()
                 {
                     { "consumer", consumer },
                     { "context", "HandleModelShutdown" }
                 };
-                model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
+                modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
             }
         }
     }
diff --git a/projects/RabbitMQ.Client/client/impl/Work.cs b/projects/RabbitMQ.Client/client/impl/Work.cs
@@ -11,11 +11,11 @@ protected Work(IBasicConsumer consumer)
             _asyncConsumer = (IAsyncBasicConsumer)consumer;
         }
 
-        public Task Execute(ModelBase model)
+        public Task Execute(IModel model)
         {
             return Execute(model, _asyncConsumer);
         }
 
-        protected abstract Task Execute(ModelBase model, IAsyncBasicConsumer consumer);
+        protected abstract Task Execute(IModel model, IAsyncBasicConsumer consumer);
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,7 @@ public AsyncConsumerWorkService(int concurrency) : base(concurrency)`
`16`	`16`	`_startNewWorkPoolFunc = model => StartNewWorkPool(model);`
`17`	`17`	`}`
`18`	`18`
`19`		`- public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work`
	`19`	`+ public void Schedule<TWork>(IModel model, TWork work) where TWork : Work`
`20`	`20`	`{`
`21`	`21`	`/*`
`22`	`22`	`* rabbitmq/rabbitmq-dotnet-client#841`
`@@ -29,7 +29,7 @@ public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work`
`29`	`29`
`30`	`30`	`private WorkPool StartNewWorkPool(IModel model)`
`31`	`31`	`{`
`32`		`- var newWorkPool = new WorkPool(model as ModelBase, _concurrency);`
	`32`	`+ var newWorkPool = new WorkPool(model, _concurrency);`
`33`	`33`	`newWorkPool.Start();`
`34`	`34`	`return newWorkPool;`
`35`	`35`	`}`
`@@ -47,13 +47,13 @@ public Task Stop(IModel model)`
`47`	`47`	`class WorkPool`
`48`	`48`	`{`
`49`	`49`	`readonly Channel<Work> _channel;`
`50`		`- readonly ModelBase _model;`
	`50`	`+ readonly IModel _model;`
`51`	`51`	`private Task _worker;`
`52`	`52`	`private readonly int _concurrency;`
`53`	`53`	`private SemaphoreSlim _limiter;`
`54`	`54`	`private CancellationTokenSource _tokenSource;`
`55`	`55`
`56`		`- public WorkPool(ModelBase model, int concurrency)`
	`56`	`+ public WorkPool(IModel model, int concurrency)`
`57`	`57`	`{`
`58`	`58`	`_concurrency = concurrency;`
`59`	`59`	`_model = model;`
`@@ -125,7 +125,7 @@ async Task LoopWithConcurrency(CancellationToken cancellationToken)`
`125`	`125`	`}`
`126`	`126`	`}`
`127`	`127`
`128`		`- static async Task HandleConcurrent(Work work, ModelBase model, SemaphoreSlim limiter)`
	`128`	`+ static async Task HandleConcurrent(Work work, IModel model, SemaphoreSlim limiter)`
`129`	`129`	`{`
`130`	`130`	`try`
`131`	`131`	`{`
Original file line number	Diff line number	Diff line change
`@@ -15,20 +15,25 @@ public BasicCancel(IBasicConsumer consumer, string consumerTag) : base(consumer)`
`15`	`15`	`_consumerTag = consumerTag;`
`16`	`16`	`}`
`17`	`17`
`18`		`- protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)`
	`18`	`+ protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)`
`19`	`19`	`{`
`20`	`20`	`try`
`21`	`21`	`{`
`22`	`22`	`await consumer.HandleBasicCancel(_consumerTag).ConfigureAwait(false);`
`23`	`23`	`}`
`24`	`24`	`catch (Exception e)`
`25`	`25`	`{`
	`26`	`+ if (!(model is ModelBase modelBase))`
	`27`	`+ {`
	`28`	`+ return;`
	`29`	`+ }`
	`30`	`+`
`26`	`31`	`var details = new Dictionary<string, object>`
`27`	`32`	`{`
`28`	`33`	`{"consumer", consumer},`
`29`	`34`	`{"context", "HandleBasicCancel"}`
`30`	`35`	`};`
`31`		`- model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
	`36`	`+ modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
`32`	`37`	`}`
`33`	`38`	`}`
`34`	`39`	`}`
Original file line number	Diff line number	Diff line change
`@@ -15,20 +15,25 @@ public BasicCancelOk(IBasicConsumer consumer, string consumerTag) : base(consume`
`15`	`15`	`_consumerTag = consumerTag;`
`16`	`16`	`}`
`17`	`17`
`18`		`- protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)`
	`18`	`+ protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)`
`19`	`19`	`{`
`20`	`20`	`try`
`21`	`21`	`{`
`22`	`22`	`await consumer.HandleBasicCancelOk(_consumerTag).ConfigureAwait(false);`
`23`	`23`	`}`
`24`	`24`	`catch (Exception e)`
`25`	`25`	`{`
	`26`	`+ if (!(model is ModelBase modelBase))`
	`27`	`+ {`
	`28`	`+ return;`
	`29`	`+ }`
	`30`	`+`
`26`	`31`	`var details = new Dictionary<string, object>()`
`27`	`32`	`{`
`28`	`33`	`{"consumer", consumer},`
`29`	`34`	`{"context", "HandleBasicCancelOk"}`
`30`	`35`	`};`
`31`		`- model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
	`36`	`+ modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
`32`	`37`	`}`
`33`	`38`	`}`
`34`	`39`	`}`
Original file line number	Diff line number	Diff line change
`@@ -15,20 +15,25 @@ public BasicConsumeOk(IBasicConsumer consumer, string consumerTag) : base(consum`
`15`	`15`	`_consumerTag = consumerTag;`
`16`	`16`	`}`
`17`	`17`
`18`		`- protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)`
	`18`	`+ protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)`
`19`	`19`	`{`
`20`	`20`	`try`
`21`	`21`	`{`
`22`	`22`	`await consumer.HandleBasicConsumeOk(_consumerTag).ConfigureAwait(false);`
`23`	`23`	`}`
`24`	`24`	`catch (Exception e)`
`25`	`25`	`{`
	`26`	`+ if (!(model is ModelBase modelBase))`
	`27`	`+ {`
	`28`	`+ return;`
	`29`	`+ }`
	`30`	`+`
`26`	`31`	`var details = new Dictionary<string, object>()`
`27`	`32`	`{`
`28`	`33`	`{"consumer", consumer},`
`29`	`34`	`{"context", "HandleBasicConsumeOk"}`
`30`	`35`	`};`
`31`		`- model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
	`36`	`+ modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
`32`	`37`	`}`
`33`	`38`	`}`
`34`	`39`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,12 +18,12 @@ sealed class BasicDeliver : Work`
`18`	`18`	`readonly IBasicProperties _basicProperties;`
`19`	`19`	`readonly ReadOnlyMemory<byte> _body;`
`20`	`20`
`21`		`- public BasicDeliver(IBasicConsumer consumer,`
`22`		`- string consumerTag,`
`23`		`- ulong deliveryTag,`
`24`		`- bool redelivered,`
`25`		`- string exchange,`
`26`		`- string routingKey,`
	`21`	`+ public BasicDeliver(IBasicConsumer consumer,`
	`22`	`+ string consumerTag,`
	`23`	`+ ulong deliveryTag,`
	`24`	`+ bool redelivered,`
	`25`	`+ string exchange,`
	`26`	`+ string routingKey,`
`27`	`27`	`IBasicProperties basicProperties,`
`28`	`28`	`ReadOnlyMemory<byte> body) : base(consumer)`
`29`	`29`	`{`
`@@ -36,7 +36,7 @@ public BasicDeliver(IBasicConsumer consumer,`
`36`	`36`	`_body = body;`
`37`	`37`	`}`
`38`	`38`
`39`		`- protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)`
	`39`	`+ protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)`
`40`	`40`	`{`
`41`	`41`	`try`
`42`	`42`	`{`
`@@ -50,12 +50,17 @@ await consumer.HandleBasicDeliver(_consumerTag,`
`50`	`50`	`}`
`51`	`51`	`catch (Exception e)`
`52`	`52`	`{`
	`53`	`+ if (!(model is ModelBase modelBase))`
	`54`	`+ {`
	`55`	`+ return;`
	`56`	`+ }`
	`57`	`+`
`53`	`58`	`var details = new Dictionary<string, object>()`
`54`	`59`	`{`
`55`	`60`	`{"consumer", consumer},`
`56`	`61`	`{"context", "HandleBasicDeliver"}`
`57`	`62`	`};`
`58`		`- model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
	`63`	`+ modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
`59`	`64`	`}`
`60`	`65`	`finally`
`61`	`66`	`{`
Original file line number	Diff line number	Diff line change
`@@ -15,20 +15,25 @@ public ModelShutdown(IBasicConsumer consumer, ShutdownEventArgs reason) : base(c`
`15`	`15`	`_reason = reason;`
`16`	`16`	`}`
`17`	`17`
`18`		`- protected override async Task Execute(ModelBase model, IAsyncBasicConsumer consumer)`
	`18`	`+ protected override async Task Execute(IModel model, IAsyncBasicConsumer consumer)`
`19`	`19`	`{`
`20`	`20`	`try`
`21`	`21`	`{`
`22`	`22`	`await consumer.HandleModelShutdown(model, _reason).ConfigureAwait(false);`
`23`	`23`	`}`
`24`	`24`	`catch (Exception e)`
`25`	`25`	`{`
	`26`	`+ if (!(model is ModelBase modelBase))`
	`27`	`+ {`
	`28`	`+ return;`
	`29`	`+ }`
	`30`	`+`
`26`	`31`	`var details = new Dictionary<string, object>()`
`27`	`32`	`{`
`28`	`33`	`{ "consumer", consumer },`
`29`	`34`	`{ "context", "HandleModelShutdown" }`
`30`	`35`	`};`
`31`		`- model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
	`36`	`+ modelBase.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));`
`32`	`37`	`}`
`33`	`38`	`}`
`34`	`39`	`}`
Original file line number	Diff line number	Diff line change
`@@ -11,11 +11,11 @@ protected Work(IBasicConsumer consumer)`
`11`	`11`	`_asyncConsumer = (IAsyncBasicConsumer)consumer;`
`12`	`12`	`}`
`13`	`13`
`14`		`- public Task Execute(ModelBase model)`
	`14`	`+ public Task Execute(IModel model)`
`15`	`15`	`{`
`16`	`16`	`return Execute(model, _asyncConsumer);`
`17`	`17`	`}`
`18`	`18`
`19`		`- protected abstract Task Execute(ModelBase model, IAsyncBasicConsumer consumer);`
	`19`	`+ protected abstract Task Execute(IModel model, IAsyncBasicConsumer consumer);`
`20`	`20`	`}`
`21`	`21`	`}`