From a8d01f7c9d6ccb08ca1695adb4df2d2f3f4f7c3e Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Mon, 28 Mar 2011 12:06:45 +0100 Subject: [PATCH] AMQP-103, AMQP-126: add tests for re-declaration and fix bug exposed in tx handling --- .../CompositeConnectionListener.java | 6 + .../rabbit/connection/ConnectionListener.java | 2 + .../connection/SingleConnectionFactory.java | 1 + .../amqp/rabbit/core/RabbitAdmin.java | 4 + .../listener/BlockingQueueConsumer.java | 2 +- .../SimpleMessageListenerContainer.java | 62 ++++++---- .../CachingConnectionFactoryTests.java | 35 +++--- .../SingleConnectionFactoryTests.java | 30 +++-- .../core/RabbitAdminIntegrationTests.java | 109 ++++++++++++++++-- ...sageListenerContainerIntegrationTests.java | 54 ++++----- .../amqp/rabbit/test/BrokerTestUtils.java | 9 ++ 11 files changed, 230 insertions(+), 84 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeConnectionListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeConnectionListener.java index f448b562..20e51006 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeConnectionListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CompositeConnectionListener.java @@ -32,6 +32,12 @@ public class CompositeConnectionListener implements ConnectionListener { } } + public void onClose(Connection connection) { + for (ConnectionListener delegate : delegates) { + delegate.onClose(connection); + } + } + public void setDelegates(List delegates) { this.delegates = new ArrayList(delegates); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionListener.java index 585b10de..ac663dd2 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionListener.java @@ -23,4 +23,6 @@ public interface ConnectionListener { void onCreate(Connection connection); + void onClose(Connection connection); + } 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 495d5f28..59b5aa44 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 @@ -147,6 +147,7 @@ public class SingleConnectionFactory implements ConnectionFactory, DisposableBea public final void destroy() { synchronized (this.connectionMonitor) { if (this.targetConnection != null) { + listener.onClose(targetConnection); RabbitUtils.closeConnection(this.targetConnection); } this.targetConnection = null; diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java index e3689151..9eefac05 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java @@ -223,6 +223,10 @@ public class RabbitAdmin implements AmqpAdmin, ApplicationContextAware, Initiali initialized = true; } + public void onClose(Connection connection) { + initialized = false; + } + }); this.running = true; 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 d99bbf4f..cafe9b6e 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 @@ -124,7 +124,7 @@ public class BlockingQueueConsumer { * @throws ShutdownSignalException if the connection is shut down while waiting */ public Message nextMessage() throws InterruptedException, ShutdownSignalException { - logger.debug("Retrieving delivery for " + this); + logger.trace("Retrieving delivery for " + this); return handle(queue.take()); } 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 6312a6cf..4f522a75 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 @@ -35,10 +35,11 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.jmx.export.annotation.ManagedMetric; import org.springframework.jmx.support.MetricType; import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; -import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; import org.springframework.transaction.interceptor.TransactionAttribute; -import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; import com.rabbitmq.client.Channel; @@ -237,16 +238,10 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } private void initializeProxy() { - if (advices.length == 0 && transactionManager == null) { + if (advices.length == 0) { return; } ProxyFactory factory = new ProxyFactory(); - if (transactionManager != null) { - MatchAlwaysTransactionAttributeSource txAttributeSource = new MatchAlwaysTransactionAttributeSource(); - txAttributeSource.setTransactionAttribute(transactionAttribute); - Advice txAdvice = new TransactionInterceptor(transactionManager, txAttributeSource); - factory.addAdvisor(new DefaultPointcutAdvisor(Pointcut.TRUE, txAdvice)); - } for (Advice advice : getAdvices()) { factory.addAdvisor(new DefaultPointcutAdvisor(Pointcut.TRUE, advice)); } @@ -391,26 +386,44 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } - private boolean receiveAndExecute(BlockingQueueConsumer consumer) throws Throwable { + private boolean receiveAndExecute(final BlockingQueueConsumer consumer) throws Throwable { + + if (transactionManager != null) { + try { + return new TransactionTemplate(transactionManager, transactionAttribute) + .execute(new TransactionCallback() { + public Boolean doInTransaction(TransactionStatus status) { + ConnectionFactoryUtils.bindResourceToTransaction( + new RabbitResourceHolder(consumer.getChannel()), getConnectionFactory(), true); + try { + return doReceiveAndExecute(consumer); + } catch (RuntimeException e) { + throw e; + } catch (Throwable e) { + throw new WrappedTransactionException(e); + } + } + }); + } catch (WrappedTransactionException e) { + throw e.getCause(); + } + } + + return doReceiveAndExecute(consumer); + + } + + private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Throwable { Channel channel = consumer.getChannel(); - int totalMsgCount = 0; - - ConnectionFactory connectionFactory = getConnectionFactory(); - if (getAcknowledgeMode().isTransactionAllowed()) { - ConnectionFactoryUtils - .bindResourceToTransaction(new RabbitResourceHolder(channel), connectionFactory, true); - } - for (int i = 0; i < txSize; i++) { - logger.debug("Waiting for message from consumer."); + logger.trace("Waiting for message from consumer."); Message message = consumer.nextMessage(receiveTimeout); if (message == null) { return false; } - totalMsgCount++; executeListener(channel, message); } @@ -525,7 +538,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta } } - + @Override protected void invokeListener(Channel channel, Message message) throws Exception { proxy.invokeListener(channel, message); @@ -549,5 +562,12 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta throw new IllegalStateException("Unrecoverable interruption on consumer restart"); } } + + @SuppressWarnings("serial") + private static class WrappedTransactionException extends RuntimeException { + public WrappedTransactionException(Throwable cause) { + super(cause); + } + } } 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 c0b381b2..409f3529 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 @@ -294,19 +294,18 @@ public class CachingConnectionFactoryTests { Connection con = ccf.createConnection(); - Channel channel1 = con.createChannel(false); // This will return a - // Spring AOP proxy that - // surpresses calls to - // close - Channel channel2 = con.createChannel(false); // " + // This will return a proxy that surpresses calls to close + Channel channel1 = con.createChannel(false); + Channel channel2 = con.createChannel(false); - channel1.close(); // should be ignored, and add last into channel cache. - channel2.close(); // " + // Should be ignored, and add last into channel cache. + channel1.close(); + channel2.close(); - Channel ch1 = con.createChannel(false); // remove first entry in cache - // (channel1) - Channel ch2 = con.createChannel(false); // remove first entry in cache - // (channel2) + // remove first entry in cache (channel1) + Channel ch1 = con.createChannel(false); + // remove first entry in cache (channel2) + Channel ch2 = con.createChannel(false); Assert.assertSame(ch1, channel1); Assert.assertSame(ch2, channel2); @@ -314,9 +313,8 @@ public class CachingConnectionFactoryTests { Channel target1 = ((ChannelProxy) ch1).getTargetChannel(); Channel target2 = ((ChannelProxy) ch2).getTargetChannel(); - Assert.assertNotSame(target1, target2); // make sure mokito returned - // different mocks for the - // channel + // make sure mokito returned different mocks for the channel + Assert.assertNotSame(target1, target2); ch1.close(); ch2.close(); @@ -331,5 +329,14 @@ public class CachingConnectionFactoryTests { // verify(mockChannel1).close(); verify(mockChannel2, times(1)).close(); + // After destroy we can get a new connection + Connection con1 = ccf.createConnection(); + Assert.assertNotSame(con, con1); + + // This will return a proxy that surpresses calls to close + Channel channel3 = con.createChannel(false); + Assert.assertNotSame(channel3, channel1); + Assert.assertNotSame(channel3, channel2); } + } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java index 2d81129f..c6d8d35a 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SingleConnectionFactoryTests.java @@ -1,6 +1,8 @@ package org.springframework.amqp.rabbit.connection; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -18,29 +20,35 @@ import org.junit.Test; public class SingleConnectionFactoryTests { @Test - public void testWithListener() throws IOException { com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class); com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class); - + when(mockConnectionFactory.newConnection()).thenReturn(mockConnection); - + final AtomicBoolean called = new AtomicBoolean(false); - SingleConnectionFactory ccf = new SingleConnectionFactory(mockConnectionFactory); - ccf.setConnectionListeners(Arrays.asList(new ConnectionListener(){ + SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); + connectionFactory.setConnectionListeners(Arrays.asList(new ConnectionListener() { public void onCreate(Connection connection) { called.set(true); - } + } + public void onClose(Connection connection) { + called.set(false); + } })); - Connection con = ccf.createConnection(); - + + Connection con = connectionFactory.createConnection(); assertTrue(called.get()); - + con.close(); - + assertTrue(called.get()); verify(mockConnection, never()).close(); - + + connectionFactory.destroy(); + assertFalse(called.get()); + verify(mockConnection, atLeastOnce()).close(); + } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java index 2ac185d5..b2e46ad7 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminIntegrationTests.java @@ -1,7 +1,12 @@ package org.springframework.amqp.rabbit.core; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.After; +import org.junit.Before; import org.junit.Rule; import org.junit.Test; import org.springframework.amqp.core.Queue; @@ -10,29 +15,113 @@ import org.springframework.amqp.rabbit.test.BrokerRunning; import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.context.support.GenericApplicationContext; +import com.rabbitmq.client.AMQP.Queue.DeclareOk; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + public class RabbitAdminIntegrationTests { - private static Queue queue = new Queue("test.queue"); - private CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); - + @Rule public BrokerRunning brokerIsRunning = BrokerRunning.isRunning(); - + + private GenericApplicationContext context; + + private RabbitAdmin rabbitAdmin; + public RabbitAdminIntegrationTests() { connectionFactory.setPort(BrokerTestUtils.getPort()); } - + + @Before + public void init() { + context = new GenericApplicationContext(); + rabbitAdmin = new RabbitAdmin(connectionFactory); + rabbitAdmin.setApplicationContext(context); + rabbitAdmin.setAutoStartup(true); + } + + @After + public void close() { + if (context != null) { + context.close(); + } + } + @Test public void testStartupWithBroker() throws Exception { - GenericApplicationContext applicationContext = new GenericApplicationContext(); - applicationContext.getBeanFactory().registerSingleton("foo", queue); - RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory); - rabbitAdmin.setApplicationContext(applicationContext); - rabbitAdmin.setAutoStartup(true); + Queue queue = new Queue("test.queue"); + context.getBeanFactory().registerSingleton("foo", queue); rabbitAdmin.deleteQueue(queue.getName()); rabbitAdmin.afterPropertiesSet(); assertTrue(rabbitAdmin.deleteQueue(queue.getName())); } + @Test + public void testStartupWithAutodelete() throws Exception { + + final Queue queue = new Queue("test.queue", false, true, true); + context.getBeanFactory().registerSingleton("foo", queue); + rabbitAdmin.deleteQueue(queue.getName()); + rabbitAdmin.afterPropertiesSet(); + + final AtomicReference connectionHolder = new AtomicReference(); + + RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); + // Force RabbitAdmin to initialize the queue + boolean exists = rabbitTemplate.execute(new ChannelCallback() { + public Boolean doInRabbit(Channel channel) throws Exception { + DeclareOk result = channel.queueDeclarePassive(queue.getName()); + connectionHolder.set(channel.getConnection()); + return result != null; + } + }); + assertTrue("Expected Queue to exist", exists); + + assertTrue(queueExists(connectionHolder.get(), queue)); + connectionFactory.destroy(); + // Broker now deletes queue (only verifiable in native API) + assertFalse(queueExists(null, queue)); + + // Broker auto-deleted queue, but it is re-created by the connection listener + exists = rabbitTemplate.execute(new ChannelCallback() { + public Boolean doInRabbit(Channel channel) throws Exception { + DeclareOk result = channel.queueDeclarePassive(queue.getName()); + connectionHolder.set(channel.getConnection()); + return result != null; + } + }); + assertTrue("Expected Queue to exist", exists); + + assertTrue(queueExists(connectionHolder.get(), queue)); + assertTrue(rabbitAdmin.deleteQueue(queue.getName())); + assertFalse(queueExists(null, queue)); + + } + + /** + * Use native Rabbit API to test queue, bypassing all the connection and channel caching and callbacks in Spring + * AMQP. + * + * @param connection the raw connection to use + * @param queue the Queue to test + * @return true if the queue exists + */ + private boolean queueExists(Connection connection, Queue queue) throws Exception { + if (connection==null) { + ConnectionFactory connectionFactory = new ConnectionFactory(); + connectionFactory.setPort(BrokerTestUtils.getPort()); + connection = connectionFactory.newConnection(); + } + Channel channel = connection.createChannel(); + try { + DeclareOk result = channel.queueDeclarePassive(queue.getName()); + return result != null; + } catch (Exception e) { + return false; + } + } + } 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 a9999ea6..12eedab1 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 @@ -12,7 +12,6 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.apache.log4j.Level; import org.junit.After; import org.junit.Before; import org.junit.Rule; @@ -30,7 +29,6 @@ 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; import org.springframework.transaction.support.AbstractPlatformTransactionManager; @@ -51,9 +49,9 @@ public class SimpleMessageListenerContainerIntegrationTests { private final AcknowledgeMode acknowledgeMode; - @Rule - public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.ERROR, RabbitTemplate.class, - SimpleMessageListenerContainer.class, BlockingQueueConsumer.class); + // @Rule + // public Log4jLevelAdjuster logLevels = new Log4jLevelAdjuster(Level.ERROR, RabbitTemplate.class, + // SimpleMessageListenerContainer.class, BlockingQueueConsumer.class); @Rule public BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueue(queue); @@ -96,7 +94,8 @@ public class SimpleMessageListenerContainerIntegrationTests { ); } - private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, boolean transactional, 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, transactional, txSize, false }; } @@ -106,7 +105,8 @@ public class SimpleMessageListenerContainerIntegrationTests { return params(i, messageCount, concurrency, acknowledgeMode, acknowledgeMode.isTransactionAllowed(), txSize); } - private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, boolean transactional) { + private static Object[] params(int i, int messageCount, int concurrency, AcknowledgeMode acknowledgeMode, + boolean transactional) { return params(i, messageCount, concurrency, acknowledgeMode, transactional, 1); } @@ -141,37 +141,37 @@ public class SimpleMessageListenerContainerIntegrationTests { CountDownLatch latch = new CountDownLatch(messageCount); doSunnyDayTest(latch, new MessageListenerAdapter(new PojoListener(latch))); } - + @Test public void testListenerSunnyDay() throws Exception { CountDownLatch latch = new CountDownLatch(messageCount); doSunnyDayTest(latch, new Listener(latch)); } - + @Test public void testChannelAwareListenerSunnyDay() throws Exception { CountDownLatch latch = new CountDownLatch(messageCount); doSunnyDayTest(latch, new ChannelAwareListener(latch)); } - + @Test public void testPojoListenerWithException() throws Exception { CountDownLatch latch = new CountDownLatch(messageCount); doListenerWithExceptionTest(latch, new MessageListenerAdapter(new PojoListener(latch, true))); } - + @Test public void testListenerWithException() throws Exception { CountDownLatch latch = new CountDownLatch(messageCount); doListenerWithExceptionTest(latch, new Listener(latch, true)); } - + @Test public void testChannelAwareListenerWithException() throws Exception { CountDownLatch latch = new CountDownLatch(messageCount); doListenerWithExceptionTest(latch, new ChannelAwareListener(latch, true)); } - + private void doSunnyDayTest(CountDownLatch latch, Object listener) throws Exception { container = createContainer(listener); for (int i = 0; i < messageCount; i++) { @@ -181,7 +181,7 @@ public class SimpleMessageListenerContainerIntegrationTests { assertTrue("Timed out waiting for message", waited); assertNull(template.receiveAndConvert(queue.getName())); } - + private void doListenerWithExceptionTest(CountDownLatch latch, Object listener) throws Exception { container = createContainer(listener); if (acknowledgeMode.isTransactionAllowed()) { @@ -248,7 +248,7 @@ public class SimpleMessageListenerContainerIntegrationTests { try { int counter = count.getAndIncrement(); if (logger.isDebugEnabled() && counter % 500 == 0) { - logger.debug(value + counter); + logger.debug("Handling: " + value + ":" + counter + " - " + latch); } if (fail) { throw new RuntimeException("Planned failure"); @@ -258,23 +258,23 @@ public class SimpleMessageListenerContainerIntegrationTests { } } } - + public static class Listener implements MessageListener { private AtomicInteger count = new AtomicInteger(); - + private final CountDownLatch latch; - + private final boolean fail; - + public Listener(CountDownLatch latch) { this(latch, false); } - + public Listener(CountDownLatch latch, boolean fail) { this.latch = latch; this.fail = fail; } - + public void onMessage(Message message) { String value = new String(message.getBody()); try { @@ -290,23 +290,23 @@ public class SimpleMessageListenerContainerIntegrationTests { } } } - + public static class ChannelAwareListener implements ChannelAwareMessageListener { private AtomicInteger count = new AtomicInteger(); - + private final CountDownLatch latch; - + private final boolean fail; - + public ChannelAwareListener(CountDownLatch latch) { this(latch, false); } - + public ChannelAwareListener(CountDownLatch latch, boolean fail) { this.latch = latch; this.fail = fail; } - + public void onMessage(Message message, Channel channel) throws Exception { String value = new String(message.getBody()); try { 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 index 06d192ee..3b6992bd 100644 --- 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 @@ -39,6 +39,15 @@ public class BrokerTestUtils { return DEFAULT_PORT; } + /** + * The port that the tracer is listening on (e.g. as input for a {@link ConnectionFactory}). + * + * @return a port number + */ + public static int getTracerPort() { + return TRACER_PORT; + } + /** * An alternative port number than can safely be used to stop and start a broker, even when one is already running * on the standard port as a privileged user. Useful for tests involving {@link RabbitBrokerAdmin} on UN*X.