Merge pull request #1324 from rabbitmq/rabbitmq-dotnet-client-1238

Fix consumer recovery with server-named queues

Author: lukebakken

.gitignore

@@ -50,7 +50,8 @@ build/
 [Oo]bj/
 *.lock.json
 
-APIApproval.Approve.received.txt
+projects/Unit/APIApproval.Approve.received.txt
+projects/Unit/APIApproval.Approve.*.received.txt
 
 # Visual Studio 2015 cache/options directory
 .vs/

projects/RabbitMQ.Client/client/api/IModel.cs

@@ -97,6 +97,14 @@ public interface IModel : IDisposable
         /// </summary>
         ulong NextPublishSeqNo { get; }
 
+        /// <summary>
+        /// The name of the last queue declared on this channel.
+        /// </summary>
+        /// <remarks>
+        /// https://www.rabbitmq.com/amqp-0-9-1-reference.html#domain.queue-name
+        /// </remarks>
+        string CurrentQueue { get; }
+
         /// <summary>
         /// Signalled when a Basic.Ack command arrives from the broker.
         /// </summary>

projects/RabbitMQ.Client/client/impl/AutorecoveringModel.cs

@@ -395,6 +395,19 @@ public ulong NextPublishSeqNo
             }
         }
 
+        public string CurrentQueue
+        {
+            get
+            {
+                if (_disposed)
+                {
+                    throw new ObjectDisposedException(GetType().FullName);
+                }
+
+                return _delegate.CurrentQueue;
+            }
+        }
+
         public void AutomaticallyRecover(AutorecoveringConnection conn, bool recoverConsumers)
         {
             if (_disposed)

projects/RabbitMQ.Client/client/impl/ModelBase.cs

@@ -170,6 +170,8 @@ public bool IsOpen
 
         public ulong NextPublishSeqNo { get; private set; }
 
+        public string CurrentQueue { get; private set; }
+
         public ISession Session { get; private set; }
 
         public Task Close(ushort replyCode, string replyText, bool abort)
@@ -1481,7 +1483,9 @@ private QueueDeclareOk QueueDeclare(string queue, bool passive, bool durable, bo
                 _Private_QueueDeclare(queue, passive, durable, exclusive, autoDelete, false, arguments);
                 k.GetReply(ContinuationTimeout);
             }
-            return k.m_result;
+            QueueDeclareOk result = k.m_result;
+            CurrentQueue = result.QueueName;
+            return result;
         }
 
 

projects/RabbitMQ.Client/client/impl/RecordedConsumer.cs

@@ -36,24 +36,33 @@ namespace RabbitMQ.Client.Impl
 {
     internal class RecordedConsumer : IRecordedConsumer
     {
+        private readonly AutorecoveringModel _model;
+
         public RecordedConsumer(AutorecoveringModel model, string queue, string consumerTag)
         {
-            if (model == null)
+            if (model is null)
             {
                 throw new ArgumentNullException(nameof(model));
             }
             else
             {
-                Model = model;
+                _model = model;
             }
 
-            if (string.IsNullOrEmpty(queue))
+            if (queue is null)
             {
-                throw new ArgumentNullException(nameof(consumerTag));
+                throw new ArgumentNullException(nameof(queue));
             }
             else
             {
-                Queue = queue;
+                if (queue == string.Empty)
+                {
+                    Queue = _model.CurrentQueue;
+                }
+                else
+                {
+                    Queue = queue;
+                }
             }
 
             if (string.IsNullOrEmpty(consumerTag))
@@ -66,20 +75,23 @@ public RecordedConsumer(AutorecoveringModel model, string queue, string consumer
             }
         }
 
-        public AutorecoveringModel Model { get; }
+        public AutorecoveringModel Model
+        {
+            get { return _model; }
+        }
+
+        public string Queue { get; set; }
+        public string ConsumerTag { get; set; }
         public IDictionary<string, object> Arguments { get; set; }
         public bool AutoAck { get; set; }
-        public IBasicConsumer Consumer { get; set; }
-        public string ConsumerTag { get; set; }
         public bool Exclusive { get; set; }
-        public string Queue { get; set; }
+        public IBasicConsumer Consumer { get; set; }
 
         public string Recover(IModel channelToUse)
         {
             ConsumerTag = channelToUse.BasicConsume(Queue, AutoAck,
                 ConsumerTag, false, Exclusive,
                 Arguments, Consumer);
-
             return ConsumerTag;
         }
 
@@ -101,29 +113,10 @@ public RecordedConsumer WithConsumer(IBasicConsumer value)
             return this;
         }
 
-        public RecordedConsumer WithConsumerTag(string value)
-        {
-            if (string.IsNullOrEmpty(value))
-            {
-                throw new System.ArgumentNullException(nameof(value));
-            }
-            else
-            {
-                ConsumerTag = value;
-            }
-            return this;
-        }
-
         public RecordedConsumer WithExclusive(bool value)
         {
             Exclusive = value;
             return this;
         }
-
-        public RecordedConsumer WithQueue(string value)
-        {
-            Queue = value ?? throw new System.ArgumentNullException(nameof(value));
-            return this;
-        }
     }
 }

projects/Unit/APIApproval.Approve.verified.txt

@@ -385,6 +385,7 @@ namespace RabbitMQ.Client
         int ChannelNumber { get; }
         RabbitMQ.Client.ShutdownEventArgs CloseReason { get; }
         System.TimeSpan ContinuationTimeout { get; set; }
+        string CurrentQueue { get; }
         RabbitMQ.Client.IBasicConsumer DefaultConsumer { get; set; }
         bool IsClosed { get; }
         bool IsOpen { get; }

projects/Unit/TestConnectionRecovery.cs

@@ -410,6 +410,38 @@ public void TestConsumerRecoveryOnClientNamedQueueWithOneRecovery()
             }
         }
 
+        [Test(Description = "https://github.com/rabbitmq/rabbitmq-dotnet-client/issues/1238")]
+        public void TestConsumerRecoveryWithServerNamedQueue()
+        {
+            using (AutorecoveringConnection c = CreateAutorecoveringConnection())
+            {
+                IModel m = c.CreateModel();
+                QueueDeclareOk queueDeclareResult = m.QueueDeclare(queue: string.Empty, durable: false, exclusive: true, autoDelete: true, arguments: null);
+                string qname = queueDeclareResult.QueueName;
+                Assert.False(string.IsNullOrEmpty(qname));
+
+                var cons = new EventingBasicConsumer(m);
+                m.BasicConsume(string.Empty, true, cons);
+                AssertConsumerCount(m, qname, 1);
+
+                bool queueNameBeforeIsEqual = false;
+                bool queueNameChangeAfterRecoveryCalled = false;
+                string qnameAfterRecovery = null;
+                c.QueueNameChangeAfterRecovery += (source, ea) =>
+                {
+                    queueNameChangeAfterRecoveryCalled = true;
+                    queueNameBeforeIsEqual = qname.Equals(ea.NameBefore);
+                    qnameAfterRecovery = ea.NameAfter;
+                };
+
+                CloseAndWaitForRecovery(c);
+
+                AssertConsumerCount(m, qnameAfterRecovery, 1);
+                Assert.True(queueNameChangeAfterRecoveryCalled);
+                Assert.True(queueNameBeforeIsEqual);
+            }
+        }
+
         [Test]
         public void TestConsumerRecoveryWithManyConsumers()
         {