Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-3 upgraded to RabbitMQ 1.8.0 Java client

Browse Source
This commit is contained in:
Mark Fisher
2010-07-06 15:44:27 -04:00
parent cd910651dd
commit 7fd87ed71a
10 changed files with 58 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-amqp-parent/pom.xml
View File

@@ -19,7 +19,7 @@
<org.slf4j.version>1.5.10</org.slf4j.version>
<org.codehaus.jackson.version>1.4.3</org.codehaus.jackson.version>
<org.erlang.otp.version>1.5.3</org.erlang.otp.version>
<com.rabbitmq.version>1.7.2</com.rabbitmq.version>
<com.rabbitmq.version>1.8.0</com.rabbitmq.version>
<org.springframework.version>3.0.3.RELEASE</org.springframework.version>
</properties>
<profiles>

3
spring-amqp-samples/helloworld/pom.xml
View File

@@ -16,6 +16,7 @@
</description>
<properties>
<maven.test.failure.ignore>true</maven.test.failure.ignore>
<rabbitmq.version>1.8.0</rabbitmq.version>
<spring.framework.version>3.0.3.RELEASE</spring.framework.version>
<spring.amqp.version>1.0.0.BUILD-SNAPSHOT</spring.amqp.version>
</properties>
@@ -48,7 +49,7 @@
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>1.7.2</version>
<version>${rabbitmq.version}</version>
</dependency>
<dependency>

3
spring-amqp-samples/stocks/pom.xml
View File

@@ -16,6 +16,7 @@
</description>
<properties>
<maven.test.failure.ignore>true</maven.test.failure.ignore>
<rabbitmq.version>1.8.0</rabbitmq.version>
<spring.framework.version>3.0.3.RELEASE</spring.framework.version>
<spring.amqp.version>1.0.0.BUILD-SNAPSHOT</spring.amqp.version>
</properties>
@@ -48,7 +49,7 @@
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>1.7.2</version>
<version>${rabbitmq.version}</version>
</dependency>
<dependency>

15
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminTemplate.java
View File

@@ -103,8 +103,11 @@ public class RabbitAdminTemplate implements RabbitAdminOperations {
return rabbitTemplate.execute(new ChannelCallback<AMQP.Queue.DeclareOk>() {
public AMQP.Queue.DeclareOk doInRabbit(Channel channel) throws Exception {
logger.debug("Declaring queue [" + queue.getName() + "]");
return channel.queueDeclare(queue.getName(), queue.isPassive(), queue.isDurable(),
queue.isExclusive(), queue.isAutoDelete(), queue.getArguments());
if (queue.isPassive()) {
return channel.queueDeclarePassive(queue.getName());
}
return channel.queueDeclare(queue.getName(), queue.isDurable(), queue.isExclusive(),
queue.isAutoDelete(), queue.getArguments());
}
});
}
@@ -113,14 +116,16 @@ public class RabbitAdminTemplate implements RabbitAdminOperations {
return rabbitTemplate.execute(new ChannelCallback<AMQP.Exchange.DeclareOk>() {
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) {

1
spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/test/RabbitTestExecutionListener.java
View File

@@ -134,7 +134,6 @@ public class RabbitTestExecutionListener extends AbstractTestExecutionListener{
}
ccf.setUsername(username);
ccf.setPassword(password);
ccf.afterPropertiesSet();
rabbitAdminTemplate = new RabbitAdminTemplate(ccf);
}
}

112
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -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;
*
* <p><b>NOTE: This ConnectionFactory requires explicit closing of all Channels obtained form its
* shared Connection.</b> 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<Class> classes = new ArrayList<Class>(1);
List<Class<?>> classes = new ArrayList<Class<?>>(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 + "]";
}
}

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactory.java
View File

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

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

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

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
View File

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

2
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/QueueUtils.java
View File

@@ -9,7 +9,7 @@ public class QueueUtils {
// declare and bind queue
template.execute(new ChannelCallback<String>() {
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);