Add option to override client-id

Author: Havret

src/ArtemisNetClient/AutoRecovering/AutoRecoveringConnection.cs

@@ -21,6 +21,7 @@ internal class AutoRecoveringConnection : IConnection
         private IConnection _connection;
         private readonly ILoggerFactory _loggerFactory;
         private readonly Func<IMessageIdPolicy> _messageIdPolicyFactory;
+        private readonly Func<string> _clientIdFactory;
         private readonly ILogger<AutoRecoveringConnection> _logger;
         private readonly Endpoint[] _endpoints;
         private readonly ChannelReader<ConnectCommand> _reader;
@@ -30,11 +31,16 @@ internal class AutoRecoveringConnection : IConnection
         private readonly AsyncRetryPolicy<IConnection> _connectionRetryPolicy;
         private readonly Task _recoveryLoopTask;
 
-        public AutoRecoveringConnection(ILoggerFactory loggerFactory, IEnumerable<Endpoint> endpoints, IRecoveryPolicy recoveryPolicy, Func<IMessageIdPolicy> messageIdPolicyFactory)
+        public AutoRecoveringConnection(ILoggerFactory loggerFactory,
+            IEnumerable<Endpoint> endpoints,
+            IRecoveryPolicy recoveryPolicy,
+            Func<IMessageIdPolicy> messageIdPolicyFactory,
+            Func<string> clientIdFactory)
         {
             _logger = loggerFactory.CreateLogger<AutoRecoveringConnection>();
             _loggerFactory = loggerFactory;
             _messageIdPolicyFactory = messageIdPolicyFactory;
+            _clientIdFactory = clientIdFactory;
             _endpoints = endpoints.ToArray();
 
             var channel = Channel.CreateUnbounded<ConnectCommand>();
@@ -169,7 +175,7 @@ private Task<IConnection> CreateConnection(CancellationToken cancellationToken)
                 int retryCount = context.GetRetryCount();
                 var endpoint = GetNextEndpoint(retryCount);
                 context.SetEndpoint(endpoint);
-                var connectionBuilder = new ConnectionBuilder(_loggerFactory, _messageIdPolicyFactory);
+                var connectionBuilder = new ConnectionBuilder(_loggerFactory, _messageIdPolicyFactory, _clientIdFactory);
                 var connection = await connectionBuilder.CreateAsync(endpoint, ct);
 
                 if (retryCount > 0)

src/ArtemisNetClient/Builders/ConnectionBuilder.cs

@@ -11,14 +11,21 @@ namespace ActiveMQ.Artemis.Client.Builders
 {
     internal class ConnectionBuilder
     {
+        const uint DefaultMaxFrameSize = 256 * 1024;
+        const ushort ChannelMax = 255;
+
         private readonly ILoggerFactory _loggerFactory;
         private readonly Func<IMessageIdPolicy> _messageIdPolicyFactory;
+        private readonly Func<string> _clientIdFactory;
         private readonly TaskCompletionSource<bool> _tcs;
 
-        public ConnectionBuilder(ILoggerFactory loggerFactory, Func<IMessageIdPolicy> messageIdPolicyFactory)
+        public ConnectionBuilder(ILoggerFactory loggerFactory,
+            Func<IMessageIdPolicy> messageIdPolicyFactory,
+            Func<string> clientIdFactory)
         {
             _loggerFactory = loggerFactory;
             _messageIdPolicyFactory = messageIdPolicyFactory;
+            _clientIdFactory = clientIdFactory;
             _tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
         }
 
@@ -30,7 +37,8 @@ public async Task<IConnection> CreateAsync(Endpoint endpoint, CancellationToken
             var connectionFactory = new Amqp.ConnectionFactory();
             try
             {
-                var connection = await connectionFactory.CreateAsync(endpoint.Address, null, OnOpened).ConfigureAwait(false);
+                var open = GetOpenFrame(endpoint);
+                var connection = await connectionFactory.CreateAsync(endpoint.Address, open, OnOpened).ConfigureAwait(false);
                 connection.AddClosedCallback(OnClosed);
                 await _tcs.Task.ConfigureAwait(false);
                 connection.Closed -= OnClosed;
@@ -42,6 +50,22 @@ public async Task<IConnection> CreateAsync(Endpoint endpoint, CancellationToken
             }
         }
 
+        private Open GetOpenFrame(Endpoint endpoint)
+        {
+            if (_clientIdFactory != null)
+            {
+                return new Open
+                {
+                    ContainerId = _clientIdFactory(),
+                    HostName = endpoint.Host,
+                    MaxFrameSize = DefaultMaxFrameSize,
+                    ChannelMax = ChannelMax
+                };
+            }
+
+            return null;
+        }
+
         private void OnOpened(Amqp.IConnection connection, Open open)
         {
             if (connection != null)

src/ArtemisNetClient/ConnectionFactory.cs

@@ -17,6 +17,7 @@ public class ConnectionFactory
     {
         private IRecoveryPolicy _recoveryPolicy;
         private Func<IMessageIdPolicy> _messageIdPolicyFactory;
+        private Func<string> _clientIdFactory;
 
         public async Task<IConnection> CreateAsync(IEnumerable<Endpoint> endpoints, CancellationToken cancellationToken)
         {
@@ -29,13 +30,13 @@ public async Task<IConnection> CreateAsync(IEnumerable<Endpoint> endpoints, Canc
 
             if (AutomaticRecoveryEnabled)
             {
-                var autoRecoveringConnection = new AutoRecoveringConnection(LoggerFactory, endpointsList, RecoveryPolicy, MessageIdPolicyFactory);
+                var autoRecoveringConnection = new AutoRecoveringConnection(LoggerFactory, endpointsList, RecoveryPolicy, MessageIdPolicyFactory, ClientIdFactory);
                 await autoRecoveringConnection.InitAsync(cancellationToken).ConfigureAwait(false);
                 return autoRecoveringConnection;
             }
             else
             {
-                var connectionBuilder = new ConnectionBuilder(LoggerFactory, MessageIdPolicyFactory);
+                var connectionBuilder = new ConnectionBuilder(LoggerFactory, MessageIdPolicyFactory, ClientIdFactory);
                 return await connectionBuilder.CreateAsync(endpointsList.First(), cancellationToken).ConfigureAwait(false);
             }
         }
@@ -53,5 +54,14 @@ public Func<IMessageIdPolicy> MessageIdPolicyFactory
             get => _messageIdPolicyFactory ?? ActiveMQ.Artemis.Client.MessageIdPolicy.MessageIdPolicyFactory.DisableMessageIdPolicy;
             set => _messageIdPolicyFactory = value ?? throw new ArgumentNullException(nameof(value), "MessageId Policy Factory cannot be null.");
         }
+
+        /// <summary>
+        /// Factory function that creates unique identifier for the connection.
+        /// </summary>
+        public Func<string> ClientIdFactory
+        {
+            get => _clientIdFactory;
+            set => _clientIdFactory = value ?? throw new ArgumentNullException(nameof(value), "Client Id Factory cannot be null.");
+        }
     }
 }

test/ArtemisNetClient.UnitTests/ConnectionFactorySpec.cs

@@ -7,6 +7,7 @@
 using ActiveMQ.Artemis.Client.Exceptions;
 using ActiveMQ.Artemis.Client.MessageIdPolicy;
 using ActiveMQ.Artemis.Client.UnitTests.Utils;
+using Amqp.Framing;
 using Amqp.Handler;
 using Xunit;
 using Xunit.Abstractions;
@@ -108,5 +109,34 @@ public void Should_assign_custom_message_id_policy_factory()
             var connectionFactory = CreateConnectionFactory();
             connectionFactory.MessageIdPolicyFactory = MessageIdPolicyFactory.GuidMessageIdPolicy;
         }
+
+        [Fact]
+        public async Task Should_create_connection_with_specified_client_id()
+        {
+            // Arrange
+            var endpoint = GetUniqueEndpoint();
+
+            var tcs = new TaskCompletionSource<Open>();
+            var handler = new TestHandler(@event =>
+            {
+                switch (@event.Id)
+                {
+                    case EventId.ConnectionRemoteOpen:
+                        tcs.TrySetResult((Open) @event.Context);
+                        break;
+                }
+            });
+            using var host = CreateOpenedContainerHost(endpoint, handler);
+
+            var connectionFactory = CreateConnectionFactory();
+            connectionFactory.ClientIdFactory = () => "foo";
+
+            // Act
+            await using var connection = await connectionFactory.CreateAsync(endpoint);
+
+            // Assert
+            var openFrame = await tcs.Task;
+            Assert.Equal("foo", openFrame.ContainerId);
+        }
     }
 }