diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java index f4fd8f14..ab6e5020 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java +++ b/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; } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeConnectionListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeConnectionListener.java index 35c65eb9..ce1b01a7 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeConnectionListener.java +++ b/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 delegates = new CopyOnWriteArrayList(); + @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 delegates) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionListener.java index 5d739220..7154bbf9 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionListener.java +++ b/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) { + } + } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnnectionListenerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ConnnectionListenerTests.java new file mode 100644 index 00000000..03638c6a --- /dev/null +++ b/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(); + } + +}