() {
public AMQP.Exchange.DeclareOk doInRabbit(Channel channel) throws Exception {
logger.debug("Declaring exchange [" + exchange.getName() + "]");
- return channel.exchangeDeclare(exchange.getName(), exchange.getExchangeType().name(),
- exchange.isPassive(),
+ if (exchange.isPassive()) {
+ return channel.exchangeDeclarePassive(exchange.getName());
+ }
+ return channel.exchangeDeclare(exchange.getName(),
+ exchange.getExchangeType().name(),
exchange.isDurable(),
exchange.isAutoDelete(),
exchange.getArguments());
}
});
-
}
public AMQP.Queue.BindOk declareBinding(final Binding binding) {
diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java
index a0164c62..c8981ad3 100644
--- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java
+++ b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java
@@ -134,7 +134,6 @@ public class RabbitTestExecutionListener extends AbstractTestExecutionListener{
}
ccf.setUsername(username);
ccf.setPassword(password);
- ccf.afterPropertiesSet();
rabbitAdminTemplate = new RabbitAdminTemplate(ccf);
}
}
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
index fe692a80..8fcb2e83 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
@@ -29,14 +29,13 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+
import org.springframework.amqp.rabbit.support.RabbitUtils;
import org.springframework.beans.factory.DisposableBean;
-import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionParameters;
/**
* A {@link ConnectionFactory} implementation that returns the same Connections from all
@@ -49,12 +48,13 @@ import com.rabbitmq.client.ConnectionParameters;
*
* NOTE: This ConnectionFactory requires explicit closing of all Channels obtained form its
* shared Connection. This is the usual recommendation for native Rabbit access code anyway.
- * However, with this ConnectionFactory, its use is mandator in order to actually allow for Channel reuse.
+ * However, with this ConnectionFactory, its use is mandatory in order to actually allow for Channel reuse.
*
* @author Mark Pollack
+ * @author Mark Fisher
*/
//TODO are there heartbeats and/or exception thrown if a connection is broken?
-public class CachingConnectionFactory implements ConnectionFactory, InitializingBean, DisposableBean {
+public class CachingConnectionFactory implements ConnectionFactory, DisposableBean {
protected final Log logger = LogFactory.getLog(getClass());
@@ -64,7 +64,7 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
private int channelCacheSize = 1;
- private com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory;
+ private final com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory;
/** Raw Rabbit Connection */
private Connection targetConnection;
@@ -79,13 +79,14 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
private volatile boolean active = true;
+
/**
* Create a new CachingConnectionFactory initializing the hostname to be the
* value returned from InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws
* an exception.
*/
public CachingConnectionFactory() {
- this.hostName = initializeDefaultHostName();
+ this.hostName = initializeDefaultHostName();
this.rabbitConnectionFactory = new com.rabbitmq.client.ConnectionFactory();
}
@@ -97,15 +98,6 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
this(new com.rabbitmq.client.ConnectionFactory(), hostName);
}
- /**
- * Create a new CachingConnectionFactory given ConnectionParameters and the host name.
- * @param connectionParameters the connection parameters to use when creating a connection
- * @param hostName the host name to connect to
- */
- public CachingConnectionFactory(ConnectionParameters connectionParameters, String hostName) {
- this(new com.rabbitmq.client.ConnectionFactory(connectionParameters), hostName);
- }
-
/**
* Create a new CachingConnectionFactory for the given target
* ConnectionFactory.
@@ -118,31 +110,25 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
this.rabbitConnectionFactory = rabbitConnectionFactory;
this.hostName = hostName;
}
-
- public com.rabbitmq.client.ConnectionFactory getRabbitConnectionFactory() {
- return this.rabbitConnectionFactory;
- }
+
public void setUsername(String username) {
- this.rabbitConnectionFactory.getParameters().setUsername(username);
+ this.rabbitConnectionFactory.setUsername(username);
}
public void setPassword(String password) {
- this.rabbitConnectionFactory.getParameters().setPassword(password);
+ this.rabbitConnectionFactory.setPassword(password);
}
public void setChannelCacheSize(int sessionCacheSize) {
Assert.isTrue(sessionCacheSize >= 1, "Channel cache size must be 1 or higher");
this.channelCacheSize = sessionCacheSize;
}
-
-
public int getChannelCacheSize() {
return this.channelCacheSize;
}
-
public String getHostName() {
return hostName;
}
@@ -182,17 +168,7 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
new CachedChannelInvocationHandler(target, channelList));
}
- /**
- * Make sure a Connection or ConnectionFactory has been set.
- */
- public void afterPropertiesSet() {
- if (getRabbitConnectionFactory() == null) {
- throw new IllegalArgumentException("Connection or 'RabbitConnectionFactory' is required");
- }
- }
-
public Connection createConnection() throws IOException {
-
synchronized (this.connectionMonitor) {
if (this.connection == null) {
initConnection();
@@ -200,20 +176,8 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
return this.connection;
}
}
-
- public ConnectionParameters getParameters() {
- return this.rabbitConnectionFactory.getParameters();
- }
-
-
-
-
public void initConnection() throws IOException {
- if (getRabbitConnectionFactory() == null) {
- throw new IllegalStateException(
- "'rabbitConnectionFactory' is required for lazily initializing a Connection");
- }
synchronized (this.connectionMonitor) {
if (this.targetConnection != null) {
closeConnection(this.targetConnection);
@@ -226,9 +190,7 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
this.connection = getSharedConnectionProxy(this.targetConnection);
}
}
-
-
-
+
/**
* Close the underlying shared connection.
* The provider of this ConnectionFactory needs to care for proper shutdown.
@@ -237,8 +199,7 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
*/
public void destroy() {
resetConnection();
- }
-
+ }
/**
* Reset the Channel cache and underlying shared Connection, to be reinitialized on next access.
@@ -249,14 +210,14 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
for (Channel channel : cachedChannels) {
try {
channel.close();
- } catch (Throwable ex) {
+ }
+ catch (Throwable ex) {
logger.trace("Could not close cached Rabbit Channel", ex);
}
}
this.cachedChannels.clear();
}
this.active = true;
-
synchronized (this.connectionMonitor) {
if (this.targetConnection != null) {
closeConnection(this.targetConnection);
@@ -265,49 +226,50 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
this.connection = null;
}
}
+
/**
* Close the given Connection.
- * @param con the Connection to close
+ * @param connection the Connection to close
*/
- protected void closeConnection(Connection con) {
+ protected void closeConnection(Connection connection) {
if (logger.isDebugEnabled()) {
logger.debug("Closing shared Rabbit Connection: " + this.targetConnection);
}
try {
//TODO there are other close overloads close(int closeCode, java.lang.String closeMessage, int timeout)
- con.close();
+ connection.close();
}
catch (Throwable ex) {
logger.debug("Could not close shared Rabbit Connection", ex);
}
}
-
+
/**
* Create a Rabbit Connection via this class's ConnectionFactory.
* @return the new Rabbit Connection
*/
protected Connection doCreateConnection() throws IOException {
- //TODO there are other overloaded .newConnection methods
- return getRabbitConnectionFactory().newConnection(this.hostName, this.portNumber);
+ return this.rabbitConnectionFactory.newConnection();
}
protected void prepareConnection(Connection con) throws IOException {
//TODO configure ShutdownListener, investigate reconnection exceptions
}
-
+
protected String initializeDefaultHostName() {
String temp;
try {
InetAddress localMachine = InetAddress.getLocalHost();
temp = localMachine.getHostName();
logger.debug("Using hostname [" + temp + "] for hostname.");
- } catch (UnknownHostException e) {
+ }
+ catch (UnknownHostException e) {
logger.warn("Could not get host name, using 'localhost' as default value", e);
temp = "localhost";
}
return temp;
}
-
+
/**
* Wrap the given Connection with a proxy that delegates every method call to it
* but suppresses close calls. This is useful for allowing application code to
@@ -317,14 +279,22 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
* @return the wrapped Connection
*/
protected Connection getSharedConnectionProxy(Connection target) {
- List classes = new ArrayList(1);
+ List> classes = new ArrayList>(1);
classes.add(Connection.class);
return (Connection) Proxy.newProxyInstance(
Connection.class.getClassLoader(),
classes.toArray(new Class[classes.size()]),
new SharedConnectionInvocationHandler(target));
}
-
+
+ @Override
+ public String toString() {
+ return "CachingConnectionFactory [channelCacheSize=" + channelCacheSize
+ + ", hostName=" + hostName + ", portNumber=" + portNumber
+ + ", active=" + active + "]";
+ }
+
+
/**
* Invocation handler for a cached Rabbit Connection proxy.
*/
@@ -379,8 +349,7 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
}
}
-
-
+
public class CachedChannelInvocationHandler implements InvocationHandler {
private final Channel target;
@@ -433,8 +402,8 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
throw ex.getTargetException();
}
}
+
private void logicalClose(Channel proxy) throws Exception {
-
// Allow for multiple close calls...
if (!this.channelList.contains(proxy)) {
if (logger.isTraceEnabled()) {
@@ -453,13 +422,4 @@ public class CachingConnectionFactory implements ConnectionFactory, Initializing
}
-
-
- @Override
- public String toString() {
- return "CachingConnectionFactory [channelCacheSize=" + channelCacheSize
- + ", hostName=" + hostName + ", portNumber=" + portNumber
- + ", active=" + active + "]";
- }
-
}
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 e1d259e7..4de48a05 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
@@ -19,7 +19,6 @@ package org.springframework.amqp.rabbit.connection;
import java.io.IOException;
import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionParameters;
/**
* An interface based ConnectionFactory for creating {@link com.rabbitmq.client.Connection}s.
@@ -31,8 +30,7 @@ import com.rabbitmq.client.ConnectionParameters;
public interface ConnectionFactory {
Connection createConnection() throws IOException;
-
- ConnectionParameters getParameters();
-
+
String getHostName();
+
}
diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
index 221269b4..1480e25f 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
@@ -211,7 +211,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
String[] queue = StringUtils.commaDelimitedListToStringArray(queueNames);
for (int i = 0; i < queue.length; i++) {
- channel.queueDeclare(queue[i]);
+ channel.queueDeclarePassive(queue[i]);
String consumerTag = channel.basicConsume(queue[i], autoAck, consumer);
consumer.setConsumerTag(consumerTag);
}
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
index df037cf1..fe6fc293 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
@@ -7,7 +7,6 @@ import java.io.IOException;
import junit.framework.Assert;
import org.junit.Test;
-import org.springframework.amqp.rabbit.support.RabbitUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
@@ -23,11 +22,10 @@ public class CachingConnectionFactoryTests {
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
- when(mockConnectionFactory.newConnection("localhost", RabbitUtils.DEFAULT_PORT)).thenReturn(mockConnection);
+ when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel);
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory, "localhost");
- ccf.afterPropertiesSet();
Connection con = ccf.createConnection();
Channel channel = con.createChannel();
@@ -53,13 +51,12 @@ public class CachingConnectionFactoryTests {
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
- when(mockConnectionFactory.newConnection("localhost", RabbitUtils.DEFAULT_PORT)).thenReturn(mockConnection);
+ when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel1);
when(mockConnection.createChannel()).thenReturn(mockChannel2);
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory, "localhost");
ccf.setChannelCacheSize(2);
- ccf.afterPropertiesSet();
Connection con = ccf.createConnection();
@@ -99,13 +96,12 @@ public class CachingConnectionFactoryTests {
Assert.assertNotSame(mockChannel1, mockChannel2);
- when(mockConnectionFactory.newConnection("localhost", RabbitUtils.DEFAULT_PORT)).thenReturn(mockConnection);
+ when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
//You can't repeat 'when' statements for stubbing consecutive calls to the same method to returning different values.
stub(mockConnection.createChannel()).toReturn(mockChannel1).toReturn(mockChannel2);
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory, "localhost");
ccf.setChannelCacheSize(2);
- ccf.afterPropertiesSet();
Connection con = ccf.createConnection();
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/QueueUtils.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/QueueUtils.java
index 80275b7e..7948a743 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/QueueUtils.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/QueueUtils.java
@@ -9,7 +9,7 @@ public class QueueUtils {
// declare and bind queue
template.execute(new ChannelCallback() {
public String doInRabbit(Channel channel) throws Exception {
- Queue.DeclareOk res = channel.queueDeclare(TestConstants.QUEUE_NAME);
+ Queue.DeclareOk res = channel.queueDeclarePassive(TestConstants.QUEUE_NAME);
String queueName = res.getQueue();
System.out.println("Queue Name = " + queueName);
channel.queueBind(queueName, TestConstants.EXCHANGE_NAME, routingKey);