From bda44c9861bbda7de9e9ba6502ca1a0e8d78aa51 Mon Sep 17 00:00:00 2001 From: Luke Bakken Date: Mon, 20 Mar 2023 15:57:19 -0700 Subject: [PATCH] Delete recorded consumers when autorecovering model is disposed Fixes #1302 Update dependencies Ensure RecordedConsumer has a ConsumerTag --- .../client/impl/AutorecoveringConnection.cs | 24 ++++++++-- .../client/impl/AutorecoveringModel.cs | 17 ++++--- .../client/impl/RecordedConsumer.cs | 44 ++++++++++++++++--- 3 files changed, 70 insertions(+), 15 deletions(-) diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs index 0f9b5c90f5..48fe1d1f1b 100644 --- a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs +++ b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs @@ -612,13 +612,23 @@ public void RecordBinding(RecordedBinding rb) } } - public void RecordConsumer(string name, RecordedConsumer c) + public void RecordConsumer(string consumerTag, RecordedConsumer recordedConsumer) { + if (string.IsNullOrEmpty(consumerTag)) + { + throw new ArgumentNullException(nameof(consumerTag)); + } + + if (recordedConsumer is null) + { + throw new ArgumentNullException(nameof(recordedConsumer)); + } + lock (_recordedEntitiesLock) { - if (!_recordedConsumers.ContainsKey(name)) + if (!_recordedConsumers.ContainsKey(consumerTag)) { - _recordedConsumers.Add(name, c); + _recordedConsumers.Add(consumerTag, recordedConsumer); } } } @@ -651,6 +661,14 @@ public override string ToString() public void UnregisterModel(AutorecoveringModel model) { + lock (_recordedEntitiesLock) + { + foreach (string ct in model.ConsumerTags) + { + DeleteRecordedConsumer(ct); + } + } + lock (_models) { _models.Remove(model); diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs index c4b3f0796c..5eb91599ef 100644 --- a/projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs +++ b/projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs @@ -43,6 +43,7 @@ internal sealed class AutorecoveringModel : IFullModel, IRecoverable private readonly object _eventLock = new object(); private AutorecoveringConnection _connection; private RecoveryAwareModel _delegate; + private List _recordedConsumerTags = new List(); private EventHandler _recordedBasicAckEventHandlers; private EventHandler _recordedBasicNackEventHandlers; @@ -91,6 +92,8 @@ public TimeSpan ContinuationTimeout } } + public IEnumerable ConsumerTags => _recordedConsumerTags; + public AutorecoveringModel(AutorecoveringConnection conn, RecoveryAwareModel _delegate) { _connection = conn; @@ -476,7 +479,7 @@ public void Close(ShutdownEventArgs reason, bool abort) try { - _delegate.Close(reason, abort).GetAwaiter().GetResult();; + _delegate.Close(reason, abort).GetAwaiter().GetResult(); } finally { @@ -510,6 +513,8 @@ private void Dispose(bool disposing) { Abort(); + _recordedConsumerTags.Clear(); + _recordedConsumerTags = null; _connection = null; _delegate = null; _recordedBasicAckEventHandlers = null; @@ -1184,16 +1189,16 @@ public string BasicConsume( throw new ObjectDisposedException(GetType().FullName); } - string result = _delegate.BasicConsume(queue, autoAck, consumerTag, noLocal, + string resultConsumerTag = _delegate.BasicConsume(queue, autoAck, consumerTag, noLocal, exclusive, arguments, consumer); - RecordedConsumer rc = new RecordedConsumer(this, queue). - WithConsumerTag(result). + RecordedConsumer rc = new RecordedConsumer(this, queue, resultConsumerTag). WithConsumer(consumer). WithExclusive(exclusive). WithAutoAck(autoAck). WithArguments(arguments); - _connection.RecordConsumer(result, rc); - return result; + _connection.RecordConsumer(resultConsumerTag, rc); + _recordedConsumerTags.Add(resultConsumerTag); + return resultConsumerTag; } public BasicGetResult BasicGet(string queue, diff --git a/projects/RabbitMQ.Client/client/impl/RecordedConsumer.cs b/projects/RabbitMQ.Client/client/impl/RecordedConsumer.cs index e2c802df8e..ba8d0674e3 100644 --- a/projects/RabbitMQ.Client/client/impl/RecordedConsumer.cs +++ b/projects/RabbitMQ.Client/client/impl/RecordedConsumer.cs @@ -29,16 +29,41 @@ // Copyright (c) 2007-2020 VMware, Inc. All rights reserved. //--------------------------------------------------------------------------- +using System; using System.Collections.Generic; namespace RabbitMQ.Client.Impl { internal class RecordedConsumer : IRecordedConsumer { - public RecordedConsumer(AutorecoveringModel model, string queue) + public RecordedConsumer(AutorecoveringModel model, string queue, string consumerTag) { - Model = model; - Queue = queue; + if (model == null) + { + throw new ArgumentNullException(nameof(model)); + } + else + { + Model = model ?? throw new ArgumentNullException(nameof(model)); + } + + if (string.IsNullOrEmpty(queue)) + { + throw new ArgumentNullException(nameof(consumerTag)); + } + else + { + Queue = queue; + } + + if (string.IsNullOrEmpty(consumerTag)) + { + throw new ArgumentNullException(nameof(consumerTag)); + } + else + { + ConsumerTag = consumerTag; + } } public AutorecoveringModel Model { get; } @@ -72,13 +97,20 @@ public RecordedConsumer WithAutoAck(bool value) public RecordedConsumer WithConsumer(IBasicConsumer value) { - Consumer = value; + Consumer = value ?? throw new System.ArgumentNullException(nameof(value)); return this; } public RecordedConsumer WithConsumerTag(string value) { - ConsumerTag = value; + if (string.IsNullOrEmpty(value)) + { + throw new System.ArgumentNullException(nameof(value)); + } + else + { + ConsumerTag = value; + } return this; } @@ -90,7 +122,7 @@ public RecordedConsumer WithExclusive(bool value) public RecordedConsumer WithQueue(string value) { - Queue = value; + Queue = value ?? throw new System.ArgumentNullException(nameof(value)); return this; } }