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

INT-2791 temp-channel-transacted for AMQP Channel

Browse Source 
JIRA: https://jira.spring.io/browse/INT-2791

* Add `template-channel-transacted` attribute for the `<amqp:channel>`s to separate configuration for
container and RabbitTemplate
* Make container's `channelTransacted` as `false` by default
* Upgrade to `S-AMQP-1.4.0`
* Add `AmqpHeaders.PUBLISH_CONFIRM_NACK_CAUSE` according to the new `RabbitTemplate.ConfirmCallback#confirm` signature

Conflicts:
	src/reference/docbook/whats-new.xml

Conflicts:
	src/reference/docbook/whats-new.xml

INT-2791: Polishing

Upgrade to S-AMQP-2.0

Fix for mock publisher confirm/return tests, since `channel` is now physically closed, if the `ConnectionFactory` isn't `CachingConnectionFactory`
This commit is contained in:
Artem Bilan
2014-08-01 15:59:49 +03:00
committed by Gary Russell
parent d0cf1a8080
commit d43b89dedc
11 changed files with 78 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java
View File

@@ -79,6 +79,8 @@ public abstract class AmqpHeaders {
public static final String PUBLISH_CONFIRM = PREFIX + "publishConfirm";
public static final String PUBLISH_CONFIRM_NACK_CAUSE = PREFIX + "publishConfirmNackCause";
public static final String RETURN_REPLY_CODE = PREFIX + "returnReplyCode";
public static final String RETURN_REPLY_TEXT = PREFIX + "returnReplyText";

17
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
View File

@@ -109,10 +109,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
private volatile AcknowledgeMode acknowledgeMode;
/**
* This value differs from the container implementations' default (which is false).
*/
private volatile boolean channelTransacted = true;
private volatile boolean channelTransacted;
private volatile Executor taskExecutor;
@@ -202,15 +199,21 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
}
}
public void setTemplateChannelTransacted(boolean channelTransacted) {
if (this.amqpTemplate instanceof RabbitTemplate) {
((RabbitTemplate) this.amqpTemplate).setChannelTransacted(channelTransacted);
}
else if (logger.isInfoEnabled()) {
logger.info("AmqpTemplate is not a RabbitTemplate, so configured 'channelTransacted' will be ignored.");
}
}
/*
* Template and Container properties
*/
public void setChannelTransacted(boolean channelTransacted) {
this.channelTransacted = channelTransacted;
if (this.amqpTemplate instanceof RabbitTemplate) {
((RabbitTemplate) this.amqpTemplate).setChannelTransacted(channelTransacted);
}
}
public void setConnectionFactory(ConnectionFactory connectionFactory) {

1
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelParser.java
View File

@@ -56,6 +56,7 @@ public class AmqpChannelParser extends AbstractChannelParser {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "amqp-admin");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "channel-transacted");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "template-channel-transacted");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "concurrent-consumers");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "encoding");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-handler");

13
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
View File

@@ -13,6 +13,7 @@
package org.springframework.integration.amqp.outbound;
import java.util.HashMap;
import java.util.Map;
import org.springframework.amqp.AmqpException;
@@ -43,6 +44,7 @@ import org.springframework.integration.support.AbstractIntegrationMessageBuilder
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Adapter that converts and sends Messages to an AMQP Exchange.
@@ -318,13 +320,20 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
}
@Override
public void confirm(CorrelationData correlationData, boolean ack) {
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
Object userCorrelationData = correlationData;
if (correlationData instanceof CorrelationDataWrapper) {
userCorrelationData = ((CorrelationDataWrapper) correlationData).getUserData();
}
Map<String, Object> headers = new HashMap<String, Object>();
headers.put(AmqpHeaders.PUBLISH_CONFIRM, ack);
if (!ack && StringUtils.hasText(cause)) {
headers.put(AmqpHeaders.PUBLISH_CONFIRM_NACK_CAUSE, cause);
}
Message<Object> confirmMessage = this.getMessageBuilderFactory().withPayload(userCorrelationData)
.setHeader(AmqpHeaders.PUBLISH_CONFIRM, ack)
.copyHeaders(headers)
.build();
if (ack && this.confirmAckChannel != null) {
this.confirmAckChannel.send(confirmMessage);

23
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.1.xsd
View File

@@ -621,13 +621,6 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
</xsd:attributeGroup>
<xsd:attributeGroup name="containerAndTemplateSharedAttributes">
<xsd:attribute name="channel-transacted" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Flag to indicate that channels created by this component will be transactional.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
@@ -663,6 +656,14 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
SimpleMessageListenerContainer, such as channelTransacted, connectionFactory, and messagePropertiesConverter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="template-channel-transacted" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Flag to indicate that channels created by this component will be transactional.
Only applies to messages sent to this channel, or when 'message-driven' is 'false'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="encoding" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
@@ -693,6 +694,14 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
connectionFactory, and messsagePropertiesConverter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="channel-transacted" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Flag to indicate that channels created by this component will be transactional.
Only applies to outbound messages when 'message-driven' is 'true'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="acknowledge-mode">
<xsd:annotation>
<xsd:documentation>

3
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml
View File

@@ -17,7 +17,8 @@
<bean id="rabbitConnectionFactory" class="org.springframework.integration.amqp.StubRabbitConnectionFactory"/>
<amqp:channel id="channelWithSubscriberLimit" max-subscribers="1" missing-queues-fatal="false" />
<amqp:channel id="channelWithSubscriberLimit" max-subscribers="1" missing-queues-fatal="false"
template-channel-transacted="true"/>
<amqp:publish-subscribe-channel id="pubSub" />

4
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java
View File

@@ -60,6 +60,8 @@ public class AmqpChannelParserTests {
assertSame(mbf, TestUtils.getPropertyValue(channel, "dispatcher.messageBuilderFactory"));
assertSame(mbf, TestUtils.getPropertyValue(channel, "container.messageListener.messageBuilderFactory"));
assertTrue(TestUtils.getPropertyValue(channel, "container.missingQueuesFatal", Boolean.class));
assertFalse(TestUtils.getPropertyValue(channel, "container.transactional", Boolean.class));
assertFalse(TestUtils.getPropertyValue(channel, "amqpTemplate.transactional", Boolean.class));
}
@Test
@@ -68,6 +70,8 @@ public class AmqpChannelParserTests {
assertEquals(1, TestUtils.getPropertyValue(
TestUtils.getPropertyValue(channel, "dispatcher"), "maxSubscribers", Integer.class).intValue());
assertFalse(TestUtils.getPropertyValue(channel, "container.missingQueuesFatal", Boolean.class));
assertFalse(TestUtils.getPropertyValue(channel, "container.transactional", Boolean.class));
assertTrue(TestUtils.getPropertyValue(channel, "amqpTemplate.transactional", Boolean.class));
}

13
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
View File

@@ -24,6 +24,7 @@ import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -38,11 +39,14 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Matchers;
import org.mockito.Mockito;
import org.mockito.internal.stubbing.answers.DoesNothing;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -82,9 +86,6 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ReflectionUtils;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
@@ -186,7 +187,8 @@ public class AmqpOutboundChannelAdapterParserTests {
Channel mockChannel = mock(Channel.class);
when(connectionFactory.createConnection()).thenReturn(mockConnection);
PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
PublisherCallbackChannelImpl publisherCallbackChannel = spy(new PublisherCallbackChannelImpl(mockChannel));
doAnswer(new DoesNothing()).when(publisherCallbackChannel).close();
when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
MessageChannel requestChannel = context.getBean("pcRequestChannel", MessageChannel.class);
@@ -246,7 +248,8 @@ public class AmqpOutboundChannelAdapterParserTests {
Channel mockChannel = mock(Channel.class);
when(connectionFactory.createConnection()).thenReturn(mockConnection);
PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
PublisherCallbackChannelImpl publisherCallbackChannel = spy(new PublisherCallbackChannelImpl(mockChannel));
doAnswer(new DoesNothing()).when(publisherCallbackChannel).close();
when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
MessageChannel requestChannel = context.getBean("returnRequestChannel", MessageChannel.class);