Unify on IModel

danielmarbach · danielmarbach · commit 2a9990134411 · 2020-06-02T12:46:24.000+02:00
diff --git a/projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs b/projects/RabbitMQ.Client/client/impl/AsyncConsumerWorkService.cs
@@ -11,7 +11,7 @@ internal sealed class AsyncConsumerWorkService : ConsumerWorkService
         private readonly ConcurrentDictionary<IModel, WorkPool> _workPools = new ConcurrentDictionary<IModel, WorkPool>();
         private readonly Func<IModel, WorkPool> _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
@@ -24,7 +24,7 @@ public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work
 
         private static WorkPool StartNewWorkPool(IModel model)
         {
-            var newWorkPool = new WorkPool(model as ModelBase);
+            var newWorkPool = new WorkPool(model);
             newWorkPool.Start();
             return newWorkPool;
         }
@@ -42,10 +42,10 @@ public Task Stop(IModel model)
         class WorkPool
         {
             readonly Channel<Work> _channel;
-            readonly ModelBase _model;
+            readonly IModel _model;
             private Task _worker;
 
-            public WorkPool(ModelBase model)
+            public WorkPool(IModel model)
             {
                 _model = model;
                 _channel = Channel.CreateUnbounded<Work>(new UnboundedChannelOptions { SingleReader = true, SingleWriter = false, AllowSynchronousContinuations = false });
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
@@ -12,7 +12,7 @@ protected Work(IBasicConsumer consumer)
             _asyncConsumer = (IAsyncBasicConsumer)consumer;
         }
 
-        public async Task Execute(ModelBase model)
+        public async Task Execute(IModel model)
         {
             try
             {
@@ -25,6 +25,6 @@ public async Task Execute(ModelBase model)
             }
         }
 
-        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
`@@ -11,7 +11,7 @@ internal sealed class AsyncConsumerWorkService : ConsumerWorkService`
`11`	`11`	`private readonly ConcurrentDictionary<IModel, WorkPool> _workPools = new ConcurrentDictionary<IModel, WorkPool>();`
`12`	`12`	`private readonly Func<IModel, WorkPool> _startNewWorkPoolFunc = model => StartNewWorkPool(model);`
`13`	`13`
`14`		`- public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work`
	`14`	`+ public void Schedule<TWork>(IModel model, TWork work) where TWork : Work`
`15`	`15`	`{`
`16`	`16`	`/*`
`17`	`17`	`* rabbitmq/rabbitmq-dotnet-client#841`
`@@ -24,7 +24,7 @@ public void Schedule<TWork>(ModelBase model, TWork work) where TWork : Work`
`24`	`24`
`25`	`25`	`private static WorkPool StartNewWorkPool(IModel model)`
`26`	`26`	`{`
`27`		`- var newWorkPool = new WorkPool(model as ModelBase);`
	`27`	`+ var newWorkPool = new WorkPool(model);`
`28`	`28`	`newWorkPool.Start();`
`29`	`29`	`return newWorkPool;`
`30`	`30`	`}`
`@@ -42,10 +42,10 @@ public Task Stop(IModel model)`
`42`	`42`	`class WorkPool`
`43`	`43`	`{`
`44`	`44`	`readonly Channel<Work> _channel;`
`45`		`- readonly ModelBase _model;`
	`45`	`+ readonly IModel _model;`
`46`	`46`	`private Task _worker;`
`47`	`47`
`48`		`- public WorkPool(ModelBase model)`
	`48`	`+ public WorkPool(IModel model)`
`49`	`49`	`{`
`50`	`50`	`_model = model;`
`51`	`51`	`_channel = Channel.CreateUnbounded<Work>(new UnboundedChannelOptions { SingleReader = true, SingleWriter = false, AllowSynchronousContinuations = false });`
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
`@@ -12,7 +12,7 @@ protected Work(IBasicConsumer consumer)`
`12`	`12`	`_asyncConsumer = (IAsyncBasicConsumer)consumer;`
`13`	`13`	`}`
`14`	`14`
`15`		`- public async Task Execute(ModelBase model)`
	`15`	`+ public async Task Execute(IModel model)`
`16`	`16`	`{`
`17`	`17`	`try`
`18`	`18`	`{`
`@@ -25,6 +25,6 @@ public async Task Execute(ModelBase model)`
`25`	`25`	`}`
`26`	`26`	`}`
`27`	`27`
`28`		`- protected abstract Task Execute(ModelBase model, IAsyncBasicConsumer consumer);`
	`28`	`+ protected abstract Task Execute(IModel model, IAsyncBasicConsumer consumer);`
`29`	`29`	`}`
`30`	`30`	`}`