Connection recovery should not try to handle UnauthorizedAccess #221

Havret · May 13, 2021 · 5cdf309 · 5cdf309
1 parent 889c6ae
commit 5cdf309
Show file tree

Hide file tree

Showing 4 changed files with 61 additions and 11 deletions.
diff --git a/src/ActiveMQ.Artemis.Client/AutoRecovering/AutoRecoveringConnection.cs b/src/ActiveMQ.Artemis.Client/AutoRecovering/AutoRecoveringConnection.cs
@@ -6,8 +6,10 @@
 using System.Threading.Tasks;
 using ActiveMQ.Artemis.Client.AutoRecovering.RecoveryPolicy;
 using ActiveMQ.Artemis.Client.Builders;
+using ActiveMQ.Artemis.Client.Exceptions;
 using ActiveMQ.Artemis.Client.InternalUtilities;
 using ActiveMQ.Artemis.Client.MessageIdPolicy;
+using Amqp;
 using Microsoft.Extensions.Logging;
 using Polly;
 using Polly.Retry;
@@ -52,7 +54,11 @@ public AutoRecoveringConnection(ILoggerFactory loggerFactory, IEnumerable<Endpoi
         private AsyncRetryPolicy<IConnection> CreateConnectionRetryPolicy(IRecoveryPolicy recoveryPolicy)
         {
             return Policy<IConnection>
-                   .Handle<Exception>()
+                   .Handle<Exception>(exception => exception switch
+                   {
+                       CreateConnectionException { ErrorCode: ErrorCode.UnauthorizedAccess } => false,
+                       _ => true
+                   })
                    .WaitAndRetryAsync(recoveryPolicy.RetryCount, (retryAttempt, context) =>
                    {
                        context.SetRetryCount(retryAttempt);
@@ -85,7 +91,7 @@ private Task StartRecoveryLoop()
                         if (!IsOpened)
                         {
                             await DisposeInnerConnection().ConfigureAwait(false);
-                            
+
                             foreach (var recoverable in _recoverables.Values)
                             {
                                 recoverable.Suspend();
@@ -110,7 +116,7 @@ private Task StartRecoveryLoop()
                             _connection.ConnectionClosed += OnConnectionClosed;
 
                             Log.ConnectionRecovered(_logger);
-                            
+
                             ConnectionRecovered?.Invoke(this, new ConnectionRecoveredEventArgs(_connection.Endpoint));
                         }
                         else
@@ -135,7 +141,7 @@ private Task StartRecoveryLoop()
                     {
                         await recoverable.TerminateAsync(e).ConfigureAwait(false);
                     }
-                    
+
                     ConnectionRecoveryError?.Invoke(this, new ConnectionRecoveryErrorEventArgs(e));
                 }
             });
@@ -197,7 +203,7 @@ public async Task<IAnonymousProducer> CreateAnonymousProducerAsync(AnonymousProd
         }
 
         public event EventHandler<ConnectionClosedEventArgs> ConnectionClosed;
-        
+
         public event EventHandler<ConnectionRecoveredEventArgs> ConnectionRecovered;
         public event EventHandler<ConnectionRecoveryErrorEventArgs> ConnectionRecoveryError;
 

diff --git a/src/ActiveMQ.Artemis.Client/Builders/ConnectionBuilder.cs b/src/ActiveMQ.Artemis.Client/Builders/ConnectionBuilder.cs
@@ -28,11 +28,18 @@ public async Task<IConnection> CreateAsync(Endpoint endpoint, CancellationToken
             cancellationToken.Register(() => _tcs.TrySetCanceled());
 
             var connectionFactory = new Amqp.ConnectionFactory();
-            var connection = await connectionFactory.CreateAsync(endpoint.Address, null, OnOpened).ConfigureAwait(false);
-            connection.AddClosedCallback(OnClosed);
-            await _tcs.Task.ConfigureAwait(false);
-            connection.Closed -= OnClosed;
-            return new Connection(_loggerFactory, endpoint, connection, _messageIdPolicyFactory);
+            try
+            {
+                var connection = await connectionFactory.CreateAsync(endpoint.Address, null, OnOpened).ConfigureAwait(false);
+                connection.AddClosedCallback(OnClosed);
+                await _tcs.Task.ConfigureAwait(false);
+                connection.Closed -= OnClosed;
+                return new Connection(_loggerFactory, endpoint, connection, _messageIdPolicyFactory);
+            }
+            catch (AmqpException exception)
+            {
+                throw new CreateConnectionException(message: exception.Error?.Description ?? exception.Message, errorCode: exception.Error?.Condition);
+            }
         }
 
         private void OnOpened(Amqp.IConnection connection, Open open)

diff --git a/test/ActiveMQ.Artemis.Client.IntegrationTests/ActiveMQNetIntegrationSpec.cs b/test/ActiveMQ.Artemis.Client.IntegrationTests/ActiveMQNetIntegrationSpec.cs
@@ -32,7 +32,7 @@ protected Task<IConnection> CreateConnection()
             return connectionFactory.CreateAsync(endpoint);
         }
 
-        private static Endpoint GetEndpoint()
+        protected static Endpoint GetEndpoint()
         {
             string userName = Environment.GetEnvironmentVariable("ARTEMIS_USERNAME") ?? "guest";
             string password = Environment.GetEnvironmentVariable("ARTEMIS_PASSWORD") ?? "guest";

diff --git a/test/ActiveMQ.Artemis.Client.IntegrationTests/CreateConnectionSpec.cs b/test/ActiveMQ.Artemis.Client.IntegrationTests/CreateConnectionSpec.cs
@@ -0,0 +1,37 @@
+using System;
+using System.Reflection;
+using System.Threading.Tasks;
+using ActiveMQ.Artemis.Client.Exceptions;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace ActiveMQ.Artemis.Client.IntegrationTests
+{
+    public class CreateConnectionSpec : ActiveMQNetIntegrationSpec
+    {
+        public CreateConnectionSpec(ITestOutputHelper output) : base(output)
+        {
+        }
+
+        [Fact]
+        public async Task Should_throw_security_exception_when_wrong_credentials_provided_and_connection_auto_recovery_enabled()
+        {
+            var defaultEndpoint = GetEndpoint();
+            var endpoint = Endpoint.Create(
+                password: "wrongPassword",
+                user: defaultEndpoint.User,
+                host: defaultEndpoint.Host,
+                port: defaultEndpoint.Port,
+                scheme: defaultEndpoint.Scheme
+            );
+
+            var connectionFactory = new ConnectionFactory
+            {
+                AutomaticRecoveryEnabled = true
+            };
+
+            var createConnectionException = await Assert.ThrowsAnyAsync<CreateConnectionException>(() => connectionFactory.CreateAsync(endpoint));
+            Assert.Equal("amqp:unauthorized-access", createConnectionException.ErrorCode);
+        }
+    }
+}