From 7fd87ed71a2ca97484e53eda4a536001ce5672d1 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 6 Jul 2010 15:44:27 -0400 Subject: [PATCH] AMQP-3 upgraded to RabbitMQ 1.8.0 Java client --- spring-amqp-parent/pom.xml | 2 +- spring-amqp-samples/helloworld/pom.xml | 3 +- spring-amqp-samples/stocks/pom.xml | 3 +- .../rabbit/admin/RabbitAdminTemplate.java | 15 ++- .../test/RabbitTestExecutionListener.java | 1 - .../connection/CachingConnectionFactory.java | 112 ++++++------------ .../rabbit/connection/ConnectionFactory.java | 6 +- .../SimpleMessageListenerContainer.java | 2 +- .../CachingConnectionFactoryTests.java | 10 +- .../amqp/rabbit/core/QueueUtils.java | 2 +- 10 files changed, 58 insertions(+), 98 deletions(-) diff --git a/spring-amqp-parent/pom.xml b/spring-amqp-parent/pom.xml index cf3c72d9..2c8bf045 100644 --- a/spring-amqp-parent/pom.xml +++ b/spring-amqp-parent/pom.xml @@ -19,7 +19,7 @@ 1.5.10 1.4.3 1.5.3 - 1.7.2 + 1.8.0 3.0.3.RELEASE diff --git a/spring-amqp-samples/helloworld/pom.xml b/spring-amqp-samples/helloworld/pom.xml index 3bf0f38a..32ca1791 100644 --- a/spring-amqp-samples/helloworld/pom.xml +++ b/spring-amqp-samples/helloworld/pom.xml @@ -16,6 +16,7 @@ true + 1.8.0 3.0.3.RELEASE 1.0.0.BUILD-SNAPSHOT @@ -48,7 +49,7 @@ com.rabbitmq amqp-client - 1.7.2 + ${rabbitmq.version} diff --git a/spring-amqp-samples/stocks/pom.xml b/spring-amqp-samples/stocks/pom.xml index e782d245..838e4a1d 100644 --- a/spring-amqp-samples/stocks/pom.xml +++ b/spring-amqp-samples/stocks/pom.xml @@ -16,6 +16,7 @@ true + 1.8.0 3.0.3.RELEASE 1.0.0.BUILD-SNAPSHOT @@ -48,7 +49,7 @@ com.rabbitmq amqp-client - 1.7.2 + ${rabbitmq.version} diff --git a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminTemplate.java b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminTemplate.java index b039b832..4d1a047b 100644 --- a/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminTemplate.java +++ b/spring-rabbit-admin/src/main/java/org/springframework/amqp/rabbit/admin/RabbitAdminTemplate.java @@ -103,8 +103,11 @@ public class RabbitAdminTemplate implements RabbitAdminOperations { return rabbitTemplate.execute(new ChannelCallback() { 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() { 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);