diff --git a/spring-eai-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java b/spring-eai-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java index d5994be637..c4ba7668a3 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java @@ -50,7 +50,7 @@ public class ConsumerPolicy { private int initialDelay = 0; - private int period = -1; + private int period = 5; private TimeUnit timeUnit = TimeUnit.MILLISECONDS; diff --git a/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 6d7631ea2d..09f7d7d402 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -58,6 +58,8 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life private ScheduledThreadPoolExecutor dispatcherExecutor; + private MessageChannel invalidMessageChannel; + private boolean autoCreateChannels; private boolean running; @@ -76,6 +78,10 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life this.autoCreateChannels = autoCreateChannels; } + public void setInvalidMessageChannel(MessageChannel invalidMessageChannel) { + this.invalidMessageChannel = invalidMessageChannel; + } + @SuppressWarnings("unchecked") private void registerChannels(ApplicationContext context) { Map channelBeans = @@ -121,6 +127,10 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life return this.channels.get(channelName); } + public MessageChannel getInvalidMessageChannel() { + return this.invalidMessageChannel; + } + public void registerChannel(String name, MessageChannel channel) { this.channels.put(name, channel); } @@ -162,6 +172,10 @@ public class MessageBus implements ChannelMapping, ApplicationContextAware, Life this.dispatcherTasks.add(dispatcherTask); if (this.isRunning()) { scheduleDispatcherTask(dispatcherTask); + if (this.logger.isInfoEnabled()) { + logger.info("scheduled dispatcher task: channel='" + + channelName + "' endpoint='" + endpointName + "'"); + } } } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/ChannelMapping.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/ChannelMapping.java index eeede640ce..a5ed893be3 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/ChannelMapping.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/channel/ChannelMapping.java @@ -25,4 +25,6 @@ public interface ChannelMapping { MessageChannel getChannel(String channelName); + MessageChannel getInvalidMessageChannel(); + } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/config/AbstractChannelAdapterParser.java b/spring-eai-core/src/main/java/org/springframework/integration/config/AbstractChannelAdapterParser.java new file mode 100644 index 0000000000..91fb7bb5ce --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/config/AbstractChannelAdapterParser.java @@ -0,0 +1,70 @@ +/* + * Copyright 2002-2007 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.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.parsing.BeanComponentDefinition; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.beans.factory.xml.BeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter; +import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter; +import org.springframework.util.StringUtils; + +/** + * Base parser for inbound and outbound channel adapters. + * + * @author Mark Fisher + */ +public abstract class AbstractChannelAdapterParser implements BeanDefinitionParser { + + private static final String ID_ATTRIBUTE = "id"; + + private static final String REF_ATTRIBUTE = "ref"; + + private static final String METHOD_ATTRIBUTE = "method"; + + + public BeanDefinition parse(Element element, ParserContext parserContext) { + RootBeanDefinition adapterDef = null; + if (this.isInbound()) { + adapterDef = new RootBeanDefinition(InboundMethodInvokingChannelAdapter.class); + } + else { + adapterDef = new RootBeanDefinition(OutboundMethodInvokingChannelAdapter.class); + } + adapterDef.setSource(parserContext.extractSource(element)); + String ref = element.getAttribute(REF_ATTRIBUTE); + String method = element.getAttribute(METHOD_ATTRIBUTE); + if (!StringUtils.hasText(ref) || !StringUtils.hasText(method)) { + } + adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref)); + adapterDef.getPropertyValues().addPropertyValue("method", method); + String beanName = element.getAttribute(ID_ATTRIBUTE); + if (!StringUtils.hasText(beanName)) { + beanName = parserContext.getReaderContext().generateBeanName(adapterDef); + } + parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, beanName)); + return adapterDef; + } + + protected abstract boolean isInbound(); + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java b/spring-eai-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java new file mode 100644 index 0000000000..f32482caf8 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java @@ -0,0 +1,64 @@ +/* + * Copyright 2002-2007 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.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.parsing.BeanComponentDefinition; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.beans.factory.xml.BeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; + +/** + * @author Mark Fisher + */ +public class AnnotationDrivenParser implements BeanDefinitionParser { + + private static final String PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME = + "internal.PublisherAnnotationPostProcessor"; + + private static final String SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME = + "internal.SubscriberAnnotationPostProcessor"; + + + public BeanDefinition parse(Element element, ParserContext parserContext) { + this.createPublisherPostProcessor(parserContext); + this.createSubscriberPostProcessor(parserContext); + return null; + } + + private void createPublisherPostProcessor(ParserContext parserContext) { + BeanDefinition bd = new RootBeanDefinition(PublisherAnnotationPostProcessor.class); + bd.getPropertyValues().addPropertyValue("channelMapping", + new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); + BeanComponentDefinition bcd = new BeanComponentDefinition( + bd, PUBLISHER_ANNOTATION_POST_PROCESSOR_BEAN_NAME); + parserContext.registerBeanComponent(bcd); + } + + private void createSubscriberPostProcessor(ParserContext parserContext) { + BeanDefinition bd = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class); + bd.getPropertyValues().addPropertyValue("messageBus", + new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); + BeanComponentDefinition bcd = new BeanComponentDefinition( + bd, SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME); + parserContext.registerBeanComponent(bcd); + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/config/InboundChannelAdapterParser.java b/spring-eai-core/src/main/java/org/springframework/integration/config/InboundChannelAdapterParser.java new file mode 100644 index 0000000000..362f9469df --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/config/InboundChannelAdapterParser.java @@ -0,0 +1,29 @@ +/* + * Copyright 2002-2007 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.config; + +/** + * @author Mark Fisher + */ +public class InboundChannelAdapterParser extends AbstractChannelAdapterParser { + + @Override + protected boolean isInbound() { + return true; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java b/spring-eai-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java index 4d6f0c5b11..ae7fab3a6f 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java @@ -19,14 +19,18 @@ package org.springframework.integration.config; import org.springframework.beans.factory.xml.NamespaceHandlerSupport; /** - * Handler for the integration namespace. + * Namespace handler for the integration namespace. * * @author Mark Fisher */ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport { public void init() { + registerBeanDefinitionParser("message-bus", new MessageBusParser()); + registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenParser()); registerBeanDefinitionParser("channel", new ChannelParser()); + registerBeanDefinitionParser("inbound-channel-adapter", new InboundChannelAdapterParser()); + registerBeanDefinitionParser("outbound-channel-adapter", new OutboundChannelAdapterParser()); registerBeanDefinitionParser("endpoint", new EndpointParser()); } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/config/MessageBusParser.java b/spring-eai-core/src/main/java/org/springframework/integration/config/MessageBusParser.java new file mode 100644 index 0000000000..5046c9e526 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/config/MessageBusParser.java @@ -0,0 +1,47 @@ +/* + * Copyright 2002-2007 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.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.BeanDefinitionStoreException; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.bus.MessageBus; + +/** + * Parser for the message-bus element of the integration namespace. + * + * @author Mark Fisher + */ +public class MessageBusParser extends AbstractSingleBeanDefinitionParser { + + public static final String MESSAGE_BUS_BEAN_NAME = "internal.messageBus"; + + @Override + protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) + throws BeanDefinitionStoreException { + return MESSAGE_BUS_BEAN_NAME; + } + + @Override + protected Class getBeanClass(Element element) { + return MessageBus.class; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/config/OutboundChannelAdapterParser.java b/spring-eai-core/src/main/java/org/springframework/integration/config/OutboundChannelAdapterParser.java new file mode 100644 index 0000000000..154718dfe2 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/config/OutboundChannelAdapterParser.java @@ -0,0 +1,29 @@ +/* + * Copyright 2002-2007 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.config; + +/** + * @author Mark Fisher + */ +public class OutboundChannelAdapterParser extends AbstractChannelAdapterParser { + + @Override + protected boolean isInbound() { + return false; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationPostProcessor.java b/spring-eai-core/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java similarity index 95% rename from spring-eai-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationPostProcessor.java rename to spring-eai-core/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java index 672dc3219f..9dfd7d7940 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationPostProcessor.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.aop; +package org.springframework.integration.config; import java.lang.annotation.Annotation; @@ -25,6 +25,8 @@ import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.integration.aop.Publisher; +import org.springframework.integration.aop.PublisherAnnotationAdvisor; import org.springframework.integration.channel.ChannelMapping; import org.springframework.util.Assert; diff --git a/spring-eai-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java b/spring-eai-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java new file mode 100644 index 0000000000..cf90f9090d --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java @@ -0,0 +1,103 @@ +/* + * Copyright 2002-2007 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.config; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.aop.framework.Advised; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.endpoint.MessageHandlerAdapter; +import org.springframework.integration.endpoint.annotation.Subscriber; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.ReflectionUtils; + +/** + * A {@link BeanPostProcessor} that creates a method-invoking handler adapter + * when it discovers methods annotated with {@link Subscriber @Subscriber}. + * + * @author Mark Fisher + */ +public class SubscriberAnnotationPostProcessor implements BeanPostProcessor { + + private Log logger = LogFactory.getLog(this.getClass()); + + private Class subscriberAnnotationType = Subscriber.class; + + private String channelNameAttribute = "channel"; + + private MessageBus messageBus; + + + public void setSubscriberAnnotationType(Class subscriberAnnotationType) { + Assert.notNull(subscriberAnnotationType, "subscriberAnnotationType must not be null"); + this.subscriberAnnotationType = subscriberAnnotationType; + } + + public void setChannelNameAttribute(String channelNameAttribute) { + Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null"); + this.channelNameAttribute = channelNameAttribute; + } + + public void setMessageBus(MessageBus messageBus) { + Assert.notNull(messageBus, "messageBus must not be null"); + this.messageBus = messageBus; + } + + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + public Object postProcessAfterInitialization(final Object bean, String beanName) throws BeansException { + final Class targetClass = bean instanceof Advised ? + ((Advised) bean).getTargetSource().getTargetClass() : bean.getClass(); + if (targetClass == null) { + return bean; + } + ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + Annotation annotation = method.getAnnotation(subscriberAnnotationType); + if (annotation != null) { + String channelName = (String) AnnotationUtils.getValue(annotation, channelNameAttribute); + MessageHandlerAdapter adapter = new MessageHandlerAdapter(); + adapter.setMethod(method.getName()); + adapter.setObject(bean); + adapter.afterPropertiesSet(); + GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); + endpoint.setInputChannelName(channelName); + endpoint.setChannelMapping(messageBus); + endpoint.setHandler(adapter); + String endpointName = ClassUtils.getShortNameAsProperty(targetClass) + + "-" + method.getName() + "-endpoint"; + messageBus.registerEndpoint(endpointName, endpoint); + if (logger.isInfoEnabled()) { + logger.info("registered endpoint: " + endpointName); + } + } + } + }); + return bean; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/spring-eai-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index 037e5f1e50..f752215167 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/spring-eai-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -17,6 +17,28 @@ ]]> + + + + + Defines a message bus. + + + + + + + + + + + + Enables the publisher annotation post-processor. + + + + + @@ -32,6 +54,38 @@ + + + + + Defines an inbound channel adapter. + + + + + + + + + + + + + + + + Defines an outbound channel adapter. + + + + + + + + + + + diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java index df7f373552..8a01fdbf55 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java @@ -47,7 +47,7 @@ public class GenericMessageEndpoint implements MessageEndpoint { private ChannelMapping channelMapping; - private ConsumerPolicy consumerPolicy; + private ConsumerPolicy consumerPolicy = new ConsumerPolicy(); /** @@ -94,7 +94,7 @@ public class GenericMessageEndpoint implements MessageEndpoint { } - public void messageReceived(Message message) { + public void messageReceived(Message message) { if (this.handler == null) { if (this.defaultOutputChannelName == null) { throw new MessagingConfigurationException( @@ -104,7 +104,7 @@ public class GenericMessageEndpoint implements MessageEndpoint { replyChannel.send(message); return; } - Message replyMessage = handler.handle(message); + Message replyMessage = handler.handle(message); if (replyMessage != null) { MessageChannel replyChannel = this.resolveReplyChannel(message); if (replyChannel == null) { @@ -116,7 +116,7 @@ public class GenericMessageEndpoint implements MessageEndpoint { } } - private MessageChannel resolveReplyChannel(Message message) { + private MessageChannel resolveReplyChannel(Message message) { if (this.channelMapping == null) { return null; } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index c615cf9503..b2213b43ca 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -37,6 +37,6 @@ public interface MessageEndpoint { void setChannelMapping(ChannelMapping channelMapping); - void messageReceived(Message message); + void messageReceived(Message message); } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/AnnotationHandlerCreator.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/AnnotationHandlerCreator.java new file mode 100644 index 0000000000..5a7ac09dc8 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/AnnotationHandlerCreator.java @@ -0,0 +1,31 @@ +/* + * Copyright 2002-2007 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.endpoint.annotation; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; + +import org.springframework.integration.handler.MessageHandler; + +/** + * @author Mark Fisher + */ +public interface AnnotationHandlerCreator { + + MessageHandler createHandler(Object object, Method method, Annotation annotation); + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/DefaultAnnotationHandlerCreator.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/DefaultAnnotationHandlerCreator.java new file mode 100644 index 0000000000..4f6dfceade --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/DefaultAnnotationHandlerCreator.java @@ -0,0 +1,44 @@ +/* + * Copyright 2002-2007 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.endpoint.annotation; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; + +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.core.annotation.Order; +import org.springframework.integration.endpoint.MessageHandlerAdapter; +import org.springframework.integration.handler.MessageHandler; + +/** + * @author Mark Fisher + */ +public class DefaultAnnotationHandlerCreator implements AnnotationHandlerCreator { + + public MessageHandler createHandler(Object object, Method method, Annotation annotation) { + MessageHandlerAdapter adapter = new MessageHandlerAdapter(); + adapter.setObject(object); + adapter.setMethod(method.getName()); + Order orderAnnotation = (Order) AnnotationUtils.getAnnotation(method, Order.class); + if (orderAnnotation != null) { + adapter.setOrder(orderAnnotation.value()); + } + adapter.afterPropertiesSet(); + return adapter; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/EndpointAnnotationPostProcessor.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/EndpointAnnotationPostProcessor.java new file mode 100644 index 0000000000..c2902d022d --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/EndpointAnnotationPostProcessor.java @@ -0,0 +1,172 @@ +/* + * Copyright 2002-2007 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.endpoint.annotation; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.core.OrderComparator; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter; +import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.handler.MessageHandlerChain; +import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.StringUtils; + +/** + * A {@link BeanPostProcessor} implementation that generates endpoints for + * classes annotated with {@link MessageEndpoint @MessageEndpoint}. + * + * @author Mark Fisher + */ +public class EndpointAnnotationPostProcessor implements BeanPostProcessor, InitializingBean { + + private Map, AnnotationHandlerCreator> handlerCreators = + new ConcurrentHashMap, AnnotationHandlerCreator>(); + + private MessageBus messageBus; + + + public void setMessageBus(MessageBus messageBus) { + Assert.notNull(messageBus, "messageBus must not be null"); + this.messageBus = messageBus; + } + + public void setCustomHandlerCreators( + Map, AnnotationHandlerCreator> customHandlerCreators) { + for (Map.Entry, AnnotationHandlerCreator> entry : customHandlerCreators.entrySet()) { + this.handlerCreators.put(entry.getKey(), entry.getValue()); + } + } + + public void afterPropertiesSet() { + Assert.notNull(this.messageBus, "messageBus is required"); + this.handlerCreators.put(Handler.class, new DefaultAnnotationHandlerCreator()); + } + + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + MessageEndpoint endpointAnnotation = bean.getClass().getAnnotation(MessageEndpoint.class); + if (endpointAnnotation == null) { + return bean; + } + GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); + this.configureInputChannel(bean, beanName, endpointAnnotation, endpoint); + this.configureDefaultOutputChannel(bean, beanName, endpointAnnotation, endpoint); + MessageHandlerChain handlerChain = this.createHandlerChain(bean); + if (handlerChain != null) { + endpoint.setHandler(handlerChain); + } + this.messageBus.registerEndpoint(beanName, endpoint); + return endpoint; + } + + private void configureInputChannel(final Object bean, final String beanName, + MessageEndpoint annotation, final GenericMessageEndpoint endpoint) { + String channelName = annotation.input(); + if (StringUtils.hasText(channelName)) { + endpoint.setInputChannelName(channelName); + ConsumerPolicy consumerPolicy = new ConsumerPolicy(); + consumerPolicy.setPeriod(annotation.pollPeriod()); + endpoint.setConsumerPolicy(consumerPolicy); + return; + } + ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + Annotation annotation = AnnotationUtils.getAnnotation(method, Polled.class); + if (annotation != null) { + InboundMethodInvokingChannelAdapter adapter = new InboundMethodInvokingChannelAdapter(); + adapter.setObject(bean); + adapter.setMethod(method.getName()); + adapter.afterPropertiesSet(); + String channelName = beanName + "-inputChannel"; + messageBus.registerChannel(channelName, adapter); + endpoint.setInputChannelName(channelName); + return; + } + } + }); + } + + private void configureDefaultOutputChannel(final Object bean, final String beanName, + final MessageEndpoint annotation, final GenericMessageEndpoint endpoint) { + String channelName = annotation.defaultOutput(); + if (StringUtils.hasText(channelName)) { + endpoint.setDefaultOutputChannelName(channelName); + return; + } + ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + Annotation annotation = AnnotationUtils.getAnnotation(method, DefaultOutput.class); + if (annotation != null) { + OutboundMethodInvokingChannelAdapter adapter = new OutboundMethodInvokingChannelAdapter(); + adapter.setObject(bean); + adapter.setMethod(method.getName()); + adapter.afterPropertiesSet(); + String channelName = beanName + "-defaultOutputChannel"; + messageBus.registerChannel(channelName, adapter); + endpoint.setDefaultOutputChannelName(channelName); + return; + } + } + }); + } + + @SuppressWarnings("unchecked") + private MessageHandlerChain createHandlerChain(final Object bean) { + final List handlers = new ArrayList(); + ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + for (Class annotationType : handlerCreators.keySet()) { + Annotation annotation = AnnotationUtils.getAnnotation(method, annotationType); + if (annotation != null) { + MessageHandler handler = handlerCreators.get(annotationType).createHandler(bean, method, annotation); + if (handler != null) { + handlers.add(handler); + } + } + } + } + }); + if (handlers.size() > 0) { + MessageHandlerChain handlerChain = new MessageHandlerChain(); + Collections.sort(handlers, new OrderComparator()); + for (MessageHandler handler : handlers) { + handlerChain.add(handler); + } + return handlerChain; + } + return null; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/Subscriber.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/Subscriber.java new file mode 100644 index 0000000000..35c5248a44 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/annotation/Subscriber.java @@ -0,0 +1,40 @@ +/* + * Copyright 2002-2007 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.endpoint.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Indicates that a method-invoking handler adapter should delegate to this + * method. + * + * @author Mark Fisher + */ +@Target(ElementType.METHOD) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +public @interface Subscriber { + + String channel(); + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/handler/MessageHandler.java b/spring-eai-core/src/main/java/org/springframework/integration/handler/MessageHandler.java index e30243a9ea..13a87f02b7 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/handler/MessageHandler.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/handler/MessageHandler.java @@ -27,6 +27,6 @@ import org.springframework.integration.message.Message; */ public interface MessageHandler { - Message handle(Message message); + Message handle(Message message); } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/message/AbstractMessage.java b/spring-eai-core/src/main/java/org/springframework/integration/message/GenericMessage.java similarity index 85% rename from spring-eai-core/src/main/java/org/springframework/integration/message/AbstractMessage.java rename to spring-eai-core/src/main/java/org/springframework/integration/message/GenericMessage.java index ec12e46b87..a06d2fd6e0 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/message/AbstractMessage.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/message/GenericMessage.java @@ -27,18 +27,18 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public abstract class AbstractMessage implements Message { +public class GenericMessage implements Message { private Object id; private MessageHeader header = new MessageHeader(); - private Object payload; + private T payload; private ReentrantLock lock; - protected AbstractMessage(Object id, Object payload) { + public GenericMessage(Object id, T payload) { Assert.notNull(id, "id must not be null"); Assert.notNull(payload, "payload must not be null"); this.id = id; @@ -57,11 +57,11 @@ public abstract class AbstractMessage implements Message { this.header = header; } - public Object getPayload() { + public T getPayload() { return this.payload; } - protected void setPayload(Object newPayload) { + protected void setPayload(T newPayload) { this.payload = newPayload; } @@ -85,7 +85,8 @@ public abstract class AbstractMessage implements Message { public void transformPayload(ObjectTransformer transformer) { this.lock(); try { - this.setPayload(transformer.transform(this.getPayload())); + // TODO: remove this method (probably) or parameterize transformer + this.setPayload((T) transformer.transform(this.getPayload())); } finally { this.unlock(); diff --git a/spring-eai-core/src/main/java/org/springframework/integration/message/Message.java b/spring-eai-core/src/main/java/org/springframework/integration/message/Message.java index 6083b1c1fc..92003cdf2e 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/message/Message.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/message/Message.java @@ -23,13 +23,13 @@ import org.springframework.integration.transformer.ObjectTransformer; * * @author Mark Fisher */ -public interface Message { +public interface Message { Object getId(); MessageHeader getHeader(); - Object getPayload(); + T getPayload(); void transformPayload(ObjectTransformer transformer); diff --git a/spring-eai-core/src/main/java/org/springframework/integration/message/MessageMapper.java b/spring-eai-core/src/main/java/org/springframework/integration/message/MessageMapper.java index 85687d9f17..e673868701 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/message/MessageMapper.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/message/MessageMapper.java @@ -21,16 +21,16 @@ package org.springframework.integration.message; * * @author Mark Fisher */ -public interface MessageMapper { +public interface MessageMapper { /** * Map to a {@link Message} from the given object. */ - Message toMessage(Object source); + Message toMessage(O source); /** * Map from the given {@link Message} to an object. */ - Object fromMessage(Message message); + O fromMessage(Message message); } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/message/SimplePayloadMessageMapper.java b/spring-eai-core/src/main/java/org/springframework/integration/message/SimplePayloadMessageMapper.java index 0f5f39c176..1884e7269d 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/message/SimplePayloadMessageMapper.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/message/SimplePayloadMessageMapper.java @@ -26,7 +26,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class SimplePayloadMessageMapper implements MessageMapper { +public class SimplePayloadMessageMapper implements MessageMapper { private UidGenerator uidGenerator; @@ -54,15 +54,15 @@ public class SimplePayloadMessageMapper implements MessageMapper { /** * Return the payload of the given Message. */ - public Object fromMessage(Message message) { + public T fromMessage(Message message) { return message.getPayload(); } /** * Return a {@link DocumentMessage} with the given object as its payload. */ - public Message toMessage(Object source) { - return new DocumentMessage(uidGenerator.generateUid(), source); + public Message toMessage(T source) { + return new GenericMessage(uidGenerator.generateUid(), source); } } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/message/DocumentMessage.java b/spring-eai-core/src/main/java/org/springframework/integration/message/StringMessage.java similarity index 79% rename from spring-eai-core/src/main/java/org/springframework/integration/message/DocumentMessage.java rename to spring-eai-core/src/main/java/org/springframework/integration/message/StringMessage.java index 68a37ae0aa..e9c7fd4342 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/message/DocumentMessage.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/message/StringMessage.java @@ -17,13 +17,11 @@ package org.springframework.integration.message; /** - * A simple Message implementation that encapsulates a single Object payload. - * * @author Mark Fisher */ -public class DocumentMessage extends AbstractMessage { +public class StringMessage extends GenericMessage { - public DocumentMessage(Object id, Object payload) { + public StringMessage(Object id, String payload) { super(id, payload); } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/samples/Logger.java b/spring-eai-core/src/main/java/org/springframework/integration/samples/Logger.java new file mode 100644 index 0000000000..9733cb8a5a --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/samples/Logger.java @@ -0,0 +1,31 @@ +/* + * Copyright 2002-2007 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.samples; + +import org.springframework.integration.endpoint.annotation.Subscriber; + +/** + * @author Mark Fisher + */ +public class Logger { + + //@Subscriber(channel="quotes") + public void log(Object o) { + System.out.println(o); + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/samples/Quote.java b/spring-eai-core/src/main/java/org/springframework/integration/samples/Quote.java new file mode 100644 index 0000000000..61472566eb --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/samples/Quote.java @@ -0,0 +1,39 @@ +/* + * Copyright 2002-2007 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.samples; + +import java.math.BigDecimal; + +/** + * @author Mark Fisher + */ +public class Quote { + + private String ticker; + + private BigDecimal price; + + public Quote(String ticker, BigDecimal price) { + this.ticker = ticker; + this.price = price; + } + + public String toString() { + return ticker + ": " + price; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/samples/QuoteService.java b/spring-eai-core/src/main/java/org/springframework/integration/samples/QuoteService.java new file mode 100644 index 0000000000..c25627ce19 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/samples/QuoteService.java @@ -0,0 +1,26 @@ +/* + * Copyright 2002-2007 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.samples; + +/** + * @author Mark Fisher + */ +public interface QuoteService { + + Quote lookup(String ticker); + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/samples/StockQuoteDemo.java b/spring-eai-core/src/main/java/org/springframework/integration/samples/StockQuoteDemo.java new file mode 100644 index 0000000000..2da0ce545e --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/samples/StockQuoteDemo.java @@ -0,0 +1,34 @@ +/* + * Copyright 2002-2007 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.samples; + +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; + +/** + * @author Mark Fisher + */ +public class StockQuoteDemo { + + public static void main(String[] args) throws Exception { + AbstractApplicationContext context = new ClassPathXmlApplicationContext("stockQuoteDemo.xml", StockQuoteDemo.class); + context.start(); + QuoteService service = (QuoteService) context.getBean("quoteService"); + service.lookup("SOA"); + Thread.sleep(1000); + } +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/samples/StubQuoteService.java b/spring-eai-core/src/main/java/org/springframework/integration/samples/StubQuoteService.java new file mode 100644 index 0000000000..943eeebf03 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/samples/StubQuoteService.java @@ -0,0 +1,36 @@ +/* + * Copyright 2002-2007 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.samples; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.util.Random; + +import org.springframework.integration.aop.Publisher; + +/** + * @author Mark Fisher + */ +public class StubQuoteService implements QuoteService { + + @Publisher(channel="quotes") + public Quote lookup(String ticker) { + BigDecimal price = new BigDecimal(new Random().nextDouble() * 100); + return new Quote(ticker, price.setScale(2, RoundingMode.HALF_EVEN)); + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/samples/stockQuoteDemo.xml b/spring-eai-core/src/main/java/org/springframework/integration/samples/stockQuoteDemo.xml new file mode 100644 index 0000000000..60a6239a4c --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/samples/stockQuoteDemo.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-eai-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java b/spring-eai-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java index e58f984a70..3a16bb120e 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java @@ -43,6 +43,9 @@ public class PublisherAnnotationAdvisorTests { } return null; } + public MessageChannel getInvalidMessageChannel() { + return null; + } }; PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelMapping); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); @@ -62,6 +65,9 @@ public class PublisherAnnotationAdvisorTests { } return null; } + public MessageChannel getInvalidMessageChannel() { + return null; + } }; PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelMapping); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml b/spring-eai-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml index 5d508928bd..cbce4ca45e 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml +++ b/spring-eai-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml @@ -10,7 +10,7 @@ - + diff --git a/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java index 52c922cb8d..6f07b13100 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java @@ -24,14 +24,11 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; - -import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.GenericMessage; /** * @author Mark Fisher @@ -74,16 +71,16 @@ public class EventDrivenConsumerTests { bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend - 110; i++) { - channel.send(new DocumentMessage(1, "fast-1." + (i+1))); + channel.send(new GenericMessage(1, "fast-1." + (i+1))); } int activeCountAfterFirstBurst = bus.getActiveCountForEndpoint("testEndpoint"); for (int i = 0; i < 10; i++) { - channel.send(new DocumentMessage(1, "slow-1." + (i+1))); + channel.send(new GenericMessage(1, "slow-1." + (i+1))); Thread.sleep(10); } int activeCountAfterSlowDown = bus.getActiveCountForEndpoint("testEndpoint"); for (int i = 0; i < 100; i++) { - channel.send(new DocumentMessage(1, "fast-2." + (i+1))); + channel.send(new GenericMessage(1, "fast-2." + (i+1))); } int activeCountAfterLastBurst = bus.getActiveCountForEndpoint("testEndpoint"); latch.await(100, TimeUnit.SECONDS); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java index 241ecbdc64..fe8005da61 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java @@ -24,14 +24,11 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; - -import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.GenericMessage; /** * @author Mark Fisher @@ -66,7 +63,7 @@ public class FixedDelayConsumerTests { bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { - channel.send(new DocumentMessage(1, "test " + (i+1))); + channel.send(new GenericMessage(1, "test " + (i+1))); } latch.await(250, TimeUnit.MILLISECONDS); assertEquals(messagesToSend, counter.get()); @@ -100,7 +97,7 @@ public class FixedDelayConsumerTests { bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { - channel.send(new DocumentMessage(1, "test " + (i+1))); + channel.send(new GenericMessage(1, "test " + (i+1))); } latch.await(80, TimeUnit.MILLISECONDS); assertTrue(counter.get() < 10); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java index 771800b2c9..029c007143 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java @@ -24,14 +24,11 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; - -import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.GenericMessage; /** * @author Mark Fisher @@ -63,7 +60,7 @@ public class FixedRateConsumerTests { bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { - channel.send(new DocumentMessage(1, "test " + (i+1))); + channel.send(new GenericMessage(1, "test " + (i+1))); } latch.await(250, TimeUnit.MILLISECONDS); assertEquals(messagesToSend, counter.get()); @@ -97,7 +94,7 @@ public class FixedRateConsumerTests { bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { - channel.send(new DocumentMessage(1, "test " + (i+1))); + channel.send(new GenericMessage(1, "test " + (i+1))); } latch.await(80, TimeUnit.MILLISECONDS); assertTrue(counter.get() < 10); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 0a6a2aead9..b476131bff 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -19,13 +19,10 @@ package org.springframework.integration.bus; import static org.junit.Assert.assertEquals; import org.junit.Test; - import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.GenericMessage; /** * @author Mark Fisher @@ -65,9 +62,8 @@ public class MessageBusTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("messageBusTests.xml", this.getClass()); context.start(); MessageChannel sourceChannel = (MessageChannel) context.getBean("sourceChannel"); - sourceChannel.send(new DocumentMessage("123", "test")); + sourceChannel.send(new GenericMessage("123", "test")); MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel"); - // TODO: add metadata for this MessageBus bus = (MessageBus) context.getBean("bus"); ConsumerPolicy policy = new ConsumerPolicy(); Subscription subscription = new Subscription(); @@ -75,7 +71,7 @@ public class MessageBusTests { subscription.setEndpoint("endpoint"); subscription.setPolicy(policy); bus.activateSubscription(subscription); - Message result = targetChannel.receive(10); + Message result = targetChannel.receive(10); assertEquals("test", result.getPayload()); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java b/spring-eai-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java index 5be49b8a19..30150b8303 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java @@ -29,9 +29,9 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; -import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageSelector; +import org.springframework.integration.message.GenericMessage; /** * @author Mark Fisher @@ -45,7 +45,7 @@ public class PointToPointChannelTests { final PointToPointChannel channel = new PointToPointChannel(); new Thread(new Runnable() { public void run() { - Message message = channel.receive(); + Message message = channel.receive(); if (message != null) { messageReceived.set(true); latch.countDown(); @@ -53,7 +53,7 @@ public class PointToPointChannelTests { } }).start(); assertFalse(messageReceived.get()); - channel.send(new DocumentMessage(1, "testing")); + channel.send(new GenericMessage(1, "testing")); latch.await(25, TimeUnit.MILLISECONDS); assertTrue(messageReceived.get()); } @@ -67,7 +67,7 @@ public class PointToPointChannelTests { Executor singleThreadExecutor = Executors.newSingleThreadExecutor(); Runnable receiveTask1 = new Runnable() { public void run() { - Message message = channel.receive(0); + Message message = channel.receive(0); if (message != null) { messageReceived.set(true); } @@ -76,7 +76,7 @@ public class PointToPointChannelTests { }; Runnable sendTask = new Runnable() { public void run() { - channel.send(new DocumentMessage(1, "testing")); + channel.send(new GenericMessage(1, "testing")); } }; singleThreadExecutor.execute(receiveTask1); @@ -85,7 +85,7 @@ public class PointToPointChannelTests { assertFalse(messageReceived.get()); Runnable receiveTask2 = new Runnable() { public void run() { - Message message = channel.receive(0); + Message message = channel.receive(0); if (message != null) { messageReceived.set(true); } @@ -104,7 +104,7 @@ public class PointToPointChannelTests { final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { public void run() { - Message message = channel.receive(); + Message message = channel.receive(); receiveInterrupted.set(true); assertTrue(message == null); latch.countDown(); @@ -124,7 +124,7 @@ public class PointToPointChannelTests { final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { public void run() { - Message message = channel.receive(10000); + Message message = channel.receive(10000); receiveInterrupted.set(true); assertTrue(message == null); latch.countDown(); @@ -140,26 +140,26 @@ public class PointToPointChannelTests { @Test public void testImmediateSend() { PointToPointChannel channel = new PointToPointChannel(3); - boolean result1 = channel.send(new DocumentMessage(1, "test-1")); + boolean result1 = channel.send(new GenericMessage(1, "test-1")); assertTrue(result1); - boolean result2 = channel.send(new DocumentMessage(2, "test-2"), 100); + boolean result2 = channel.send(new GenericMessage(2, "test-2"), 100); assertTrue(result2); - boolean result3 = channel.send(new DocumentMessage(3, "test-3"), 0); + boolean result3 = channel.send(new GenericMessage(3, "test-3"), 0); assertTrue(result3); - boolean result4 = channel.send(new DocumentMessage(4, "test-4"), 0); + boolean result4 = channel.send(new GenericMessage(4, "test-4"), 0); assertFalse(result4); } @Test public void testBlockingSendWithNoTimeout() throws Exception{ final PointToPointChannel channel = new PointToPointChannel(1); - boolean result1 = channel.send(new DocumentMessage(1, "test-1")); + boolean result1 = channel.send(new GenericMessage(1, "test-1")); assertTrue(result1); final AtomicBoolean sendInterrupted = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { public void run() { - channel.send(new DocumentMessage(2, "test-2")); + channel.send(new GenericMessage(2, "test-2")); sendInterrupted.set(true); latch.countDown(); } @@ -174,13 +174,13 @@ public class PointToPointChannelTests { @Test public void testBlockingSendWithTimeout() throws Exception{ final PointToPointChannel channel = new PointToPointChannel(1); - boolean result1 = channel.send(new DocumentMessage(1, "test-1")); + boolean result1 = channel.send(new GenericMessage(1, "test-1")); assertTrue(result1); final AtomicBoolean sendInterrupted = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { public void run() { - channel.send(new DocumentMessage(2, "test-2"), 10000); + channel.send(new GenericMessage(2, "test-2"), 10000); sendInterrupted.set(true); latch.countDown(); } @@ -196,7 +196,7 @@ public class PointToPointChannelTests { public void testSelectorMatchesWithinTimeout() throws Exception { final PointToPointChannel channel = new PointToPointChannel(); final CountDownLatch latch = new CountDownLatch(1); - final AtomicReference messageRef = new AtomicReference(); + final AtomicReference> messageRef = new AtomicReference>(); Thread receiver = new Thread(new Runnable() { public void run() { Message message = channel.receive(new MessageSelector() { @@ -211,13 +211,13 @@ public class PointToPointChannelTests { receiver.start(); Thread sender = new Thread(new Runnable() { public void run() { - channel.send(new DocumentMessage(1, "test-1")); + channel.send(new GenericMessage(1, "test-1")); try { Thread.sleep(5); } catch (Exception e) {} - channel.send(new DocumentMessage(2, "test-2")); + channel.send(new GenericMessage(2, "test-2")); try { Thread.sleep(5); } catch (Exception e) {} - channel.send(new DocumentMessage(3, "test-3")); + channel.send(new GenericMessage(3, "test-3")); try { Thread.sleep(100); } catch (Exception e) {} - channel.send(new DocumentMessage(4, "test-4")); + channel.send(new GenericMessage(4, "test-4")); } }); sender.start(); @@ -229,7 +229,7 @@ public class PointToPointChannelTests { public void testSelectorDoesNotMatchWithinTimeout() throws Exception { final PointToPointChannel channel = new PointToPointChannel(); final CountDownLatch latch = new CountDownLatch(1); - final AtomicReference messageRef = new AtomicReference(); + final AtomicReference> messageRef = new AtomicReference>(); Thread receiver = new Thread(new Runnable() { public void run() { Message message = channel.receive(new MessageSelector() { @@ -244,11 +244,11 @@ public class PointToPointChannelTests { receiver.start(); Thread sender = new Thread(new Runnable() { public void run() { - channel.send(new DocumentMessage(1, "test-1")); + channel.send(new GenericMessage(1, "test-1")); try { Thread.sleep(5); } catch (Exception e) {} - channel.send(new DocumentMessage(2, "test-2")); + channel.send(new GenericMessage(2, "test-2")); try { Thread.sleep(5); } catch (Exception e) {} - channel.send(new DocumentMessage(3, "test-3")); + channel.send(new GenericMessage(3, "test-3")); } }); sender.start(); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java b/spring-eai-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java index 568b5a7fc5..93235acba8 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java @@ -20,10 +20,9 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import org.junit.Test; - import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.DocumentMessage; +import org.springframework.integration.message.GenericMessage; /** * @author Mark Fisher @@ -36,9 +35,10 @@ public class ChannelParserTests { "channelParserTests.xml", this.getClass()); MessageChannel channel = (MessageChannel) context.getBean("testChannel"); for (int i = 0; i < 10; i++) { - boolean result = channel.send(new DocumentMessage(1, "test"), 10); + boolean result = channel.send(new GenericMessage(1, "test"), 10); assertTrue(result); } - assertFalse(channel.send(new DocumentMessage(1, "test"), 3)); + assertFalse(channel.send(new GenericMessage(1, "test"), 3)); } + } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/config/ComponentConfigurerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/config/ComponentConfigurerTests.java index 9f14b88289..87bdd2ceab 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/config/ComponentConfigurerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/config/ComponentConfigurerTests.java @@ -28,8 +28,8 @@ import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.GenericMessage; /** * @author Mark Fisher @@ -43,10 +43,10 @@ public class ComponentConfigurerTests { context.registerBeanDefinition("tester", new RootBeanDefinition(Tester.class)); context.registerBeanDefinition("out", new RootBeanDefinition(PointToPointChannel.class)); context.registerBeanDefinition("bus", new RootBeanDefinition(MessageBus.class)); - String name = cr.serviceActivator("in", "out", "tester", "test"); + String name = cr.serviceActivator(null, "out", "tester", "test"); context.refresh(); MessageEndpoint endpoint = (MessageEndpoint) context.getBean(name); - endpoint.messageReceived(new DocumentMessage(1, "world")); + endpoint.messageReceived(new GenericMessage(1, "world")); MessageChannel channel = (MessageChannel) context.getBean("out"); assertEquals("hello world", channel.receive().getPayload()); } @@ -58,7 +58,7 @@ public class ComponentConfigurerTests { context.registerBeanDefinition("tester", new RootBeanDefinition(Tester.class)); String adapter = cr.inboundChannelAdapter("tester", "foo"); MessageChannel channel = (MessageChannel) context.getBean(adapter); - Message message = channel.receive(); + Message message = channel.receive(); assertEquals("bar", message.getPayload()); } @@ -71,7 +71,7 @@ public class ComponentConfigurerTests { Tester tester = (Tester) context.getBean("tester"); assertNull(tester.getStoredValue()); MessageChannel channel = (MessageChannel) context.getBean(adapter); - boolean result = channel.send(new DocumentMessage(1, "foo")); + boolean result = channel.send(new GenericMessage(1, "foo")); assertTrue(result); assertEquals("foo", tester.getStoredValue()); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/spring-eai-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 5b5d0bd630..f828d7da6f 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -25,7 +25,7 @@ import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.DocumentMessage; +import org.springframework.integration.message.GenericMessage; /** * @author Mark Fisher @@ -40,7 +40,7 @@ public class EndpointParserTests { MessageChannel channel = (MessageChannel) context.getBean("testChannel"); TestHandler handler = (TestHandler) context.getBean("testHandler"); assertNull(handler.getMessageString()); - channel.send(new DocumentMessage(1, "test")); + channel.send(new GenericMessage(1, "test")); handler.getLatch().await(50, TimeUnit.MILLISECONDS); assertEquals("test", handler.getMessageString()); } @@ -53,7 +53,7 @@ public class EndpointParserTests { MessageChannel channel = (MessageChannel) context.getBean("testChannel"); TestBean bean = (TestBean) context.getBean("testBean"); assertNull(bean.getMessage()); - channel.send(new DocumentMessage(1, "test")); + channel.send(new GenericMessage(1, "test")); bean.getLatch().await(50, TimeUnit.MILLISECONDS); assertEquals("test", bean.getMessage()); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java index 78bc05c5cd..4f4576e01f 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java @@ -20,15 +20,14 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import org.junit.Test; - import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.bus.Subscription; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher @@ -40,8 +39,8 @@ public class GenericMessageEndpointTests { MessageChannel channel = new PointToPointChannel(); MessageChannel replyChannel = new PointToPointChannel(); MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { - return new DocumentMessage("123", "hello " + message.getPayload()); + public Message handle(Message message) { + return new StringMessage("123", "hello " + message.getPayload()); } }; GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); @@ -60,9 +59,9 @@ public class GenericMessageEndpointTests { subscription.setPolicy(policy); bus.activateSubscription(subscription); bus.start(); - DocumentMessage testMessage = new DocumentMessage(1, "test"); + StringMessage testMessage = new StringMessage(1, "test"); channel.send(testMessage); - Message reply = replyChannel.receive(10); + Message reply = replyChannel.receive(10); assertNotNull(reply); assertEquals("hello test", reply.getPayload()); } @@ -73,7 +72,7 @@ public class GenericMessageEndpointTests { final MessageChannel replyChannel = new PointToPointChannel(); MessageHandler handler = new MessageHandler() { public Message handle(Message message) { - return new DocumentMessage("123", "hello " + message.getPayload()); + return new StringMessage("123", "hello " + message.getPayload()); } }; GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); @@ -91,7 +90,7 @@ public class GenericMessageEndpointTests { subscription.setPolicy(policy); bus.activateSubscription(subscription); bus.start(); - DocumentMessage testMessage = new DocumentMessage(1, "test"); + StringMessage testMessage = new StringMessage(1, "test"); testMessage.getHeader().setReplyChannelName("replyChannel"); channel.send(testMessage); Message reply = replyChannel.receive(10); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/OutboundChannelAdapterTests.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/OutboundChannelAdapterTests.java index a274af7a68..4d7679f904 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/OutboundChannelAdapterTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/OutboundChannelAdapterTests.java @@ -22,7 +22,7 @@ import org.junit.Test; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter; -import org.springframework.integration.message.DocumentMessage; +import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher @@ -36,7 +36,7 @@ public class OutboundChannelAdapterTests { adapter.setObject(new TestSink()); adapter.setMethod("validMethod"); adapter.afterPropertiesSet(); - boolean result = adapter.send(new DocumentMessage(1, "test")); + boolean result = adapter.send(new StringMessage(1, "test")); assertTrue(result); } @@ -47,7 +47,7 @@ public class OutboundChannelAdapterTests { adapter.setObject(new TestSink()); adapter.setMethod("invalidMethodWithNoArgs"); adapter.afterPropertiesSet(); - adapter.send(new DocumentMessage(1, "test")); + adapter.send(new StringMessage(1, "test")); } @Test @@ -57,7 +57,7 @@ public class OutboundChannelAdapterTests { adapter.setObject(new TestSink()); adapter.setMethod("validMethodWithIgnoredReturnValue"); adapter.afterPropertiesSet(); - boolean result = adapter.send(new DocumentMessage(1, "test")); + boolean result = adapter.send(new StringMessage(1, "test")); assertTrue(result); } @@ -68,7 +68,7 @@ public class OutboundChannelAdapterTests { adapter.setObject(new TestSink()); adapter.setMethod("noSuchMethod"); adapter.afterPropertiesSet(); - adapter.send(new DocumentMessage(1, "test")); + adapter.send(new StringMessage(1, "test")); } } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/AnnotationAwareMessageEndpointTests.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/AnnotationAwareMessageEndpointTests.java index 2bc5927b3e..845d3835d8 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/AnnotationAwareMessageEndpointTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/AnnotationAwareMessageEndpointTests.java @@ -20,7 +20,7 @@ import org.junit.Test; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.DocumentMessage; +import org.springframework.integration.message.GenericMessage; /** * @author Mark Fisher @@ -37,7 +37,7 @@ public class AnnotationAwareMessageEndpointTests { MessageChannel channel = messageBus.getChannel("testChannel"); messageBus.start(); endpoint.afterPropertiesSet(); - channel.send(new DocumentMessage(1, "world"), 10); + channel.send(new GenericMessage(1, "world"), 10); } @MessageEndpoint(input="testChannel", pollPeriod=10) diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/EndpointAnnotationPostProcessorTests.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/EndpointAnnotationPostProcessorTests.java new file mode 100644 index 0000000000..80af449266 --- /dev/null +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/EndpointAnnotationPostProcessorTests.java @@ -0,0 +1,44 @@ +/* + * Copyright 2002-2007 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.endpoint.annotation; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.GenericMessage; + +/** + * @author Mark Fisher + */ +public class EndpointAnnotationPostProcessorTests { + + @Test + public void testSimpleHandler() throws InterruptedException { + AbstractApplicationContext context = new ClassPathXmlApplicationContext("simpleAnnotatedEndpointTests.xml", this.getClass()); + context.start(); + MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel"); + MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel"); + inputChannel.send(new GenericMessage(1, "world")); + Message message = outputChannel.receive(); + assertEquals("hello world", message.getPayload()); + } + +} diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java new file mode 100644 index 0000000000..a629218f2e --- /dev/null +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java @@ -0,0 +1,35 @@ +/* + * Copyright 2002-2007 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.endpoint.annotation; + +/** + * @author Mark Fisher + */ +@MessageEndpoint(defaultOutput="outputChannel") +public class InboundChannelAdapterTestBean { + + @Polled(period=100) + public String getName() { + return "world"; + } + + @Handler + public String sayHello(String name) { + return "hello " + name; + } + +} diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/OutboundChannelAdapterTestBean.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/OutboundChannelAdapterTestBean.java new file mode 100644 index 0000000000..45e69819b0 --- /dev/null +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/OutboundChannelAdapterTestBean.java @@ -0,0 +1,35 @@ +/* + * Copyright 2002-2007 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.endpoint.annotation; + +/** + * @author Mark Fisher + */ +@MessageEndpoint(input="inputChannel") +public class OutboundChannelAdapterTestBean { + + @Handler + public String sayHello(String name) { + return "hello " + name; + } + + @DefaultOutput + public void sendGreeting(String greeting) { + System.out.println(greeting); + } + +} diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java new file mode 100644 index 0000000000..5cd3feb4a2 --- /dev/null +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java @@ -0,0 +1,30 @@ +/* + * Copyright 2002-2007 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.endpoint.annotation; + +/** + * @author Mark Fisher + */ +@MessageEndpoint(input="inputChannel", defaultOutput="outputChannel", pollPeriod=10) +public class SimpleAnnotatedEndpoint { + + @Handler + public String sayHello(String name) { + return "hello " + name; + } + +} diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml new file mode 100644 index 0000000000..ff62e13814 --- /dev/null +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml @@ -0,0 +1,22 @@ + + + + + + + + + + + + + + + + diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml new file mode 100644 index 0000000000..4f54233273 --- /dev/null +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml @@ -0,0 +1,19 @@ + + + + + + + + + + + + + + + + diff --git a/spring-eai-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java b/spring-eai-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java index a4ead3cf05..e1d80a4903 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java @@ -20,8 +20,8 @@ import static org.junit.Assert.assertEquals; import org.junit.Test; -import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher @@ -35,7 +35,7 @@ public class MessageHandlerChainTests { chain.add(new TestHandler("b")); chain.add(new TestHandler("c")); chain.add(new TestHandler("d")); - Message result = chain.handle(new DocumentMessage(1, "!")); + Message result = chain.handle(new StringMessage(1, "!")); assertEquals("!abcd", result.getPayload()); } @@ -49,7 +49,7 @@ public class MessageHandlerChainTests { chain.add(new TestHandler("a")); chain.add(handler4); chain.add(new TestHandler("b")); - Message result = chain.handle(new DocumentMessage(1, "!")); + Message result = chain.handle(new StringMessage(1, "!")); assertEquals("234!a*234b", result.getPayload()); } @@ -63,7 +63,7 @@ public class MessageHandlerChainTests { } public Message handle(Message message) { - return new DocumentMessage(1, message.getPayload() + text); + return new StringMessage(1, message.getPayload() + text); } } @@ -78,8 +78,8 @@ public class MessageHandlerChainTests { } public Message handle(Message message, MessageHandler target) { - message = target.handle(new DocumentMessage(1, text + message.getPayload())); - return new DocumentMessage(1, message.getPayload() + text); + message = target.handle(new StringMessage(1, text + message.getPayload())); + return new StringMessage(1, message.getPayload() + text); } }