Support client specified connection name

Author: Daniil Fedotov

projects/client/RabbitMQ.Client/src/client/api/ConnectionFactory.cs

@@ -348,7 +348,19 @@ public AuthMechanismFactory AuthMechanismFactory(IList<string> mechanismNames)
         /// </exception>
         public virtual IConnection CreateConnection()
         {
-            return CreateConnection(new List<string>() { HostName });
+            return CreateConnection(new List<string>() { HostName }, null);
+        }
+
+        /// <summary>
+        /// Create a connection to the specified endpoint.
+        /// </summary>
+        /// <param name="connectionName">Connection name client property</param>
+        /// <exception cref="BrokerUnreachableException">
+        /// When the configured hostname was not reachable.
+        /// </exception>
+        public IConnection CreateConnection(String connectionName)
+        {
+            return CreateConnection(new List<string>() { HostName }, connectionName);
         }
 
         /// <summary>
@@ -365,20 +377,39 @@ public virtual IConnection CreateConnection()
         /// When no hostname was reachable.
         /// </exception>
         public IConnection CreateConnection(IList<string> hostnames)
+        {
+            return CreateConnection(hostnames, null);
+        }
+
+        /// <summary>
+        /// Create a connection using a list of hostnames. The first reachable
+        /// hostname will be used initially. Subsequent hostname picks are determined
+        /// by the <see cref="IHostnameSelector" /> configured.
+        /// </summary>
+        /// <param name="hostnames">
+        /// List of hostnames to use for the initial
+        /// connection and recovery.
+        /// </param>
+        /// <param name="connectionName">Connection name client property</param>
+        /// <returns>Open connection</returns>
+        /// <exception cref="BrokerUnreachableException">
+        /// When no hostname was reachable.
+        /// </exception>
+        public IConnection CreateConnection(IList<string> hostnames, String connectionName)
         {
             IConnection conn;
             try
             {
                 if (AutomaticRecoveryEnabled)
                 {
-                    var autorecoveringConnection = new AutorecoveringConnection(this);
+                    var autorecoveringConnection = new AutorecoveringConnection(this, connectionName);
                     autorecoveringConnection.Init(hostnames);
                     conn = autorecoveringConnection;
                 }
                 else
                 {
                     IProtocol protocol = Protocols.DefaultProtocol;
-                    conn = protocol.CreateConnection(this, false, CreateFrameHandler());
+                    conn = protocol.CreateConnection(this, false, CreateFrameHandler(), connectionName);
                 }
             }
             catch (Exception e)

projects/client/RabbitMQ.Client/src/client/api/IConnection.cs

@@ -152,6 +152,11 @@ public interface IConnection : NetworkConnection, IDisposable
         /// </summary>
         IList<ShutdownReportEntry> ShutdownReport { get; }
 
+        /// <summary>
+        /// Returns connection name from client properties
+        /// </summary>
+        String ConnectionName { get; }
+
         /// <summary>
         /// Signalled when an exception occurs in a callback invoked by the connection.
         /// </summary>

projects/client/RabbitMQ.Client/src/client/api/IConnectionFactory.cs

@@ -97,16 +97,31 @@ public interface IConnectionFactory
         /// </summary>
         IConnection CreateConnection();
 
+        /// <summary>
+        /// Create a connection to the specified endpoint.
+        /// </summary>
+        /// <param name="connectionName">Connection name client property</param>
+        /// <returns></returns>
+        IConnection CreateConnection(String connectionName);
+
         /// <summary>
         /// Connects to the first reachable hostname from the list.
         /// </summary>
         /// <param name="hostnames">List of host names to use</param>
         /// <returns></returns>
         IConnection CreateConnection(IList<string> hostnames);
 
+        /// <summary>
+        /// Connects to the first reachable hostname from the list.
+        /// </summary>
+        /// <param name="hostnames">List of host names to use</param>
+        /// <param name="connectionName">Connection name client property</param>
+        /// <returns></returns>
+        IConnection CreateConnection(IList<string> hostnames, String connectionName);
+
         /// <summary>
         /// Advanced option.
-        /// 
+        ///
         /// What task scheduler should consumer dispatcher use.
         /// </summary>
         TaskScheduler TaskScheduler { get; set; }

projects/client/RabbitMQ.Client/src/client/api/IProtocol.cs

@@ -95,6 +95,19 @@ public interface IProtocol
         /// </summary>
         IConnection CreateConnection(ConnectionFactory factory, IFrameHandler frameHandler, bool automaticRecoveryEnabled);
 
+        /// <summary>
+        /// Construct a connection from a given set of parameters,
+        /// a frame handler, and no automatic recovery.
+        /// The "insist" parameter is passed on to the AMQP connection.open method.
+        /// </summary>
+        IConnection CreateConnection(IConnectionFactory factory, bool insist, IFrameHandler frameHandler, String connectionName);
+
+        /// <summary>
+        /// Construct a connection from a given set of parameters,
+        /// a frame handler, and automatic recovery settings.
+        /// </summary>
+        IConnection CreateConnection(ConnectionFactory factory, IFrameHandler frameHandler, bool automaticRecoveryEnabled, String connectionName);
+
         /// <summary>
         ///  Construct a frame handler for a given endpoint.
         ///  </summary>

projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringConnection.cs

@@ -94,9 +94,10 @@ public class AutorecoveringConnection : IConnection, IRecoverable
         private EventHandler<QueueNameChangedAfterRecoveryEventArgs> m_queueNameChange;
         private EventHandler<EventArgs> m_recovery;
 
-        public AutorecoveringConnection(ConnectionFactory factory)
+        public AutorecoveringConnection(ConnectionFactory factory, String connectionName = null)
         {
             m_factory = factory;
+            ConnectionName = connectionName;
         }
 
         public event EventHandler<CallbackExceptionEventArgs> CallbackException
@@ -231,6 +232,8 @@ public event EventHandler<EventArgs> Recovery
             }
         }
 
+        public String ConnectionName { get; private set; }
+
         public bool AutoClose
         {
             get { return m_delegate.AutoClose; }
@@ -575,7 +578,8 @@ public void Init(IList<string> hostnames)
         protected void Init(string hostname)
         {
             m_delegate = new Connection(m_factory, false,
-                m_factory.CreateFrameHandlerForHostname(hostname));
+                m_factory.CreateFrameHandlerForHostname(hostname),
+                ConnectionName);
 
             AutorecoveringConnection self = this;
             EventHandler<ShutdownEventArgs> recoveryListener = (_, args) =>
@@ -792,7 +796,7 @@ protected void RecoverConnectionDelegate()
                 {
                     var nextHostname = m_factory.HostnameSelector.NextFrom(this.hostnames);
                     var fh = m_factory.CreateFrameHandler(m_factory.Endpoint.CloneWithHostname(nextHostname));
-                    m_delegate = new Connection(m_factory, false, fh);
+                    m_delegate = new Connection(m_factory, false, fh, ConnectionName);
                     recovering = false;
                 }
                 catch (Exception e)

projects/client/RabbitMQ.Client/src/client/impl/Connection.cs

@@ -113,8 +113,9 @@ public class Connection : IConnection
 
         public ConsumerWorkService ConsumerWorkService { get; private set; }
 
-        public Connection(IConnectionFactory factory, bool insist, IFrameHandler frameHandler)
+        public Connection(IConnectionFactory factory, bool insist, IFrameHandler frameHandler, String connectionName = null)
         {
+            ConnectionName = connectionName;
             KnownHosts = null;
             FrameMax = 0;
             m_factory = factory;
@@ -225,6 +226,8 @@ public event EventHandler<EventArgs> ConnectionUnblocked
             }
         }
 
+        public String ConnectionName { get; private set; }
+
         public bool AutoClose
         {
             get { return m_sessionManager.AutoClose; }
@@ -995,7 +998,7 @@ public void MaybeStartHeartbeatTimers()
                 _heartbeatWriteTimer = new Timer(HeartbeatWriteTimerCallback);
                 _heartbeatReadTimer = new Timer(HeartbeatReadTimerCallback);
 #if NETFX_CORE
-                _heartbeatWriteTimer.Change(200, m_heartbeatTimeSpan.Milliseconds);               
+                _heartbeatWriteTimer.Change(200, m_heartbeatTimeSpan.Milliseconds);
                 _heartbeatReadTimer.Change(200, m_heartbeatTimeSpan.Milliseconds);
 #else
                 _heartbeatWriteTimer.Change(TimeSpan.FromMilliseconds(200), m_heartbeatTimeSpan);
@@ -1274,6 +1277,7 @@ protected void StartAndTune()
 
             m_clientProperties = new Dictionary<string, object>(m_factory.ClientProperties);
             m_clientProperties["capabilities"] = Protocol.Capabilities;
+            m_clientProperties["connection_name"] = ConnectionName;
 
             // FIXME: parse out locales properly!
             ConnectionTuneDetails connectionTune = default(ConnectionTuneDetails);

projects/client/RabbitMQ.Client/src/client/impl/ProtocolBase.cs

@@ -131,14 +131,33 @@ public IConnection CreateConnection(IConnectionFactory factory,
             bool insist,
             IFrameHandler frameHandler)
         {
-            return new Connection(factory, insist, frameHandler);
+            return new Connection(factory, insist, frameHandler, null);
+        }
+
+
+        public IConnection CreateConnection(IConnectionFactory factory,
+            bool insist,
+            IFrameHandler frameHandler,
+            String connectionName)
+        {
+            return new Connection(factory, insist, frameHandler, connectionName);
         }
 
         public IConnection CreateConnection(ConnectionFactory factory,
             IFrameHandler frameHandler,
             bool automaticRecoveryEnabled)
         {
-            var ac = new AutorecoveringConnection(factory);
+            var ac = new AutorecoveringConnection(factory, null);
+            ac.Init();
+            return ac;
+        }
+
+        public IConnection CreateConnection(ConnectionFactory factory,
+            IFrameHandler frameHandler,
+            bool automaticRecoveryEnabled,
+            String connectionName)
+        {
+            var ac = new AutorecoveringConnection(factory, connectionName);
             ac.Init();
             return ac;
         }