From 257a2d3dab7235e95acc5b84d959a4f21581acfa Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 22 Feb 2017 17:41:06 -0500 Subject: [PATCH] INT-4226: Fix PollableAmqpChannel.receive(timeout) JIRA: https://jira.spring.io/browse/INT-4226 Since `rabbitTemplate` properly implement `receive()` with `timeout` it looks organic to delegate a `PollableChannel.receive(timeout)` there. * Refactor `PollableAmqpChannel` to properly delegate `PollableChannel.receive(timeout)` implementation * Refactor `ChannelTests` to perform `PollableAmqpChannel.receive(timeout)` directly instead of `Thread.sleep()` loop * Get rid of `BrokerRunning` in favor of the same rule from the `spring-rabbit-junit` dependency Doc Polishing Fix DSL AmqpTests --- build.gradle | 1 + .../amqp/channel/PollableAmqpChannel.java | 51 ++++--- .../amqp/channel/ChannelTests.java | 34 ++--- .../OutboundGatewayIntegrationTests.java | 6 +- .../integration/amqp/dsl/AmqpTests.java | 23 ++- .../amqp/inbound/ManualAckTests.java | 2 +- .../outbound/AmqpOutboundEndpointTests.java | 4 +- .../amqp/outbound/AsyncAmqpGatewayTests.java | 2 +- .../integration/amqp/rule/BrokerRunning.java | 144 ------------------ .../JsonConverterCompatibilityTests.java | 4 +- src/reference/asciidoc/amqp.adoc | 6 + src/reference/asciidoc/whats-new.adoc | 3 + 12 files changed, 79 insertions(+), 201 deletions(-) delete mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java diff --git a/build.gradle b/build.gradle index ed836930c1..d78cfb9ff4 100644 --- a/build.gradle +++ b/build.gradle @@ -283,6 +283,7 @@ project('spring-integration-amqp') { exclude group: 'org.springframework', module: 'spring-core' exclude group: 'org.springframework', module: 'spring-tx' } + testCompile("org.springframework.amqp:spring-rabbit-junit:$springAmqpVersion") testCompile project(":spring-integration-stream") testCompile project(":spring-integration-http") // need to test INT-2713 } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index f655f34f7d..726cf4d7de 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -43,6 +43,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Artem Bilan * @author Gary Russell + * * @since 2.1 */ public class PollableAmqpChannel extends AbstractAmqpChannel @@ -151,6 +152,16 @@ public class PollableAmqpChannel extends AbstractAmqpChannel @Override public Message receive() { + return doReceive(null); + } + + @Override + public Message receive(long timeout) { + return doReceive(timeout); + } + + + protected Message doReceive(Long timeout) { ChannelInterceptorList interceptorList = getInterceptors(); Deque interceptorStack = null; boolean counted = false; @@ -160,13 +171,13 @@ public class PollableAmqpChannel extends AbstractAmqpChannel logger.trace("preReceive on channel '" + this + "'"); } if (interceptorList.getInterceptors().size() > 0) { - interceptorStack = new ArrayDeque(); + interceptorStack = new ArrayDeque<>(); if (!interceptorList.preReceive(this, interceptorStack)) { return null; } } - Object object = doReceive(); + Object object = performReceive(timeout); if (object == null) { if (isLoggingEnabled() && logger.isTraceEnabled()) { logger.trace("postReceive on channel '" + this + "', message is null"); @@ -177,12 +188,14 @@ public class PollableAmqpChannel extends AbstractAmqpChannel getMetrics().afterReceive(); counted = true; } - Message message = null; + Message message; if (object instanceof Message) { message = (Message) object; } else { - message = getMessageBuilderFactory().withPayload(object).build(); + message = getMessageBuilderFactory() + .withPayload(object) + .build(); } if (isLoggingEnabled() && logger.isDebugEnabled()) { logger.debug("postReceive on channel '" + this + "', message: " + message); @@ -204,14 +217,25 @@ public class PollableAmqpChannel extends AbstractAmqpChannel } } - - protected Object doReceive() { + protected Object performReceive(Long timeout) { if (!isExtractPayload()) { - return getAmqpTemplate().receiveAndConvert(this.queueName); + if (timeout == null) { + return getAmqpTemplate().receiveAndConvert(this.queueName); + } + else { + return getAmqpTemplate().receiveAndConvert(this.queueName, timeout); + } } else { RabbitTemplate rabbitTemplate = getRabbitTemplate(); - org.springframework.amqp.core.Message message = rabbitTemplate.receive(this.queueName); + org.springframework.amqp.core.Message message; + if (timeout == null) { + message = rabbitTemplate.receive(this.queueName); + } + else { + message = rabbitTemplate.receive(this.queueName, timeout); + } + if (message != null) { Object payload = rabbitTemplate.getMessageConverter().fromMessage(message); Map headers = getInboundHeaderMapper() @@ -227,15 +251,6 @@ public class PollableAmqpChannel extends AbstractAmqpChannel } } - @Override - public Message receive(long timeout) { - if (isLoggingEnabled() && logger.isInfoEnabled()) { - logger.info("Calling receive with a timeout value on PollableAmqpChannel. " + - "The timeout will be ignored since no receive timeout is supported."); - } - return this.receive(); - } - @Override public void setInterceptors(List interceptors) { super.setInterceptors(interceptors); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java index 805e698800..0bdaeb7d91 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java @@ -46,6 +46,7 @@ import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.BrokerRunning; import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.support.converter.MessageConversionException; @@ -53,10 +54,8 @@ import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.amqp.config.AmqpChannelFactoryBean; -import org.springframework.integration.amqp.rule.BrokerRunning; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.support.LogAdjustingTestSupport; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; @@ -69,13 +68,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Gary Russell * @author Artem Bilan + * * @since 4.0 * */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD) -public class ChannelTests extends LogAdjustingTestSupport { +public class ChannelTests { @ClassRule public static final BrokerRunning brokerIsRunning = @@ -108,16 +108,9 @@ public class ChannelTests extends LogAdjustingTestSupport { @Autowired private AmqpHeaderMapper mapperOut; - public ChannelTests() { - super("org.springframework.integration", "org.springframework.integration.amqp", "org.springframework.amqp"); - } - @After public void tearDown() { - RabbitAdmin rabbitAdmin = new RabbitAdmin(this.factory); - rabbitAdmin.deleteExchange("si.fanout.foo"); - rabbitAdmin.deleteExchange("si.fanout.channel"); - rabbitAdmin.deleteExchange("si.fanout.pubSubWithEP"); + brokerIsRunning.deleteExchanges("si.fanout.foo", "si.fanout.channel", "si.fanout.pubSubWithEP"); brokerIsRunning.removeTestQueues(); } @@ -220,27 +213,22 @@ public class ChannelTests extends LogAdjustingTestSupport { Foo foo = new Foo("bar"); Message message = MessageBuilder.withPayload(foo).setHeader("baz", "qux").build(); this.pollableWithEP.send(message); - Message received = this.pollableWithEP.receive(); - int n = 0; - while (received == null && n++ < 100) { - Thread.sleep(100); - received = this.pollableWithEP.receive(); - } + Message received = this.pollableWithEP.receive(10000); assertNotNull(received); - assertThat((Foo) received.getPayload(), equalTo(foo)); - assertThat((String) received.getHeaders().get("baz"), equalTo("qux")); + assertThat(received.getPayload(), equalTo(foo)); + assertThat(received.getHeaders().get("baz"), equalTo("qux")); this.withEP.send(message); received = this.out.receive(10000); assertNotNull(received); - assertThat((Foo) received.getPayload(), equalTo(foo)); - assertThat((String) received.getHeaders().get("baz"), equalTo("qux")); + assertThat(received.getPayload(), equalTo(foo)); + assertThat(received.getHeaders().get("baz"), equalTo("qux")); this.pubSubWithEP.send(message); received = this.out.receive(10000); assertNotNull(received); - assertThat((Foo) received.getPayload(), equalTo(foo)); - assertThat((String) received.getHeaders().get("baz"), equalTo("qux")); + assertThat(received.getPayload(), equalTo(foo)); + assertThat(received.getHeaders().get("baz"), equalTo("qux")); assertSame(this.mapperIn, TestUtils.getPropertyValue(this.pollableWithEP, "inboundHeaderMapper")); assertSame(this.mapperOut, TestUtils.getPropertyValue(this.pollableWithEP, "outboundHeaderMapper")); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java index c2bd39bda8..424bc7e5fd 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2017 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. @@ -23,8 +23,8 @@ import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.amqp.rabbit.junit.BrokerRunning; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.integration.amqp.rule.BrokerRunning; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; @@ -64,9 +64,11 @@ public class OutboundGatewayIntegrationTests { public static class EchoBean { + String echo(String o) { return o.toUpperCase(); } + } } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java index 612ce100cc..07fe64e543 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2017 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. @@ -21,7 +21,8 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; -import org.junit.Rule; +import org.junit.AfterClass; +import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; @@ -34,13 +35,13 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.BrokerRunning; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.amqp.channel.AbstractAmqpChannel; import org.springframework.integration.amqp.inbound.AmqpInboundGateway; -import org.springframework.integration.amqp.rule.BrokerRunning; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.channel.QueueChannel; @@ -65,8 +66,8 @@ import org.springframework.test.context.junit4.SpringRunner; @DirtiesContext public class AmqpTests { - @Rule - public BrokerRunning brokerRunning = BrokerRunning.isRunning(); + @ClassRule + public static BrokerRunning brokerRunning = BrokerRunning.isRunning(); @Autowired private ConnectionFactory rabbitConnectionFactory; @@ -81,6 +82,12 @@ public class AmqpTests { @Autowired private AmqpInboundGateway amqpInboundGateway; + @AfterClass + public static void tearDown() { + brokerRunning.removeTestQueues("amqpOutboundInput", "amqpReplyChannel", "asyncReplies", "defaultReplyTo", + "si.dsl.test", "testTemplateChannelTransacted"); + } + @Test public void testAmqpInboundGatewayFlow() throws Exception { Object result = this.amqpTemplate.convertSendAndReceive(this.amqpQueue.getName(), "world"); @@ -108,7 +115,7 @@ public class AmqpTests { @Test public void testAmqpOutboundFlow() throws Exception { this.amqpOutboundInput.send(MessageBuilder.withPayload("hello through the amqp") - .setHeader("routingKey", "foo") + .setHeader("routingKey", "si.dsl.test") .build()); Message receive = null; int i = 0; @@ -219,7 +226,7 @@ public class AmqpTests { @Bean public Queue fooQueue() { - return new Queue("foo"); + return new Queue("si.dsl.test"); } @Bean @@ -257,7 +264,7 @@ public class AmqpTests { @Bean public AbstractAmqpChannel unitChannel(ConnectionFactory rabbitConnectionFactory) { return Amqp.pollableChannel(rabbitConnectionFactory) - .queueName("foo") + .queueName("si.dsl.test") .channelTransacted(true) .extractPayload(true) .inboundHeaderMapper(mapperIn()) diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java index 55acaa024a..e160b00119 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java @@ -31,6 +31,7 @@ import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.BrokerRunning; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.beans.factory.BeanFactory; @@ -39,7 +40,6 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.integration.amqp.inbound.ManualAckTests.ManualAckConfig; -import org.springframework.integration.amqp.rule.BrokerRunning; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.QueueChannel; diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java index aec0ab5919..e2a997c149 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -27,9 +27,9 @@ import org.junit.runner.RunWith; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.BrokerRunning; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.integration.amqp.rule.BrokerRunning; import org.springframework.integration.mapping.support.JsonHeaders; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java index 04c9a45800..cbed32f6fa 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java @@ -46,6 +46,7 @@ import org.springframework.amqp.rabbit.AsyncRabbitTemplate; import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.BrokerRunning; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener; @@ -53,7 +54,6 @@ import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; -import org.springframework.integration.amqp.rule.BrokerRunning; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.support.MessageBuilder; diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java deleted file mode 100644 index 71e55e4ebe..0000000000 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java +++ /dev/null @@ -1,144 +0,0 @@ -/* - * Copyright 2014-2017 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.integration.amqp.rule; - -import java.util.ArrayList; -import java.util.List; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.junit.Assume; -import org.junit.rules.TestWatcher; -import org.junit.runner.Description; -import org.junit.runners.model.Statement; - -import org.springframework.amqp.core.Queue; -import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; -import org.springframework.amqp.rabbit.core.RabbitAdmin; -import org.springframework.util.Assert; - -/** - *

- * A rule that prevents integration tests from failing if the Rabbit broker application is not running or not - * accessible. If the Rabbit broker is not running in the background all the tests here will simply be skipped because - * of a violated assumption (showing as successful). - *

- * The rule can be declared as static so that it only has to check once for all tests in the enclosing test case, but - * there isn't a lot of overhead in making it non-static. - * - * @author Dave Syer - * @author Artem Bilan - * - * @since 4.0 - */ -public class BrokerRunning extends TestWatcher { - - public static final int PORT = 5672; - - private static final Log logger = LogFactory.getLog(BrokerRunning.class); - - private static final Queue DEFAULT_QUEUE_NAME = new Queue(BrokerRunning.class.getName()); - - private final Queue[] queues; - - /** - * Ensure the broker is running and has an empty queue (which can be addressed via the default exchange). - * - * @return a new rule that assumes an existing running broker - */ - public static BrokerRunning isRunningWithEmptyQueues(Queue... queues) { - Assert.notNull(queues, "'queues' must not be null"); - Assert.noNullElements(queues, "'queues' must not contain null elements"); - return new BrokerRunning(queues); - } - - /** - * Ensure the broker is running and has an empty queue (which can be addressed via the default exchange). - * - * @return a new rule that assumes an existing running broker - */ - public static BrokerRunning isRunningWithEmptyQueues(String... queues) { - Assert.notNull(queues, "'queues' must not be null"); - Assert.noNullElements(queues, "'queues' must not contain null elements"); - return new BrokerRunning(queues); - } - - /** - * @return a new rule that assumes an existing running broker - */ - public static BrokerRunning isRunning() { - return new BrokerRunning(DEFAULT_QUEUE_NAME); - } - - - private BrokerRunning(Queue... queues) { - this.queues = queues; - } - - private BrokerRunning(String... queues) { - List queueList = new ArrayList(queues.length); - for (String queue : queues) { - queueList.add(new Queue(queue)); - } - this.queues = queueList.toArray(new Queue[queues.length]); - } - - @Override - public Statement apply(Statement base, Description description) { - CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); - connectionFactory.setHost("localhost"); - - try { - - connectionFactory.setPort(PORT); - - RabbitAdmin admin = new RabbitAdmin(connectionFactory); - - for (Queue queue : queues) { - String queueName = queue.getName(); - logger.info("Deleting queue: " + queueName); - // Delete completely - gets rid of consumers and bindings as well - admin.deleteQueue(queueName); - - if (!DEFAULT_QUEUE_NAME.getName().equals(queueName)) { - admin.declareQueue(queue); - } - } - - - } - catch (final Exception e) { - logger.warn("Not executing tests because basic connectivity test failed", e); - Assume.assumeNoException(e); - } - finally { - connectionFactory.destroy(); - } - - return super.apply(base, description); - } - - public void removeTestQueues() { - CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost"); - RabbitAdmin admin = new RabbitAdmin(connectionFactory); - for (Queue queue : this.queues) { - admin.deleteQueue(queue.getName()); - } - connectionFactory.destroy(); - } - -} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java index dbfa4b89e9..a891ad7dcd 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -27,9 +27,9 @@ import org.junit.Test; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.BrokerRunning; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; -import org.springframework.integration.amqp.rule.BrokerRunning; import org.springframework.integration.json.ObjectToJsonTransformer; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index c3e7327fe6..7f2fe5608d 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -1223,6 +1223,12 @@ IMPORTANT: Just as with other persistence-backed channels, AMQP-backed channels persistence to avoid message loss. They are not intended to distribute work to other peer applications; for that purpose, use channel adapters instead. +IMPORTANT: Starting with _version 5.0_, the pollable channel now blocks the poller thread for the specified `receiveTimeout` (default 1 second). +Previously, unlike other `PollableChannel` s, the thread returned immediately to the scheduler if no message was available, regardless of the receive timeout. +Blocking is a little more expensive than just using a `basicGet()` to retrieve a message (with no timeout) because a consumer has to be created to receive each message. +To restore the previous behavior, set the poller `receiveTimeout` to 0. + + ==== Configuring with Java Configuration The following provides an example of configuring the channels using Java configuration: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 8554f7f041..6dcb4fb97e 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -105,6 +105,9 @@ See <> for more information. The inbound endpoints now support the Spring AMQP `DirectMessageListenerContainer`. See <> for more information. +Pollable AMQP-backed channels now block the poller thread for the poller's configured `receiveTimeout` (default 1 second). +See <> for more information. + ==== HTTP Changes The `DefaultHttpHeaderMapper.userDefinedHeaderPrefix` property is now an empty string by default instead of `X-`.