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

AMQP-68: Add @Ignored tests for mixed tx/notx

Browse Source
This commit is contained in:
Dave Syer
2010-11-16 09:57:56 +00:00
parent 31a164e4c0
commit a26a32fe2e
3 changed files with 298 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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;
@@ -31,35 +28,30 @@ import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
* NOTE: this ConnectionFactory implementation is considered <b>experimental</b>
* at this stage. There are concerns to be addressed in relation to the
* statefulness of channels. Therefore, we recommend using
* {@link SingleConnectionFactory} for now.
* NOTE: this ConnectionFactory implementation is considered <b>experimental</b> at this stage. There are concerns to be
* addressed in relation to the statefulness of channels. Therefore, we recommend using {@link SingleConnectionFactory}
* for now.
*
* A {@link ConnectionFactory} implementation that returns the same Connections
* from all {@link #createConnection()} calls, and ignores calls to
* {@link com.rabbitmq.client.Connection#close()} and caches
* A {@link ConnectionFactory} implementation that returns the same Connections from all {@link #createConnection()}
* calls, and ignores calls to {@link com.rabbitmq.client.Connection#close()} and caches
* {@link com.rabbitmq.client.Channel}.
*
* <p>
* By default, only one single Session will be cached, with further requested
* Channels being created and disposed on demand. Consider raising the
* {@link #setChannelCacheSize(int) "channelCacheSize" value} in case of a
* high-concurrency environment.
* By default, only one single Session will be cached, with further requested Channels being created and disposed on
* demand. Consider raising the {@link #setChannelCacheSize(int) "channelCacheSize" value} in case of a high-concurrency
* environment.
*
* <p>
* <b>NOTE: This ConnectionFactory requires explicit closing of all Channels
* obtained form its shared Connection.</b> This is the usual recommendation for
* native Rabbit access code anyway. However, with this ConnectionFactory, its
* use is mandatory in order to actually allow for Channel reuse.
* <b>NOTE: This ConnectionFactory requires explicit closing of all Channels obtained form its shared Connection.</b>
* This is the usual recommendation for native Rabbit access code anyway. However, with this ConnectionFactory, its use
* is mandatory in order to actually allow for Channel reuse.
*
* @author Mark Pollack
* @author Mark Fisher
* @author Dave Syer
*/
// TODO are there heartbeats and/or exception thrown if a connection is broken?
public class CachingConnectionFactory extends SingleConnectionFactory implements
DisposableBean {
public class CachingConnectionFactory extends SingleConnectionFactory implements DisposableBean {
private int channelCacheSize = 1;
@@ -68,9 +60,8 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
private volatile boolean active = true;
/**
* Create a new CachingConnectionFactory initializing the hostname to be the
* value returned from InetAddress.getLocalHost(), or "localhost" if
* getLocalHost() throws an exception.
* Create a new CachingConnectionFactory initializing the hostname to be the value returned from
* InetAddress.getLocalHost(), or "localhost" if getLocalHost() throws an exception.
*/
public CachingConnectionFactory() {
super();
@@ -79,28 +70,23 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
/**
* 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);
}
/**
* Create a new CachingConnectionFactory for the given target
* ConnectionFactory.
* Create a new CachingConnectionFactory for the given target ConnectionFactory.
*
* @param rabbitConnectionFactory
* the target ConnectionFactory
* @param rabbitConnectionFactory the target ConnectionFactory
*/
public CachingConnectionFactory(
com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory) {
public CachingConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory) {
super(rabbitConnectionFactory);
}
public void setChannelCacheSize(int sessionCacheSize) {
Assert.isTrue(sessionCacheSize >= 1,
"Channel cache size must be 1 or higher");
Assert.isTrue(sessionCacheSize >= 1, "Channel cache size must be 1 or higher");
this.channelCacheSize = sessionCacheSize;
}
@@ -126,15 +112,13 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
return channel;
}
protected ChannelProxy getCachedChannelProxy(Connection connection,
LinkedList<ChannelProxy> channelList) {
protected ChannelProxy getCachedChannelProxy(Connection connection, LinkedList<ChannelProxy> channelList) {
Channel targetChannel = createBareChannel(connection);
if (logger.isDebugEnabled()) {
logger.debug("Creating cached Rabbit Channel");
logger.debug("Creating cached Rabbit Channel from " + targetChannel);
}
return (ChannelProxy) Proxy.newProxyInstance(ChannelProxy.class
.getClassLoader(), new Class[] { ChannelProxy.class },
new CachedChannelInvocationHandler(connection, targetChannel,
return (ChannelProxy) Proxy.newProxyInstance(ChannelProxy.class.getClassLoader(),
new Class[] { ChannelProxy.class }, new CachedChannelInvocationHandler(connection, targetChannel,
channelList));
}
@@ -147,8 +131,7 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
}
/**
* Reset the Channel cache and underlying shared Connection, to be
* reinitialized on next access.
* Reset the Channel cache and underlying shared Connection, to be reinitialized on next access.
*/
public void resetConnection() {
this.active = false;
@@ -168,9 +151,8 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
@Override
public String toString() {
return "CachingConnectionFactory [channelCacheSize=" + channelCacheSize
+ ", host=" + this.getHost() + ", port=" + this.getPort()
+ ", active=" + active + "]";
return "CachingConnectionFactory [channelCacheSize=" + channelCacheSize + ", host=" + this.getHost()
+ ", port=" + this.getPort() + ", active=" + active + "]";
}
private class CachedChannelInvocationHandler implements InvocationHandler {
@@ -183,15 +165,14 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
private final Object targetMonitor = new Object();
public CachedChannelInvocationHandler(Connection connection,
Channel target, LinkedList<ChannelProxy> channelList) {
public CachedChannelInvocationHandler(Connection connection, Channel target,
LinkedList<ChannelProxy> channelList) {
this.connection = connection;
this.target = target;
this.channelList = channelList;
}
public Object invoke(Object proxy, Method method, Object[] args)
throws Throwable {
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
if (methodName.equals("equals")) {
// Only consider equal when proxies are identical.
@@ -217,13 +198,14 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
physicalClose();
return null;
} else if (methodName.equals("getTargetChannel")) {
// Handle getTargetSession method: return underlying Channel.
// Handle getTargetChannel method: return underlying Channel.
return this.target;
}
try {
} try {
return method.invoke(this.target, args);
} catch (InvocationTargetException ex) {
if (!this.target.isOpen()) {
// Basic re-connection logic...
logger.debug("Detected closed channel on exception. Re-initializing: " + target);
synchronized (targetMonitor) {
if (!this.target.isOpen()) {
this.target = createBareChannel(connection);
@@ -237,8 +219,7 @@ public class CachingConnectionFactory extends SingleConnectionFactory implements
/**
* GUARDED by channelList
*
* @param proxy
* the channel to close
* @param proxy the channel to close
*/
private void logicalClose(ChannelProxy proxy) throws Exception {
// Allow for multiple close calls...

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

@@ -3,16 +3,20 @@ package org.springframework.amqp.rabbit.connection;
import static org.junit.Assert.assertEquals;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.springframework.amqp.AmqpIOException;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
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 com.rabbitmq.client.Channel;
public class CachingConnectionFactoryIntegrationTests {
private CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
@@ -23,6 +27,11 @@ public class CachingConnectionFactoryIntegrationTests {
@Rule
public ExpectedException exception = ExpectedException.none();
@Before
public void open() {
connectionFactory.setPort(5673);
}
@After
public void close() {
// Release resources
@@ -62,4 +71,32 @@ public class CachingConnectionFactoryIntegrationTests {
}
@Test
@Ignore // TODO: add this feature
public void testMixTransactionalAndNonTransactional() throws Exception {
RabbitTemplate template1 = new RabbitTemplate(connectionFactory);
RabbitTemplate template2 = new RabbitTemplate(connectionFactory);
template1.setChannelTransacted(true);
RabbitAdmin admin = new RabbitAdmin(template1);
Queue queue = admin.declareQueue();
template1.convertAndSend(queue.getName(), "message");
String result = (String) template2.receiveAndConvert(queue.getName());
assertEquals("message", result);
// The channel is not transactional
exception.expect(AmqpIOException.class);
template2.execute(new ChannelCallback<Void>() {
public Void doInRabbit(Channel channel) throws Exception {
// Should be an exception because the channel is not transactional
channel.txRollback();
return null;
}
});
}
}

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

@@ -1,15 +1,22 @@
package org.springframework.amqp.rabbit.connection;
import static org.mockito.Mockito.*;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
import junit.framework.Assert;
import org.junit.Ignore;
import org.junit.Test;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.GetResponse;
/**
* @author Mark Pollack
@@ -21,120 +28,267 @@ public class CachingConnectionFactoryTests {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel);
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
Connection con = ccf.createConnection();
Channel channel = con.createChannel();
channel.close(); // should be ignored, and placed into channel cache.
con.close(); // should be ignored
con.close(); // should be ignored
Connection con2 = ccf.createConnection();
Channel channel2 = con2.createChannel(); // will retrieve same channel object that was just put into channel cache
Channel channel2 = con2.createChannel(); // will retrieve same channel object that was just put into channel
// cache
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);
Connection mockConnection = mock(Connection.class);
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel1);
when(mockConnection.createChannel()).thenReturn(mockChannel2);
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));
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
ccf.setChannelCacheSize(2);
Connection con = ccf.createConnection();
Channel channel1 = con.createChannel();
Channel channel2 = con.createChannel();
channel1.basicGet("foo", true);
channel2.basicGet("bar", true);
channel1.close(); // should be ignored, and add last into channel cache.
channel2.close(); // should be ignored, and add last into channel cache.
Channel ch1 = con.createChannel(); // remove first entry in cache (channel1)
Channel ch2 = con.createChannel(); // remove first entry in cache (channel2)
Channel ch1 = con.createChannel(); // remove first entry in cache (channel1)
Channel ch2 = con.createChannel(); // remove first entry in cache (channel2)
Assert.assertNotSame(ch1, ch2);
Assert.assertSame(ch1, channel1);
Assert.assertSame(ch2, channel2);
ch1.close();
ch2.close();
verify(mockConnection, times(2)).createChannel();
con.close(); // should be ignored
verify(mockConnection, never()).close();
verify(mockConnection, times(2)).createChannel();
con.close(); // should be ignored
verify(mockConnection, never()).close();
verify(mockChannel1, never()).close();
verify(mockChannel2, never()).close();
}
@Test
public void testWithConnectionFactoryDestory() throws IOException {
public void testCacheSizeExceeded() throws IOException {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
Assert.assertNotSame(mockChannel1, mockChannel2);
Channel mockChannel3 = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
//You can't repeat 'when' statements for stubbing consecutive calls to the same method to returning different values.
stub(mockConnection.createChannel()).toReturn(mockChannel1).toReturn(mockChannel2);
when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2).thenReturn(mockChannel3);
// Called during physical close
when(mockChannel2.isOpen()).thenReturn(true);
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
ccf.setChannelCacheSize(1);
Connection con = ccf.createConnection();
Channel channel1 = con.createChannel();
// cache size is 1, but the other connection is not released yet so this creates a new one
Channel channel2 = con.createChannel();
Assert.assertNotSame(channel1, channel2);
channel1.close(); // should be ignored, and add last into channel cache.
channel2.close(); // should be physically closed
Channel ch1 = con.createChannel(); // remove first entry in cache (channel1)
Channel ch2 = con.createChannel(); // create anew channel
Assert.assertNotSame(ch1, ch2);
Assert.assertSame(ch1, channel1);
Assert.assertNotSame(ch2, channel2);
ch1.close();
ch2.close();
verify(mockConnection, times(3)).createChannel();
con.close(); // should be ignored
verify(mockConnection, never()).close();
verify(mockChannel1, never()).close();
verify(mockChannel2, atLeastOnce()).close();
verify(mockChannel3, never()).close();
}
@Test
public void testCacheSizeExceededAfterClose() throws IOException {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
// Called during physical close
when(mockChannel2.isOpen()).thenReturn(true);
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
ccf.setChannelCacheSize(1);
Connection con = ccf.createConnection();
Channel channel1 = con.createChannel();
channel1.close(); // should be ignored, and add last into channel cache.
Channel channel2 = con.createChannel();
channel2.close(); // should be ignored, and add last into channel cache.
Assert.assertSame(channel1, channel2);
Channel ch1 = con.createChannel(); // remove first entry in cache (channel1)
Channel ch2 = con.createChannel(); // create new channel
Assert.assertNotSame(ch1, ch2);
Assert.assertSame(ch1, channel1);
Assert.assertNotSame(ch2, channel2);
ch1.close();
ch2.close();
verify(mockConnection, times(2)).createChannel();
con.close(); // should be ignored
verify(mockConnection, never()).close();
verify(mockChannel1, never()).close();
verify(mockChannel2, atLeastOnce()).close();
}
@Test
@Ignore // TODO: add this feature
public void testTransactionalAndNonTransactional() throws IOException {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
// Called during physical close
when(mockChannel2.isOpen()).thenReturn(true);
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
ccf.setChannelCacheSize(1);
Connection con = ccf.createConnection();
Channel channel1 = con.createChannel();
channel1.txSelect();
channel1.close(); // should be ignored, and add last into channel cache.
/*
* When a channel is created we can't assume it should be transactional, so this should create a new one.
*/
Channel channel2 = con.createChannel();
channel2.close(); // should be ignored, and add last into channel cache.
Assert.assertNotSame(channel1, channel2);
Channel ch1 = con.createChannel(); // remove first entry in cache (channel1)
Channel ch2 = con.createChannel(); // create new channel
Assert.assertNotSame(ch1, ch2);
Assert.assertSame(ch1, channel2); // The non-transactional one
Assert.assertNotSame(ch2, channel2);
ch1.close();
ch2.close();
verify(mockConnection, times(2)).createChannel();
con.close(); // should be ignored
verify(mockConnection, never()).close();
verify(mockChannel1, never()).close();
verify(mockChannel2, atLeastOnce()).close();
}
@Test
public void testWithConnectionFactoryDestroy() throws IOException {
com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
Assert.assertNotSame(mockChannel1, mockChannel2);
when(mockConnectionFactory.newConnection()).thenReturn(mockConnection);
// You can't repeat 'when' statements for stubbing consecutive calls to the same method to returning different
// values.
when(mockConnection.createChannel()).thenReturn(mockChannel1).thenReturn(mockChannel2);
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
ccf.setChannelCacheSize(2);
Connection con = ccf.createConnection();
Channel channel1 = con.createChannel(); // This will return a Spring AOP proxy that surpresses calls to close
Channel channel2 = con.createChannel(); // "
Channel channel1 = con.createChannel(); // This will return a Spring AOP proxy that surpresses calls to close
Channel channel2 = con.createChannel(); // "
channel1.close(); // should be ignored, and add last into channel cache.
channel2.close(); // "
Channel ch1 = con.createChannel(); // remove first entry in cache (channel1)
Channel ch2 = con.createChannel(); // remove first entry in cache (channel2)
channel2.close(); // "
Channel ch1 = con.createChannel(); // remove first entry in cache (channel1)
Channel ch2 = con.createChannel(); // remove first entry in cache (channel2)
Assert.assertSame(ch1, channel1);
Assert.assertSame(ch2, channel2);
Channel target1 = ((ChannelProxy)ch1).getTargetChannel();
Channel target2 = ((ChannelProxy)ch2).getTargetChannel();
Assert.assertNotSame(target1, target2); // make sure mokito returned different mocks for the channel
Channel target1 = ((ChannelProxy) ch1).getTargetChannel();
Channel target2 = ((ChannelProxy) ch2).getTargetChannel();
Assert.assertNotSame(target1, target2); // make sure mokito returned different mocks for the channel
ch1.close();
ch2.close();
con.close(); // should be ignored
ccf.destroy(); // should call close on connection and channels in cache
ch2.close();
con.close(); // should be ignored
ccf.destroy(); // should call close on connection and channels in cache
verify(mockConnection, times(2)).createChannel();
verify(mockConnection, times(1)).close();
//verify(mockChannel1).close();
verify(mockConnection, times(1)).close();
// verify(mockChannel1).close();
verify(mockChannel2, times(1)).close();
}
}