From 9052377bf537b086667a4cc5e421d49aaa78148b Mon Sep 17 00:00:00 2001 From: Gary russell Date: Thu, 24 Apr 2014 18:00:52 +0300 Subject: [PATCH] INT-3385 AMQP Support MANUAL Ack Mode JIRA:https://jira.spring.io/browse/INT-3385 MANUAL was supported in the schema, but there was no mechanism for the application to ack. Polishing - Fix tests - Add integration test - More documentation --- .../integration/amqp/AmqpHeaders.java | 6 +- .../inbound/AmqpInboundChannelAdapter.java | 17 +- .../amqp/inbound/AmqpInboundGateway.java | 17 +- .../AmqpInboundChannelAdapterParserTests.java | 48 +++-- .../config/AmqpInboundGatewayParserTests.java | 7 +- .../amqp/inbound/InboundEndpointTests.java | 35 +++- .../amqp/inbound/ManualAckTests.java | 161 ++++++++++++++++ src/reference/docbook/amqp.xml | 180 ++++++++++++------ 8 files changed, 371 insertions(+), 100 deletions(-) create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java index 0b17413950..8a59c41c6e 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2014 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. @@ -16,8 +16,8 @@ package org.springframework.integration.amqp; -import org.springframework.messaging.MessageHeaders; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.messaging.MessageHeaders; /** * Pre-defined names and prefixes to be used for setting and/or retrieving AMQP @@ -87,6 +87,8 @@ public abstract class AmqpHeaders { public static final String RETURN_ROUTING_KEY = PREFIX + "returnRoutingKey"; + public static final String CHANNEL = PREFIX + "channel"; + /** * Compatibility with Spring-AMQP 1.1 * This was previously in RabbitTemplate diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index aabf3b3645..2af63205e2 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -18,17 +18,21 @@ package org.springframework.integration.amqp.inbound; import java.util.Map; +import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Message; -import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.integration.amqp.AmqpHeaders; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.util.Assert; +import com.rabbitmq.client.Channel; + /** * Adapter that receives Messages from an AMQP Queue, converts them into * Spring Integration Messages, and sends the results to a Message Channel. @@ -73,11 +77,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements @Override protected void onInit() { - this.messageListenerContainer.setMessageListener(new MessageListener() { + this.messageListenerContainer.setMessageListener(new ChannelAwareMessageListener() { + @Override - public void onMessage(Message message) { + public void onMessage(Message message, Channel channel) throws Exception { Object payload = messageConverter.fromMessage(message); - Map headers = headerMapper.toHeadersFromRequest(message.getMessageProperties()); + Map headers = headerMapper.toHeadersFromRequest(message.getMessageProperties()); + if (messageListenerContainer.getAcknowledgeMode() == AcknowledgeMode.MANUAL) { + headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag()); + headers.put(AmqpHeaders.CHANNEL, channel); + } sendMessage(AmqpInboundChannelAdapter.this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build()); } }); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index 0ae8349c2b..e9f9711345 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java @@ -19,21 +19,25 @@ package org.springframework.integration.amqp.inbound; import java.util.Map; import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Address; import org.springframework.amqp.core.Message; -import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.integration.amqp.AmqpHeaders; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.util.Assert; import org.springframework.util.StringUtils; +import com.rabbitmq.client.Channel; + /** * Adapter that receives Messages from an AMQP Queue, converts them into * Spring Integration Messages, and sends the results to a Message Channel. @@ -82,11 +86,15 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { @Override protected void onInit() throws Exception { - this.messageListenerContainer.setMessageListener(new MessageListener() { + this.messageListenerContainer.setMessageListener(new ChannelAwareMessageListener() { @Override - public void onMessage(Message message) { + public void onMessage(Message message, Channel channel) { Object payload = amqpMessageConverter.fromMessage(message); - Map headers = headerMapper.toHeadersFromRequest(message.getMessageProperties()); + Map headers = headerMapper.toHeadersFromRequest(message.getMessageProperties()); + if (messageListenerContainer.getAcknowledgeMode() == AcknowledgeMode.MANUAL) { + headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag()); + headers.put(AmqpHeaders.CHANNEL, channel); + } org.springframework.messaging.Message request = AmqpInboundGateway.this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build(); final org.springframework.messaging.Message reply = sendAndReceiveMessage(request); @@ -119,6 +127,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport { }); } } + }); this.messageListenerContainer.afterPropertiesSet(); this.amqpTemplate.afterPropertiesSet(); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java index 9afec9b10c..c3f3ad24a0 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -26,8 +26,8 @@ import org.junit.runner.RunWith; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Message; -import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; @@ -44,6 +44,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Mark Fisher * @author Artem Bilan + * @author Gary Russell * @since 2.1 */ @ContextConfiguration @@ -72,12 +73,14 @@ public class AmqpInboundChannelAdapterParserTests { } @Test - public void withHeaderMapperStandardAndCustomHeaders() { - AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperStandardAndCustomHeaders", AmqpInboundChannelAdapter.class); + public void withHeaderMapperStandardAndCustomHeaders() throws Exception { + AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperStandardAndCustomHeaders", + AmqpInboundChannelAdapter.class); AbstractMessageListenerContainer mlc = TestUtils.getPropertyValue(adapter, "messageListenerContainer", AbstractMessageListenerContainer.class); - MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class); + ChannelAwareMessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", + ChannelAwareMessageListener.class); MessageProperties amqpProperties = new MessageProperties(); amqpProperties.setAppId("test.appId"); amqpProperties.setClusterId("test.clusterId"); @@ -87,7 +90,7 @@ public class AmqpInboundChannelAdapterParserTests { amqpProperties.setHeader("foo", "foo"); amqpProperties.setHeader("bar", "bar"); Message amqpMessage = new Message("hello".getBytes(), amqpProperties); - listener.onMessage(amqpMessage); + listener.onMessage(amqpMessage, null); QueueChannel requestChannel = context.getBean("requestChannel", QueueChannel.class); org.springframework.messaging.Message siMessage = requestChannel.receive(0); assertEquals("foo", siMessage.getHeaders().get("foo")); @@ -99,12 +102,14 @@ public class AmqpInboundChannelAdapterParserTests { } @Test - public void withHeaderMapperOnlyCustomHeaders() { - AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperOnlyCustomHeaders", AmqpInboundChannelAdapter.class); + public void withHeaderMapperOnlyCustomHeaders() throws Exception { + AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperOnlyCustomHeaders", + AmqpInboundChannelAdapter.class); AbstractMessageListenerContainer mlc = TestUtils.getPropertyValue(adapter, "messageListenerContainer", AbstractMessageListenerContainer.class); - MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class); + ChannelAwareMessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", + ChannelAwareMessageListener.class); MessageProperties amqpProperties = new MessageProperties(); amqpProperties.setAppId("test.appId"); amqpProperties.setClusterId("test.clusterId"); @@ -114,7 +119,7 @@ public class AmqpInboundChannelAdapterParserTests { amqpProperties.setHeader("foo", "foo"); amqpProperties.setHeader("bar", "bar"); Message amqpMessage = new Message("hello".getBytes(), amqpProperties); - listener.onMessage(amqpMessage); + listener.onMessage(amqpMessage, null); QueueChannel requestChannel = context.getBean("requestChannel", QueueChannel.class); org.springframework.messaging.Message siMessage = requestChannel.receive(0); assertEquals("foo", siMessage.getHeaders().get("foo")); @@ -126,12 +131,14 @@ public class AmqpInboundChannelAdapterParserTests { } @Test - public void withHeaderMapperNothingToMap() { - AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperNothingToMap", AmqpInboundChannelAdapter.class); + public void withHeaderMapperNothingToMap() throws Exception { + AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperNothingToMap", + AmqpInboundChannelAdapter.class); AbstractMessageListenerContainer mlc = TestUtils.getPropertyValue(adapter, "messageListenerContainer", AbstractMessageListenerContainer.class); - MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class); + ChannelAwareMessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", + ChannelAwareMessageListener.class); MessageProperties amqpProperties = new MessageProperties(); amqpProperties.setAppId("test.appId"); amqpProperties.setClusterId("test.clusterId"); @@ -141,7 +148,7 @@ public class AmqpInboundChannelAdapterParserTests { amqpProperties.setHeader("foo", "foo"); amqpProperties.setHeader("bar", "bar"); Message amqpMessage = new Message("hello".getBytes(), amqpProperties); - listener.onMessage(amqpMessage); + listener.onMessage(amqpMessage, null); QueueChannel requestChannel = context.getBean("requestChannel", QueueChannel.class); org.springframework.messaging.Message siMessage = requestChannel.receive(0); @@ -154,12 +161,14 @@ public class AmqpInboundChannelAdapterParserTests { } @Test - public void withHeaderMapperDefaultMapping() { - AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperDefaultMapping", AmqpInboundChannelAdapter.class); + public void withHeaderMapperDefaultMapping() throws Exception { + AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperDefaultMapping", + AmqpInboundChannelAdapter.class); AbstractMessageListenerContainer mlc = TestUtils.getPropertyValue(adapter, "messageListenerContainer", AbstractMessageListenerContainer.class); - MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class); + ChannelAwareMessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", + ChannelAwareMessageListener.class); MessageProperties amqpProperties = new MessageProperties(); amqpProperties.setAppId("test.appId"); amqpProperties.setClusterId("test.clusterId"); @@ -169,7 +178,7 @@ public class AmqpInboundChannelAdapterParserTests { amqpProperties.setHeader("foo", "foo"); amqpProperties.setHeader("bar", "bar"); Message amqpMessage = new Message("hello".getBytes(), amqpProperties); - listener.onMessage(amqpMessage); + listener.onMessage(amqpMessage, null); QueueChannel requestChannel = context.getBean("requestChannel", QueueChannel.class); org.springframework.messaging.Message siMessage = requestChannel.receive(0); assertNull(siMessage.getHeaders().get("bar")); @@ -183,7 +192,8 @@ public class AmqpInboundChannelAdapterParserTests { @Test public void testInt2971HeaderMapperAndMappedHeadersExclusivity() { try { - new ClassPathXmlApplicationContext("AmqpInboundChannelAdapterParserTests-headerMapper-fail-context.xml", this.getClass()); + new ClassPathXmlApplicationContext("AmqpInboundChannelAdapterParserTests-headerMapper-fail-context.xml", + this.getClass()); } catch (BeanDefinitionParsingException e) { assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java index cc37e874d4..a6716e95c0 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java @@ -29,8 +29,8 @@ import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.amqp.core.Message; -import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.support.CorrelationData; @@ -123,7 +123,8 @@ public class AmqpInboundGatewayParserTests { AbstractMessageListenerContainer mlc = TestUtils.getPropertyValue(gateway, "messageListenerContainer", AbstractMessageListenerContainer.class); - MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class); + ChannelAwareMessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", + ChannelAwareMessageListener.class); MessageProperties amqpProperties = new MessageProperties(); amqpProperties.setAppId("test.appId"); amqpProperties.setClusterId("test.clusterId"); @@ -134,7 +135,7 @@ public class AmqpInboundGatewayParserTests { amqpProperties.setHeader("foo", "foo"); amqpProperties.setHeader("bar", "bar"); Message amqpMessage = new Message("hello".getBytes(), amqpProperties); - listener.onMessage(amqpMessage); + listener.onMessage(amqpMessage, null); Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(Message.class), Mockito.any(CorrelationData.class)); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index c87b5aec1d..41dcf1b95b 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -19,6 +19,7 @@ package org.springframework.integration.amqp.inbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.anyBoolean; import static org.mockito.Mockito.doAnswer; @@ -32,10 +33,11 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; -import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.support.CorrelationData; @@ -43,6 +45,7 @@ import org.springframework.amqp.support.converter.JsonMessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.amqp.AmqpHeaders; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; @@ -67,7 +70,7 @@ import com.rabbitmq.client.Channel; public class InboundEndpointTests { @Test - public void testInt2809JavaTypePropertiesToAmqp() { + public void testInt2809JavaTypePropertiesToAmqp() throws Exception { Connection connection = mock(Connection.class); doAnswer(new Answer() { @Override @@ -79,6 +82,7 @@ public class InboundEndpointTests { when(connectionFactory.createConnection()).thenReturn(connection); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); + container.setAcknowledgeMode(AcknowledgeMode.MANUAL); AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container); adapter.setMessageConverter(new JsonMessageConverter()); @@ -95,19 +99,24 @@ public class InboundEndpointTests { Message jsonMessage = objectToJsonTransformer.transform(new GenericMessage(payload)); MessageProperties amqpMessageProperties = new MessageProperties(); + amqpMessageProperties.setDeliveryTag(123L); org.springframework.amqp.core.Message amqpMessage = new SimpleMessageConverter().toMessage(jsonMessage.getPayload(), amqpMessageProperties); new DefaultAmqpHeaderMapper().fromHeadersToRequest(jsonMessage.getHeaders(), amqpMessageProperties); - MessageListener listener = (MessageListener) container.getMessageListener(); - listener.onMessage(amqpMessage); + ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener(); + Channel rabbitChannel = mock(Channel.class); + listener.onMessage(amqpMessage, rabbitChannel); Message result = channel.receive(1000); assertEquals(payload, result.getPayload()); + + assertSame(rabbitChannel, result.getHeaders().get(AmqpHeaders.CHANNEL)); + assertEquals(123L, result.getHeaders().get(AmqpHeaders.DELIVERY_TAG)); } @Test - public void testInt2809JavaTypePropertiesFromAmqp() { + public void testInt2809JavaTypePropertiesFromAmqp() throws Exception { Connection connection = mock(Connection.class); doAnswer(new Answer() { @Override @@ -133,8 +142,8 @@ public class InboundEndpointTests { MessageProperties amqpMessageProperties = new MessageProperties(); org.springframework.amqp.core.Message amqpMessage = new JsonMessageConverter().toMessage(payload, amqpMessageProperties); - MessageListener listener = (MessageListener) container.getMessageListener(); - listener.onMessage(amqpMessage); + ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener(); + listener.onMessage(amqpMessage, null); Message receive = channel.receive(1000); @@ -144,7 +153,7 @@ public class InboundEndpointTests { } @Test - public void testMessageConverterJsonHeadersHavePrecedenceOverMessageHeaders() { + public void testMessageConverterJsonHeadersHavePrecedenceOverMessageHeaders() throws Exception { Connection connection = mock(Connection.class); doAnswer(new Answer() { @Override @@ -156,13 +165,18 @@ public class InboundEndpointTests { when(connectionFactory.createConnection()).thenReturn(connection); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); + container.setAcknowledgeMode(AcknowledgeMode.MANUAL); DirectChannel channel = new DirectChannel(); + final Channel rabbitChannel = mock(Channel.class); + channel.subscribe(new MessageTransformingHandler(new Transformer() { @Override public Message transform(Message message) { + assertSame(rabbitChannel, message.getHeaders().get(AmqpHeaders.CHANNEL)); + assertEquals(123L, message.getHeaders().get(AmqpHeaders.DELIVERY_TAG)); return MessageBuilder.fromMessage(message) .setHeader(JsonHeaders.TYPE_ID, "foo") .setHeader(JsonHeaders.CONTENT_TYPE_ID, "bar") @@ -207,10 +221,11 @@ public class InboundEndpointTests { MessageProperties amqpMessageProperties = new MessageProperties(); amqpMessageProperties.setReplyTo("test"); + amqpMessageProperties.setDeliveryTag(123L); org.springframework.amqp.core.Message amqpMessage = new JsonMessageConverter().toMessage(payload, amqpMessageProperties); - MessageListener listener = (MessageListener) container.getMessageListener(); - listener.onMessage(amqpMessage); + ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener(); + listener.onMessage(amqpMessage, rabbitChannel); } 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 new file mode 100644 index 0000000000..c1cde909dc --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java @@ -0,0 +1,161 @@ +/* + * Copyright 2014 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.inbound; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.mockito.Mockito.mock; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.AnonymousQueue; +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.listener.SimpleMessageListenerContainer; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.amqp.AmqpHeaders; +import org.springframework.integration.amqp.inbound.ManualAckTests.ManualAckConfig; +import org.springframework.integration.amqp.rule.BrokerRunning; +import org.springframework.integration.annotation.Header; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Payload; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.rabbitmq.client.Channel; + +/** + * @author Gary Russell + * @since 4.0 + * + */ +@ContextConfiguration(classes = ManualAckConfig.class) +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class ManualAckTests { + + @Rule + public BrokerRunning brokerRunning = BrokerRunning.isRunning(); + + @Autowired + private MessageChannel foo; + + @Autowired + private PollableChannel bar; + + @Autowired + private SimpleMessageListenerContainer container; + + @Autowired + private RabbitTemplate template; + + @Test + public void testManual() { + AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(this.container); + adapter.setBeanFactory(mock(BeanFactory.class)); + adapter.setOutputChannel(foo); + adapter.afterPropertiesSet(); + adapter.start(); + this.template.convertAndSend("Hello, world"); + Message out = bar.receive(5000); + assertNotNull(out); + assertEquals(1, out.getPayload()); + out = bar.receive(5000); + assertNotNull(out); + assertEquals(2, out.getPayload()); + out = bar.receive(5000); + assertNotNull(out); + assertEquals(3, out.getPayload()); + out = bar.receive(1000); + assertNull(out); + adapter.stop(); + } + + @Configuration + @EnableIntegration + @ComponentScan + @MessageEndpoint + public static class ManualAckConfig { + + private int called; + + @ServiceActivator(inputChannel = "foo", outputChannel = "bar") + public Integer handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Channel channel, + @Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag) throws Exception { + if (++called > 2) { + channel.basicAck(deliveryTag, false); + } + else { + channel.basicNack(deliveryTag, false, true); + } + return called; + } + + @Bean + public QueueChannel bar() { + return new QueueChannel(); + } + + @Bean + public CachingConnectionFactory connectionFactory() { + return new CachingConnectionFactory(); + } + + @Bean + public Queue queue() { + return new AnonymousQueue(); + } + + @Bean + public SimpleMessageListenerContainer container() { + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory()); + container.setQueues(queue()); + container.setAcknowledgeMode(AcknowledgeMode.MANUAL); + container.setAutoStartup(false); + return container; + } + + @Bean + public RabbitTemplate template() { + RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory()); + rabbitTemplate.setRoutingKey(queue().getName()); + return rabbitTemplate; + } + + @Bean + public RabbitAdmin admin() { + return new RabbitAdmin(connectionFactory()); + } + } + +} diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index 798b118295..9ae0c62b84 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -52,7 +52,8 @@ A configuration sample for an AMQP Inbound Channel Adapter is shown below. - Acknowledge Mode for the MessageListenerContainer. - Optional (Defaults to AUTO). + When set to MANUAL, the delivery tag and channel are provided in + message headers amqp_deliveryTag and + amqp_channel respectively; the user application is + responsible for acknowledgement. NONE means no acknowledgements + (autoAck); AUTO means the adapter's container will acknowledge + when the downstream flow completes. + Optional (Defaults to AUTO) see . Extra AOP Advice(s) to handle cross cutting behavior associated with this Inbound Channel Adapter. @@ -282,6 +289,119 @@ this list can also be simple patterns to be matched against the header names (e. +
+ Inbound Gateway + A configuration sample for an AMQP Inbound Gateway is shown + below. + ]]> + + + + + Unique ID for this adapter. + Optional. + + + Message Channel to which converted Messages should be sent. + Required. + + + Names of the AMQP Queues from which Messages should be consumed (comma-separated list). + Required. + + + Extra AOP Advice(s) to handle cross cutting behavior associated with this Inbound Gateway. + Optional. + + + Specify the number of concurrent consumers to + create. Default is 1. Raising the number of concurrent + consumers is recommended in order to scale the + consumption of messages coming in from a queue. + However, note that any ordering guarantees are lost + once multiple consumers are registered. In general, + stick with 1 consumer for low-volume queues. + Optional (Defaults to 1). + + + Bean reference to the RabbitMQ ConnectionFactory. + Optional (Defaults to 'connectionFactory'). + + + Acknowledge Mode for the MessageListenerContainer. + When set to MANUAL, the delivery tag and channel are provided in + message headers amqp_deliveryTag and + amqp_channel respectively; the user application is + responsible for acknowledgement. NONE means no acknowledgements + (autoAck); AUTO means the adapter's container will acknowledge + when the downstream flow completes. + Optional (Defaults to AUTO) see . + + + Message Channel where reply Messages will be expected. + Optional. + + + + + See the note in about configuring the listener-container + attribute. + +
+ +
+ Inbound Endpoint Acknowledge Mode + + By default the inbound endpoints use acknowledge mode AUTO, which means the container + automatically acks the message when the downstream integration flow completes (or a message is + handed off to another thread using a QueueChannel or ExecutorChannel). + Setting the mode to NONE configures the consumer such that acks are not used at all + (the broker automatically acks the message as soon as it is sent). Setting the mode to + MANUAL allows user code to ack the message at some other point during processing. + To support this, with this mode, the endpoints provide the Channel and + deliveryTag in the amqp_channel and amqp_deliveryTag + headers respectively. + + + You can perform any valid rabbit command on the Channel but, generally, only + basicAck and basicNack (or basicReject) would be used. + In order to not interfere with the operation of the container, you should not retain a reference + to the channel and just use it in the context of the current message. + + + Since the Channel is a reference to a "live" object, it cannot be serialized + and will be lost if a message is persisted. + + + This is an example of how you might use MANUAL acknowledgement: + + +
+
Outbound Channel Adapter @@ -389,62 +509,6 @@ this list can also be simple patterns to be matched against the header names (e.
-
- Inbound Gateway - A configuration sample for an AMQP Inbound Gateway is shown - below. - ]]> - - - - - Unique ID for this adapter. - Optional. - - - Message Channel to which converted Messages should be sent. - Required. - - - Names of the AMQP Queues from which Messages should be consumed (comma-separated list). - Required. - - - Extra AOP Advice(s) to handle cross cutting behavior associated with this Inbound Gateway. - Optional. - - - Specify the number of concurrent consumers to - create. Default is 1. Raising the number of concurrent - consumers is recommended in order to scale the - consumption of messages coming in from a queue. - However, note that any ordering guarantees are lost - once multiple consumers are registered. In general, - stick with 1 consumer for low-volume queues. - Optional (Defaults to 1). - - - Bean reference to the RabbitMQ ConnectionFactory. - Optional (Defaults to 'connectionFactory'). - - - Message Channel where reply Messages will be expected. - Optional. - - - - - See the note in about configuring the listener-container - attribute. - -
-
Outbound Gateway A configuration sample for an AMQP Outbound Gateway is shown