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
This commit is contained in:
Gary russell
committed by
Artem Bilan
Artem Bilan
parent
d2e83e8280
commit
9052377bf5
@@ -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
|
||||
|
||||
@@ -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<String, ?> headers = headerMapper.toHeadersFromRequest(message.getMessageProperties());
|
||||
Map<String, Object> 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());
|
||||
}
|
||||
});
|
||||
|
||||
@@ -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<String, ?> headers = headerMapper.toHeadersFromRequest(message.getMessageProperties());
|
||||
Map<String, Object> 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();
|
||||
|
||||
@@ -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 " +
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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<Channel>() {
|
||||
@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<Object>(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<Channel>() {
|
||||
@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<Channel>() {
|
||||
@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);
|
||||
|
||||
}
|
||||
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -52,7 +52,8 @@
|
||||
|
||||
<para>A configuration sample for an AMQP Inbound Channel Adapter is shown
|
||||
below.</para>
|
||||
<programlisting language="xml"><![CDATA[<int-amqp:inbound-channel-adapter id="inboundAmqp"]]><co id="amqp-inbound-channel-adapter-xml-01-co" linkends="amqp-inbound-channel-adapter-xml-01" /><![CDATA[
|
||||
<programlisting language="xml"><![CDATA[<int-amqp:inbound-channel-adapter
|
||||
id="inboundAmqp"]]><co id="amqp-inbound-channel-adapter-xml-01-co" linkends="amqp-inbound-channel-adapter-xml-01" /><![CDATA[
|
||||
channel="inboundChannel"]]><co id="amqp-inbound-channel-adapter-xml-02-co" linkends="amqp-inbound-channel-adapter-xml-02" /><![CDATA[
|
||||
queue-names="si.test.queue"]]><co id="amqp-inbound-channel-adapter-xml-03-co" linkends="amqp-inbound-channel-adapter-xml-03" /><![CDATA[
|
||||
acknowledge-mode="AUTO"]]><co id="amqp-inbound-channel-adapter-xml-04-co" linkends="amqp-inbound-channel-adapter-xml-04" /><![CDATA[
|
||||
@@ -95,7 +96,13 @@
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-channel-adapter-xml-04-co" id="amqp-inbound-channel-adapter-xml-04">
|
||||
<para>Acknowledge Mode for the <interface>MessageListenerContainer</interface>.
|
||||
<emphasis>Optional (Defaults to AUTO)</emphasis>.</para>
|
||||
When set to MANUAL, the delivery tag and channel are provided in
|
||||
message headers <code>amqp_deliveryTag</code> and
|
||||
<code>amqp_channel</code> 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.
|
||||
<emphasis>Optional (Defaults to AUTO)</emphasis> see <xref linkend="amqp-inbound-ack"/>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-channel-adapter-xml-05-co" id="amqp-inbound-channel-adapter-xml-05">
|
||||
<para>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.
|
||||
</important>
|
||||
</section>
|
||||
|
||||
<section id="amqp-inbound-gateway">
|
||||
<title>Inbound Gateway</title>
|
||||
<para>A configuration sample for an AMQP Inbound Gateway is shown
|
||||
below.</para>
|
||||
<programlisting language="xml"><![CDATA[<int-amqp:inbound-gateway
|
||||
id="inboundGateway"]]><co id="amqp-inbound-gateway-adapter-xml-1-co" linkends="amqp-inbound-gateway-adapter-xml-1" /><![CDATA[
|
||||
request-channel="myRequestChannel"]]><co id="amqp-inbound-gateway-adapter-xml-2-co" linkends="amqp-inbound-gateway-adapter-xml-2" /><![CDATA[
|
||||
queue-names="si.test.queue"]]><co id="amqp-inbound-gateway-adapter-xml-3-co" linkends="amqp-inbound-gateway-adapter-xml-3" /><![CDATA[
|
||||
advice-chain=""]]><co id="amqp-inbound-gateway-adapter-xml-4-co" linkends="amqp-inbound-gateway-adapter-xml-4" /><![CDATA[
|
||||
concurrent-consumers="1"]]><co id="amqp-inbound-gateway-adapter-xml-5-co" linkends="amqp-inbound-gateway-adapter-xml-5" /><![CDATA[
|
||||
connection-factory="connectionFactory"]]><co id="amqp-inbound-gateway-adapter-xml-6-co" linkends="amqp-inbound-gateway-adapter-xml-6" /><![CDATA[
|
||||
acknowledge-mode="AUTO"]]><co id="amqp-inbound-gateway-adapter-xml-6a-co" linkends="amqp-inbound-gateway-adapter-xml-6a" /><![CDATA[
|
||||
reply-channel="myReplyChannel"]]><co id="amqp-inbound-gateway-adapter-xml-7-co" linkends="amqp-inbound-gateway-adapter-xml-7" /><![CDATA[/>]]>
|
||||
</programlisting>
|
||||
<para>
|
||||
<calloutlist>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-1-co" id="amqp-inbound-gateway-adapter-xml-1">
|
||||
<para>Unique ID for this adapter.
|
||||
<emphasis>Optional</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-2-co" id="amqp-inbound-gateway-adapter-xml-2">
|
||||
<para>Message Channel to which converted Messages should be sent.
|
||||
<emphasis>Required</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-3-co" id="amqp-inbound-gateway-adapter-xml-3">
|
||||
<para>Names of the AMQP Queues from which Messages should be consumed (comma-separated list).
|
||||
<emphasis>Required</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-4-co" id="amqp-inbound-gateway-adapter-xml-4">
|
||||
<para>Extra AOP Advice(s) to handle cross cutting behavior associated with this Inbound Gateway.
|
||||
<emphasis>Optional</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-5-co" id="amqp-inbound-gateway-adapter-xml-5">
|
||||
<para>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.
|
||||
<emphasis>Optional (Defaults to 1)</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-6-co" id="amqp-inbound-gateway-adapter-xml-6">
|
||||
<para>Bean reference to the RabbitMQ ConnectionFactory.
|
||||
<emphasis>Optional (Defaults to 'connectionFactory')</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-6a-co" id="amqp-inbound-gateway-adapter-xml-6a">
|
||||
<para>Acknowledge Mode for the <interface>MessageListenerContainer</interface>.
|
||||
When set to MANUAL, the delivery tag and channel are provided in
|
||||
message headers <code>amqp_deliveryTag</code> and
|
||||
<code>amqp_channel</code> 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.
|
||||
<emphasis>Optional (Defaults to AUTO) see <xref linkend="amqp-inbound-ack"/></emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-7-co" id="amqp-inbound-gateway-adapter-xml-7">
|
||||
<para>Message Channel where reply Messages will be expected.
|
||||
<emphasis>Optional</emphasis>.</para>
|
||||
</callout>
|
||||
</calloutlist>
|
||||
</para>
|
||||
<para>
|
||||
See the note in <xref linkend="amqp-inbound-channel-adapter"/> about configuring the <code>listener-container</code>
|
||||
attribute.
|
||||
</para>
|
||||
</section>
|
||||
|
||||
<section id="amqp-inbound-ack">
|
||||
<title>Inbound Endpoint Acknowledge Mode</title>
|
||||
<para>
|
||||
By default the inbound endpoints use acknowledge mode <code>AUTO</code>, which means the container
|
||||
automatically <emphasis>acks</emphasis> the message when the downstream integration flow completes (or a message is
|
||||
handed off to another thread using a <classname>QueueChannel</classname> or <classname>ExecutorChannel</classname>).
|
||||
Setting the mode to <code>NONE</code> 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
|
||||
<code>MANUAL</code> allows user code to ack the message at some other point during processing.
|
||||
To support this, with this mode, the endpoints provide the <classname>Channel</classname> and
|
||||
<code>deliveryTag</code> in the <code>amqp_channel</code> and <code>amqp_deliveryTag</code>
|
||||
headers respectively.
|
||||
</para>
|
||||
<para>
|
||||
You can perform any valid rabbit command on the <classname>Channel</classname> but, generally, only
|
||||
<code>basicAck</code> and <code>basicNack</code> (or <code>basicReject</code>) 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.
|
||||
</para>
|
||||
<note>
|
||||
Since the <classname>Channel</classname> is a reference to a "live" object, it cannot be serialized
|
||||
and will be lost if a message is persisted.
|
||||
</note>
|
||||
<para>
|
||||
This is an example of how you might use <code>MANUAL</code> acknowledgement:
|
||||
</para>
|
||||
<programlisting language="java"><![CDATA[@ServiceActivator(inputChannel = "foo", outputChannel = "bar")
|
||||
public Object handle(@Payload String payload, @Header(AmqpHeaders.CHANNEL) Channel channel,
|
||||
@Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag) throws Exception {
|
||||
|
||||
// Do some processing
|
||||
|
||||
if (allOK) {
|
||||
channel.basicAck(deliveryTag, false);
|
||||
|
||||
// perhaps do some more processing
|
||||
|
||||
}
|
||||
else {
|
||||
channel.basicNack(deliveryTag, false, true);
|
||||
}
|
||||
return someResultForDownStreamProcessing;
|
||||
}]]></programlisting>
|
||||
</section>
|
||||
|
||||
<section id="amqp-outbound-channel-adapter">
|
||||
<title>Outbound Channel Adapter</title>
|
||||
|
||||
@@ -389,62 +509,6 @@ this list can also be simple patterns to be matched against the header names (e.
|
||||
</para>
|
||||
</section>
|
||||
|
||||
<section id="amqp-inbound-gateway">
|
||||
<title>Inbound Gateway</title>
|
||||
<para>A configuration sample for an AMQP Inbound Gateway is shown
|
||||
below.</para>
|
||||
<programlisting language="xml"><![CDATA[<int-amqp:inbound-gateway id="inboundGateway"]]><co id="amqp-inbound-gateway-adapter-xml-1-co" linkends="amqp-inbound-gateway-adapter-xml-1" /><![CDATA[
|
||||
request-channel="myRequestChannel"]]><co id="amqp-inbound-gateway-adapter-xml-2-co" linkends="amqp-inbound-gateway-adapter-xml-2" /><![CDATA[
|
||||
queue-names="si.test.queue"]]><co id="amqp-inbound-gateway-adapter-xml-3-co" linkends="amqp-inbound-gateway-adapter-xml-3" /><![CDATA[
|
||||
advice-chain=""]]><co id="amqp-inbound-gateway-adapter-xml-4-co" linkends="amqp-inbound-gateway-adapter-xml-4" /><![CDATA[
|
||||
concurrent-consumers="1"]]><co id="amqp-inbound-gateway-adapter-xml-5-co" linkends="amqp-inbound-gateway-adapter-xml-5" /><![CDATA[
|
||||
connection-factory="connectionFactory"]]><co id="amqp-inbound-gateway-adapter-xml-6-co" linkends="amqp-inbound-gateway-adapter-xml-6" /><![CDATA[
|
||||
reply-channel="myReplyChannel"]]><co id="amqp-inbound-gateway-adapter-xml-7-co" linkends="amqp-inbound-gateway-adapter-xml-7" /><![CDATA[/>]]>
|
||||
</programlisting>
|
||||
<para>
|
||||
<calloutlist>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-1-co" id="amqp-inbound-gateway-adapter-xml-1">
|
||||
<para>Unique ID for this adapter.
|
||||
<emphasis>Optional</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-2-co" id="amqp-inbound-gateway-adapter-xml-2">
|
||||
<para>Message Channel to which converted Messages should be sent.
|
||||
<emphasis>Required</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-3-co" id="amqp-inbound-gateway-adapter-xml-3">
|
||||
<para>Names of the AMQP Queues from which Messages should be consumed (comma-separated list).
|
||||
<emphasis>Required</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-4-co" id="amqp-inbound-gateway-adapter-xml-4">
|
||||
<para>Extra AOP Advice(s) to handle cross cutting behavior associated with this Inbound Gateway.
|
||||
<emphasis>Optional</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-5-co" id="amqp-inbound-gateway-adapter-xml-5">
|
||||
<para>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.
|
||||
<emphasis>Optional (Defaults to 1)</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-6-co" id="amqp-inbound-gateway-adapter-xml-6">
|
||||
<para>Bean reference to the RabbitMQ ConnectionFactory.
|
||||
<emphasis>Optional (Defaults to 'connectionFactory')</emphasis>.</para>
|
||||
</callout>
|
||||
<callout arearefs="amqp-inbound-gateway-adapter-xml-7-co" id="amqp-inbound-gateway-adapter-xml-7">
|
||||
<para>Message Channel where reply Messages will be expected.
|
||||
<emphasis>Optional</emphasis>.</para>
|
||||
</callout>
|
||||
</calloutlist>
|
||||
</para>
|
||||
<para>
|
||||
See the note in <xref linkend="amqp-inbound-channel-adapter"/> about configuring the <code>listener-container</code>
|
||||
attribute.
|
||||
</para>
|
||||
</section>
|
||||
|
||||
<section id="amqp-outbound-gateway">
|
||||
<title>Outbound Gateway</title>
|
||||
<para>A configuration sample for an AMQP Outbound Gateway is shown
|
||||
|
||||
Reference in New Issue
Block a user