diff --git a/build.gradle b/build.gradle index ae13dc455e..c2d35b093d 100644 --- a/build.gradle +++ b/build.gradle @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2011 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. @@ -122,6 +122,7 @@ configure(javaprojects) { log4jVersion = '1.2.12' mockitoVersion = '1.8.4' springVersion = '3.1.0.M2' + springAmqpVersion = '1.0.0.RC2' springSecurityVersion = '3.0.5.RELEASE' springWsVersion = '2.0.2.RELEASE' @@ -164,6 +165,15 @@ configure(javaprojects) { // // @see configure(javaprojects) above for general config // ----------------------------------------------------------------------------- +project('spring-integration-amqp') { + description = 'Spring Integration AMQP Support' + dependencies { + compile project(":spring-integration-core") + compile "org.springframework:spring-context:$springVersion" + compile "org.springframework.amqp:spring-rabbit:$springAmqpVersion" + } +} + project('spring-integration-core') { description = 'Spring Integration Core' dependencies { diff --git a/settings.gradle b/settings.gradle index e9dca2f6bc..b45d2f2a24 100644 --- a/settings.gradle +++ b/settings.gradle @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2011 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. @@ -17,6 +17,7 @@ rootProject.name = 'spring-integration' include 'docs' +include 'spring-integration-amqp' include 'spring-integration-core' include 'spring-integration-event' include 'spring-integration-feed' diff --git a/spring-integration-amqp/pom.xml b/spring-integration-amqp/pom.xml new file mode 100644 index 0000000000..804d69136b --- /dev/null +++ b/spring-integration-amqp/pom.xml @@ -0,0 +1,173 @@ + + + 4.0.0 + org.springframework.integration + spring-integration-amqp + 2.1.0.BUILD-SNAPSHOT + Spring Integration AMQP Support + + + The Apache Software License, Version 2.0 + http://www.apache.org/licenses/LICENSE-2.0.txt + repo + + + + + + src/main/java + + **/* + + + **/*.java + + + + src/main/resources + + **/* + + + + + + src/test/java + + **/* + + + **/*.java + + + + src/test/resources + + **/* + + + + + + maven-compiler-plugin + + 1.5 + 1.5 + + + + maven-surefire-plugin + + + **/*Tests.java + + + **/*Abstract*.java + + + + + + + + http://download.java.net/maven/2 + http://download.java.net/maven/2 + + + http://maven.springframework.org/milestone + http://maven.springframework.org/milestone + + + http://maven.springframework.org/release + http://maven.springframework.org/release + + + http://maven.springframework.org/snapshot + http://maven.springframework.org/snapshot + + + http://repository.springsource.com/maven/bundles/external + http://repository.springsource.com/maven/bundles/external + + + http://repository.springsource.com/maven/bundles/milestone + http://repository.springsource.com/maven/bundles/milestone + + + http://repository.springsource.com/maven/bundles/release + http://repository.springsource.com/maven/bundles/release + + + + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + + + org.easymock + easymock + 2.3 + test + + + org.easymock + easymockclassextension + 2.3 + test + + + org.springframework.amqp + spring-rabbit + 1.0.0.RC2 + compile + + + org.springframework + spring-test + 3.1.0.M2 + test + + + org.springframework + spring-context + 3.1.0.M2 + compile + + + org.hamcrest + hamcrest-all + 1.1 + test + + + org.mockito + mockito-all + 1.8.4 + test + + + log4j + log4j + 1.2.12 + test + + + cglib + cglib-nodep + 2.2 + test + + + junit + junit-dep + 4.8.2 + test + + + + UTF8 + + 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 new file mode 100644 index 0000000000..0d1b3460a2 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java @@ -0,0 +1,75 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp; + +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; + +/** + * Pre-defined names and prefixes to be used for setting and/or retrieving AMQP + * MessageProperties from/to integration Message Headers. + * + * @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 = PREFIX + "contentType"; + + 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"; + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java new file mode 100644 index 0000000000..a68e5fe6b7 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AbstractAmqpInboundAdapterParser.java @@ -0,0 +1,137 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.config; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * Base class for inbound adapter parsers for the AMQP namespace. + * + * @author Mark Fisher + * @since 2.1 + */ +abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefinitionParser { + + private static final String[] CONTAINER_VALUE_ATTRIBUTES = { + "acknowledge-mode", + "channel-transacted", + "concurrent-consumers", + "expose-listener-channel", + "phase", + "prefetch-count", + "queue-names", + "recovery-interval", + "receive-timeout", + "shutdown-timeout", + "tx-size" + }; + + private static final String[] CONTAINER_REFERENCE_ATTRIBUTES = { + "advice-chain", + "connection-factory", + "error-handler", + "message-properties-converter", + "task-executor", + "transaction-attribute", + "transaction-manager" + }; + + + private final String adapterClassName; + + + AbstractAmqpInboundAdapterParser(String adapterClassName) { + Assert.hasText(adapterClassName, "adapterClassName is required"); + this.adapterClassName = adapterClassName; + } + + + @Override + protected final String getBeanClassName(Element element) { + return this.adapterClassName; + } + + @Override + protected final boolean shouldGenerateId() { + return false; + } + + @Override + protected final boolean shouldGenerateIdAsFallback() { + return true; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + String listenerContainerRef = element.getAttribute("listener-container"); + if (StringUtils.hasText(listenerContainerRef)) { + assertNoContainerAttributes(element, parserContext); + builder.addConstructorArgReference(listenerContainerRef); + } + else { + BeanDefinition listenerContainerBeanDef = this.buildListenerContainer(element, parserContext); + builder.addConstructorArgValue(listenerContainerBeanDef); + } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "header-mapper"); + this.configureChannels(element, parserContext, builder); + } + + protected abstract void configureChannels(Element element, ParserContext parserContext, BeanDefinitionBuilder builder); + + private BeanDefinition buildListenerContainer(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( + "org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer"); + String connectionFactoryRef = element.getAttribute("connection-factory"); + if (!StringUtils.hasText(connectionFactoryRef)) { + connectionFactoryRef = "connectionFactory"; + } + builder.addConstructorArgReference(connectionFactoryRef); + for (String attributeName : CONTAINER_VALUE_ATTRIBUTES) { + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, attributeName); + } + for (String attributeName : CONTAINER_REFERENCE_ATTRIBUTES) { + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, attributeName); + } + return builder.getBeanDefinition(); + } + + private void assertNoContainerAttributes(Element element, ParserContext parserContext) { + Object source = parserContext.extractSource(element); + List allContainerAttributes = new ArrayList(Arrays.asList(CONTAINER_VALUE_ATTRIBUTES)); + allContainerAttributes.addAll(Arrays.asList(CONTAINER_REFERENCE_ATTRIBUTES)); + for (String attributeName : allContainerAttributes) { + if (StringUtils.hasText(element.getAttribute(attributeName))) { + parserContext.getReaderContext().error("Attribute '" + attributeName + + "' is not allowed when a 'listener-container' reference has been provided", source); + } + } + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParser.java new file mode 100644 index 0000000000..04d3d71a07 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParser.java @@ -0,0 +1,44 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; + +/** + * Parser for the AMQP 'inbound-channel-adapter' element. + * + * @author Mark Fisher + * @since 2.1 + */ +public class AmqpInboundChannelAdapterParser extends AbstractAmqpInboundAdapterParser { + + AmqpInboundChannelAdapterParser() { + super("org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter"); + } + + + @Override + protected void configureChannels(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "outputChannel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParser.java new file mode 100644 index 0000000000..201d0fa696 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParser.java @@ -0,0 +1,44 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; + +/** + * Parser for the AMQP 'inbound-gateway' element. + * + * @author Mark Fisher + * @since 2.1 + */ +public class AmqpInboundGatewayParser extends AbstractAmqpInboundAdapterParser { + + AmqpInboundGatewayParser() { + super("org.springframework.integration.amqp.inbound.AmqpInboundGateway"); + } + + + @Override + protected void configureChannels(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "request-channel"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel"); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java new file mode 100644 index 0000000000..8f70f8ef91 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpNamespaceHandler.java @@ -0,0 +1,37 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.config; + +import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler; + +/** + * Namespace handler for the AMQP schema. + * + * @author Mark Fisher + * @since 2.1 + */ +public class AmqpNamespaceHandler extends AbstractIntegrationNamespaceHandler { + + @Override + public void init() { + this.registerBeanDefinitionParser("inbound-channel-adapter", new AmqpInboundChannelAdapterParser()); + this.registerBeanDefinitionParser("inbound-gateway", new AmqpInboundGatewayParser()); + this.registerBeanDefinitionParser("outbound-channel-adapter", new AmqpOutboundChannelAdapterParser()); + this.registerBeanDefinitionParser("outbound-gateway", new AmqpOutboundGatewayParser()); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java new file mode 100644 index 0000000000..064bccb6e5 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java @@ -0,0 +1,50 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.config; + +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; +import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; +import org.w3c.dom.Element; + +/** + * Parser for the AMQP 'outbound-channel-adapter' element. + * + * @author Mark Fisher + * @since 2.1 + */ +public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { + + @Override + protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(AmqpOutboundEndpoint.class); + String amqpTemplateRef = element.getAttribute("amqp-template"); + if (!StringUtils.hasText(amqpTemplateRef)) { + amqpTemplateRef = "amqpTemplate"; + } + builder.addConstructorArgReference(amqpTemplateRef); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression"); + return builder.getBeanDefinition(); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java new file mode 100644 index 0000000000..c513e86194 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java @@ -0,0 +1,54 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.amqp.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; +import org.springframework.integration.config.xml.AbstractConsumerEndpointParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; + +/** + * Parser for the AMQP 'outbound-channel-adapter' element. + * + * @author Mark Fisher + * @since 2.1 + */ +public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser { + + @Override + protected String getInputChannelAttributeName() { + return "request-channel"; + } + + @Override + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(AmqpOutboundEndpoint.class); + String amqpTemplateRef = element.getAttribute("amqp-template"); + if (!StringUtils.hasText(amqpTemplateRef)) { + amqpTemplateRef = "amqpTemplate"; + } + builder.addConstructorArgReference(amqpTemplateRef); + builder.addPropertyValue("expectReply", true); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression"); + return builder; + } + +} 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 new file mode 100644 index 0000000000..09343da75d --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -0,0 +1,90 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.inbound; + +import java.util.Map; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.util.Assert; + +/** + * 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 + * @since 2.1 + */ +public class AmqpInboundChannelAdapter extends MessageProducerSupport { + + private final AbstractMessageListenerContainer messageListenerContainer; + + private volatile MessageConverter messageConverter = new SimpleMessageConverter(); + + private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper(); + + + 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."); + this.messageListenerContainer = listenerContainer; + this.messageListenerContainer.setAutoStartup(false); + } + + + public void setMessageConverter(MessageConverter messageConverter) { + Assert.notNull(messageConverter, "messageConverter must not be null"); + this.messageConverter = messageConverter; + } + + public void setHeaderMapper(AmqpHeaderMapper headerMapper) { + Assert.notNull(headerMapper, "headerMapper must not be null"); + this.headerMapper = headerMapper; + } + + @Override + protected void onInit() { + this.messageListenerContainer.setMessageListener(new MessageListener() { + public void onMessage(Message message) { + Object payload = messageConverter.fromMessage(message); + Map headers = headerMapper.toHeaders(message.getMessageProperties()); + sendMessage(MessageBuilder.withPayload(payload).copyHeaders(headers).build()); + } + }); + this.messageListenerContainer.afterPropertiesSet(); + super.onInit(); + } + + @Override + protected void doStart() { + this.messageListenerContainer.start(); + } + + @Override + protected void doStop() { + this.messageListenerContainer.stop(); + } + +} 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 new file mode 100644 index 0000000000..66a44aee5f --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java @@ -0,0 +1,116 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.inbound; + +import java.util.Map; + +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.core.MessagePostProcessor; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.util.Assert; + +/** + * Adapter that receives Messages from an AMQP Queue, converts them into + * Spring Integration Messages, and sends the results to a Message Channel. + * If a reply Message is received, it will be converted and sent back to + * the AMQP 'replyTo'. + * + * @author Mark Fisher + * @since 2.1 + */ +public class AmqpInboundGateway extends MessagingGatewaySupport { + + private final SimpleMessageListenerContainer messageListenerContainer; + + private final SimpleMessageConverter messageConverter = new SimpleMessageConverter(); + + private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper(); + + private final RabbitTemplate amqpTemplate; + + + public AmqpInboundGateway(ConnectionFactory connectionFactory) { + Assert.notNull(connectionFactory, "ConnectionFactory must not be null"); + this.messageListenerContainer = new SimpleMessageListenerContainer(connectionFactory); + this.messageListenerContainer.setAutoStartup(false); + this.amqpTemplate = new RabbitTemplate(connectionFactory); + } + + + public void setQueueNames(String queueName) { + this.messageListenerContainer.setQueueNames(queueName); + } + + public void setHeaderMapper(AmqpHeaderMapper headerMapper) { + Assert.notNull(headerMapper, "headerMapper must not be null"); + this.headerMapper = headerMapper; + } + + public void setConcurrentConsumers(int concurrentConsumers) { + this.messageListenerContainer.setConcurrentConsumers(concurrentConsumers); + } + + @Override + protected void onInit() throws Exception { + this.messageListenerContainer.setMessageListener(new MessageListener() { + public void onMessage(Message message) { + Object payload = messageConverter.fromMessage(message); + Map headers = headerMapper.toHeaders(message.getMessageProperties()); + org.springframework.integration.Message request = + MessageBuilder.withPayload(payload).copyHeaders(headers).build(); + final org.springframework.integration.Message reply = sendAndReceiveMessage(request); + if (reply != null) { + // TODO: fallback to a reply address property of this gateway + Address replyTo = message.getMessageProperties().getReplyTo(); + Assert.notNull(replyTo, "The replyTo header must not be null on a " + + "request Message being handled by the AMQP inbound gateway."); + amqpTemplate.convertAndSend(replyTo.getExchangeName(), replyTo.getRoutingKey(), reply.getPayload(), + new MessagePostProcessor() { + public Message postProcessMessage(Message message) throws AmqpException { + headerMapper.fromHeaders(reply.getHeaders(), message.getMessageProperties()); + return message; + } + }); + } + } + }); + this.messageListenerContainer.afterPropertiesSet(); + this.amqpTemplate.afterPropertiesSet(); + super.onInit(); + } + + @Override + protected void doStart() { + this.messageListenerContainer.start(); + } + + @Override + protected void doStop() { + this.messageListenerContainer.stop(); + } + +} 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 new file mode 100644 index 0000000000..acecf083f7 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -0,0 +1,137 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.amqp.outbound; + +import java.util.Map; + +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.MessagePostProcessor; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.support.converter.MessageConverter; +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.Message; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.util.Assert; + +/** + * Adapter that converts and sends Messages to an AMQP Exchange. + * + * @author Mark Fisher + * @since 2.1 + */ +public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler { + + private final AmqpTemplate amqpTemplate; + + private volatile String exchangeName = ""; + + private volatile String routingKey = ""; + + private volatile boolean expectReply; + + private static final ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true)); + + private volatile ExpressionEvaluatingMessageProcessor routingKeyGenerator; + + private volatile String routingKeyExpression; + + private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper(); + + + @Override + protected void onInit() { + super.onInit(); + Assert.state(routingKeyExpression == null || "".equals(routingKey), + "Either a routingKey or a routingKeyExpression can be provided, but not both"); + if (routingKeyExpression!=null) { + Expression expression = expressionParser.parseExpression(this.routingKeyExpression); + this.routingKeyGenerator = new ExpressionEvaluatingMessageProcessor(expression, String.class); + } + } + + public AmqpOutboundEndpoint(AmqpTemplate amqpTemplate) { + Assert.notNull(amqpTemplate, "AmqpTemplate must not be null"); + this.amqpTemplate = amqpTemplate; + } + + public void setExchangeName(String exchangeName) { + this.exchangeName = exchangeName; + } + + public void setRoutingKey(String routingKey) { + this.routingKey = routingKey; + } + + public void setRoutingKeyExpression(String routingKeyExpression) { + this.routingKeyExpression = routingKeyExpression; + } + + public void setExpectReply(boolean expectReply) { + this.expectReply = expectReply; + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + String routingKey = this.routingKey; + if (this.routingKeyGenerator!=null) { + routingKey = this.routingKeyGenerator.processMessage(requestMessage); + } + if (this.expectReply) { + return this.sendAndReceive(requestMessage, routingKey); + } + else { + this.send(requestMessage, routingKey); + return null; + } + } + + private void send(final Message requestMessage, String routingKey) { + this.amqpTemplate.convertAndSend(this.exchangeName, routingKey, requestMessage.getPayload(), + new MessagePostProcessor() { + public org.springframework.amqp.core.Message postProcessMessage( + org.springframework.amqp.core.Message message) throws AmqpException { + headerMapper.fromHeaders(requestMessage.getHeaders(), message.getMessageProperties()); + return message; + } + }); + } + + private Message sendAndReceive(Message requestMessage, String routingKey) { + // TODO: add a convertSendAndReceive method that accepts a MessagePostProcessor so we can map headers? + Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for send and receive"); + MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter(); + MessageProperties amqpMessageProperties = new MessageProperties(); + this.headerMapper.fromHeaders(requestMessage.getHeaders(), amqpMessageProperties); + org.springframework.amqp.core.Message amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties); + org.springframework.amqp.core.Message amqpReplyMessage = this.amqpTemplate.sendAndReceive(this.exchangeName, routingKey, amqpMessage); + Object replyObject = converter.fromMessage(amqpReplyMessage); + MessageBuilder builder = (replyObject instanceof Message) + ? MessageBuilder.fromMessage((Message) replyObject) + : MessageBuilder.withPayload(replyObject); + Map headers = this.headerMapper.toHeaders(amqpReplyMessage.getMessageProperties()); + // TODO: copyHeadersIfAbsent has instead of + builder.copyHeaders(headers); + return builder.build(); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpHeaderMapper.java new file mode 100644 index 0000000000..696aa85b06 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/AmqpHeaderMapper.java @@ -0,0 +1,29 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.support; + +import org.springframework.amqp.core.MessageProperties; +import org.springframework.integration.mapping.HeaderMapper; + +/** + * A convenience interface that extends {@link HeaderMapper} + * but parameterized with {@link MessageProperties}. + * + * @author Mark Fisher + */ +public interface AmqpHeaderMapper extends HeaderMapper { +} 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 new file mode 100644 index 0000000000..82960658e1 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java @@ -0,0 +1,332 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.support; + +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.MessageDeliveryMode; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.integration.MessageHeaders; +import org.springframework.integration.amqp.AmqpHeaders; +import org.springframework.util.CollectionUtils; +import org.springframework.util.StringUtils; + +/** + * Default implementation of {@link AmqpHeaderMapper}. + *

+ * This implementation copies AMQP properties (e.g. contentType) to and from + * Spring Integration MessageHeaders. Any user-defined headers within the AMQP + * MessageProperties will also be copied from an AMQP Message to a Spring Integration + * Message, and any other headers on a Spring Integration Message (beyond the standard + * AMQP properties) will likewise be copied to an AMQP Message. + *

+ * Constants for the AMQP header keys are defined in {@link AmqpHeaders}. + * + * @author Mark Fisher + */ +public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private volatile String inboundPrefix = ""; + + private volatile String outboundPrefix = ""; + + + /** + * Specify a prefix to be appended to the integration message header name for + * any user-defined AMQP header that is being mapped into the MessageHeaders. + * The Default is an empty string (no prefix). + *

+ * This does not affect the standard AMQP properties, such as contentType, etc. + * The header names used for mapping such properties are all defined in the + * {@link AmqpHeaders} class as constants. + */ + public void setInboundPrefix(String inboundPrefix) { + this.inboundPrefix = (inboundPrefix != null) ? inboundPrefix : ""; + } + + /** + * Specify a prefix to be appended to the AMQP header name for any + * integration message header that is being mapped into the AMQP Message. + * The Default is an empty string (no prefix). + *

+ * This does not affect the standard AMQP properties, such as contentType, etc. + * The header names used for mapping such properties are all defined in + * the {@link AmqpHeaders} class as constants. + */ + public void setOutboundPrefix(String outboundPrefix) { + this.outboundPrefix = (outboundPrefix != null) ? outboundPrefix : ""; + } + + /** + * Maps headers from a Spring Integration MessageHeaders instance to the MessageProperties + * of an AMQP Message. + */ + public void fromHeaders(MessageHeaders headers, MessageProperties amqpMessageProperties) { + try { + String appId = getHeaderIfAvailable(headers, AmqpHeaders.APP_ID, String.class); + if (StringUtils.hasText(appId)) { + amqpMessageProperties.setAppId(appId); + } + String clusterId = getHeaderIfAvailable(headers, AmqpHeaders.CLUSTER_ID, String.class); + if (StringUtils.hasText(clusterId)) { + amqpMessageProperties.setClusterId(clusterId); + } + String contentEncoding = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_ENCODING, String.class); + if (StringUtils.hasText(contentEncoding)) { + amqpMessageProperties.setContentEncoding(contentEncoding); + } + Long contentLength = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_LENGTH, Long.class); + if (contentLength != null) { + amqpMessageProperties.setContentLength(contentLength); + } + String contentType = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_TYPE, String.class); + if (StringUtils.hasText(contentType)) { + amqpMessageProperties.setContentType(contentType); + } + Object correlationId = headers.get(AmqpHeaders.CORRELATION_ID); + if (correlationId instanceof byte[]) { + amqpMessageProperties.setCorrelationId((byte[]) correlationId); + } + MessageDeliveryMode deliveryMode = getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.class); + if (deliveryMode != null) { + amqpMessageProperties.setDeliveryMode(deliveryMode); + } + Long deliveryTag = getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_TAG, Long.class); + if (deliveryTag != null) { + amqpMessageProperties.setDeliveryTag(deliveryTag); + } + String expiration = getHeaderIfAvailable(headers, AmqpHeaders.EXPIRATION, String.class); + if (StringUtils.hasText(expiration)) { + amqpMessageProperties.setExpiration(expiration); + } + Integer messageCount = getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_COUNT, Integer.class); + if (messageCount != null) { + amqpMessageProperties.setMessageCount(messageCount); + } + String messageId = getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_ID, String.class); + if (StringUtils.hasText(messageId)) { + amqpMessageProperties.setMessageId(messageId); + } + Integer priority = headers.getPriority(); + if (priority != null) { + amqpMessageProperties.setPriority(priority); + } + String receivedExchange = getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_EXCHANGE, String.class); + if (StringUtils.hasText(receivedExchange)) { + amqpMessageProperties.setReceivedExchange(receivedExchange); + } + String receivedRoutingKey = getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_ROUTING_KEY, String.class); + if (StringUtils.hasText(receivedRoutingKey)) { + amqpMessageProperties.setReceivedRoutingKey(receivedRoutingKey); + } + Boolean redelivered = getHeaderIfAvailable(headers, AmqpHeaders.REDELIVERED, Boolean.class); + if (redelivered != null) { + amqpMessageProperties.setRedelivered(redelivered); + } + Address replyTo = getHeaderIfAvailable(headers, AmqpHeaders.REPLY_TO, Address.class); + if (replyTo != null) { + amqpMessageProperties.setReplyTo(replyTo); + } + Date timestamp = getHeaderIfAvailable(headers, AmqpHeaders.TIMESTAMP, Date.class); + if (timestamp != null) { + amqpMessageProperties.setTimestamp(timestamp); + } + String type = getHeaderIfAvailable(headers, AmqpHeaders.TYPE, String.class); + if (type != null) { + amqpMessageProperties.setType(type); + } + String userId = getHeaderIfAvailable(headers, AmqpHeaders.USER_ID, String.class); + if (StringUtils.hasText(userId)) { + amqpMessageProperties.setUserId(userId); + } + // now map to the user-defined headers, if any, within the AMQP MessageProperties + Set headerNames = headers.keySet(); + for (String headerName : headerNames) { + if (StringUtils.hasText(headerName) && !headerName.startsWith(AmqpHeaders.PREFIX)) { + Object value = headers.get(headerName); + if (value != null) { + try { + String key = this.fromHeaderName(headerName); + amqpMessageProperties.setHeader(key, value); + } + catch (Exception e) { + if (logger.isWarnEnabled()) { + logger.warn("failed to map Message header '" + headerName + "' to AMQP header", e); + } + } + } + } + } + } + catch (Exception e) { + if (logger.isWarnEnabled()) { + logger.warn("error occurred while mapping from MessageHeaders to AMQP properties", e); + } + } + } + + /** + * Maps headers from an AMQP MessageProperties instance to the MessageHeaders of a + * Spring Integration Message. + */ + public Map toHeaders(MessageProperties amqpMessageProperties) { + Map headers = new HashMap(); + try { + String appId = amqpMessageProperties.getAppId(); + if (StringUtils.hasText(appId)) { + headers.put(AmqpHeaders.APP_ID, appId); + } + String clusterId = amqpMessageProperties.getClusterId(); + if (StringUtils.hasText(clusterId)) { + headers.put(AmqpHeaders.CLUSTER_ID, clusterId); + } + String contentEncoding = amqpMessageProperties.getContentEncoding(); + if (StringUtils.hasText(contentEncoding)) { + headers.put(AmqpHeaders.CONTENT_ENCODING, contentEncoding); + } + long contentLength = amqpMessageProperties.getContentLength(); + if (contentLength > 0) { + headers.put(AmqpHeaders.CONTENT_LENGTH, contentLength); + } + String contentType = amqpMessageProperties.getContentType(); + if (StringUtils.hasText(contentType)) { + headers.put(AmqpHeaders.CONTENT_TYPE, contentType); + } + byte[] correlationId = amqpMessageProperties.getCorrelationId(); + if (correlationId != null && correlationId.length > 0) { + headers.put(AmqpHeaders.CORRELATION_ID, correlationId); + } + MessageDeliveryMode deliveryMode = amqpMessageProperties.getDeliveryMode(); + if (deliveryMode != null) { + headers.put(AmqpHeaders.DELIVERY_MODE, deliveryMode); + } + long deliveryTag = amqpMessageProperties.getDeliveryTag(); + if (deliveryTag > 0) { + headers.put(AmqpHeaders.DELIVERY_TAG, deliveryTag); + } + String expiration = amqpMessageProperties.getExpiration(); + if (StringUtils.hasText(expiration)) { + headers.put(AmqpHeaders.EXPIRATION, expiration); + } + Integer messageCount = amqpMessageProperties.getMessageCount(); + if (messageCount != null && messageCount > 0) { + headers.put(AmqpHeaders.MESSAGE_COUNT, messageCount); + } + String messageId = amqpMessageProperties.getMessageId(); + if (StringUtils.hasText(messageId)) { + headers.put(AmqpHeaders.MESSAGE_ID, messageId); + } + Integer priority = amqpMessageProperties.getPriority(); + if (priority != null && priority > 0) { + headers.put(MessageHeaders.PRIORITY, priority); + } + String receivedExchange = amqpMessageProperties.getReceivedExchange(); + if (StringUtils.hasText(receivedExchange)) { + headers.put(AmqpHeaders.RECEIVED_EXCHANGE, receivedExchange); + } + String receivedRoutingKey = amqpMessageProperties.getReceivedRoutingKey(); + if (StringUtils.hasText(receivedRoutingKey)) { + headers.put(AmqpHeaders.RECEIVED_ROUTING_KEY, receivedRoutingKey); + } + Boolean redelivered = amqpMessageProperties.isRedelivered(); + if (redelivered != null) { + headers.put(AmqpHeaders.REDELIVERED, redelivered); + } + Address replyTo = amqpMessageProperties.getReplyTo(); + if (replyTo != null) { + headers.put(AmqpHeaders.REPLY_TO, replyTo); + } + Date timestamp = amqpMessageProperties.getTimestamp(); + if (timestamp != null) { + headers.put(AmqpHeaders.TIMESTAMP, timestamp); + } + String type = amqpMessageProperties.getType(); + if (StringUtils.hasText(type)) { + headers.put(AmqpHeaders.TYPE, type); + } + String userId = amqpMessageProperties.getUserId(); + if (StringUtils.hasText(userId)) { + headers.put(AmqpHeaders.USER_ID, userId); + } + Map amqpHeaders = amqpMessageProperties.getHeaders(); + if (!CollectionUtils.isEmpty(amqpHeaders)) { + for (Map.Entry entry : amqpHeaders.entrySet()) { + try { + String headerName = this.toHeaderName(entry.getKey()); + headers.put(headerName, entry.getValue()); + } + catch (Exception e) { + if (logger.isWarnEnabled()) { + logger.warn("error occurred while mapping AMQP header '" + + entry.getKey() + "' to Message header", e); + } + } + } + } + } + catch (Exception e) { + if (logger.isWarnEnabled()) { + logger.warn("error occurred while mapping from AMQP properties to MessageHeaders", e); + } + } + return headers; + } + + private T getHeaderIfAvailable(MessageHeaders headers, String name, Class type) { + try { + return headers.get(name, type); + } + catch (IllegalArgumentException e) { + if (logger.isWarnEnabled()) { + logger.warn("skipping header '" + name + "' since it is not of expected type [" + type + "]", e); + } + return null; + } + } + + /** + * Adds the outbound prefix if necessary. + */ + private String fromHeaderName(String headerName) { + String propertyName = headerName; + if (StringUtils.hasText(this.outboundPrefix) && !propertyName.startsWith(this.outboundPrefix)) { + propertyName = this.outboundPrefix + headerName; + } + return propertyName; + } + + /** + * Adds the inbound prefix if necessary. + */ + private String toHeaderName(String propertyName) { + String headerName = propertyName; + if (StringUtils.hasText(this.inboundPrefix) && !headerName.startsWith(this.inboundPrefix)) { + headerName = this.inboundPrefix + propertyName; + } + return headerName; + } + +} diff --git a/spring-integration-amqp/src/main/resources/META-INF/spring.handlers b/spring-integration-amqp/src/main/resources/META-INF/spring.handlers new file mode 100644 index 0000000000..b0847f108a --- /dev/null +++ b/spring-integration-amqp/src/main/resources/META-INF/spring.handlers @@ -0,0 +1 @@ +http\://www.springframework.org/schema/integration/amqp=org.springframework.integration.amqp.config.AmqpNamespaceHandler diff --git a/spring-integration-amqp/src/main/resources/META-INF/spring.schemas b/spring-integration-amqp/src/main/resources/META-INF/spring.schemas new file mode 100644 index 0000000000..55c9d2f603 --- /dev/null +++ b/spring-integration-amqp/src/main/resources/META-INF/spring.schemas @@ -0,0 +1,2 @@ +http\://www.springframework.org/schema/integration/amqp/spring-integration-amqp-2.1.xsd=org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd +http\://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd=org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd new file mode 100644 index 0000000000..92919f100d --- /dev/null +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-2.1.xsd @@ -0,0 +1,479 @@ + + + + + + + + + + + Configures an endpoint that will publish an AMQP Message to the provided Exchange. + + + + + + + Unique ID for this adapter. + + + + + + + Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange. + + + + + + + + + + + + The name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange. + + + + + + + The fixed routing-key to use when sending Messages. By default, this will be an empty String. + + + + + + + The routing-key to use when sending Messages evealuated as an expression on the message (e.g. 'payload.key'). By default, this will be an empty String. + + + + + + + + + + + + + + + + + + + Configures an endpoint that will receive AMQP Messages sent to a given queue and then forward those messages to a Message Channel. + + + + + + + Unique ID for this adapter. + + + + + + + Acknowledge Mode for the MessageListenerContainer. + + + + + + + + + + + + + + + + + + Array of AOP Advice instances to be applied to the MessageListener. + + + + + + + + Message Channel to which converted Messages should be sent. + + + + + + + + + + + + Flag to indicate that channels created by this component will be transactional. + + + + + + + Specify the number of concurrent consumers to create. Default is 1. + Raising the number of concurrent consumers is recommended in order to scale the consumption of messages coming in + from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In + general, stick with 1 consumer for low-volume queues. + + + + + + + + + + + + + + + + Message Channel to which error Messages should be sent. + + + + + + + + + + + + + Set whether to expose the listener Rabbit Channel to a registered ChannelAwareMessageListener as well as + to RabbitTemplate calls. + + + + + + + + HeaderMapper to use when receiving AMQP Messages. + + + + + + + + + + + + Reference to the MessageListener container to use for receiving AMQP Messages. If this attribute is provided, + then no other attribute related to the listener container configuration should be provided. In other words, by + setting this reference, you must take full responsibility of the listener container configuration. The only + exception is the MessageListener itself. Since that is actually the core responsibility of this Channel Adapter + implementation, the referenced listener container must NOT already have its own MessageListener configured. + + + + + + + + + + + + MessageConverter to use when receiving AMQP Messages. + + + + + + + + + + + + MessagePropertiesConverter to use when receiving AMQP Messages. + + + + + + + + + + + + + The lifeycle phase determining the start/stop order of the underlying listener container. + + + + + + + + + Specifies how many messages to send to each consumer in a single request. Often this can be set quite high + to improve throughput. It should be greater than or equal to the tx-size value. + + + + + + + + Names of the AMQP Queues from which Messages should be consumed (comma-separated list). + + + + + + + + The timeout for each attempt by a consumer to receive the next message. + + + + + + + + + The interval between recovery attempts, in milliseconds. The default is 5000 ms, that is, 5 seconds. + + + + + + + + + The time to wait for workers in milliseconds after the container is stopped, and before the connection is forced closed. + + + + + + + + + How many messages to process in a single transaction (if the channel is transactional). For best results it should be + less than or equal to the prefetch count. + + + + + + + + + Reference to the Executor to be used for running Consumer threads. + + + + + + + + + + + + The TransactionAttribute to use when the Consumer receives the AMQP Message and the Listener is invoked + within a transaction. This is only applicable when a TransactionManager has been configured. + + + + + + + + + + + + The PlatformTransactionManager to use when the Consumer receives the AMQP Message and the Listener is invoked. + + + + + + + + + + + + + + Configures a gateway that will publish an AMQP Message to the provided Exchange + and expect a reply Message. + + + + + + + Unique ID for this gateway. + + + + + + + Message Channel to which Messages should be sent in order to have them converted and published to an AMQP Exchange. + + + + + + + + + + + + Message Channel to which replies should be sent after being received from an AQMP Queue and converted. + + + + + + + + + + + + The name of the AMQP Exchange to which Messages should be sent. If not provided, Messages will be sent to the default, no-name Exchange. + + + + + + + The routing-key to use when sending Messages. By default, this will be an empty String. + + + + + + + The routing-key to use when sending Messages evealuated as an expression on the message (e.g. 'payload.key'). By default, this will be an empty String. + + + + + + + + + + + + + + + + + + + Configures a gateway that will receive AMQP Messages sent to a given queue and then forward those messages to a Message Channel. + If a reply Message is returned, it will also send that to the 'replyTo' provide by the AMQP request Message. + + + + + + + Unique ID for this adapter. + + + + + + + Message Channel to which converted Messages should be sent. + + + + + + + + + + + + Message Channel where reply Messages will be expected. + + + + + + + + + + + + Names of the AMQP Queues from which Messages should be consumed (comma-separated list). + + + + + + + Specify the number of concurrent consumers to create. Default is 1. + Raising the number of concurrent consumers is recommended in order to scale the consumption of messages coming in + from a queue. However, note that any ordering guarantees are lost once multiple consumers are registered. In + general, stick with 1 consumer for low-volume queues. + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample-context.xml new file mode 100644 index 0000000000..797ebd320a --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample-context.xml @@ -0,0 +1,49 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample.java new file mode 100644 index 0000000000..7b2ade3541 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample.java @@ -0,0 +1,31 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.config; + +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author Mark Fisher + * @since 2.1 + */ +public class EchoSample { + + public static void main(String[] args) { + new ClassPathXmlApplicationContext("EchoSample-context.xml", EchoSample.class); + } + +} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml new file mode 100644 index 0000000000..944c47f076 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml @@ -0,0 +1,47 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java new file mode 100644 index 0000000000..23b5927f65 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java @@ -0,0 +1,48 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.config; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Mark Fisher + * @since 2.1 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class OutboundGatewayIntegrationTests { + + @Test + public void run() throws Exception { + SingleConnectionFactory connectionFactory = new SingleConnectionFactory(); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + template.convertAndSend("si.test.exchange", "si.test.binding", "foo"); + } + + + public static class EchoBean { + String echo(String o) { + return o.toUpperCase(); + } + } + +} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests-context.xml new file mode 100644 index 0000000000..07c654c75c --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests-context.xml @@ -0,0 +1,31 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java new file mode 100644 index 0000000000..5f5af7d2dd --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java @@ -0,0 +1,49 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.config; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.springframework.beans.factory.xml.XmlBeanFactory; +import org.springframework.core.io.ClassPathResource; +import org.springframework.test.util.ReflectionTestUtils; + +/** + * @author Mark Fisher + * @author Dave Syer + * @since 2.1 + */ +public class OutboundGatewayTests { + + private XmlBeanFactory beanFactory = new XmlBeanFactory(new ClassPathResource(getClass().getSimpleName()+"-context.xml", getClass())); + + @Test + public void testVanillaConfiguration() throws Exception { + assertTrue(beanFactory.containsBeanDefinition("vanilla")); + beanFactory.getBean("vanilla"); + } + + @Test + public void testExpressionBasedConfiguration() throws Exception { + assertTrue(beanFactory.containsBeanDefinition("expression")); + Object target = beanFactory.getBean("expression"); + assertNotNull(ReflectionTestUtils.getField(ReflectionTestUtils.getField(target, "handler"), "routingKeyGenerator")); + } + +} diff --git a/spring-integration-amqp/src/test/resources/log4j.properties b/spring-integration-amqp/src/test/resources/log4j.properties new file mode 100644 index 0000000000..dcc21d1995 --- /dev/null +++ b/spring-integration-amqp/src/test/resources/log4j.properties @@ -0,0 +1,9 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - <%m>%n + +log4j.category.org.springframework=WARN +log4j.category.org.springframework.amqp=WARN +log4j.category.org.springframework.integration.amqp=WARN diff --git a/spring-integration-amqp/template.mf b/spring-integration-amqp/template.mf new file mode 100644 index 0000000000..618b310949 --- /dev/null +++ b/spring-integration-amqp/template.mf @@ -0,0 +1,15 @@ +Bundle-SymbolicName: org.springframework.integration.amqp +Bundle-Name: Spring Integration AMQP Support +Bundle-Vendor: SpringSource +Bundle-ManifestVersion: 2 +Import-Template: + org.apache.commons.logging;version="[1.1.1, 2.0.0)", + org.apache.commons.lang.*;version="[2.5.0, 3.0.0)", + org.springframework.integration.*;version="[2.1.0, 2.1.1)", + org.springframework.beans.*;version="[3.1.0, 4.0.0)", + org.springframework.context;version="[3.1.0, 4.0.0)", + org.springframework.core.*;version="[3.1.0, 4.0.0)", + org.springframework.util;version="[3.1.0, 4.0.0)", + org.springframework.amqp.*;version="[1.0.0, 2.0.0)", + javax.*;version="0", + org.w3c.dom.*;version="0" diff --git a/spring-integration-event/pom.xml b/spring-integration-event/pom.xml index 33e14d905c..95d556f52c 100644 --- a/spring-integration-event/pom.xml +++ b/spring-integration-event/pom.xml @@ -100,6 +100,12 @@ + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -160,12 +166,6 @@ 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-feed/pom.xml b/spring-integration-feed/pom.xml index ec8005e950..5ea54a86b8 100644 --- a/spring-integration-feed/pom.xml +++ b/spring-integration-feed/pom.xml @@ -100,12 +100,6 @@ - - org.springframework.integration - spring-integration-test - 2.1.0.BUILD-SNAPSHOT - test - org.springframework spring-context @@ -136,6 +130,12 @@ 1.2.12 test + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -148,6 +148,12 @@ 2.3 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test @@ -166,12 +172,6 @@ 1.8.4 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - net.java.dev.rome rome diff --git a/spring-integration-file/pom.xml b/spring-integration-file/pom.xml index 747e4eb1af..2972c24f38 100644 --- a/spring-integration-file/pom.xml +++ b/spring-integration-file/pom.xml @@ -100,6 +100,12 @@ + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -160,12 +166,6 @@ 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-ftp/pom.xml b/spring-integration-ftp/pom.xml index 6008464101..ed607196ab 100644 --- a/spring-integration-ftp/pom.xml +++ b/spring-integration-ftp/pom.xml @@ -102,9 +102,9 @@ org.springframework.integration - spring-integration-test + spring-integration-file 2.1.0.BUILD-SNAPSHOT - test + compile cglib @@ -136,18 +136,18 @@ 2.3 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test 3.1.0.M2 test - - org.springframework.integration - spring-integration-file - 2.1.0.BUILD-SNAPSHOT - compile - org.springframework spring-context-support diff --git a/spring-integration-groovy/pom.xml b/spring-integration-groovy/pom.xml index 2c7043bc9e..1a4f227663 100644 --- a/spring-integration-groovy/pom.xml +++ b/spring-integration-groovy/pom.xml @@ -100,6 +100,12 @@ + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -160,12 +166,6 @@ 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-http/pom.xml b/spring-integration-http/pom.xml index bbc9067a03..63075d05eb 100644 --- a/spring-integration-http/pom.xml +++ b/spring-integration-http/pom.xml @@ -100,12 +100,6 @@ - - org.springframework.integration - spring-integration-test - 2.1.0.BUILD-SNAPSHOT - test - javax.servlet servlet-api @@ -130,6 +124,12 @@ 3.1.0.M2 compile + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -155,6 +155,12 @@ 2.3 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test @@ -173,12 +179,6 @@ 1.8.4 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - junit junit-dep diff --git a/spring-integration-ip/pom.xml b/spring-integration-ip/pom.xml index 8fa6faa639..4825e4f197 100644 --- a/spring-integration-ip/pom.xml +++ b/spring-integration-ip/pom.xml @@ -100,6 +100,12 @@ + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -124,6 +130,12 @@ 2.1.0.BUILD-SNAPSHOT test + + org.springframework.integration + spring-integration-stream + 2.1.0.BUILD-SNAPSHOT + runtime + org.springframework spring-context @@ -154,24 +166,12 @@ 2.2 test - - org.springframework.integration - spring-integration-stream - 2.1.0.BUILD-SNAPSHOT - runtime - junit junit-dep 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-jdbc/pom.xml b/spring-integration-jdbc/pom.xml index 5f3b5780fe..e9a80b7a46 100644 --- a/spring-integration-jdbc/pom.xml +++ b/spring-integration-jdbc/pom.xml @@ -110,12 +110,6 @@ 3.1.0.M2 compile - - org.springframework.integration - spring-integration-test - 2.1.0.BUILD-SNAPSHOT - test - org.springframework spring-context @@ -146,6 +140,12 @@ 2.2 test + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + com.h2database h2 @@ -176,6 +176,12 @@ 3.1.0.M2 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.hamcrest hamcrest-all @@ -206,12 +212,6 @@ 1.6.8 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - junit junit-dep diff --git a/spring-integration-jms/pom.xml b/spring-integration-jms/pom.xml index 50278979e8..4cecff8020 100644 --- a/spring-integration-jms/pom.xml +++ b/spring-integration-jms/pom.xml @@ -106,12 +106,6 @@ 5.3.0 test - - org.springframework.integration - spring-integration-test - 2.1.0.BUILD-SNAPSHOT - test - org.springframework spring-context @@ -142,6 +136,12 @@ 3.1.0.M2 compile + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -160,6 +160,12 @@ 2.3 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test @@ -190,12 +196,6 @@ 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-jmx/pom.xml b/spring-integration-jmx/pom.xml index 9ac1d463e6..38742aff75 100644 --- a/spring-integration-jmx/pom.xml +++ b/spring-integration-jmx/pom.xml @@ -100,12 +100,6 @@ - - org.springframework.integration - spring-integration-test - 2.1.0.BUILD-SNAPSHOT - test - org.springframework spring-context @@ -130,6 +124,12 @@ 1.2.12 test + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -142,6 +142,12 @@ 2.3 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test @@ -166,12 +172,6 @@ 1.6.8 compile - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - junit junit-dep diff --git a/spring-integration-mail/pom.xml b/spring-integration-mail/pom.xml index 305656b28e..ee68b7be4e 100644 --- a/spring-integration-mail/pom.xml +++ b/spring-integration-mail/pom.xml @@ -100,12 +100,6 @@ - - org.springframework.integration - spring-integration-test - 2.1.0.BUILD-SNAPSHOT - test - cglib cglib-nodep @@ -118,6 +112,12 @@ 1.2.12 test + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -130,6 +130,12 @@ 2.3 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test @@ -161,12 +167,6 @@ 1.8.4 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - javax.mail mail diff --git a/spring-integration-rmi/pom.xml b/spring-integration-rmi/pom.xml index 7a22065d6d..91f0ba08d3 100644 --- a/spring-integration-rmi/pom.xml +++ b/spring-integration-rmi/pom.xml @@ -100,6 +100,12 @@ + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -166,12 +172,6 @@ 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-security/pom.xml b/spring-integration-security/pom.xml index bb8abf628e..640f0e820d 100644 --- a/spring-integration-security/pom.xml +++ b/spring-integration-security/pom.xml @@ -124,6 +124,12 @@ 1.2.12 test + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -178,12 +184,6 @@ 3.1.0.M2 compile - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - junit junit-dep diff --git a/spring-integration-sftp/pom.xml b/spring-integration-sftp/pom.xml index 8aef64fbef..92bf86854e 100644 --- a/spring-integration-sftp/pom.xml +++ b/spring-integration-sftp/pom.xml @@ -108,9 +108,15 @@ org.springframework.integration - spring-integration-test + spring-integration-stream 2.1.0.BUILD-SNAPSHOT - test + compile + + + org.springframework.integration + spring-integration-file + 2.1.0.BUILD-SNAPSHOT + compile log4j @@ -126,7 +132,7 @@ org.springframework.integration - spring-integration-stream + spring-integration-core 2.1.0.BUILD-SNAPSHOT compile @@ -142,18 +148,18 @@ 2.3 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test 3.1.0.M2 test - - org.springframework.integration - spring-integration-file - 2.1.0.BUILD-SNAPSHOT - compile - org.springframework spring-context-support @@ -185,12 +191,6 @@ 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-stream/pom.xml b/spring-integration-stream/pom.xml index a852578392..98b447ad3a 100644 --- a/spring-integration-stream/pom.xml +++ b/spring-integration-stream/pom.xml @@ -100,6 +100,12 @@ + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -154,12 +160,6 @@ 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-test/pom.xml b/spring-integration-test/pom.xml index 7c46aee3a9..58c5766eba 100644 --- a/spring-integration-test/pom.xml +++ b/spring-integration-test/pom.xml @@ -100,6 +100,12 @@ + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -154,12 +160,6 @@ 4.8.2 compile - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-twitter/pom.xml b/spring-integration-twitter/pom.xml index 31d7bdee9e..d899a911d9 100644 --- a/spring-integration-twitter/pom.xml +++ b/spring-integration-twitter/pom.xml @@ -100,12 +100,6 @@ - - org.springframework.integration - spring-integration-test - 2.1.0.BUILD-SNAPSHOT - test - org.twitter4j twitter4j-core @@ -124,6 +118,12 @@ 1.2.12 test + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -136,6 +136,12 @@ 2.3 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test @@ -167,12 +173,6 @@ 1.8.4 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - junit junit-dep diff --git a/spring-integration-ws/pom.xml b/spring-integration-ws/pom.xml index aa8d5a7c87..20214b7a41 100644 --- a/spring-integration-ws/pom.xml +++ b/spring-integration-ws/pom.xml @@ -100,12 +100,6 @@ - - org.springframework.integration - spring-integration-test - 2.1.0.BUILD-SNAPSHOT - test - com.sun.xml.messaging.saaj saaj-impl @@ -150,6 +144,12 @@ true + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -174,6 +174,12 @@ 1.0.1 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test @@ -199,12 +205,6 @@ 1.8.4 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - junit junit-dep diff --git a/spring-integration-xml/pom.xml b/spring-integration-xml/pom.xml index 0edc9c5989..bc53f3b173 100644 --- a/spring-integration-xml/pom.xml +++ b/spring-integration-xml/pom.xml @@ -100,12 +100,6 @@ - - org.springframework.integration - spring-integration-test - 2.1.0.BUILD-SNAPSHOT - test - org.springframework.ws spring-xml @@ -148,6 +142,12 @@ 2.0EA3 test + + org.springframework.integration + spring-integration-core + 2.1.0.BUILD-SNAPSHOT + compile + org.easymock easymock @@ -166,6 +166,12 @@ 3.1.0.M2 compile + + org.springframework + spring-test + 3.1.0.M2 + test + stax stax-api @@ -173,9 +179,9 @@ test - org.springframework - spring-test - 3.1.0.M2 + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT test @@ -203,12 +209,6 @@ 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8 diff --git a/spring-integration-xmpp/pom.xml b/spring-integration-xmpp/pom.xml index 22c7c1d792..f282bbb4d0 100644 --- a/spring-integration-xmpp/pom.xml +++ b/spring-integration-xmpp/pom.xml @@ -102,7 +102,7 @@ org.springframework.integration - spring-integration-test + spring-integration-stream 2.1.0.BUILD-SNAPSHOT test @@ -120,9 +120,9 @@ org.springframework.integration - spring-integration-stream + spring-integration-core 2.1.0.BUILD-SNAPSHOT - test + compile org.easymock @@ -136,6 +136,12 @@ 2.3 test + + org.springframework.integration + spring-integration-test + 2.1.0.BUILD-SNAPSHOT + test + org.springframework spring-test @@ -185,12 +191,6 @@ 4.8.2 test - - org.springframework.integration - spring-integration-core - 2.1.0.BUILD-SNAPSHOT - compile - UTF8