From 9f396e888bbcc42b4c7ec5ebe643542354fb0257 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Wed, 23 Feb 2011 09:22:47 +0000 Subject: [PATCH] AMQP-44: implement some basic re-connect logic mainly to support async use cases - re-connect in listener container if channel is closed - re-connect in SingleConnectionFactory if connection is closed Both of these features are immediate (no back off) because they apply to common scenarios where the broker is fine, but there was a client problem (e.g. misuse of API) that caused the connection to be closed by the broker. Clients using RabbitTemplate will also benefit: if an operation fails they can retry it immediately and if the only problem was a closed Channel or Connection it will succeed. --- .../connection/CachingConnectionFactory.java | 127 +++++++-- .../amqp/rabbit/connection/Connection.java | 18 +- .../rabbit/connection/ConnectionFactory.java | 31 +-- .../rabbit/connection/ConnectionProxy.java | 34 +++ .../rabbit/connection/SimpleConnection.java | 24 +- .../connection/SingleConnectionFactory.java | 202 ++++++-------- .../AbstractMessageListenerContainer.java | 21 +- .../listener/BlockingQueueConsumer.java | 15 +- .../SimpleMessageListenerContainer.java | 62 +++-- ...hingConnectionFactoryIntegrationTests.java | 7 +- .../CachingConnectionFactoryTests.java | 56 ++-- .../core/RabbitBindingIntegrationTests.java | 2 +- ...itTemplatePerformanceIntegrationTests.java | 3 +- ...enerCachingConnectionIntegrationTests.java | 247 ++++++++++++++++++ ...nerContainerLifecycleIntegrationTests.java | 3 +- ...sageListenerManualAckIntegrationTests.java | 3 +- ...tenerSingleConnectionIntegrationTests.java | 15 ++ ...sageListenerContainerIntegrationTests.java | 43 +-- .../listener/UnackedRawIntegrationTests.java | 3 +- .../amqp/rabbit/test/BrokerTestUtils.java | 31 +++ 20 files changed, 713 insertions(+), 234 deletions(-) create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionProxy.java create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerCachingConnectionIntegrationTests.java create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerSingleConnectionIntegrationTests.java create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java 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 1f4b6337..cf6c7eb6 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 @@ -13,7 +13,6 @@ package org.springframework.amqp.rabbit.connection; -import java.io.IOException; import java.lang.reflect.InvocationHandler; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; @@ -22,7 +21,8 @@ import java.util.LinkedList; import javax.naming.OperationNotSupportedException; -import org.springframework.amqp.AmqpIOException; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; import org.springframework.util.Assert; @@ -54,6 +54,8 @@ import com.rabbitmq.client.Channel; // TODO are there heartbeats and/or exception thrown if a connection is broken? public class CachingConnectionFactory extends SingleConnectionFactory implements DisposableBean { + private final Log logger = LogFactory.getLog(getClass()); + private int channelCacheSize = 1; private final LinkedList cachedChannelsNonTransactional = new LinkedList(); @@ -62,6 +64,8 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements private volatile boolean active = true; + private ChannelCachingConnectionProxy targetConnection; + /** * Create a new CachingConnectionFactory initializing the hostname to be the value returned from * InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception. @@ -97,8 +101,9 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements return this.channelCacheSize; } - protected Channel getChannel(Connection connection, boolean transactional) throws IOException { - LinkedList channelList = transactional ? this.cachedChannelsTransactional : this.cachedChannelsNonTransactional; + private Channel getChannel(boolean transactional) { + LinkedList channelList = transactional ? this.cachedChannelsTransactional + : this.cachedChannelsNonTransactional; Channel channel = null; synchronized (channelList) { if (!channelList.isEmpty()) { @@ -110,33 +115,35 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements logger.trace("Found cached Rabbit Channel"); } } else { - channel = getCachedChannelProxy(connection, channelList, transactional); + channel = getCachedChannelProxy(channelList, transactional); } return channel; } - protected ChannelProxy getCachedChannelProxy(Connection connection, LinkedList channelList, boolean transactional) { - Channel targetChannel = createBareChannel(connection, transactional); + private ChannelProxy getCachedChannelProxy(LinkedList channelList, boolean transactional) { + Channel targetChannel = createBareChannel(transactional); if (logger.isDebugEnabled()) { logger.debug("Creating cached Rabbit Channel from " + targetChannel); } return (ChannelProxy) Proxy.newProxyInstance(ChannelProxy.class.getClassLoader(), - new Class[] { ChannelProxy.class }, new CachedChannelInvocationHandler(connection, targetChannel, - channelList, transactional)); + new Class[] { ChannelProxy.class }, new CachedChannelInvocationHandler(targetChannel, channelList, + transactional)); } - private Channel createBareChannel(Connection connection, boolean transactional) { - try { - return connection.createChannel(transactional); - } catch (IOException e) { - throw new AmqpIOException(e); - } + private Channel createBareChannel(boolean transactional) { + return this.targetConnection.createBareChannel(transactional); + } + + @Override + protected Connection doCreateConnection() { + targetConnection = new ChannelCachingConnectionProxy(super.doCreateConnection()); + return targetConnection; } /** * Reset the Channel cache and underlying shared Connection, to be reinitialized on next access. */ - public void resetConnection() { + protected void reset() { this.active = false; synchronized (this.cachedChannelsNonTransactional) { for (ChannelProxy channel : cachedChannelsNonTransactional) { @@ -149,7 +156,7 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements this.cachedChannelsNonTransactional.clear(); } this.active = true; - super.resetConnection(); + super.reset(); } @Override @@ -164,15 +171,12 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements private final LinkedList channelList; - private final Connection connection; - private final Object targetMonitor = new Object(); private final boolean transactional; - public CachedChannelInvocationHandler(Connection connection, Channel target, - LinkedList channelList, boolean transactional) { - this.connection = connection; + public CachedChannelInvocationHandler(Channel target, LinkedList channelList, + boolean transactional) { this.target = target; this.channelList = channelList; this.transactional = transactional; @@ -209,7 +213,13 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements } else if (methodName.equals("getTargetChannel")) { // Handle getTargetChannel method: return underlying Channel. return this.target; - } try { + } + try { + synchronized (targetMonitor) { + if (this.target == null) { + this.target = createBareChannel(transactional); + } + } return method.invoke(this.target, args); } catch (InvocationTargetException ex) { if (!this.target.isOpen()) { @@ -217,7 +227,7 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements logger.debug("Detected closed channel on exception. Re-initializing: " + target); synchronized (targetMonitor) { if (!this.target.isOpen()) { - this.target = createBareChannel(connection, transactional); + this.target = createBareChannel(transactional); } } } @@ -231,6 +241,14 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements * @param proxy the channel to close */ private void logicalClose(ChannelProxy proxy) throws Exception { + if (!this.target.isOpen()) { + synchronized (targetMonitor) { + if (!this.target.isOpen()) { + this.target = null; + return; + } + } + } // Allow for multiple close calls... if (!this.channelList.contains(proxy)) { if (logger.isTraceEnabled()) { @@ -244,15 +262,76 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements if (logger.isDebugEnabled()) { logger.debug("Closing cached Channel: " + this.target); } + if (this.target == null) { + return; + } if (this.target.isOpen()) { synchronized (targetMonitor) { if (this.target.isOpen()) { this.target.close(); } + this.target = null; } } } } + private class ChannelCachingConnectionProxy implements Connection, ConnectionProxy { + + private volatile Connection target; + + public ChannelCachingConnectionProxy(Connection target) { + this.target = target; + } + + private Channel createBareChannel(boolean transactional) { + return target.createChannel(transactional); + } + + public Channel createChannel(boolean transactional) { + Channel channel = getChannel(transactional); + return channel; + } + + public void close() { + target.close(); + } + + public boolean isOpen() { + return target!=null && target.isOpen(); + } + + public Connection getTargetConnection() { + return target; + } + + @Override + public int hashCode() { + return 31 + ((target == null) ? 0 : target.hashCode()); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) + return true; + if (obj == null) + return false; + if (getClass() != obj.getClass()) + return false; + ChannelCachingConnectionProxy other = (ChannelCachingConnectionProxy) obj; + if (target == null) { + if (other.target != null) + return false; + } else if (!target.equals(other.target)) + return false; + return true; + } + + @Override + public String toString() { + return "Shared Rabbit Connection: " + this.target; + } + + } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java index 9f4f7540..33365112 100755 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/Connection.java @@ -3,7 +3,7 @@ */ package org.springframework.amqp.rabbit.connection; -import java.io.IOException; +import org.springframework.amqp.AmqpException; import com.rabbitmq.client.Channel; @@ -17,9 +17,9 @@ public interface Connection { * Create a new channel, using an internally allocated channel number. * @param transactional true if the channel should support transactions * @return a new channel descriptor, or null if none is available - * @throws IOException if an I/O problem is encountered + * @throws AmqpException if an I/O problem is encountered */ - Channel createChannel(boolean transactional) throws IOException; + Channel createChannel(boolean transactional) throws AmqpException; /** * Close this connection and all its channels @@ -28,9 +28,15 @@ public interface Connection { * * Waits for all the close operations to complete. * - * @throws IOException if an I/O problem is encountered + * @throws AmqpException if an I/O problem is encountered */ - // TODO: throws AmqpException - void close() throws IOException; + void close() throws AmqpException; + + /** + * Flag to indicate the status of the connection. + * + * @return true if the connection is open + */ + boolean isOpen(); } 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 0887e7e4..672ec7b7 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 @@ -1,17 +1,14 @@ /* * 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. + * + * 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; @@ -19,18 +16,22 @@ package org.springframework.amqp.rabbit.connection; import java.io.IOException; /** - * An interface based ConnectionFactory for creating {@link com.rabbitmq.client.Connection}s. + * An interface based ConnectionFactory for creating {@link com.rabbitmq.client.Connection Connections}. * - *

NOTE: The Rabbit API contains a ConnectionFactory class (same name). + *

+ * NOTE: The Rabbit API contains a ConnectionFactory class (same name). * * @author Mark Fisher + * @author Dave Syer */ public interface ConnectionFactory { Connection createConnection() throws IOException; String getHost(); - + + int getPort(); + String getVirtualHost(); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionProxy.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionProxy.java new file mode 100644 index 00000000..d247054c --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionProxy.java @@ -0,0 +1,34 @@ +/* + * 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; + +/** + * Subinterface of {@link Connection} to be implemented by + * Connection proxies. Allows access to the underlying target Connection + * + * @author Dave Syer + * @see CachingConnectionFactory + */ +public interface ConnectionProxy extends Connection { + + /** + * Return the target Channel of this proxy. + *

This will typically be the native provider Connection + * @return the underlying Connection (never null) + */ + Connection getTargetConnection(); +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java index d10c30fd..3a4fa226 100755 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SimpleConnection.java @@ -17,6 +17,8 @@ package org.springframework.amqp.rabbit.connection; import java.io.IOException; +import org.springframework.amqp.rabbit.support.RabbitUtils; + import com.rabbitmq.client.Channel; public class SimpleConnection implements Connection { @@ -27,12 +29,26 @@ public class SimpleConnection implements Connection { this.delegate = delegate; } - public Channel createChannel(boolean transactional) throws IOException { - return delegate.createChannel(); + // TODO: expose the transactional flag + public Channel createChannel(boolean transactional) { + try { + return delegate.createChannel(); + } catch (IOException e) { + throw RabbitUtils.convertRabbitAccessException(e); + } } - public void close() throws IOException { - delegate.close(); + public void close() { + try { + delegate.close(); + } catch (IOException e) { + throw RabbitUtils.convertRabbitAccessException(e); + } } + public boolean isOpen() { + return delegate!=null && delegate.isOpen(); + } + + } 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 c6fdb3a9..0c12ff33 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 @@ -1,17 +1,14 @@ /* * 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. + * + * 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; @@ -30,40 +27,35 @@ import org.springframework.util.StringUtils; import com.rabbitmq.client.Channel; /** - * A {@link ConnectionFactory} implementation that returns the same Connections from all - * {@link #createConnection()} calls, and ignores calls to {@link com.rabbitmq.client.Connection#close()}. + * A {@link ConnectionFactory} implementation that returns the same Connections from all {@link #createConnection()} + * calls, and ignores calls to {@link com.rabbitmq.client.Connection#close()}. * * @author Mark Fisher * @author Mark Pollack * @author Dave Syer */ -//TODO are there heartbeats and/or exception thrown if a connection is broken? +// TODO are there heartbeats and/or exception thrown if a connection is broken? public class SingleConnectionFactory implements ConnectionFactory, DisposableBean { - protected final Log logger = LogFactory.getLog(getClass()); - - private volatile int port = RabbitUtils.DEFAULT_PORT; + private final Log logger = LogFactory.getLog(getClass()); private final com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory; - + /** Raw Rabbit Connection */ private Connection targetConnection; /** Proxy Connection */ private Connection connection; - + /** Synchronization monitor for the shared Connection */ private final Object connectionMonitor = new Object(); - /** - * Create a new SingleConnectionFactory initializing the hostname to be the - * value returned from InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws - * an exception. + * Create a new SingleConnectionFactory initializing the hostname to be the value returned from + * InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception. */ public SingleConnectionFactory() { - this.rabbitConnectionFactory = new com.rabbitmq.client.ConnectionFactory(); - this.rabbitConnectionFactory.setHost(this.getDefaultHostName()); + this((String) null); } /** @@ -87,7 +79,6 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea this.rabbitConnectionFactory = rabbitConnectionFactory; } - public void setUsername(String username) { this.rabbitConnectionFactory.setUsername(username); } @@ -112,92 +103,65 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea this.rabbitConnectionFactory.setPort(port); } - protected int getPort() { - return this.port; + public int getPort() { + return this.rabbitConnectionFactory.getPort(); } - protected Channel getChannel(Connection connection, boolean transactional) throws IOException { - return this.createChannel(connection); - } - - private Channel createChannel(Connection connection) throws IOException { - //TODO overload with channel number. - return connection.createChannel(false); - } - - public Connection createConnection() throws IOException { + public final Connection createConnection() throws IOException { synchronized (this.connectionMonitor) { if (this.connection == null) { - initConnection(); + if (this.targetConnection != null) { + RabbitUtils.closeConnection(this.targetConnection); + } + this.targetConnection = doCreateConnection(); + if (logger.isInfoEnabled()) { + logger.info("Established shared Rabbit Connection: " + this.targetConnection); + } + this.connection = new SharedConnectionProxy(this.targetConnection); } return this.connection; } } - public void initConnection() throws IOException { + /** + * Close the underlying shared connection. The provider of this ConnectionFactory needs to care for proper shutdown. + *

+ * As this bean implements DisposableBean, a bean factory will automatically invoke this on destruction of its + * cached singletons. + */ + public final void destroy() { synchronized (this.connectionMonitor) { if (this.targetConnection != null) { - closeConnection(this.targetConnection); - } - this.targetConnection = doCreateConnection(); - prepareConnection(this.targetConnection); - if (logger.isInfoEnabled()) { - logger.info("Established shared Rabbit Connection: " + this.targetConnection); - } - this.connection = getSharedConnectionProxy(this.targetConnection); - } - } - - /** - * Close the underlying shared connection. - * The provider of this ConnectionFactory needs to care for proper shutdown. - *

As this bean implements DisposableBean, a bean factory will - * automatically invoke this on destruction of its cached singletons. - */ - public void destroy() { - resetConnection(); - } - - /** - * Reset the underlying shared Connection, to be reinitialized on next access. - */ - public void resetConnection() { - synchronized (this.connectionMonitor) { - if (this.targetConnection != null) { - closeConnection(this.targetConnection); + RabbitUtils.closeConnection(this.targetConnection); } this.targetConnection = null; this.connection = null; } + reset(); } /** - * Close the given Connection. - * @param connection the Connection to close + * Default implementation does nothing. Called on {@link #destroy()}. */ - protected void closeConnection(Connection connection) { - if (logger.isDebugEnabled()) { - logger.debug("Closing shared Rabbit Connection: " + this.targetConnection); - } + protected void reset() { + } + + /** + * Create a Connection. This implementation just delegates to the underlying Rabbit ConnectionFactory. Subclasses + * typically will decorate the result to provide additional features. + * + * @return the new Connection + */ + protected Connection doCreateConnection() { + return createBareConnection(); + } + + private Connection createBareConnection() { try { - //TODO there are other close overloads close(int closeCode, java.lang.String closeMessage, int timeout) - connection.close(); + return new SimpleConnection(this.rabbitConnectionFactory.newConnection()); + } catch (IOException e) { + throw RabbitUtils.convertRabbitAccessException(e); } - 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 { - return new SimpleConnection(this.rabbitConnectionFactory.newConnection()); - } - - protected void prepareConnection(Connection con) throws IOException { - //TODO configure ShutdownListener, investigate reconnection exceptions } private String getDefaultHostName() { @@ -206,45 +170,54 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea 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 - * handle a special framework Connection just like an ordinary Connection from a - * Rabbit ConnectionFactory. - * @param target the original Connection to wrap - * @return the wrapped Connection - */ - protected Connection getSharedConnectionProxy(Connection target) { - return new SharedConnectionProxy(target); - } - @Override public String toString() { - return "SingleConnectionFactory [host=" + rabbitConnectionFactory.getHost() + ", port=" + port + "]"; + return "SingleConnectionFactory [host=" + rabbitConnectionFactory.getHost() + ", port=" + + rabbitConnectionFactory.getPort() + "]"; } - private class SharedConnectionProxy implements Connection { + /** + * Wrap a raw Connection with a proxy that delegates every method call to it but suppresses close calls. This is + * useful for allowing application code to handle a special framework Connection just like an ordinary Connection + * from a Rabbit ConnectionFactory. + */ + private class SharedConnectionProxy implements Connection, ConnectionProxy { - private final Connection target; + private volatile Connection target; public SharedConnectionProxy(Connection target) { this.target = target; } - public Channel createChannel(boolean transactional) throws IOException { - Channel channel = getChannel(this.target, transactional); + public Channel createChannel(boolean transactional) { + if (!target.isOpen()) { + synchronized (this) { + if (!target.isOpen()) { + logger.debug("Detected closed connection. Opening a new one before creating Channel."); + target = createBareConnection(); + } + } + } + Channel channel = target.createChannel(transactional); return channel; } - public void close() throws IOException { + public void close() { + } + + public boolean isOpen() { + return target != null && target.isOpen(); + } + + public Connection getTargetConnection() { + return target; } @Override @@ -264,12 +237,11 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea if (target == null) { if (other.target != null) return false; - } - else if (!target.equals(other.target)) + } else if (!target.equals(other.target)) return false; return true; } - + @Override public String toString() { return "Shared Rabbit Connection: " + this.target; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index 567a2528..90152b1b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -691,6 +691,10 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im // Not locally transacted but it is transacted so it // could be synchronized with an external transaction ConnectionFactoryUtils.registerDeliveryTag(getConnectionFactory(), channel, deliveryTag); + } else if (ackRequired) { + if (ackRequired) { + channel.basicAck(deliveryTag, false); + } } } @@ -700,7 +704,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im * @param channel the Rabbit Channel to roll back */ protected void rollbackIfNecessary(Channel channel) { - if (this.isChannelLocallyTransacted(channel)) { + boolean ackRequired = !getAcknowledgeMode().isAutoAck() && !getAcknowledgeMode().isManual(); + if (ackRequired) { /* * Re-queue messages and don't get them re-delivered to the same consumer, otherwise the broker just spins * trying to get us to accept the same message over and over @@ -710,6 +715,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im } catch (IOException e) { throw new AmqpIOException(e); } + } + if (this.isChannelLocallyTransacted(channel)) { // Transacted channel enabled by this container -> rollback. RabbitUtils.rollbackIfNecessary(channel); } @@ -723,16 +730,22 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im */ protected void rollbackOnExceptionIfNecessary(Channel channel, Message message, Throwable ex) throws Exception { - // TODO not sure if the exception at this point if only from the - // application or from Rabbit. + boolean ackRequired = !getAcknowledgeMode().isAutoAck() && !getAcknowledgeMode().isManual(); try { if (this.isChannelTransacted()) { if (logger.isDebugEnabled()) { logger.debug("Initiating transaction rollback on application exception: " + ex); } RabbitUtils.rollbackIfNecessary(channel); - if (message != null) { + } + if (message != null) { + if (ackRequired) { + if (logger.isDebugEnabled()) { + logger.debug("Rejecting message"); + } channel.basicReject(message.getMessageProperties().getDeliveryTag(), true); + } + if (this.isChannelTransacted()) { // Need to commit the reject (=nack) RabbitUtils.commitIfNecessary(channel); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index f2339628..500c31ee 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -44,6 +44,8 @@ public class BlockingQueueConsumer { private final int prefetchCount; + private final boolean transactional; + private final Channel channel; private final AtomicBoolean cancelled = new AtomicBoolean(false); @@ -52,9 +54,10 @@ public class BlockingQueueConsumer { private final AcknowledgeMode acknowledgeMode; - public BlockingQueueConsumer(Channel channel, AcknowledgeMode acknowledgeMode, int prefetchCount, String... queues) { + public BlockingQueueConsumer(Channel channel, AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, String... queues) { this.channel = channel; this.acknowledgeMode = acknowledgeMode; + this.transactional = transactional; this.prefetchCount = prefetchCount; this.queues = queues; this.consumer = new InternalConsumer(channel); @@ -72,8 +75,9 @@ public class BlockingQueueConsumer { * Check if we are in shutdown mode and if so throw an exception. */ private void checkShutdown() { - if (shutdown != null) + if (shutdown != null) { throw Utility.fixStackTrace(shutdown); + } } /** @@ -150,7 +154,7 @@ public class BlockingQueueConsumer { public void stop() { cancelled.set(true); logger.debug("Closing Rabbit Channel: " + channel); - RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), acknowledgeMode.isTransactionAllowed()); + RabbitUtils.closeMessageConsumer(consumer.getChannel(), consumer.getConsumerTag(), transactional); RabbitUtils.closeChannel(channel); } @@ -162,9 +166,11 @@ public class BlockingQueueConsumer { @Override public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) { + if (logger.isDebugEnabled()) { + logger.debug("Received shutdown for consumer tag=" + consumerTag, sig); + } shutdown = sig; // TODO: interrupt? - // TODO: is this ever used? } @Override @@ -177,7 +183,6 @@ public class BlockingQueueConsumer { } // TODO: do we want to pass on 'consumerTag'? logger.debug("Storing delivery for " + BlockingQueueConsumer.this); - checkShutdown(); try { // TODO: If transactional we could use a bounded queue and offer() here with a timeout // in which case if it fails we could nack the message and have it requeued. 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 28c21b14..4e27b13d 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 @@ -14,8 +14,6 @@ package org.springframework.amqp.rabbit.listener; import java.io.IOException; -import java.util.ArrayList; -import java.util.Collection; import java.util.HashSet; import java.util.Set; import java.util.concurrent.CountDownLatch; @@ -39,7 +37,6 @@ import org.springframework.util.Assert; import org.springframework.util.StringUtils; import com.rabbitmq.client.Channel; -import com.rabbitmq.client.ShutdownSignalException; /** * @author Mark Pollack @@ -60,7 +57,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private volatile Executor taskExecutor = new SimpleAsyncTaskExecutor(); - private volatile int concurrentConsumers = 1; + private volatile int concurrentConsumers = 0; private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; @@ -161,7 +158,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta "CachingConnectionFactory's channelCacheSize can not be less than the number of concurrentConsumers"); } // Default setting - if (concurrentConsumers == 1) { + if (concurrentConsumers < 1) { + concurrentConsumers = 1; // Set concurrent consumers to size of connection factory // channel cache. if (cf.getChannelCacheSize() > 1) { @@ -204,10 +202,16 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta protected void doStart() throws Exception { super.doStart(); initializeConsumers(); - cancellationLock = new CountDownLatch(this.consumers.size()); - for (BlockingQueueConsumer consumer : this.consumers) { - this.taskExecutor - .execute(new AsyncMessageProcessingConsumer(consumer, this.txSize, this, cancellationLock)); + synchronized (this.consumersMonitor) { + if (this.consumers == null) { + logger.info("Consumers were initialized and then cleared (presumably the container was stopped concurrently)"); + return; + } + cancellationLock = new CountDownLatch(this.consumers.size()); + for (BlockingQueueConsumer consumer : this.consumers) { + this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, this.txSize, this, + cancellationLock)); + } } } @@ -236,20 +240,18 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta logger.debug("Interrupted waiting for workers. Continuing with shutdown."); } - this.consumers = null; + synchronized (this.consumersMonitor) { + this.consumers = null; + } } protected void initializeConsumers() throws IOException { synchronized (this.consumersMonitor) { if (this.consumers == null) { - Collection channels = new ArrayList(); + this.consumers = new HashSet(this.concurrentConsumers); for (int i = 0; i < this.concurrentConsumers; i++) { Channel channel = getTransactionalResourceHolder().getChannel(); - channels.add(channel); - } - this.consumers = new HashSet(this.concurrentConsumers); - for (Channel channel : channels) { BlockingQueueConsumer consumer = createBlockingQueueConsumer(channel); this.consumers.add(consumer); } @@ -261,14 +263,34 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta return super.isChannelLocallyTransacted(channel) && this.transactionManager == null; } - protected BlockingQueueConsumer createBlockingQueueConsumer(final Channel channel) throws IOException { + protected BlockingQueueConsumer createBlockingQueueConsumer(final Channel channel) { BlockingQueueConsumer consumer; String queueNames = getRequiredQueueName(); String[] queues = StringUtils.commaDelimitedListToStringArray(queueNames); - consumer = new BlockingQueueConsumer(channel, getAcknowledgeMode(), prefetchCount, queues); + consumer = new BlockingQueueConsumer(channel, getAcknowledgeMode(), isChannelTransacted(), prefetchCount, + queues); return consumer; } + private void restart(BlockingQueueConsumer consumer) { + synchronized (this.consumersMonitor) { + if (this.consumers != null) { + try { + // Need to recycle the channel in this consumer + consumer.stop(); + } catch (Exception e) { + // Ignore + } + this.consumers.remove(consumer); + Channel channel = getTransactionalResourceHolder().getChannel(); + consumer = createBlockingQueueConsumer(channel); + this.consumers.add(consumer); + this.taskExecutor.execute(new AsyncMessageProcessingConsumer(consumer, this.txSize, this, + cancellationLock)); + } + } + } + private class AsyncMessageProcessingConsumer implements Runnable { private final BlockingQueueConsumer consumer; @@ -317,8 +339,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } catch (InterruptedException e) { logger.debug("Consumer thread interrupted, processing stopped."); Thread.currentThread().interrupt(); - } catch (ShutdownSignalException e) { - logger.debug("Consumer received ShutdownSignal, processing stopped.", e); } catch (Throwable t) { logger.debug("Consumer received fatal exception, processing stopped.", t); } finally { @@ -326,8 +346,12 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta if (!isActive()) { logger.debug("Cancelling " + consumer); consumer.stop(); + } else { + logger.debug("Restarting " + consumer); + restart(consumer); } } + } private boolean transactionalReceiveAndExecute() throws Exception { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java index 21ee48f9..a747fd21 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java @@ -14,6 +14,7 @@ import org.springframework.amqp.rabbit.core.ChannelCallback; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; import com.rabbitmq.client.Channel; @@ -29,13 +30,13 @@ public class CachingConnectionFactoryIntegrationTests { @Before public void open() { - // connectionFactory.setPort(5673); + connectionFactory.setPort(BrokerTestUtils.getPort()); } @After public void close() { // Release resources - connectionFactory.resetConnection(); + connectionFactory.reset(); } @Test @@ -61,7 +62,7 @@ public class CachingConnectionFactoryIntegrationTests { template.convertAndSend(queue.getName(), "message"); // Force a physical close of the channel - connectionFactory.resetConnection(); + connectionFactory.destroy(); // The queue was removed when the channel was closed exception.expect(AmqpIOException.class); 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 7eda6cce..c0b381b2 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 @@ -25,37 +25,38 @@ import com.rabbitmq.client.GetResponse; public class CachingConnectionFactoryTests { @Test + public void testWithConnectionFactoryDefaults() throws IOException { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); Channel mockChannel = mock(Channel.class); - + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); when(mockConnection.createChannel()).thenReturn(mockChannel); - + when(mockChannel.isOpen()).thenReturn(true); + when(mockConnection.isOpen()).thenReturn(true); + CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); Connection con = ccf.createConnection(); - + Channel channel = con.createChannel(false); channel.close(); // should be ignored, and placed into channel cache. con.close(); // should be ignored - + Connection con2 = ccf.createConnection(); - Channel channel2 = con2.createChannel(false); // will retrieve same - // channel object that - // was just put into - // channel - // cache + /* + * will retrieve same channel object that was just put into channel cache + */ + Channel channel2 = con2.createChannel(false); channel2.close(); // should be ignored con2.close(); // should be ignored - + Assert.assertSame(con, con2); Assert.assertSame(channel, channel2); verify(mockConnection, never()).close(); verify(mockChannel, never()).close(); - + } - @Test public void testWithConnectionFactoryCacheSize() throws IOException { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); @@ -64,10 +65,13 @@ public class CachingConnectionFactoryTests { Channel mockChannel2 = mock(Channel.class); when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2); when(mockChannel1.basicGet("foo", false)).thenReturn(new GetResponse(null, null, null, 1)); when(mockChannel2.basicGet("bar", false)).thenReturn(new GetResponse(null, null, null, 1)); + when(mockChannel1.isOpen()).thenReturn(true); + when(mockChannel2.isOpen()).thenReturn(true); CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setChannelCacheSize(2); @@ -115,9 +119,12 @@ public class CachingConnectionFactoryTests { when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2).thenReturn(mockChannel3); + when(mockConnection.isOpen()).thenReturn(true); // Called during physical close + when(mockChannel1.isOpen()).thenReturn(true); when(mockChannel2.isOpen()).thenReturn(true); + when(mockChannel3.isOpen()).thenReturn(true); CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setChannelCacheSize(1); @@ -130,12 +137,15 @@ public class CachingConnectionFactoryTests { Channel channel2 = con.createChannel(false); Assert.assertNotSame(channel1, channel2); - channel1.close(); // should be ignored, and add last into channel cache. - channel2.close(); // should be physically closed + // should be ignored, and added last into channel cache. + channel1.close(); + // should be physically closed + channel2.close(); - Channel ch1 = con.createChannel(false); // remove first entry in cache - // (channel1) - Channel ch2 = con.createChannel(false); // create anew channel + // remove first entry in cache (channel1) + Channel ch1 = con.createChannel(false); + // create a new channel + Channel ch2 = con.createChannel(false); Assert.assertNotSame(ch1, ch2); Assert.assertSame(ch1, channel1); @@ -151,7 +161,7 @@ public class CachingConnectionFactoryTests { verify(mockConnection, never()).close(); verify(mockChannel1, never()).close(); verify(mockChannel2, atLeastOnce()).close(); - verify(mockChannel3, never()).close(); + verify(mockChannel3, atLeastOnce()).close(); } @@ -164,8 +174,10 @@ public class CachingConnectionFactoryTests { when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2); + when(mockConnection.isOpen()).thenReturn(true); // Called during physical close + when(mockChannel1.isOpen()).thenReturn(true); when(mockChannel2.isOpen()).thenReturn(true); CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); @@ -209,8 +221,10 @@ public class CachingConnectionFactoryTests { when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2); + when(mockConnection.isOpen()).thenReturn(true); // Called during physical close + when(mockChannel1.isOpen()).thenReturn(true); when(mockChannel2.isOpen()).thenReturn(true); CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); @@ -245,7 +259,7 @@ public class CachingConnectionFactoryTests { verify(mockConnection, never()).close(); verify(mockChannel1, never()).close(); verify(mockChannel2, never()).close(); - + @SuppressWarnings("unchecked") List notxlist = (List) ReflectionTestUtils.getField(ccf, "cachedChannelsNonTransactional"); assertEquals(1, notxlist.size()); @@ -270,6 +284,10 @@ public class CachingConnectionFactoryTests { // the same method to returning different // values. when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2); + when(mockConnection.isOpen()).thenReturn(true); + // Called during physical close + when(mockChannel1.isOpen()).thenReturn(true); + when(mockChannel2.isOpen()).thenReturn(true); CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setChannelCacheSize(2); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java index fae46ce5..53412394 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java @@ -228,7 +228,7 @@ public class RabbitBindingIntegrationTests { } private BlockingQueueConsumer createConsumer(Channel channel) { - BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, AcknowledgeMode.AUTO, 1, queue.getName()); + BlockingQueueConsumer consumer = new BlockingQueueConsumer(channel, AcknowledgeMode.AUTO, true, 1, queue.getName()); consumer.start(); return consumer; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java index 5b20d196..7b09cb34 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePerformanceIntegrationTests.java @@ -9,6 +9,7 @@ import org.junit.Rule; import org.junit.Test; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; import org.springframework.amqp.rabbit.test.RepeatProcessor; import org.springframework.test.annotation.Repeat; @@ -40,7 +41,7 @@ public class RabbitTemplatePerformanceIntegrationTests { } connectionFactory = new CachingConnectionFactory(); connectionFactory.setChannelCacheSize(repeat.getConcurrency()); - // connectionFactory.setPort(5673); + connectionFactory.setPort(BrokerTestUtils.getPort()); template.setConnectionFactory(connectionFactory); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerCachingConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerCachingConnectionIntegrationTests.java new file mode 100644 index 00000000..5d39b0f4 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerCachingConnectionIntegrationTests.java @@ -0,0 +1,247 @@ +package org.springframework.amqp.rabbit.listener; + +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.log4j.Level; +import org.junit.After; +import org.junit.Rule; +import org.junit.Test; +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.Connection; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionProxy; +import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; +import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; +import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; + +import com.rabbitmq.client.Channel; + +public class MessageListenerCachingConnectionIntegrationTests { + + private static Log logger = LogFactory.getLog(MessageListenerCachingConnectionIntegrationTests.class); + + private Queue queue = new Queue("test.queue"); + + private int concurrentConsumers = 1; + + private int messageCount = 10; + + private int txSize = 1; + + private boolean transactional = false; + + private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO; + + private SimpleMessageListenerContainer container; + + @Rule + public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.DEBUG, RabbitTemplate.class, + SimpleMessageListenerContainer.class, BlockingQueueConsumer.class); + + @Rule + public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue); + + protected ConnectionFactory createConnectionFactory() { + CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); + connectionFactory.setChannelCacheSize(concurrentConsumers); + connectionFactory.setPort(BrokerTestUtils.getPort()); + return connectionFactory; + } + + @After + public void clear() throws Exception { + // Wait for broker communication to finish before trying to stop container + Thread.sleep(300L); + logger.debug("Shutting down at end of test"); + if (container != null) { + container.shutdown(); + } + } + + @Test + public void testListenerRecoversFromBogusDoubleAck() throws Exception { + + RabbitTemplate template = new RabbitTemplate(createConnectionFactory()); + + acknowledgeMode = AcknowledgeMode.MANUAL; + + CountDownLatch latch = new CountDownLatch(messageCount); + container = createContainer(new ManualAckListener(latch), createConnectionFactory()); + for (int i = 0; i < messageCount; i++) { + template.convertAndSend(queue.getName(), i + "foo"); + } + + int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + boolean waited = latch.await(timeout, TimeUnit.SECONDS); + assertTrue("Timed out waiting for message", waited); + + assertNull(template.receiveAndConvert(queue.getName())); + + } + + @Test + public void testListenerRecoversFromClosedChannel() throws Exception { + + RabbitTemplate template = new RabbitTemplate(createConnectionFactory()); + + CountDownLatch latch = new CountDownLatch(messageCount); + container = createContainer(new AbortChannelListener(latch), createConnectionFactory()); + for (int i = 0; i < messageCount; i++) { + template.convertAndSend(queue.getName(), i + "foo"); + } + + int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + boolean waited = latch.await(timeout, TimeUnit.SECONDS); + assertTrue("Timed out waiting for message", waited); + + assertNull(template.receiveAndConvert(queue.getName())); + + } + + @Test + public void testListenerRecoversFromClosedConnection() throws Exception { + + RabbitTemplate template = new RabbitTemplate(createConnectionFactory()); + + CountDownLatch latch = new CountDownLatch(messageCount); + ConnectionFactory connectionFactory = createConnectionFactory(); + container = createContainer(new CloseConnectionListener((ConnectionProxy) connectionFactory.createConnection(), + latch), connectionFactory); + for (int i = 0; i < messageCount; i++) { + template.convertAndSend(queue.getName(), i + "foo"); + } + + int timeout = Math.min(4 + messageCount / (4 * concurrentConsumers), 30); + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + boolean waited = latch.await(timeout, TimeUnit.SECONDS); + assertTrue("Timed out waiting for message", waited); + + assertNull(template.receiveAndConvert(queue.getName())); + + } + + @Test + public void testListenerRecoversAndTemplateSharesConnectionFactory() throws Exception { + + ConnectionFactory connectionFactory = createConnectionFactory(); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + + acknowledgeMode = AcknowledgeMode.MANUAL; + + CountDownLatch latch = new CountDownLatch(messageCount); + container = createContainer(new ManualAckListener(latch), connectionFactory); + for (int i = 0; i < messageCount; i++) { + template.convertAndSend(queue.getName(), i + "foo"); + } + + int timeout = Math.min(1 + messageCount / (4 * concurrentConsumers), 30); + logger.debug("Waiting for messages with timeout = " + timeout + " (s)"); + boolean waited = latch.await(timeout, TimeUnit.SECONDS); + assertTrue("Timed out waiting for message", waited); + + // TODO: is there a race condition here where the template gets a closed connection and this receive fails? + assertNull(template.receiveAndConvert(queue.getName())); + + } + + private SimpleMessageListenerContainer createContainer(Object listener, ConnectionFactory connectionFactory) { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); + container.setMessageListener(new MessageListenerAdapter(listener)); + container.setQueueName(queue.getName()); + container.setTxSize(txSize); + container.setPrefetchCount(txSize); + container.setConcurrentConsumers(concurrentConsumers); + container.setChannelTransacted(transactional); + container.setAcknowledgeMode(acknowledgeMode); + container.afterPropertiesSet(); + container.start(); + return container; + } + + public static class ManualAckListener implements ChannelAwareMessageListener { + + private AtomicBoolean failed = new AtomicBoolean(false); + + private final CountDownLatch latch; + + public ManualAckListener(CountDownLatch latch) { + this.latch = latch; + } + + public void onMessage(Message message, Channel channel) throws Exception { + String value = new String(message.getBody()); + try { + logger.debug("Acking: " + value); + channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); + if (failed.compareAndSet(false, true)) { + // intentional error (causes exception on connection thread): + channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); + } + } finally { + latch.countDown(); + } + } + } + + public static class AbortChannelListener implements ChannelAwareMessageListener { + + private AtomicBoolean failed = new AtomicBoolean(false); + + private final CountDownLatch latch; + + public AbortChannelListener(CountDownLatch latch) { + this.latch = latch; + } + + public void onMessage(Message message, Channel channel) throws Exception { + String value = new String(message.getBody()); + logger.debug("Receiving: " + value); + if (failed.compareAndSet(false, true)) { + // intentional error (causes exception on connection thread): + channel.abort(); + } else { + latch.countDown(); + } + } + } + + public static class CloseConnectionListener implements ChannelAwareMessageListener { + + private AtomicBoolean failed = new AtomicBoolean(false); + + private final CountDownLatch latch; + + private final Connection connection; + + public CloseConnectionListener(ConnectionProxy connection, CountDownLatch latch) { + this.connection = connection.getTargetConnection(); + this.latch = latch; + } + + public void onMessage(Message message, Channel channel) throws Exception { + String value = new String(message.getBody()); + logger.debug("Receiving: " + value); + if (failed.compareAndSet(false, true)) { + // intentional error (causes exception on connection thread): + connection.close(); + } else { + latch.countDown(); + } + } + } +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java index f761b024..8dec2e9b 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerLifecycleIntegrationTests.java @@ -20,6 +20,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; public class MessageListenerContainerLifecycleIntegrationTests { @@ -86,7 +87,7 @@ public class MessageListenerContainerLifecycleIntegrationTests { // SingleConnectionFactory connectionFactory = new SingleConnectionFactory(); CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); connectionFactory.setChannelCacheSize(concurrentConsumers); - // connectionFactory.setPort(5673); // For Tracer + connectionFactory.setPort(BrokerTestUtils.getPort()); template.setConnectionFactory(connectionFactory); return template; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java index bb747bc9..d260764d 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java @@ -21,6 +21,7 @@ import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; import com.rabbitmq.client.Channel; @@ -54,7 +55,7 @@ public class MessageListenerManualAckIntegrationTests { public void createConnectionFactory() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); connectionFactory.setChannelCacheSize(concurrentConsumers); - // connectionFactory.setPort(5673); + connectionFactory.setPort(BrokerTestUtils.getPort()); template.setConnectionFactory(connectionFactory); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerSingleConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerSingleConnectionIntegrationTests.java new file mode 100644 index 00000000..45d4fb62 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerSingleConnectionIntegrationTests.java @@ -0,0 +1,15 @@ +package org.springframework.amqp.rabbit.listener; + +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; + +public class MessageListenerSingleConnectionIntegrationTests extends MessageListenerCachingConnectionIntegrationTests { + + protected ConnectionFactory createConnectionFactory() { + SingleConnectionFactory connectionFactory = new SingleConnectionFactory(); + connectionFactory.setPort(BrokerTestUtils.getPort()); + return connectionFactory; + } + +} diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java index 47cf1707..aff18f97 100755 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegrationTests.java @@ -26,6 +26,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.rabbit.test.Log4jLevelAdjuster; import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionException; @@ -60,11 +61,14 @@ public class SimpleMessageListenerContainerIntegrationTests { private final boolean externalTransaction; + private final boolean transactional; + public SimpleMessageListenerContainerIntegrationTests(int messageCount, int concurrency, - AcknowledgeMode acknowledgeMode, int txSize, boolean externalTransaction) { + AcknowledgeMode acknowledgeMode, boolean transactional, int txSize, boolean externalTransaction) { this.messageCount = messageCount; this.concurrentConsumers = concurrency; this.acknowledgeMode = acknowledgeMode; + this.transactional = transactional; this.txSize = txSize; this.externalTransaction = externalTransaction; } @@ -76,19 +80,29 @@ public class SimpleMessageListenerContainerIntegrationTests { params(1, 1, 1, AcknowledgeMode.NONE), // params(2, 4, 1, AcknowledgeMode.AUTO), // extern(3, 4, 1, AcknowledgeMode.AUTO), // - params(4, 2, 2, AcknowledgeMode.AUTO), // - params(5, 2, 2, AcknowledgeMode.NONE), // - params(6, 20, 4, AcknowledgeMode.AUTO), // - params(7, 20, 4, AcknowledgeMode.NONE), // - params(8, 1000, 4, AcknowledgeMode.AUTO), // - params(9, 1000, 4, AcknowledgeMode.NONE), // - params(10, 1000, 4, AcknowledgeMode.AUTO, 10) // + params(4, 4, 1, AcknowledgeMode.AUTO, false), // + params(5, 2, 2, AcknowledgeMode.AUTO), // + params(6, 2, 2, AcknowledgeMode.NONE), // + params(7, 20, 4, AcknowledgeMode.AUTO), // + params(8, 20, 4, AcknowledgeMode.NONE), // + params(9, 1000, 4, AcknowledgeMode.AUTO), // + params(10, 1000, 4, AcknowledgeMode.NONE), // + params(11, 1000, 4, AcknowledgeMode.AUTO, 10) // ); } - private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, int txSize) { + private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, boolean transactional, int txSize) { // "i" is just a counter to make it easier to identify the test in the log - return new Object[] { messageCount, concurrency, acknowledgeMode, txSize, false }; + return new Object[] { messageCount, concurrency, acknowledgeMode, transactional, txSize, false }; + } + + private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, int txSize) { + // For this test always us a transaction if it makes sense... + return params(i, messageCount, concurrency, acknowledgeMode, acknowledgeMode.isTransactionAllowed(), txSize); + } + + private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, boolean transactional) { + return params(i, messageCount, concurrency, acknowledgeMode, transactional, 1); } private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode) { @@ -96,14 +110,14 @@ public class SimpleMessageListenerContainerIntegrationTests { } private static Object[] extern(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode) { - return new Object[] { messageCount, concurrency, acknowledgeMode, 1, true }; + return new Object[] { messageCount, concurrency, acknowledgeMode, true, 1, true }; } @Before public void declareQueue() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); connectionFactory.setChannelCacheSize(concurrentConsumers); - // connectionFactory.setPort(5673); + connectionFactory.setPort(BrokerTestUtils.getPort()); template.setConnectionFactory(connectionFactory); } @@ -124,7 +138,7 @@ public class SimpleMessageListenerContainerIntegrationTests { for (int i = 0; i < messageCount; i++) { template.convertAndSend(queue.getName(), i + "foo"); } - boolean waited = latch.await(Math.max(1, messageCount / 50), TimeUnit.SECONDS); + boolean waited = latch.await(Math.max(2, messageCount / 50), TimeUnit.SECONDS); assertTrue("Timed out waiting for message", waited); assertNull(template.receiveAndConvert(queue.getName())); } @@ -167,8 +181,7 @@ public class SimpleMessageListenerContainerIntegrationTests { container.setTxSize(txSize); container.setPrefetchCount(txSize); container.setConcurrentConsumers(concurrentConsumers); - // For this test always us a transaction if it makes sense... - container.setChannelTransacted(acknowledgeMode.isTransactionAllowed()); + container.setChannelTransacted(transactional); container.setAcknowledgeMode(acknowledgeMode); if (externalTransaction) { container.setTransactionManager(new TestTransactionManager()); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java index db274d29..1a8bdff7 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/UnackedRawIntegrationTests.java @@ -20,6 +20,7 @@ import org.junit.After; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; +import org.springframework.amqp.rabbit.test.BrokerTestUtils; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; @@ -46,7 +47,7 @@ public class UnackedRawIntegrationTests { public void init() throws Exception { factory.setHost("localhost"); - // factory.setPort(5673); + factory.setPort(BrokerTestUtils.getPort()); conn = factory.newConnection(); noTxChannel = conn.createChannel(); txChannel = conn.createChannel(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java new file mode 100644 index 00000000..0d617daa --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/test/BrokerTestUtils.java @@ -0,0 +1,31 @@ +/* + * 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.test; + +/** + * Global convenience class for all integration tests, carrying constants and other utilities for broker set up. + * + * @author Dave Syer + * + */ +public class BrokerTestUtils { + + public static final int DEFAULT_PORT = 5672; + + public static final int TRACER_PORT = 5673; + + public static int getPort() { + return DEFAULT_PORT; + } + +}