* Remove AmqpConnection from ConsumerConfiguration.

* Add TODO to detect RMQ version when stream filters are used.

Author: lukebakken

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -20,6 +20,7 @@ private enum PauseStatus
             PAUSED,
         }
 
+        private readonly AmqpConnection _amqpConnection;
         private readonly Guid _id = Guid.NewGuid();
 
         private ReceiverLink? _receiverLink;
@@ -28,10 +29,12 @@ private enum PauseStatus
         private readonly UnsettledMessageCounter _unsettledMessageCounter = new();
         private readonly ConsumerConfiguration _configuration;
 
-        internal AmqpConsumer(ConsumerConfiguration configuration)
+        internal AmqpConsumer(AmqpConnection amqpConnection, ConsumerConfiguration configuration)
         {
+            _amqpConnection = amqpConnection;
             _configuration = configuration;
-            _configuration.Connection.AddConsumer(_id, this);
+
+            _amqpConnection.AddConsumer(_id, this);
         }
 
         public override async Task OpenAsync()
@@ -74,7 +77,7 @@ void onAttached(ILink argLink, Attach argAttach)
                 ReceiverLink? tmpReceiverLink = null;
                 Task receiverLinkTask = Task.Run(async () =>
                 {
-                    Session session = await _configuration.Connection._nativePubSubSessions.GetOrCreateSessionAsync()
+                    Session session = await _amqpConnection._nativePubSubSessions.GetOrCreateSessionAsync()
                         .ConfigureAwait(false);
                     tmpReceiverLink = new ReceiverLink(session, _id.ToString(), attach, onAttached);
                 });
@@ -250,14 +253,14 @@ await _receiverLink.CloseAsync(TimeSpan.FromSeconds(5))
 
             _receiverLink = null;
             OnNewStatus(State.Closed, null);
-            _configuration.Connection.RemoveConsumer(_id);
+            _amqpConnection.RemoveConsumer(_id);
         }
 
         public override string ToString()
         {
             return $"Consumer{{Address='{_configuration.Address}', " +
                    $"id={_id}, " +
-                   $"Connection='{_configuration.Connection}', " +
+                   $"Connection='{_amqpConnection}', " +
                    $"State='{State}'}}";
         }
     }

RabbitMQ.AMQP.Client/Impl/AmqpConsumerBuilder.cs

@@ -16,7 +16,6 @@ namespace RabbitMQ.AMQP.Client.Impl
     /// </summary>
     internal sealed class ConsumerConfiguration
     {
-        public AmqpConnection Connection { get; set; } = null!;
         public string Address { get; set; } = "";
         public int InitialCredits { get; set; } = 100; // TODO use constant, check with Java lib
         public Map Filters { get; set; } = new();
@@ -32,10 +31,11 @@ internal sealed class ConsumerConfiguration
     public class AmqpConsumerBuilder : IConsumerBuilder
     {
         private readonly ConsumerConfiguration _configuration = new();
+        private readonly AmqpConnection _amqpConnection;
 
         public AmqpConsumerBuilder(AmqpConnection connection)
         {
-            _configuration.Connection = connection;
+            _amqpConnection = connection;
         }
 
         public IConsumerBuilder Queue(IQueueSpecification queueSpec)
@@ -80,7 +80,7 @@ public async Task<IConsumer> BuildAndStartAsync(CancellationToken cancellationTo
                 throw new ConsumerException("Message handler is not set");
             }
 
-            AmqpConsumer consumer = new(_configuration);
+            AmqpConsumer consumer = new(_amqpConnection, _configuration);
 
             // TODO pass cancellationToken
             await consumer.OpenAsync()
@@ -174,6 +174,13 @@ private void SetOffsetSpecificationFilter(object value)
 
         public IConsumerBuilder.IStreamFilterOptions Filter()
         {
+            /*
+             * TODO detect RMQ version
+             *    if (!this.builder.connection.filterExpressionsSupported()) {
+             *      throw new IllegalArgumentException(
+             *          "AMQP filter expressions requires at least RabbitMQ 4.1.0");
+             *    }
+             */
             // TODO Should this be Consumer / Listener?
             return new StreamFilterOptions(this, _filters);
         }