diff --git a/build.gradle b/build.gradle index 3ab6959233..8d05d44b00 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java index 3bffbdd9ff..a3cc548ed6 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java @@ -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 { } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index 2af63205e2..0d02c49ec0 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -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(); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index e9f9711345..7fa1d7ac43 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java @@ -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; } + }); } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java index 31b4b3ea70..4e39855b9c 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -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 { + implements RabbitTemplate.ConfirmCallback, ReturnCallback, ApplicationListener { 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 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()); } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java index 8f36d38e55..c4f6ea6058 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java @@ -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 extractUserDefinedHeaders(MessageProperties amqpMessageProperties) { - Map 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 headers){ + private String extractContentTypeAsString(Map 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) { diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java index ed132817d2..1efb02dfd0 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java @@ -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)); } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java index 33012e93b9..e0b70f90a0 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java @@ -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 { } } + } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index 71c75b94c7..34e5b2a778 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -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 { } } + } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index c5df36b344..5778c020b7 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -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; diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java index 32f4af910f..d7dd4e5f8e 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/ManualAckTests.java @@ -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()); } + } } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java index 7480dd7756..47dc7599c0 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java @@ -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 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 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)); diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java index f4bac2f576..3613fa2ef0 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -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 SYSLOG_PAYLOAD_ENTRIES = new HashSet( 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() {