GH-1315: Add onFailed() to ConnectionListener

Resolves https://github.com/spring-projects/spring-amqp/issues/1315

**cherry-pick to 2.2.x**

* Fix `@since` in the new test class

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -598,8 +598,10 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 
 			return connection;
 		}
-		catch (IOException | TimeoutException e) {
-			throw RabbitExceptionTranslator.convertRabbitAccessException(e);
+		catch (IOException | TimeoutException ex) {
+			RuntimeException converted = RabbitExceptionTranslator.convertRabbitAccessException(ex);
+			this.connectionListener.onFailed(ex);
+			throw converted;
 		}
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeConnectionListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,6 +23,8 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import com.rabbitmq.client.ShutdownSignalException;
 
 /**
+ * A composite listener that invokes its delegages in turn.
+ *
  * @author Dave Syer
  * @author Gary Russell
  *
@@ -31,23 +33,24 @@ public class CompositeConnectionListener implements ConnectionListener {
 
 	private List<ConnectionListener> delegates = new CopyOnWriteArrayList<ConnectionListener>();
 
+	@Override
 	public void onCreate(Connection connection) {
-		for (ConnectionListener delegate : this.delegates) {
-			delegate.onCreate(connection);
-		}
+		this.delegates.forEach(delegate -> delegate.onCreate(connection));
 	}
 
+	@Override
 	public void onClose(Connection connection) {
-		for (ConnectionListener delegate : this.delegates) {
-			delegate.onClose(connection);
-		}
+		this.delegates.forEach(delegate -> delegate.onClose(connection));
 	}
 
 	@Override
 	public void onShutDown(ShutdownSignalException signal) {
-		for (ConnectionListener delegate : this.delegates) {
-			delegate.onShutDown(signal);
-		}
+		this.delegates.forEach(delegate -> delegate.onShutDown(signal));
+	}
+
+	@Override
+	public void onFailed(Exception exception) {
+		this.delegates.forEach(delegate -> delegate.onFailed(exception));
 	}
 
 	public void setDelegates(List<? extends ConnectionListener> delegates) {

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -50,4 +50,12 @@ public interface ConnectionListener {
 	default void onShutDown(ShutdownSignalException signal) {
 	}
 
+	/**
+	 * Called when a connection couldn't be established.
+	 * @param exception the exception thrown.
+	 * @since 2.2.17
+	 */
+	default void onFailed(Exception exception) {
+	}
+
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnnectionListenerTests.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.connection;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.amqp.AmqpIOException;
+
+/**
+ * @author Gary Russell
+ * @since 2.2.17
+ *
+ */
+public class ConnnectionListenerTests {
+
+	@Test
+	void cantConnectCCF() {
+		CachingConnectionFactory ccf = new CachingConnectionFactory(rcf());
+		cantConnect(ccf);
+	}
+
+	@Test
+	void cantConnectTCCF() {
+		ThreadChannelConnectionFactory tccf = new ThreadChannelConnectionFactory(rcf());
+		cantConnect(tccf);
+	}
+
+	@Test
+	void cantConnectPCCF() {
+		PooledChannelConnectionFactory pccf = new PooledChannelConnectionFactory(rcf());
+		cantConnect(pccf);
+	}
+
+	private com.rabbitmq.client.ConnectionFactory rcf() {
+		com.rabbitmq.client.ConnectionFactory rcf = new com.rabbitmq.client.ConnectionFactory();
+		rcf.setHost("junk.host");
+		return rcf;
+	}
+
+	private void cantConnect(ConnectionFactory cf) {
+		AtomicBoolean failed = new AtomicBoolean();
+		cf.addConnectionListener(new ConnectionListener() {
+
+			@Override
+			public void onCreate(Connection connection) {
+			}
+
+			@Override
+			public void onFailed(Exception exception) {
+				failed.set(true);
+			}
+
+		});
+		assertThatExceptionOfType(AmqpIOException.class).isThrownBy(() -> cf.createConnection());
+		assertThat(failed.get()).isTrue();
+	}
+
+}