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 new file mode 100644 index 00000000..f448b562 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeConnectionListener.java @@ -0,0 +1,43 @@ +/* + * Copyright 2002-2010 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 + * + * http://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 java.util.ArrayList; +import java.util.List; + +/** + * @author Dave Syer + * + */ +public class CompositeConnectionListener implements ConnectionListener { + + private List delegates = new ArrayList(); + + public void onCreate(Connection connection) { + for (ConnectionListener delegate : delegates) { + delegate.onCreate(connection); + } + } + + public void setDelegates(List delegates) { + this.delegates = new ArrayList(delegates); + } + + public void addDelegate(ConnectionListener delegate) { + this.delegates.add(delegate); + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactory.java index 672ec7b7..adea3bcd 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactory.java @@ -13,7 +13,7 @@ package org.springframework.amqp.rabbit.connection; -import java.io.IOException; +import org.springframework.amqp.AmqpException; /** * An interface based ConnectionFactory for creating {@link com.rabbitmq.client.Connection Connections}. @@ -26,7 +26,7 @@ import java.io.IOException; */ public interface ConnectionFactory { - Connection createConnection() throws IOException; + Connection createConnection() throws AmqpException; String getHost(); 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 new file mode 100644 index 00000000..585b10de --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionListener.java @@ -0,0 +1,26 @@ +/* + * Copyright 2002-2010 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 + * + * http://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; + +/** + * @author Dave Syer + * + */ +public interface ConnectionListener { + + void onCreate(Connection connection); + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java index 228534db..c63b4178 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java @@ -16,9 +16,11 @@ package org.springframework.amqp.rabbit.connection; import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; +import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.amqp.AmqpException; import org.springframework.amqp.rabbit.support.RabbitUtils; import org.springframework.beans.factory.DisposableBean; import org.springframework.util.Assert; @@ -49,6 +51,8 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); + + private final CompositeConnectionListener listener = new CompositeConnectionListener(); /** * Create a new SingleConnectionFactory initializing the hostname to be the value returned from @@ -111,7 +115,15 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea return this.rabbitConnectionFactory.getPort(); } - public final Connection createConnection() throws IOException { + public void setConnectionListeners(List listeners) { + this.listener.setDelegates(listeners); + } + + public void addConnectionListener(ConnectionListener listener) { + this.listener.addDelegate(listener); + } + + public final Connection createConnection() throws AmqpException { synchronized (this.connectionMonitor) { if (this.connection == null) { if (this.targetConnection != null) { @@ -123,8 +135,9 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea } this.connection = new SharedConnectionProxy(this.targetConnection); } - return this.connection; } + this.listener.onCreate(connection); + return this.connection; } /** diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java new file mode 100644 index 00000000..2d81129f --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java @@ -0,0 +1,46 @@ +package org.springframework.amqp.rabbit.connection; + +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.Arrays; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Test; + +/** + * @author Dave Syer + */ +public class SingleConnectionFactoryTests { + + @Test + + public void testWithListener() throws IOException { + + com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); + com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); + + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + + final AtomicBoolean called = new AtomicBoolean(false); + SingleConnectionFactory ccf = new SingleConnectionFactory(mockConnectionFactory); + ccf.setConnectionListeners(Arrays.asList(new ConnectionListener(){ + public void onCreate(Connection connection) { + called.set(true); + } + })); + Connection con = ccf.createConnection(); + + assertTrue(called.get()); + + con.close(); + + verify(mockConnection, never()).close(); + + } + +}