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

INT-4226: Fix PollableAmqpChannel.receive(timeout)

Browse Source 
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
This commit is contained in:
Artem Bilan
2017-02-22 17:41:06 -05:00
committed by Gary Russell
parent 2d4a7cf92f
commit 257a2d3dab
12 changed files with 79 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
build.gradle
View File

@@ -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
}

51
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java
View File

@@ -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<ChannelInterceptor> 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<ChannelInterceptor>();
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<String, Object> 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<ChannelInterceptor> interceptors) {
super.setInterceptors(interceptors);

34
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java
View File

@@ -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"));

6
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java
View File

@@ -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();
}
}
}

23
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java
View File

@@ -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())

2
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java
View File

@@ -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;

4
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java
View File

@@ -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;

2
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java
View File

@@ -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;

144
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/rule/BrokerRunning.java
View File

@@ -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;
/**
* <p>
* 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).
* <p>
* 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<Queue> queueList = new ArrayList<Queue>(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();
}
}

4
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java
View File

@@ -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;

6
src/reference/asciidoc/amqp.adoc
View File

@@ -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:

3
src/reference/asciidoc/whats-new.adoc
View File

@@ -105,6 +105,9 @@ See <<amqp-delay>> for more information.
The inbound endpoints now support the Spring AMQP `DirectMessageListenerContainer`.
See <<amqp-inbound-channel-adapter>> for more information.
Pollable AMQP-backed channels now block the poller thread for the poller's configured `receiveTimeout` (default 1 second).
See <<amqp-channels>> for more information.
==== HTTP Changes
The `DefaultHttpHeaderMapper.userDefinedHeaderPrefix` property is now an empty string by default instead of `X-`.