* Ignore OperationCanceledException in consumer dispatchers if cancellation is requested

lukebakken · lukebakken · commit 0f4dfbb5722e · 2024-05-09T10:55:53.000-07:00
diff --git a/projects/RabbitMQ.Client/client/impl/ConsumerDispatching/AsyncConsumerDispatcher.cs b/projects/RabbitMQ.Client/client/impl/ConsumerDispatching/AsyncConsumerDispatcher.cs
@@ -16,39 +16,49 @@ internal AsyncConsumerDispatcher(ChannelBase channel, int concurrency)
 
         protected override async Task ProcessChannelAsync(CancellationToken token)
         {
-            while (await _reader.WaitToReadAsync(token).ConfigureAwait(false))
+            try
             {
-                while (_reader.TryRead(out WorkStruct work))
+                while (await _reader.WaitToReadAsync(token).ConfigureAwait(false))
                 {
-                    using (work)
+                    while (_reader.TryRead(out WorkStruct work))
                     {
-                        try
+                        using (work)
                         {
-                            Task task = work.WorkType switch
+                            try
                             {
-                                WorkType.Deliver => work.AsyncConsumer.HandleBasicDeliver(
-                                    work.ConsumerTag, work.DeliveryTag, work.Redelivered,
-                                    work.Exchange, work.RoutingKey, work.BasicProperties, work.Body.Memory),
+                                Task task = work.WorkType switch
+                                {
+                                    WorkType.Deliver => work.AsyncConsumer.HandleBasicDeliver(
+                                        work.ConsumerTag, work.DeliveryTag, work.Redelivered,
+                                        work.Exchange, work.RoutingKey, work.BasicProperties, work.Body.Memory),
 
-                                WorkType.Cancel => work.AsyncConsumer.HandleBasicCancel(work.ConsumerTag),
+                                    WorkType.Cancel => work.AsyncConsumer.HandleBasicCancel(work.ConsumerTag),
 
-                                WorkType.CancelOk => work.AsyncConsumer.HandleBasicCancelOk(work.ConsumerTag),
+                                    WorkType.CancelOk => work.AsyncConsumer.HandleBasicCancelOk(work.ConsumerTag),
 
-                                WorkType.ConsumeOk => work.AsyncConsumer.HandleBasicConsumeOk(work.ConsumerTag),
+                                    WorkType.ConsumeOk => work.AsyncConsumer.HandleBasicConsumeOk(work.ConsumerTag),
 
-                                WorkType.Shutdown => work.AsyncConsumer.HandleChannelShutdown(_channel, work.Reason),
+                                    WorkType.Shutdown => work.AsyncConsumer.HandleChannelShutdown(_channel, work.Reason),
 
-                                _ => Task.CompletedTask
-                            };
-                            await task.ConfigureAwait(false);
-                        }
-                        catch (Exception e)
-                        {
-                            _channel.OnCallbackException(CallbackExceptionEventArgs.Build(e, work.WorkType.ToString(), work.Consumer));
+                                    _ => Task.CompletedTask
+                                };
+                                await task.ConfigureAwait(false);
+                            }
+                            catch (Exception e)
+                            {
+                                _channel.OnCallbackException(CallbackExceptionEventArgs.Build(e, work.WorkType.ToString(), work.Consumer));
+                            }
                         }
                     }
                 }
             }
+            catch (OperationCanceledException)
+            {
+                if (false == token.IsCancellationRequested)
+                {
+                    throw;
+                }
+            }
         }
     }
 }
diff --git a/projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcher.cs b/projects/RabbitMQ.Client/client/impl/ConsumerDispatching/ConsumerDispatcher.cs
@@ -16,45 +16,55 @@ internal ConsumerDispatcher(ChannelBase channel, int concurrency)
 
         protected override async Task ProcessChannelAsync(CancellationToken token)
         {
-            while (await _reader.WaitToReadAsync(token).ConfigureAwait(false))
+            try
             {
-                while (_reader.TryRead(out var work))
+                while (await _reader.WaitToReadAsync(token).ConfigureAwait(false))
                 {
-                    using (work)
+                    while (_reader.TryRead(out WorkStruct work))
                     {
-                        try
+                        using (work)
                         {
-                            IBasicConsumer consumer = work.Consumer;
-                            string? consumerTag = work.ConsumerTag;
-                            switch (work.WorkType)
+                            try
                             {
-                                case WorkType.Deliver:
-                                    await consumer.HandleBasicDeliverAsync(
-                                        consumerTag, work.DeliveryTag, work.Redelivered,
-                                        work.Exchange, work.RoutingKey, work.BasicProperties, work.Body.Memory)
-                                        .ConfigureAwait(false);
-                                    break;
-                                case WorkType.Cancel:
-                                    consumer.HandleBasicCancel(consumerTag);
-                                    break;
-                                case WorkType.CancelOk:
-                                    consumer.HandleBasicCancelOk(consumerTag);
-                                    break;
-                                case WorkType.ConsumeOk:
-                                    consumer.HandleBasicConsumeOk(consumerTag);
-                                    break;
-                                case WorkType.Shutdown:
-                                    consumer.HandleChannelShutdown(_channel, work.Reason);
-                                    break;
+                                IBasicConsumer consumer = work.Consumer;
+                                string? consumerTag = work.ConsumerTag;
+                                switch (work.WorkType)
+                                {
+                                    case WorkType.Deliver:
+                                        await consumer.HandleBasicDeliverAsync(
+                                            consumerTag, work.DeliveryTag, work.Redelivered,
+                                            work.Exchange, work.RoutingKey, work.BasicProperties, work.Body.Memory)
+                                            .ConfigureAwait(false);
+                                        break;
+                                    case WorkType.Cancel:
+                                        consumer.HandleBasicCancel(consumerTag);
+                                        break;
+                                    case WorkType.CancelOk:
+                                        consumer.HandleBasicCancelOk(consumerTag);
+                                        break;
+                                    case WorkType.ConsumeOk:
+                                        consumer.HandleBasicConsumeOk(consumerTag);
+                                        break;
+                                    case WorkType.Shutdown:
+                                        consumer.HandleChannelShutdown(_channel, work.Reason);
+                                        break;
+                                }
+                            }
+                            catch (Exception e)
+                            {
+                                _channel.OnCallbackException(CallbackExceptionEventArgs.Build(e, work.WorkType.ToString(), work.Consumer));
                             }
-                        }
-                        catch (Exception e)
-                        {
-                            _channel.OnCallbackException(CallbackExceptionEventArgs.Build(e, work.WorkType.ToString(), work.Consumer));
                         }
                     }
                 }
             }
+            catch (OperationCanceledException)
+            {
+                if (false == token.IsCancellationRequested)
+                {
+                    throw;
+                }
+            }
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -16,39 +16,49 @@ internal AsyncConsumerDispatcher(ChannelBase channel, int concurrency)`
`16`	`16`
`17`	`17`	`protected override async Task ProcessChannelAsync(CancellationToken token)`
`18`	`18`	`{`
`19`		`- while (await _reader.WaitToReadAsync(token).ConfigureAwait(false))`
	`19`	`+ try`
`20`	`20`	`{`
`21`		`- while (_reader.TryRead(out WorkStruct work))`
	`21`	`+ while (await _reader.WaitToReadAsync(token).ConfigureAwait(false))`
`22`	`22`	`{`
`23`		`- using (work)`
	`23`	`+ while (_reader.TryRead(out WorkStruct work))`
`24`	`24`	`{`
`25`		`- try`
	`25`	`+ using (work)`
`26`	`26`	`{`
`27`		`- Task task = work.WorkType switch`
	`27`	`+ try`
`28`	`28`	`{`
`29`		`- WorkType.Deliver => work.AsyncConsumer.HandleBasicDeliver(`
`30`		`- work.ConsumerTag, work.DeliveryTag, work.Redelivered,`
`31`		`- work.Exchange, work.RoutingKey, work.BasicProperties, work.Body.Memory),`
	`29`	`+ Task task = work.WorkType switch`
	`30`	`+ {`
	`31`	`+ WorkType.Deliver => work.AsyncConsumer.HandleBasicDeliver(`
	`32`	`+ work.ConsumerTag, work.DeliveryTag, work.Redelivered,`
	`33`	`+ work.Exchange, work.RoutingKey, work.BasicProperties, work.Body.Memory),`
`32`	`34`
`33`		`- WorkType.Cancel => work.AsyncConsumer.HandleBasicCancel(work.ConsumerTag),`
	`35`	`+ WorkType.Cancel => work.AsyncConsumer.HandleBasicCancel(work.ConsumerTag),`
`34`	`36`
`35`		`- WorkType.CancelOk => work.AsyncConsumer.HandleBasicCancelOk(work.ConsumerTag),`
	`37`	`+ WorkType.CancelOk => work.AsyncConsumer.HandleBasicCancelOk(work.ConsumerTag),`
`36`	`38`
`37`		`- WorkType.ConsumeOk => work.AsyncConsumer.HandleBasicConsumeOk(work.ConsumerTag),`
	`39`	`+ WorkType.ConsumeOk => work.AsyncConsumer.HandleBasicConsumeOk(work.ConsumerTag),`
`38`	`40`
`39`		`- WorkType.Shutdown => work.AsyncConsumer.HandleChannelShutdown(_channel, work.Reason),`
	`41`	`+ WorkType.Shutdown => work.AsyncConsumer.HandleChannelShutdown(_channel, work.Reason),`
`40`	`42`
`41`		`- _ => Task.CompletedTask`
`42`		`- };`
`43`		`- await task.ConfigureAwait(false);`
`44`		`- }`
`45`		`- catch (Exception e)`
`46`		`- {`
`47`		`- _channel.OnCallbackException(CallbackExceptionEventArgs.Build(e, work.WorkType.ToString(), work.Consumer));`
	`43`	`+ _ => Task.CompletedTask`
	`44`	`+ };`
	`45`	`+ await task.ConfigureAwait(false);`
	`46`	`+ }`
	`47`	`+ catch (Exception e)`
	`48`	`+ {`
	`49`	`+ _channel.OnCallbackException(CallbackExceptionEventArgs.Build(e, work.WorkType.ToString(), work.Consumer));`
	`50`	`+ }`
`48`	`51`	`}`
`49`	`52`	`}`
`50`	`53`	`}`
`51`	`54`	`}`
	`55`	`+ catch (OperationCanceledException)`
	`56`	`+ {`
	`57`	`+ if (false == token.IsCancellationRequested)`
	`58`	`+ {`
	`59`	`+ throw;`
	`60`	`+ }`
	`61`	`+ }`
`52`	`62`	`}`
`53`	`63`	`}`
`54`	`64`	`}`