AMQP-122: Add ConnectionListener

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<ConnectionListener> delegates = new ArrayList<ConnectionListener>();
+
+	public void onCreate(Connection connection) {
+		for (ConnectionListener delegate : delegates) {
+			delegate.onCreate(connection);
+		}	
+	}
+
+	public void setDelegates(List<? extends ConnectionListener> delegates) {
+		this.delegates = new ArrayList<ConnectionListener>(delegates);
+	}
+	
+	public void addDelegate(ConnectionListener delegate) {
+		this.delegates.add(delegate);
+	}
+	
+}

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();
 

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);
+
+}

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<? extends ConnectionListener> 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;
 	}
 
 	/**

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();
+		
+	}
+
+}