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

AMQP-172, AMQP-171: fix bug introduced by earlier commit (consumer not clearing its tags), and refactor connection factory implementations

Browse Source
This commit is contained in:
Dave Syer
2011-06-02 12:08:39 +01:00
parent 297f6a85c2
commit 4ed7cbe511
11 changed files with 219 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-rabbit/pom.xml
View File

@@ -47,11 +47,6 @@
</dependency>
<!-- Other -->
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>

117
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java Normal file
View File

@@ -0,0 +1,117 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* specific language governing permissions and limitations under the License.
*/
package org.springframework.amqp.rabbit.connection;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.Assert;
/**
* @author Dave Syer
*
*/
public abstract class AbstractConnectionFactory implements ConnectionFactory, DisposableBean {
protected final Log logger = LogFactory.getLog(getClass());
private final com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory;
private final CompositeConnectionListener connectionListener = new CompositeConnectionListener();
/**
* Create a new SingleConnectionFactory for the given target ConnectionFactory.
* @param rabbitConnectionFactory the target ConnectionFactory
*/
public AbstractConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory) {
Assert.notNull(rabbitConnectionFactory, "Target ConnectionFactory must not be null");
this.rabbitConnectionFactory = rabbitConnectionFactory;
}
public void setUsername(String username) {
this.rabbitConnectionFactory.setUsername(username);
}
public void setPassword(String password) {
this.rabbitConnectionFactory.setPassword(password);
}
public void setHost(String host) {
this.rabbitConnectionFactory.setHost(host);
}
public String getHost() {
return this.rabbitConnectionFactory.getHost();
}
public void setVirtualHost(String virtualHost) {
this.rabbitConnectionFactory.setVirtualHost(virtualHost);
}
public String getVirtualHost() {
return rabbitConnectionFactory.getVirtualHost();
}
public void setPort(int port) {
this.rabbitConnectionFactory.setPort(port);
}
public int getPort() {
return this.rabbitConnectionFactory.getPort();
}
/**
* A composite connection listener to be used by subclasses when creating and closing connections.
*
* @return the connection listener
*/
protected CompositeConnectionListener getConnectionListener() {
return connectionListener;
}
public void setConnectionListeners(List<? extends ConnectionListener> listeners) {
this.connectionListener.setDelegates(listeners);
}
public void addConnectionListener(ConnectionListener listener) {
this.connectionListener.addDelegate(listener);
}
final protected Connection createBareConnection() {
try {
return new SimpleConnection(this.rabbitConnectionFactory.newConnection());
} catch (IOException e) {
throw RabbitUtils.convertRabbitAccessException(e);
}
}
final protected String getDefaultHostName() {
String temp;
try {
InetAddress localMachine = InetAddress.getLocalHost();
temp = localMachine.getHostName();
logger.debug("Using hostname [" + temp + "] for hostname.");
} catch (UnknownHostException e) {
logger.warn("Could not get host name, using 'localhost' as default value", e);
temp = "localhost";
}
return temp;
}
public void destroy() {
}
}

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

@@ -20,10 +20,10 @@ import java.lang.reflect.Proxy;
import java.util.LinkedList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import com.rabbitmq.client.Channel;
@@ -46,9 +46,7 @@ import com.rabbitmq.client.Channel;
* @author Mark Fisher
* @author Dave Syer
*/
public class CachingConnectionFactory extends SingleConnectionFactory implements DisposableBean {
private final Log logger = LogFactory.getLog(getClass());
public class CachingConnectionFactory extends AbstractConnectionFactory implements DisposableBean {
private int channelCacheSize = 1;
@@ -58,7 +56,10 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
private volatile boolean active = true;
private ChannelCachingConnectionProxy targetConnection;
private ChannelCachingConnectionProxy connection;
/** Synchronization monitor for the shared Connection */
private final Object connectionMonitor = new Object();
private final CompositeChannelListener channelListener = new CompositeChannelListener();
@@ -67,16 +68,21 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
* InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.
*/
public CachingConnectionFactory() {
super();
this((String) null);
}
/**
* Create a new CachingConnectionFactory given a host name.
*
* @param hostName the host name to connect to
* @param hostname the host name to connect to
*/
public CachingConnectionFactory(String hostName, int port) {
super(hostName, port);
public CachingConnectionFactory(String hostname, int port) {
super(new com.rabbitmq.client.ConnectionFactory());
if (!StringUtils.hasText(hostname)) {
hostname = getDefaultHostName();
}
setHost(hostname);
setPort(port);
}
/**
@@ -85,16 +91,16 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
* @param hostName the host name to connect to
*/
public CachingConnectionFactory(int port) {
super(port);
this(null, port);
}
/**
* Create a new CachingConnectionFactory given a host name.
*
* @param hostName the host name to connect to
* @param hostname the host name to connect to
*/
public CachingConnectionFactory(String hostName) {
super(hostName);
public CachingConnectionFactory(String hostname) {
this(hostname, com.rabbitmq.client.ConnectionFactory.DEFAULT_AMQP_PORT);
}
/**
@@ -154,17 +160,39 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
}
private Channel createBareChannel(boolean transactional) {
if (!this.targetConnection.isOpen()) {
if (this.connection==null || !this.connection.isOpen()) {
this.connection = null;
// Use createConnection here not doCreateConnection so that the old one is properly disposed
createConnection();
}
return this.targetConnection.createBareChannel(transactional);
return this.connection.createBareChannel(transactional);
}
@Override
protected Connection doCreateConnection() {
targetConnection = new ChannelCachingConnectionProxy(super.doCreateConnection());
return targetConnection;
public final Connection createConnection() throws AmqpException {
synchronized (this.connectionMonitor) {
if (this.connection == null) {
this.connection = new ChannelCachingConnectionProxy(super.createBareConnection());
// invoke the listener *after* this.connection is assigned
getConnectionListener().onCreate(connection);
}
}
return this.connection;
}
/**
* Close the underlying shared connection. The provider of this ConnectionFactory needs to care for proper shutdown.
* <p>
* 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 (connection != null) {
this.connection.destroy();
this.connection = null;
}
}
reset();
}
/**
@@ -193,8 +221,7 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
this.cachedChannelsTransactional.clear();
}
this.active = true;
super.reset();
this.targetConnection = null;
this.connection = null;
}
@Override
@@ -333,8 +360,15 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
}
public void close() {
target.close();
}
public void destroy() {
if (this.target != null) {
getConnectionListener().onClose(target);
RabbitUtils.closeConnection(this.target);
}
reset();
this.target = null;
}
public boolean isOpen() {

36
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java
View File

@@ -24,6 +24,7 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
import org.springframework.util.Assert;
import com.rabbitmq.client.Channel;
/**
* Helper class for managing a Spring based Rabbit {@link org.springframework.amqp.rabbit.connection.ConnectionFactory},
* in particular for obtaining transactional Rabbit resources for a given ConnectionFactory.
@@ -81,29 +82,26 @@ public class ConnectionFactoryUtils {
final boolean synchedLocalTransactionAllowed) {
RabbitResourceHolder holder = doGetTransactionalResourceHolder(connectionFactory, new ResourceFactory() {
public Channel getChannel(RabbitResourceHolder holder) {
return holder.getChannel();
}
public Channel getChannel(RabbitResourceHolder holder) {
return holder.getChannel();
}
public Connection getConnection(RabbitResourceHolder holder) {
return holder.getConnection();
}
public Connection getConnection(RabbitResourceHolder holder) {
return holder.getConnection();
}
public Connection createConnection() throws IOException {
return connectionFactory.createConnection();
}
public Connection createConnection() throws IOException {
return connectionFactory.createConnection();
}
public Channel createChannel(Connection con) throws IOException {
return con.createChannel(synchedLocalTransactionAllowed);
}
public Channel createChannel(Connection con) throws IOException {
return con.createChannel(synchedLocalTransactionAllowed);
}
public boolean isSynchedLocalTransactionAllowed() {
return synchedLocalTransactionAllowed;
}
});
if (synchedLocalTransactionAllowed) {
// holder.declareTransactional();
}
public boolean isSynchedLocalTransactionAllowed() {
return synchedLocalTransactionAllowed;
}
});
return holder;
}

12
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java
View File

@@ -176,18 +176,6 @@ public class RabbitResourceHolder extends ResourceHolderSupport {
}
}
/**
* Call this method once the channel {@link Channel#txSelect()} has been called.
*/
public void declareTransactional() {
if (!isSynchronizedWithTransaction() && !transactional) {
for (Channel channel : this.channels) {
RabbitUtils.declareTransactional(channel);
}
this.transactional = !channels.isEmpty();
}
}
/**
* @return true if the channels in this holder are transactional
*/

14
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java
View File

@@ -15,13 +15,7 @@ package org.springframework.amqp.rabbit.connection;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.net.ConnectException;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -30,19 +24,11 @@ import org.springframework.amqp.AmqpException;
import org.springframework.amqp.AmqpIOException;
import org.springframework.amqp.AmqpUnsupportedEncodingException;
import org.springframework.amqp.UncategorizedAmqpException;
import org.springframework.amqp.core.Address;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.impl.LongString;
/**
* @author Mark Fisher

99
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactory.java
View File

@@ -13,16 +13,9 @@
package org.springframework.amqp.rabbit.connection;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import com.rabbitmq.client.Channel;
@@ -35,11 +28,7 @@ import com.rabbitmq.client.Channel;
* @author Mark Pollack
* @author Dave Syer
*/
public class SingleConnectionFactory implements ConnectionFactory, DisposableBean {
private final Log logger = LogFactory.getLog(getClass());
private final com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory;
public class SingleConnectionFactory extends AbstractConnectionFactory {
/** Proxy Connection */
private SharedConnectionProxy connection;
@@ -47,8 +36,6 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
/** Synchronization monitor for the shared Connection */
private final Object connectionMonitor = new Object();
private final CompositeConnectionListener connectionListener = new CompositeConnectionListener();
/**
* Create a new SingleConnectionFactory initializing the hostname to be the value returned from
* InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.
@@ -79,12 +66,12 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
* @param port the port number to connect to
*/
public SingleConnectionFactory(String hostname, int port) {
super(new com.rabbitmq.client.ConnectionFactory());
if (!StringUtils.hasText(hostname)) {
hostname = getDefaultHostName();
}
this.rabbitConnectionFactory = new com.rabbitmq.client.ConnectionFactory();
this.rabbitConnectionFactory.setHost(hostname);
this.rabbitConnectionFactory.setPort(port);
setHost(hostname);
setPort(port);
}
/**
@@ -92,52 +79,19 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
* @param rabbitConnectionFactory the target ConnectionFactory
*/
public SingleConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory) {
Assert.notNull(rabbitConnectionFactory, "Target ConnectionFactory must not be null");
this.rabbitConnectionFactory = rabbitConnectionFactory;
}
public void setUsername(String username) {
this.rabbitConnectionFactory.setUsername(username);
}
public void setPassword(String password) {
this.rabbitConnectionFactory.setPassword(password);
}
public void setHost(String host) {
this.rabbitConnectionFactory.setHost(host);
}
public String getHost() {
return this.rabbitConnectionFactory.getHost();
}
public void setVirtualHost(String virtualHost) {
this.rabbitConnectionFactory.setVirtualHost(virtualHost);
}
public String getVirtualHost() {
return rabbitConnectionFactory.getVirtualHost();
}
public void setPort(int port) {
this.rabbitConnectionFactory.setPort(port);
}
public int getPort() {
return this.rabbitConnectionFactory.getPort();
super(rabbitConnectionFactory);
}
public void setConnectionListeners(List<? extends ConnectionListener> listeners) {
this.connectionListener.setDelegates(listeners);
super.setConnectionListeners(listeners);
// If the connection is already alive we assume that the new listeners want to be notified
if (this.connection != null) {
this.connectionListener.onCreate(this.connection);
this.getConnectionListener().onCreate(this.connection);
}
}
public void addConnectionListener(ConnectionListener listener) {
this.connectionListener.addDelegate(listener);
super.addConnectionListener(listener);
// If the connection is already alive we assume that the new listener wants to be notified
if (this.connection != null) {
listener.onCreate(this.connection);
@@ -150,7 +104,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
Connection target = doCreateConnection();
this.connection = new SharedConnectionProxy(target);
// invoke the listener *after* this.connection is assigned
connectionListener.onCreate(target);
getConnectionListener().onCreate(target);
}
}
return this.connection;
@@ -169,13 +123,6 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
this.connection = null;
}
}
reset();
}
/**
* Default implementation does nothing. Called on {@link #destroy()}.
*/
protected void reset() {
}
/**
@@ -189,31 +136,9 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
return connection;
}
private Connection createBareConnection() {
try {
return new SimpleConnection(this.rabbitConnectionFactory.newConnection());
} catch (IOException e) {
throw RabbitUtils.convertRabbitAccessException(e);
}
}
private String getDefaultHostName() {
String temp;
try {
InetAddress localMachine = InetAddress.getLocalHost();
temp = localMachine.getHostName();
logger.debug("Using hostname [" + temp + "] for hostname.");
} catch (UnknownHostException e) {
logger.warn("Could not get host name, using 'localhost' as default value", e);
temp = "localhost";
}
return temp;
}
@Override
public String toString() {
return "SingleConnectionFactory [host=" + rabbitConnectionFactory.getHost() + ", port="
+ rabbitConnectionFactory.getPort() + "]";
return "SingleConnectionFactory [host=" + getHost() + ", port=" + getPort() + "]";
}
/**
@@ -235,7 +160,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
if (!isOpen()) {
logger.debug("Detected closed connection. Opening a new one before creating Channel.");
target = createBareConnection();
connectionListener.onCreate(target);
getConnectionListener().onCreate(target);
}
}
}
@@ -248,7 +173,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea
public void destroy() {
if (this.target != null) {
connectionListener.onClose(target);
getConnectionListener().onClose(target);
RabbitUtils.closeConnection(this.target);
}
this.target = null;

9
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

@@ -18,7 +18,8 @@ package org.springframework.amqp.rabbit.listener;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@@ -80,7 +81,7 @@ public class BlockingQueueConsumer {
private final ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter;
private List<Long> deliveryTags = new ArrayList<Long>();
private Set<Long> deliveryTags = new LinkedHashSet<Long>();
/**
* Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker
@@ -175,6 +176,7 @@ public class BlockingQueueConsumer {
.getChannel();
this.consumer = new InternalConsumer(channel);
this.activeObjectCounter.add(this);
this.deliveryTags.clear();
try {
// Set basicQos before calling basicConsume (it is ignored if we are not transactional and the broker will
// send blocks of 100 messages)
@@ -207,6 +209,7 @@ public class BlockingQueueConsumer {
logger.debug("Closing Rabbit Channel: " + channel);
// This one never throws exceptions...
RabbitUtils.closeChannel(channel);
deliveryTags.clear();
}
private class InternalConsumer extends DefaultConsumer {
@@ -353,7 +356,7 @@ public class BlockingQueueConsumer {
} else {
if (!deliveryTags.isEmpty()) {
long deliveryTag = deliveryTags.get(deliveryTags.size()-1);
long deliveryTag = new ArrayList<Long>(deliveryTags).get(deliveryTags.size()-1);
channel.basicAck(deliveryTag, true);
}

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java
View File

@@ -40,6 +40,7 @@ import org.springframework.amqp.core.HeadersExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -165,7 +166,7 @@ public class AmqpAppender extends AppenderSkeleton {
/**
* RabbitMQ ConnectionFactory.
*/
private CachingConnectionFactory connectionFactory;
private AbstractConnectionFactory connectionFactory;
/**
* RabbitMQ host to connect to.
*/

5
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java
View File

@@ -14,7 +14,6 @@ import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.AmqpIOException;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.ChannelCallback;
@@ -151,8 +150,8 @@ public class CachingConnectionFactoryIntegrationTests {
return null;
}
});
fail("Expected AmqpConnectException");
} catch (AmqpConnectException e) {
fail("Expected AmqpIOException");
} catch (AmqpIOException e) {
// expected
}
template.convertAndSend(route, "message");

11
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitBindingIntegrationTests.java
View File

@@ -18,6 +18,7 @@ import org.springframework.amqp.rabbit.listener.ActiveObjectCounter;
import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.test.BrokerRunning;
import org.springframework.amqp.rabbit.test.BrokerTestUtils;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import com.rabbitmq.client.Channel;
@@ -26,7 +27,7 @@ public class RabbitBindingIntegrationTests {
private static Queue queue = new Queue("test.queue");
private ConnectionFactory connectionFactory = new CachingConnectionFactory();
private ConnectionFactory connectionFactory = new CachingConnectionFactory(BrokerTestUtils.getPort());
private RabbitTemplate template = new RabbitTemplate(connectionFactory );
@@ -145,7 +146,7 @@ public class RabbitBindingIntegrationTests {
result = getResult(consumer);
assertEquals("message", result);
consumer.getChannel().basicCancel(consumer.getConsumerTag());
consumer.stop();
}
@@ -171,7 +172,7 @@ public class RabbitBindingIntegrationTests {
String result = getResult(consumer);
assertEquals(null, result);
} finally {
channel.basicCancel(tag);
consumer.stop();
}
return null;
@@ -191,7 +192,7 @@ public class RabbitBindingIntegrationTests {
String result = getResult(consumer);
assertEquals("message", result);
} finally {
channel.basicCancel(tag);
consumer.stop();
}
return null;
@@ -223,7 +224,7 @@ public class RabbitBindingIntegrationTests {
String result = getResult(consumer);
assertEquals("message", result);
} finally {
channel.basicCancel(tag);
consumer.stop();
}
return null;