direct reply to restart

Gsantomaggio · Gsantomaggio · commit 16cda8b402c3 · 2025-11-06T12:01:55.000+01:00
Signed-off-by: Gabriele Santomaggio &lt;G.santomaggio@gmail.com&gt;
diff --git a/RabbitMQ.AMQP.Client/IConsumer.cs b/RabbitMQ.AMQP.Client/IConsumer.cs
@@ -39,9 +39,11 @@ public interface IConsumer : ILifeCycle
         long UnsettledMessageCount { get; }
 
         /// <summary>
-        /// Returns queue address the consumer is consuming from.
+        /// Returns queue name the consumer is consuming from.
+        /// The Queue name is usually configured by the user via the <see cref="IConsumerBuilder"/>,
+        /// but can also be generated by the client the special direct-reply-to queue.
         /// </summary>
-        string? QueueAddress { get; }
+        string Queue { get; }
     }
 
     public interface IContext
diff --git a/RabbitMQ.AMQP.Client/IRequester.cs b/RabbitMQ.AMQP.Client/IRequester.cs
@@ -105,6 +105,6 @@ public interface IRequester : ILifeCycle
         /// - The server uses this address to send the reply message. with direct-reply-to
         /// </summary>
         /// <returns></returns>
-        public string ReplyToQueueAddress();
+        public string GetReplyToQueue();
     }
 }
diff --git a/RabbitMQ.AMQP.Client/Impl/AddressBuilder.cs b/RabbitMQ.AMQP.Client/Impl/AddressBuilder.cs
@@ -21,36 +21,21 @@ public T Exchange(IExchangeSpecification exchangeSpec)
         public T Exchange(string? exchangeName)
         {
             _exchange = exchangeName;
-            if (_owner == null)
-            {
-                throw new InvalidOperationException("Owner is null");
-            }
-
-            return _owner;
+            return _owner ?? throw new InvalidOperationException("Owner is null");
         }
 
         public T Queue(IQueueSpecification queueSpec) => Queue(queueSpec.QueueName);
 
         public T Queue(string? queueName)
         {
             _queue = queueName;
-            if (_owner == null)
-            {
-                throw new InvalidOperationException("Owner is null");
-            }
-
-            return _owner;
+            return _owner ?? throw new InvalidOperationException("Owner is null");
         }
 
         public T Key(string? key)
         {
             _key = key;
-            if (_owner == null)
-            {
-                throw new InvalidOperationException("Owner is null");
-            }
-
-            return _owner;
+            return _owner ?? throw new InvalidOperationException("Owner is null");
         }
 
         public string Address()
@@ -85,12 +70,19 @@ public string Address()
                 return "";
             }
 
-            if (string.IsNullOrEmpty(_queue))
-            {
-                throw new InvalidAddressException("Queue must be set");
-            }
+            return string.IsNullOrEmpty(_queue)
+                ? throw new InvalidAddressException("Queue must be set")
+                : $"/{Consts.Queues}/{Utils.EncodePathSegment(_queue)}";
+        }
 
-            return $"/{Consts.Queues}/{Utils.EncodePathSegment(_queue)}";
+        public string DecodeQueuePathSegment(string path)
+        {
+            string? v = Utils.DecodePathSegment(path);
+            return v == null
+                ? throw new InvalidAddressException("Invalid path segment")
+                :
+                // remove the /queues prefix to the path
+                v.Substring($"/{Consts.Queues}/".Length);
         }
     }
 
@@ -127,6 +119,7 @@ public IMessage Build()
     public class RequesterAddressBuilder : DefaultAddressBuilder<IRequesterAddressBuilder>, IRequesterAddressBuilder
     {
         readonly AmqpRequesterBuilder _builder;
+
         public RequesterAddressBuilder(AmqpRequesterBuilder builder)
         {
             _builder = builder;
diff --git a/RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs b/RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs
@@ -256,12 +256,18 @@ ref Unsafe.As<PauseStatus, int>(ref _pauseStatus),
         /// </summary>
         public long UnsettledMessageCount => _unsettledMessageCounter.Get();
 
-        public string? QueueAddress
+        public string Queue
         {
             get
             {
-                string? x = _attach?.Source is not Source source ? null : source.Address;
-                return x;
+                string? sourceAddress = _attach?.Source is not Source source ? null : source.Address;
+                if (sourceAddress is null)
+                {
+                    throw new InvalidOperationException(
+                        "_attach.Source.Address is null");
+                }
+
+                return AddressBuilderHelper.AddressBuilder().DecodeQueuePathSegment(sourceAddress);
             }
         }
 
@@ -318,7 +324,7 @@ await _receiverLink.CloseAsync(TimeSpan.FromSeconds(ConsumerDefaults.CloseTimeou
 
         public override string ToString()
         {
-            return $"Consumer{{Address='{QueueAddress}', " +
+            return $"Consumer{{Address='{Queue}', " +
                    $"id={_id}, " +
                    $"Connection='{_amqpConnection}', " +
                    $"State='{State}'}}";
diff --git a/RabbitMQ.AMQP.Client/Impl/AmqpPublisher.cs b/RabbitMQ.AMQP.Client/Impl/AmqpPublisher.cs
@@ -138,9 +138,9 @@ public Task<PublishResult> PublishAsync(IMessage message, CancellationToken canc
             void OutcomeCallback(ILink sender, Message inMessage, Outcome outcome, object state)
             {
                 // Note: sometimes `inMessage` is null 🤔
-                Debug.Assert(Object.ReferenceEquals(this, state));
+                Debug.Assert(ReferenceEquals(this, state));
 
-                if (false == Object.ReferenceEquals(_senderLink, sender))
+                if (!ReferenceEquals(_senderLink, sender))
                 {
                     // TODO log this case?
                 }
diff --git a/RabbitMQ.AMQP.Client/Impl/AmqpRequester.cs b/RabbitMQ.AMQP.Client/Impl/AmqpRequester.cs
@@ -14,7 +14,6 @@ public class RequesterConfiguration
         public AmqpConnection Connection { get; set; } = null!;
         public string ReplyToQueue { get; set; } = "";
 
-        public string RequestAddress { get; set; } = "";
         public TimeSpan Timeout { get; set; } = TimeSpan.FromSeconds(10);
 
         public Func<object>? CorrelationIdSupplier { get; set; } = null;
@@ -79,7 +78,6 @@ public IRequesterBuilder Timeout(TimeSpan timeout)
 
         public async Task<IRequester> BuildAsync()
         {
-            _configuration.RequestAddress = _addressBuilder.Address();
             _configuration.Connection = _connection;
             var rpcClient = new AmqpRequester(_configuration);
             await rpcClient.OpenAsync().ConfigureAwait(false);
@@ -135,7 +133,7 @@ private IMessage RequestPostProcess(IMessage request, object correlationId)
                 return _configuration.RequestPostProcessor(request, correlationId);
             }
 
-            string s = ReplyToQueueAddress();
+            string s = GetReplyToQueue();
             return request.ReplyTo(s)
                 .MessageId(correlationId);
         }
@@ -230,7 +228,8 @@ public async Task<IMessage> PublishAsync(IMessage message, CancellationToken can
                 if (_publisher != null)
                 {
                     PublishResult pr = await _publisher.PublishAsync(
-                        message.To(_configuration.RequestAddress), cancellationToken).ConfigureAwait(false);
+                        message.To(AddressBuilderHelper.AddressBuilder().Queue(GetReplyToQueue()).Address()),
+                        cancellationToken).ConfigureAwait(false);
 
                     if (pr.Outcome.State != OutcomeState.Accepted)
                     {
@@ -250,14 +249,14 @@ await _pendingRequests[correlationId].Task.WaitAsync(_configuration.Timeout)
             }
         }
 
-        public string ReplyToQueueAddress()
+        public string GetReplyToQueue()
         {
             if (_consumer == null)
             {
                 throw new InvalidOperationException("Requester is not opened");
             }
 
-            return _consumer.QueueAddress ??
+            return _consumer.Queue ??
                    throw new InvalidOperationException("ReplyToQueueAddress is not available");
         }
     }
diff --git a/RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt b/RabbitMQ.AMQP.Client/PublicAPI.Unshipped.txt
@@ -171,7 +171,7 @@ RabbitMQ.AMQP.Client.IConnection.RequesterBuilder() -> RabbitMQ.AMQP.Client.IReq
 RabbitMQ.AMQP.Client.IConnection.ResponderBuilder() -> RabbitMQ.AMQP.Client.IResponderBuilder!
 RabbitMQ.AMQP.Client.IConsumer
 RabbitMQ.AMQP.Client.IConsumer.Pause() -> void
-RabbitMQ.AMQP.Client.IConsumer.QueueAddress.get -> string?
+RabbitMQ.AMQP.Client.IConsumer.Queue.get -> string!
 RabbitMQ.AMQP.Client.IConsumer.Unpause() -> void
 RabbitMQ.AMQP.Client.IConsumer.UnsettledMessageCount.get -> long
 RabbitMQ.AMQP.Client.IConsumerBuilder
@@ -366,7 +366,7 @@ RabbitMQ.AMQP.Client.Impl.AmqpConnection.RequesterBuilder() -> RabbitMQ.AMQP.Cli
 RabbitMQ.AMQP.Client.Impl.AmqpConnection.ResponderBuilder() -> RabbitMQ.AMQP.Client.IResponderBuilder!
 RabbitMQ.AMQP.Client.Impl.AmqpConsumer
 RabbitMQ.AMQP.Client.Impl.AmqpConsumer.Pause() -> void
-RabbitMQ.AMQP.Client.Impl.AmqpConsumer.QueueAddress.get -> string?
+RabbitMQ.AMQP.Client.Impl.AmqpConsumer.Queue.get -> string!
 RabbitMQ.AMQP.Client.Impl.AmqpConsumer.Unpause() -> void
 RabbitMQ.AMQP.Client.Impl.AmqpConsumer.UnsettledMessageCount.get -> long
 RabbitMQ.AMQP.Client.Impl.AmqpConsumerBuilder
@@ -498,8 +498,8 @@ RabbitMQ.AMQP.Client.Impl.AmqpQuorumSpecification.QuorumInitialGroupSize(int siz
 RabbitMQ.AMQP.Client.Impl.AmqpQuorumSpecification.QuorumTargetGroupSize(int size) -> RabbitMQ.AMQP.Client.IQuorumQueueSpecification!
 RabbitMQ.AMQP.Client.Impl.AmqpRequester
 RabbitMQ.AMQP.Client.Impl.AmqpRequester.AmqpRequester(RabbitMQ.AMQP.Client.Impl.RequesterConfiguration! configuration) -> void
+RabbitMQ.AMQP.Client.Impl.AmqpRequester.GetReplyToQueue() -> string!
 RabbitMQ.AMQP.Client.Impl.AmqpRequester.PublishAsync(RabbitMQ.AMQP.Client.IMessage! message, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.AMQP.Client.IMessage!>!
-RabbitMQ.AMQP.Client.Impl.AmqpRequester.ReplyToQueueAddress() -> string!
 RabbitMQ.AMQP.Client.Impl.AmqpRequesterBuilder
 RabbitMQ.AMQP.Client.Impl.AmqpRequesterBuilder.AmqpRequesterBuilder(RabbitMQ.AMQP.Client.Impl.AmqpConnection! connection) -> void
 RabbitMQ.AMQP.Client.Impl.AmqpRequesterBuilder.BuildAsync() -> System.Threading.Tasks.Task<RabbitMQ.AMQP.Client.IRequester!>!
@@ -549,6 +549,7 @@ RabbitMQ.AMQP.Client.Impl.ConsumerBuilderStreamOptions
 RabbitMQ.AMQP.Client.Impl.ConsumerBuilderStreamOptions.ConsumerBuilderStreamOptions(RabbitMQ.AMQP.Client.IConsumerBuilder! consumerBuilder, Amqp.Types.Map! filters) -> void
 RabbitMQ.AMQP.Client.Impl.DefaultAddressBuilder<T>
 RabbitMQ.AMQP.Client.Impl.DefaultAddressBuilder<T>.Address() -> string!
+RabbitMQ.AMQP.Client.Impl.DefaultAddressBuilder<T>.DecodeQueuePathSegment(string! path) -> string!
 RabbitMQ.AMQP.Client.Impl.DefaultAddressBuilder<T>.DefaultAddressBuilder() -> void
 RabbitMQ.AMQP.Client.Impl.DefaultAddressBuilder<T>.Exchange(RabbitMQ.AMQP.Client.IExchangeSpecification! exchangeSpec) -> T
 RabbitMQ.AMQP.Client.Impl.DefaultAddressBuilder<T>.Exchange(string? exchangeName) -> T
@@ -586,8 +587,6 @@ RabbitMQ.AMQP.Client.Impl.RequesterConfiguration.CorrelationIdSupplier.get -> Sy
 RabbitMQ.AMQP.Client.Impl.RequesterConfiguration.CorrelationIdSupplier.set -> void
 RabbitMQ.AMQP.Client.Impl.RequesterConfiguration.ReplyToQueue.get -> string!
 RabbitMQ.AMQP.Client.Impl.RequesterConfiguration.ReplyToQueue.set -> void
-RabbitMQ.AMQP.Client.Impl.RequesterConfiguration.RequestAddress.get -> string!
-RabbitMQ.AMQP.Client.Impl.RequesterConfiguration.RequestAddress.set -> void
 RabbitMQ.AMQP.Client.Impl.RequesterConfiguration.RequesterConfiguration() -> void
 RabbitMQ.AMQP.Client.Impl.RequesterConfiguration.RequestPostProcessor.get -> System.Func<RabbitMQ.AMQP.Client.IMessage!, object!, RabbitMQ.AMQP.Client.IMessage!>?
 RabbitMQ.AMQP.Client.Impl.RequesterConfiguration.RequestPostProcessor.set -> void
@@ -694,8 +693,8 @@ RabbitMQ.AMQP.Client.IRecoveryConfiguration.IsActivated() -> bool
 RabbitMQ.AMQP.Client.IRecoveryConfiguration.IsTopologyActive() -> bool
 RabbitMQ.AMQP.Client.IRecoveryConfiguration.Topology(bool activated) -> RabbitMQ.AMQP.Client.IRecoveryConfiguration!
 RabbitMQ.AMQP.Client.IRequester
+RabbitMQ.AMQP.Client.IRequester.GetReplyToQueue() -> string!
 RabbitMQ.AMQP.Client.IRequester.PublishAsync(RabbitMQ.AMQP.Client.IMessage! message, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.AMQP.Client.IMessage!>!
-RabbitMQ.AMQP.Client.IRequester.ReplyToQueueAddress() -> string!
 RabbitMQ.AMQP.Client.IRequesterAddressBuilder
 RabbitMQ.AMQP.Client.IRequesterAddressBuilder.Requester() -> RabbitMQ.AMQP.Client.IRequesterBuilder!
 RabbitMQ.AMQP.Client.IRequesterBuilder
diff --git a/RabbitMQ.AMQP.Client/Utils.cs b/RabbitMQ.AMQP.Client/Utils.cs
@@ -191,6 +191,11 @@ internal static Attach CreateAttach(string? address,
         {
             return PercentCodec.EncodePathSegment(url);
         }
+        
+        internal static string? DecodePathSegment(string url)
+        {
+            return PercentCodec.DecodePathSegment(url);
+        }
 
         internal static string EncodeHttpParameter(string url)
         {
@@ -374,7 +379,66 @@ static PercentCodec()
             s_unreserved['_'] = true;
             s_unreserved['~'] = true;
         }
+        
+        private static int HexValue(char ch)
+        {
+            return ch switch
+            {
+                >= '0' and <= '9' => ch - '0',
+                >= 'A' and <= 'F' => ch - 'A' + 10,
+                >= 'a' and <= 'f' => ch - 'a' + 10,
+                _ => -1
+            };
+        }
+
 
+        internal static string? DecodePathSegment(string? segment)
+        {
+            if (segment == null)
+            {
+                return null;
+            }
+        
+            var bytes = new List<byte>(segment.Length);
+            for (int i = 0; i < segment.Length; i++)
+            {
+                char c = segment[i];
+                if (c == '%')
+                {
+                    if (i + 2 >= segment.Length)
+                    {
+                        throw new FormatException("Invalid percent-encoding: incomplete escape sequence.");
+                    }
+        
+                    int hi = HexValue(segment[i + 1]);
+                    int lo = HexValue(segment[i + 2]);
+                    if (hi < 0 || lo < 0)
+                    {
+                        throw new FormatException($"Invalid percent-encoding: '{segment.Substring(i, 3)}'.");
+                    }
+        
+                    bytes.Add((byte)((hi << 4) | lo));
+                    i += 2;
+                }
+                else
+                {
+                    // Append UTF-8 encoding of the character (handles non-ASCII chars)
+                    if (c <= 0x7F)
+                    {
+                        bytes.Add((byte)c);
+                    }
+                    else
+                    {
+                        bytes.AddRange(Encoding.UTF8.GetBytes(new[] { c }));
+                    }
+                }
+            }
+        
+            return Encoding.UTF8.GetString(bytes.ToArray());
+        }
+        
+        
+        
         internal static string? EncodePathSegment(string? segment)
         {
             if (segment == null)
diff --git a/Tests/AddressBuilderTests.cs b/Tests/AddressBuilderTests.cs
@@ -0,0 +1,30 @@
+// This source code is dual-licensed under the Apache License, version 2.0,
+// and the Mozilla Public License, version 2.0.
+// Copyright (c) 2017-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+
+using RabbitMQ.AMQP.Client.Impl;
+using Xunit;
+
+namespace Tests
+{
+    public class AddressBuilderTests
+    {
+        [Theory]
+        [InlineData("myQueue", "/queues/myQueue")]
+        [InlineData("queue/with/slash", "/queues/queue%2Fwith%2Fslash")]
+        [InlineData("queue with spaces", "/queues/queue%20with%20spaces")]
+        [InlineData("queue+with+plus", "/queues/queue%2Bwith%2Bplus")]
+        [InlineData("queue?with?question", "/queues/queue%3Fwith%3Fquestion")]
+        [InlineData("特殊字符", "/queues/%E7%89%B9%E6%AE%8A%E5%AD%97%E7%AC%A6")]
+        [InlineData("emoji😊queue", "/queues/emoji%F0%9F%98%8Aqueue")]
+        [InlineData("!@#$%^&*()", "/queues/%21%40%23%24%25%5E%26%2A%28%29")]
+        public void AddressBuilder_EncodeAndDecode(string queue, string queuePath)
+        {
+            AddressBuilder addressBuilder = new();
+            string fullAddress = addressBuilder.Queue(queue).Address();
+            Assert.Equal(queuePath, fullAddress);
+            string decodedQueue = addressBuilder.DecodeQueuePathSegment(fullAddress);
+            Assert.Equal(queue, decodedQueue);
+        }
+    }
+}
diff --git a/Tests/Consumer/BasicConsumerTests.cs b/Tests/Consumer/BasicConsumerTests.cs
@@ -371,17 +371,13 @@ public async Task ConsumerShouldThrowWhenQueueDoesNotExist()
 
         IConsumerBuilder consumerBuilder = _connection.ConsumerBuilder()
             .Queue(doesNotExist)
-            .MessageHandler((context, message) =>
-            {
-                return Task.CompletedTask;
-            }
-        );
+            .MessageHandler((context, message) => Task.CompletedTask);
 
         // TODO these are timeout exceptions under the hood, compare
         // with the Java client
         ConsumerException ex = await Assert.ThrowsAsync<ConsumerException>(
             () => consumerBuilder.BuildAndStartAsync());
-        Assert.Contains(doesNotExist, ex.Message);
+        Assert.Contains("amqp:not-found", ex.Message);
     }
 
     [Fact]
diff --git a/Tests/DirectReply/DirectReplyTests.cs b/Tests/DirectReply/DirectReplyTests.cs
diff --git a/Tests/RequesterResponser/RecoveryRPCTests.cs b/Tests/RequesterResponser/RecoveryRPCTests.cs
diff --git a/Tests/RequesterResponser/ResponderTests.cs b/Tests/RequesterResponser/ResponderTests.cs
diff --git a/docs/Examples/Rpc/Program.cs b/docs/Examples/Rpc/Program.cs

Original file line number	Diff line number	Diff line change
`@@ -105,6 +105,6 @@ public interface IRequester : ILifeCycle`
`105`	`105`	`/// - The server uses this address to send the reply message. with direct-reply-to`
`106`	`106`	`/// </summary>`
`107`	`107`	`/// <returns></returns>`
`108`		`- public string ReplyToQueueAddress();`
	`108`	`+ public string GetReplyToQueue();`
`109`	`109`	`}`
`110`	`110`	`}`
Original file line number	Diff line number	Diff line change
`@@ -256,12 +256,18 @@ ref Unsafe.As<PauseStatus, int>(ref _pauseStatus),`
`256`	`256`	`/// </summary>`
`257`	`257`	`public long UnsettledMessageCount => _unsettledMessageCounter.Get();`
`258`	`258`
`259`		`- public string? QueueAddress`
	`259`	`+ public string Queue`
`260`	`260`	`{`
`261`	`261`	`get`
`262`	`262`	`{`
`263`		`- string? x = _attach?.Source is not Source source ? null : source.Address;`
`264`		`- return x;`
	`263`	`+ string? sourceAddress = _attach?.Source is not Source source ? null : source.Address;`
	`264`	`+ if (sourceAddress is null)`
	`265`	`+ {`
	`266`	`+ throw new InvalidOperationException(`
	`267`	`+ "_attach.Source.Address is null");`
	`268`	`+ }`
	`269`	`+`
	`270`	`+ return AddressBuilderHelper.AddressBuilder().DecodeQueuePathSegment(sourceAddress);`
`265`	`271`	`}`
`266`	`272`	`}`
`267`	`273`
`@@ -318,7 +324,7 @@ await _receiverLink.CloseAsync(TimeSpan.FromSeconds(ConsumerDefaults.CloseTimeou`
`318`	`324`
`319`	`325`	`public override string ToString()`
`320`	`326`	`{`
`321`		`- return $"Consumer{{Address='{QueueAddress}', " +`
	`327`	`+ return $"Consumer{{Address='{Queue}', " +`
`322`	`328`	`$"id={_id}, " +`
`323`	`329`	`$"Connection='{_amqpConnection}', " +`
`324`	`330`	`$"State='{State}'}}";`
Original file line number	Diff line number	Diff line change
`@@ -138,9 +138,9 @@ public Task<PublishResult> PublishAsync(IMessage message, CancellationToken canc`
`138`	`138`	`void OutcomeCallback(ILink sender, Message inMessage, Outcome outcome, object state)`
`139`	`139`	`{`
`140`	`140`	// Note: sometimes `inMessage` is null 🤔
`141`		`- Debug.Assert(Object.ReferenceEquals(this, state));`
	`141`	`+ Debug.Assert(ReferenceEquals(this, state));`
`142`	`142`
`143`		`- if (false == Object.ReferenceEquals(_senderLink, sender))`
	`143`	`+ if (!ReferenceEquals(_senderLink, sender))`
`144`	`144`	`{`
`145`	`145`	`// TODO log this case?`
`146`	`146`	`}`
Original file line number	Diff line number	Diff line change
`@@ -14,7 +14,6 @@ public class RequesterConfiguration`
`14`	`14`	`public AmqpConnection Connection { get; set; } = null!;`
`15`	`15`	`public string ReplyToQueue { get; set; } = "";`
`16`	`16`
`17`		`- public string RequestAddress { get; set; } = "";`
`18`	`17`	`public TimeSpan Timeout { get; set; } = TimeSpan.FromSeconds(10);`
`19`	`18`
`20`	`19`	`public Func<object>? CorrelationIdSupplier { get; set; } = null;`
`@@ -79,7 +78,6 @@ public IRequesterBuilder Timeout(TimeSpan timeout)`
`79`	`78`
`80`	`79`	`public async Task<IRequester> BuildAsync()`
`81`	`80`	`{`
`82`		`- _configuration.RequestAddress = _addressBuilder.Address();`
`83`	`81`	`_configuration.Connection = _connection;`
`84`	`82`	`var rpcClient = new AmqpRequester(_configuration);`
`85`	`83`	`await rpcClient.OpenAsync().ConfigureAwait(false);`
`@@ -135,7 +133,7 @@ private IMessage RequestPostProcess(IMessage request, object correlationId)`
`135`	`133`	`return _configuration.RequestPostProcessor(request, correlationId);`
`136`	`134`	`}`
`137`	`135`
`138`		`- string s = ReplyToQueueAddress();`
	`136`	`+ string s = GetReplyToQueue();`
`139`	`137`	`return request.ReplyTo(s)`
`140`	`138`	`.MessageId(correlationId);`
`141`	`139`	`}`
`@@ -230,7 +228,8 @@ public async Task<IMessage> PublishAsync(IMessage message, CancellationToken can`
`230`	`228`	`if (_publisher != null)`
`231`	`229`	`{`
`232`	`230`	`PublishResult pr = await _publisher.PublishAsync(`
`233`		`- message.To(_configuration.RequestAddress), cancellationToken).ConfigureAwait(false);`
	`231`	`+ message.To(AddressBuilderHelper.AddressBuilder().Queue(GetReplyToQueue()).Address()),`
	`232`	`+ cancellationToken).ConfigureAwait(false);`
`234`	`233`
`235`	`234`	`if (pr.Outcome.State != OutcomeState.Accepted)`
`236`	`235`	`{`
`@@ -250,14 +249,14 @@ await _pendingRequests[correlationId].Task.WaitAsync(_configuration.Timeout)`
`250`	`249`	`}`
`251`	`250`	`}`
`252`	`251`
`253`		`- public string ReplyToQueueAddress()`
	`252`	`+ public string GetReplyToQueue()`
`254`	`253`	`{`
`255`	`254`	`if (_consumer == null)`
`256`	`255`	`{`
`257`	`256`	`throw new InvalidOperationException("Requester is not opened");`
`258`	`257`	`}`
`259`	`258`
`260`		`- return _consumer.QueueAddress ??`
	`259`	`+ return _consumer.Queue ??`
`261`	`260`	`throw new InvalidOperationException("ReplyToQueueAddress is not available");`
`262`	`261`	`}`
`263`	`262`	`}`