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

INT-3504: Deprecate AmqpHeaders

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

* Deprecate `AmqpHeaders` and extend it from `org.springframework.amqp.support.AmqpHeaders` for backward compatibility
* Change AMQP dependency to `1.4.0`
* Remove usage of deprecated `spring_reply_correlation` and `spring_reply_to` AMQP headers
* Fix syslog `DefaultMessageConverter` JavaDoc

Reformat code
This commit is contained in:
Artem Bilan
2014-08-28 13:21:38 +03:00
committed by Gary Russell
parent 93b1224e99
commit 0b917622a6
13 changed files with 168 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -112,7 +112,7 @@ subprojects { subproject ->
slf4jVersion = "1.7.6"
smack3Version = '3.2.1'
smackVersion = '4.0.0'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.0.0.BUILD-SNAPSHOT'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.4.0.BUILD-SNAPSHOT'
springDataMongoVersion = '1.5.0.RELEASE'
springDataRedisVersion = '1.3.0.RELEASE'
springGemfireVersion = '1.4.0.RELEASE'

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

@@ -16,95 +16,14 @@
package org.springframework.integration.amqp;
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
* MessageProperties from/to integration Message Headers.
* @deprecated in favor of {@link org.springframework.amqp.support.AmqpHeaders}.
* Will be removed in a future release.
*
* @author Mark Fisher
*/
public abstract class AmqpHeaders {
/**
* Prefix used for AMQP related headers in order to distinguish from
* user-defined headers and other internal headers (e.g. replyTo).
* @see DefaultAmqpHeaderMapper
*/
public static final String PREFIX = "amqp_";
// Header Name Constants
public static final String APP_ID = PREFIX + "appId";
public static final String CLUSTER_ID = PREFIX + "clusterId";
public static final String CONTENT_ENCODING = PREFIX + "contentEncoding";
public static final String CONTENT_LENGTH = PREFIX + "contentLength";
public static final String CONTENT_TYPE = MessageHeaders.CONTENT_TYPE;
public static final String CORRELATION_ID = PREFIX + "correlationId";
public static final String DELIVERY_MODE = PREFIX + "deliveryMode";
public static final String DELIVERY_TAG = PREFIX + "deliveryTag";
public static final String EXPIRATION = PREFIX + "expiration";
public static final String MESSAGE_COUNT = PREFIX + "messageCount";
public static final String MESSAGE_ID = PREFIX + "messageId";
public static final String RECEIVED_EXCHANGE = PREFIX + "receivedExchange";
public static final String RECEIVED_ROUTING_KEY = PREFIX + "receivedRoutingKey";
public static final String REDELIVERED = PREFIX + "redelivered";
public static final String REPLY_TO = PREFIX + "replyTo";
public static final String TIMESTAMP = PREFIX + "timestamp";
public static final String TYPE = PREFIX + "type";
public static final String USER_ID = PREFIX + "userId";
public static final String SPRING_REPLY_CORRELATION = PREFIX + "springReplyCorrelation";
public static final String SPRING_REPLY_TO_STACK = PREFIX + "springReplyToStack";
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";
public static final String RETURN_EXCHANGE = PREFIX + "returnExchange";
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
* @deprecated Use the standard Rabbit CorrelationId header (mapped to {@link #CORRELATION_ID}).
*/
@Deprecated
public static final String STACKED_CORRELATION_HEADER = "spring_reply_correlation";
/**
* Compatibility with Spring-AMQP 1.1
* This was previously in RabbitTemplate
* @deprecated No longer used by Spring-AMQP 1.2 and above.
*/
@Deprecated
public static final String STACKED_REPLY_TO_HEADER = "spring_reply_to";
@Deprecated
public abstract class AmqpHeaders extends org.springframework.amqp.support.AmqpHeaders {
}

18
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -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.
@@ -18,27 +18,28 @@ package org.springframework.integration.amqp.inbound;
import java.util.Map;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.support.AmqpHeaders;
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.
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
* @since 2.1
*/
public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
@@ -53,8 +54,10 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
public AmqpInboundChannelAdapter(AbstractMessageListenerContainer listenerContainer) {
Assert.notNull(listenerContainer, "listenerContainer must not be null");
Assert.isNull(listenerContainer.getMessageListener(), "The listenerContainer provided to an AMQP inbound Channel Adapter " +
"must not have a MessageListener configured since the adapter needs to configure its own listener implementation.");
Assert.isNull(listenerContainer.getMessageListener(),
"The listenerContainer provided to an AMQP inbound Channel Adapter " +
"must not have a MessageListener configured since the adapter " +
"configure its own listener implementation.");
this.messageListenerContainer = listenerContainer;
this.messageListenerContainer.setAutoStartup(false);
}
@@ -87,8 +90,9 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag());
headers.put(AmqpHeaders.CHANNEL, channel);
}
sendMessage(AmqpInboundChannelAdapter.this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build());
sendMessage(getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build());
}
});
this.messageListenerContainer.afterPropertiesSet();
super.onInit();

17
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java
View File

@@ -18,6 +18,8 @@ package org.springframework.integration.amqp.inbound;
import java.util.Map;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Address;
@@ -27,17 +29,15 @@ 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.AmqpHeaders;
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.
@@ -45,6 +45,7 @@ import com.rabbitmq.client.Channel;
* the AMQP 'replyTo'.
*
* @author Mark Fisher
* @author Artem Bilan
* @since 2.1
*/
public class AmqpInboundGateway extends MessagingGatewaySupport {
@@ -60,8 +61,10 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
public AmqpInboundGateway(AbstractMessageListenerContainer listenerContainer) {
Assert.notNull(listenerContainer, "listenerContainer must not be null");
Assert.isNull(listenerContainer.getMessageListener(), "The listenerContainer provided to an AMQP inbound Gateway " +
"must not have a MessageListener configured since the adapter needs to configure its own listener implementation.");
Assert.isNull(listenerContainer.getMessageListener(),
"The listenerContainer provided to an AMQP inbound Gateway " +
"must not have a MessageListener configured since " +
"the adapter needs to configure its own listener implementation.");
this.messageListenerContainer = listenerContainer;
this.messageListenerContainer.setAutoStartup(false);
this.amqpTemplate = new RabbitTemplate(this.messageListenerContainer.getConnectionFactory());
@@ -96,7 +99,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
headers.put(AmqpHeaders.CHANNEL, channel);
}
org.springframework.messaging.Message<?> request =
AmqpInboundGateway.this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build();
getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build();
final org.springframework.messaging.Message<?> reply = sendAndReceiveMessage(request);
if (reply != null) {
// TODO: fallback to a reply address property of this gateway
@@ -105,6 +108,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
"request Message being handled by the AMQP inbound gateway.");
amqpTemplate.convertAndSend(replyTo.getExchangeName(), replyTo.getRoutingKey(), reply.getPayload(),
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
MessageProperties messageProperties = message.getMessageProperties();
@@ -124,6 +128,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
}
return message;
}
});
}
}

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

@@ -26,6 +26,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationListener;
@@ -34,7 +35,6 @@ import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.SpelParserConfiguration;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.channel.NullChannel;
@@ -56,7 +56,7 @@ import org.springframework.util.StringUtils;
* @since 2.1
*/
public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
implements RabbitTemplate.ConfirmCallback, ReturnCallback, ApplicationListener<ContextRefreshedEvent> {
implements RabbitTemplate.ConfirmCallback, ReturnCallback, ApplicationListener<ContextRefreshedEvent> {
private static final ExpressionParser expressionParser =
new SpelExpressionParser(new SpelParserConfiguration(true, true));
@@ -238,7 +238,8 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
if (userCorrelationData != null) {
if (userCorrelationData instanceof CorrelationData) {
correlationData = (CorrelationData) userCorrelationData;
} else {
}
else {
correlationData = new CorrelationDataWrapper(requestMessage
.getHeaders().getId().toString(), userCorrelationData);
}
@@ -367,10 +368,10 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
: this.getMessageBuilderFactory().withPayload(returnedObject);
Map<String, ?> headers = this.headerMapper.toHeadersFromReply(message.getMessageProperties());
builder.copyHeadersIfAbsent(headers)
.setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode)
.setHeader(AmqpHeaders.RETURN_REPLY_TEXT, replyText)
.setHeader(AmqpHeaders.RETURN_EXCHANGE, exchange)
.setHeader(AmqpHeaders.RETURN_ROUTING_KEY, routingKey);
.setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode)
.setHeader(AmqpHeaders.RETURN_REPLY_TEXT, replyText)
.setHeader(AmqpHeaders.RETURN_EXCHANGE, exchange)
.setHeader(AmqpHeaders.RETURN_ROUTING_KEY, routingKey);
this.returnChannel.send(builder.build());
}

29
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
View File

@@ -24,8 +24,8 @@ import java.util.Map;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.mapping.AbstractHeaderMapper;
import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.util.StringUtils;
@@ -176,20 +176,6 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
}
}
@SuppressWarnings("deprecation")
Object replyCorrelation = amqpMessageProperties.getHeaders().get(AmqpHeaders.STACKED_CORRELATION_HEADER);
if (replyCorrelation instanceof String) {
if (StringUtils.hasText((String) replyCorrelation)) {
headers.put(AmqpHeaders.SPRING_REPLY_CORRELATION, replyCorrelation);
}
}
@SuppressWarnings("deprecation")
Object replyToStack = amqpMessageProperties.getHeaders().get(AmqpHeaders.STACKED_REPLY_TO_HEADER);
if (replyToStack instanceof String) {
if (StringUtils.hasText((String) replyToStack)) {
headers.put(AmqpHeaders.SPRING_REPLY_TO_STACK, replyToStack);
}
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
@@ -202,14 +188,9 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
/**
* Extract user-defined headers from an AMQP MessageProperties instance.
*/
@SuppressWarnings("deprecation")
@Override
protected Map<String, Object> extractUserDefinedHeaders(MessageProperties amqpMessageProperties) {
Map<String, Object> headers = amqpMessageProperties.getHeaders();
headers.remove(AmqpHeaders.STACKED_CORRELATION_HEADER);
headers.remove(AmqpHeaders.STACKED_REPLY_TO_HEADER);
return headers;
return amqpMessageProperties.getHeaders();
}
/**
@@ -342,15 +323,15 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
* see INT-2713 for more details
*
*/
private String extractContentTypeAsString(Map<String, Object> headers){
private String extractContentTypeAsString(Map<String, Object> headers) {
String contentTypeStringValue = null;
Object contentType = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_TYPE, Object.class);
if (contentType != null){
if (contentType != null) {
String contentTypeClassName = contentType.getClass().getName();
if (contentTypeClassName.equals("org.springframework.http.MediaType")){ // see INT-2713
if (contentTypeClassName.equals("org.springframework.http.MediaType")) { // see INT-2713
contentTypeStringValue = contentType.toString();
}
else if (contentType instanceof String) {

5
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java
View File

@@ -30,11 +30,11 @@ import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -71,7 +71,8 @@ public class AmqpInboundChannelAdapterParserTests {
Object adapter = context.getBean("autoStartFalse.adapter");
assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(adapter, "autoStartup"));
assertEquals(123, TestUtils.getPropertyValue(adapter, "phase"));
assertEquals(AcknowledgeMode.NONE, TestUtils.getPropertyValue(adapter, "messageListenerContainer.acknowledgeMode"));
assertEquals(AcknowledgeMode.NONE,
TestUtils.getPropertyValue(adapter, "messageListenerContainer.acknowledgeMode"));
assertFalse(TestUtils.getPropertyValue(adapter, "messageListenerContainer.missingQueuesFatal", Boolean.class));
}

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

@@ -48,6 +48,9 @@ import org.mockito.internal.stubbing.answers.DoesNothing;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.Connection;
@@ -56,6 +59,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannel;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.BeansException;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -64,7 +68,6 @@ import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.channel.DirectChannel;
@@ -84,9 +87,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
@@ -127,7 +127,8 @@ public class AmqpOutboundChannelAdapterParserTests {
public void withHeaderMapperCustomHeaders() {
Object eventDrivenConsumer = context.getBean("withHeaderMapperCustomHeaders");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class);
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler",
AmqpOutboundEndpoint.class);
assertNotNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
assertFalse(TestUtils.getPropertyValue(endpoint, "lazyConnect", Boolean.class));
@@ -156,7 +157,11 @@ public class AmqpOutboundChannelAdapterParserTests {
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("requestChannel", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").setHeader("bar", "bar").setHeader("foobar", "foobar").build();
Message<?> message = MessageBuilder.withPayload("hello")
.setHeader("foo", "foo")
.setHeader("bar", "bar")
.setHeader("foobar", "foobar")
.build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class));
@@ -174,7 +179,8 @@ public class AmqpOutboundChannelAdapterParserTests {
@Test
public void parseWithPublisherConfirms() {
Object eventDrivenConsumer = context.getBean("withPublisherConfirms");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler", AmqpOutboundEndpoint.class);
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler",
AmqpOutboundEndpoint.class);
NullChannel nullChannel = context.getBean(NullChannel.class);
MessageChannel ackChannel = context.getBean("ackChannel", MessageChannel.class);
assertSame(ackChannel, TestUtils.getPropertyValue(endpoint, "confirmAckChannel"));
@@ -231,14 +237,16 @@ public class AmqpOutboundChannelAdapterParserTests {
return null;
}
})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class),
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class),
Mockito.any(CorrelationData.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("amqpOutboundChannelAdapterWithinChain", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class),
Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class),
Mockito.any(CorrelationData.class));
}
@@ -276,13 +284,14 @@ public class AmqpOutboundChannelAdapterParserTests {
@Test
public void testInt2718FailForOutboundAdapterChannelAttribute() {
try {
new ClassPathXmlApplicationContext("AmqpOutboundChannelAdapterWithinChainParserTests-fail-context.xml", this.getClass());
new ClassPathXmlApplicationContext("AmqpOutboundChannelAdapterWithinChainParserTests-fail-context.xml",
this.getClass());
fail("Expected BeanDefinitionParsingException");
}
catch (BeansException e) {
assertTrue(e instanceof BeanDefinitionParsingException);
assertTrue(e.getMessage().contains("The 'channel' attribute isn't allowed for 'amqp:outbound-channel-adapter' " +
"when it is used as a nested element"));
assertTrue(e.getMessage().contains("The 'channel' attribute isn't allowed for " +
"'amqp:outbound-channel-adapter' when it is used as a nested element"));
}
}
@@ -298,7 +307,8 @@ public class AmqpOutboundChannelAdapterParserTests {
MessageChannel requestChannel = context.getBean("toRabbitOnlyWithTemplateChannel", MessageChannel.class);
requestChannel.send(MessageBuilder.withPayload("test").build());
Mockito.verify(mockChannel, Mockito.times(1)).basicPublish(Mockito.eq("default.test.exchange"), Mockito.eq("default.routing.key"),
Mockito.verify(mockChannel, Mockito.times(1)).basicPublish(Mockito.eq("default.test.exchange"),
Mockito.eq("default.routing.key"),
Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
}
@@ -312,7 +322,8 @@ public class AmqpOutboundChannelAdapterParserTests {
PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
MessageChannel requestChannel = context.getBean("withDefaultAmqpTemplateExchangeAndRoutingKey", MessageChannel.class);
MessageChannel requestChannel = context.getBean("withDefaultAmqpTemplateExchangeAndRoutingKey",
MessageChannel.class);
requestChannel.send(MessageBuilder.withPayload("test").build());
Mockito.verify(mockChannel, Mockito.times(1)).basicPublish(Mockito.eq(""), Mockito.eq(""),
Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class));
@@ -337,7 +348,8 @@ public class AmqpOutboundChannelAdapterParserTests {
@Test
public void testInt2971HeaderMapperAndMappedHeadersExclusivity() {
try {
new ClassPathXmlApplicationContext("AmqpOutboundChannelAdapterParserTests-headerMapper-fail-context.xml", this.getClass());
new ClassPathXmlApplicationContext("AmqpOutboundChannelAdapterParserTests-headerMapper-fail-context.xml",
this.getClass());
}
catch (BeanDefinitionParsingException e) {
assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " +
@@ -347,7 +359,8 @@ public class AmqpOutboundChannelAdapterParserTests {
@Test
public void testInt2971AmqpOutboundChannelAdapterWithCustomHeaderMapper() {
AmqpHeaderMapper headerMapper = TestUtils.getPropertyValue(this.amqpMessageHandlerWithCustomHeaderMapper, "headerMapper", AmqpHeaderMapper.class);
AmqpHeaderMapper headerMapper = TestUtils.getPropertyValue(this.amqpMessageHandlerWithCustomHeaderMapper,
"headerMapper", AmqpHeaderMapper.class);
assertSame(this.context.getBean("customHeaderMapper"), headerMapper);
}
@@ -384,4 +397,5 @@ public class AmqpOutboundChannelAdapterParserTests {
}
}
}

130
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
View File

@@ -34,10 +34,10 @@ import org.mockito.stubbing.Answer;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
@@ -61,7 +61,7 @@ public class AmqpOutboundGatewayParserTests {
private static volatile int adviceCalled;
@Test
public void testGatewayConfig(){
public void testGatewayConfig() {
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object edc = context.getBean("rabbitGateway");
@@ -88,7 +88,8 @@ public class AmqpOutboundGatewayParserTests {
"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomRequestResponse");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler",
AmqpOutboundEndpoint.class);
assertNotNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
assertFalse(TestUtils.getPropertyValue(endpoint, "lazyConnect", Boolean.class));
@@ -101,30 +102,32 @@ public class AmqpOutboundGatewayParserTests {
final AtomicBoolean shouldBePersistent = new AtomicBoolean();
Mockito.doAnswer(new Answer() {
@Override
@Override
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertEquals("foo", properties.getHeaders().get("foo"));
assertEquals(shouldBePersistent.get() ? MessageDeliveryMode.PERSISTENT
: MessageDeliveryMode.NON_PERSISTENT, properties.getDeliveryMode());
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
org.springframework.amqp.core.Message amqpReplyMessage = new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
return amqpReplyMessage;
}})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertEquals("foo", properties.getHeaders().get("foo"));
assertEquals(shouldBePersistent.get() ? MessageDeliveryMode.PERSISTENT
: MessageDeliveryMode.NON_PERSISTENT, properties.getDeliveryMode());
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
}
})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("toRabbit1", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class),
Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
// verify reply
QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class);
@@ -156,7 +159,8 @@ public class AmqpOutboundGatewayParserTests {
"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomAndStandardResponse");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler",
AmqpOutboundEndpoint.class);
assertNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
@@ -165,29 +169,31 @@ public class AmqpOutboundGatewayParserTests {
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
@Override
@Override
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertEquals("foo", properties.getHeaders().get("foo"));
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
assertEquals(MessageDeliveryMode.PERSISTENT, properties.getDeliveryMode());
org.springframework.amqp.core.Message amqpReplyMessage = new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
return amqpReplyMessage;
}})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertEquals("foo", properties.getHeaders().get("foo"));
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
assertEquals(MessageDeliveryMode.PERSISTENT, properties.getDeliveryMode());
return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
}
})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("toRabbit2", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class),
Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
// verify reply
QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class);
@@ -206,9 +212,10 @@ public class AmqpOutboundGatewayParserTests {
public void withHeaderMapperNothingToMap() {
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivernConsumer = context.getBean("withHeaderMapperNothingToMap");
Object eventDrivenConsumer = context.getBean("withHeaderMapperNothingToMap");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivenConsumer, "handler",
AmqpOutboundEndpoint.class);
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
@@ -216,28 +223,30 @@ public class AmqpOutboundGatewayParserTests {
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
@Override
@Override
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertNull(properties.getHeaders().get("foo"));
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
org.springframework.amqp.core.Message amqpReplyMessage = new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
return amqpReplyMessage;
}})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertNull(properties.getHeaders().get("foo"));
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
}
})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("toRabbit3", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class),
Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
// verify reply
QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class);
@@ -280,8 +289,10 @@ public class AmqpOutboundGatewayParserTests {
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
}})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
}
})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class),
Mockito.any(org.springframework.amqp.core.Message.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
@@ -289,7 +300,8 @@ public class AmqpOutboundGatewayParserTests {
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class),
Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
// verify reply
QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class);
@@ -308,7 +320,8 @@ public class AmqpOutboundGatewayParserTests {
@Test
public void testInt2971HeaderMapperAndMappedHeadersExclusivity() {
try {
new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-headerMapper-fail-context.xml", this.getClass());
new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-headerMapper-fail-context.xml",
this.getClass());
}
catch (BeanDefinitionParsingException e) {
assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " +
@@ -326,4 +339,5 @@ public class AmqpOutboundGatewayParserTests {
}
}
}

5
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -28,12 +28,13 @@ import static org.mockito.Mockito.when;
import java.util.Map;
import com.rabbitmq.client.Channel;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.Connection;
@@ -42,11 +43,11 @@ 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;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
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;

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

@@ -13,6 +13,7 @@
* 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;
@@ -24,6 +25,8 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AnonymousQueue;
import org.springframework.amqp.core.Queue;
@@ -31,29 +34,27 @@ 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.amqp.support.AmqpHeaders;
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.messaging.handler.annotation.Header;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.messaging.handler.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.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
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
@@ -105,7 +106,7 @@ public class ManualAckTests {
@EnableIntegration
@ComponentScan
@MessageEndpoint
public static class ManualAckConfig {
public static class ManualAckConfig {
private int called;
@@ -158,6 +159,7 @@ public class ManualAckTests {
public RabbitAdmin admin() {
return new RabbitAdmin(connectionFactory());
}
}
}

18
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
View File

@@ -31,11 +31,11 @@ import org.junit.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.http.MediaType;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.messaging.MessageHeaders;
/**
@@ -47,7 +47,6 @@ import org.springframework.messaging.MessageHeaders;
*/
public class DefaultAmqpHeaderMapperTests {
@SuppressWarnings("deprecation")
@Test
public void fromHeaders() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
@@ -57,7 +56,7 @@ public class DefaultAmqpHeaderMapperTests {
headerMap.put(AmqpHeaders.CONTENT_ENCODING, "test.contentEncoding");
headerMap.put(AmqpHeaders.CONTENT_LENGTH, 99L);
headerMap.put(AmqpHeaders.CONTENT_TYPE, "test.contentType");
byte[] testCorrelationId = new byte[] {1,2,3};
byte[] testCorrelationId = new byte[] {1, 2, 3};
headerMap.put(AmqpHeaders.CORRELATION_ID, testCorrelationId);
headerMap.put(AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.NON_PERSISTENT);
headerMap.put(AmqpHeaders.DELIVERY_TAG, 1234L);
@@ -82,7 +81,7 @@ public class DefaultAmqpHeaderMapperTests {
headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
Set<String> headerKeys = amqpProperties.getHeaders().keySet();
for (String headerKey : headerKeys) {
if (headerKey.startsWith(AmqpHeaders.PREFIX)){
if (headerKey.startsWith(AmqpHeaders.PREFIX)) {
fail();
}
}
@@ -103,8 +102,6 @@ public class DefaultAmqpHeaderMapperTests {
assertEquals(testTimestamp, amqpProperties.getTimestamp());
assertEquals("test.type", amqpProperties.getType());
assertEquals("test.userId", amqpProperties.getUserId());
assertEquals("test.correlation", amqpProperties.getHeaders().get(AmqpHeaders.STACKED_CORRELATION_HEADER));
assertEquals("test.replyTo2", amqpProperties.getHeaders().get(AmqpHeaders.STACKED_REPLY_TO_HEADER));
assertNull(amqpProperties.getHeaders().get(MessageHeaders.ERROR_CHANNEL));
assertNull(amqpProperties.getHeaders().get(MessageHeaders.REPLY_CHANNEL));
@@ -125,7 +122,6 @@ public class DefaultAmqpHeaderMapperTests {
assertEquals("text/html", amqpProperties.getContentType());
}
@SuppressWarnings("deprecation")
@Test
public void toHeaders() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
@@ -135,7 +131,7 @@ public class DefaultAmqpHeaderMapperTests {
amqpProperties.setContentEncoding("test.contentEncoding");
amqpProperties.setContentLength(99L);
amqpProperties.setContentType("test.contentType");
byte[] testCorrelationId = new byte[] {1,2,3};
byte[] testCorrelationId = new byte[] {1, 2, 3};
amqpProperties.setCorrelationId(testCorrelationId);
amqpProperties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
amqpProperties.setDeliveryTag(1234L);
@@ -151,8 +147,8 @@ public class DefaultAmqpHeaderMapperTests {
amqpProperties.setTimestamp(testTimestamp);
amqpProperties.setType("test.type");
amqpProperties.setUserId("test.userId");
amqpProperties.setHeader(AmqpHeaders.STACKED_CORRELATION_HEADER, "test.correlation");
amqpProperties.setHeader(AmqpHeaders.STACKED_REPLY_TO_HEADER, "test.replyTo2");
amqpProperties.setHeader(AmqpHeaders.SPRING_REPLY_CORRELATION, "test.correlation");
amqpProperties.setHeader(AmqpHeaders.SPRING_REPLY_TO_STACK, "test.replyTo2");
Map<String, Object> headerMap = headerMapper.toHeadersFromReply(amqpProperties);
assertEquals("test.appId", headerMap.get(AmqpHeaders.APP_ID));
assertEquals("test.clusterId", headerMap.get(AmqpHeaders.CLUSTER_ID));
@@ -163,7 +159,7 @@ public class DefaultAmqpHeaderMapperTests {
assertEquals(MessageDeliveryMode.NON_PERSISTENT, headerMap.get(AmqpHeaders.DELIVERY_MODE));
assertEquals(1234L, headerMap.get(AmqpHeaders.DELIVERY_TAG));
assertEquals("test.expiration", headerMap.get(AmqpHeaders.EXPIRATION));
assertEquals(new Integer(42), headerMap.get(AmqpHeaders.MESSAGE_COUNT));
assertEquals(42, headerMap.get(AmqpHeaders.MESSAGE_COUNT));
assertEquals("test.messageId", headerMap.get(AmqpHeaders.MESSAGE_ID));
assertEquals("test.receivedExchange", headerMap.get(AmqpHeaders.RECEIVED_EXCHANGE));
assertEquals("test.receivedRoutingKey", headerMap.get(AmqpHeaders.RECEIVED_ROUTING_KEY));

12
spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java
View File

@@ -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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.syslog;
import java.util.Arrays;
@@ -34,10 +35,10 @@ import org.springframework.messaging.Message;
* Default {@link MessageConverter}; delegates to a {@link SyslogToMapTransformer} to
* convert the payload to a map of values and also provides some of the map contents as
* message headers. See @link {@link SyslogHeaders} for the headers that are mapped.
*
* @author Gary Russell
* @author David Liu
* @since 3.0
*
*/
public class DefaultMessageConverter implements MessageConverter, BeanFactoryAware {
@@ -46,15 +47,13 @@ public class DefaultMessageConverter implements MessageConverter, BeanFactoryAwa
public static final Set<String> SYSLOG_PAYLOAD_ENTRIES = new HashSet<String>(
Arrays.asList(new String[] {SyslogToMapTransformer.MESSAGE, SyslogToMapTransformer.UNDECODED}));
private volatile BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean asMap = true;
/**
* Set false will leave the payload as the original complete syslog.
* @param asMap
* @param asMap boolean flag.
*/
public void setAsMap(boolean asMap) {
this.asMap = asMap;
@@ -62,8 +61,7 @@ public class DefaultMessageConverter implements MessageConverter, BeanFactoryAwa
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {