Unify on IModel

rabbitmq · Jul 7, 2020 · 33b2a40 · 33b2a40
1 parent d0c136c
commit 33b2a40
Show file tree

Hide file tree

Showing 7 changed files with 48 additions and 23 deletions.
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);
     }
 }