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

The @Splitter annotation no longer has a "channel" attribute. Instead it uses the @MessageEndpoint's "defaultOutput" attribute (INT-189).

Browse Source
This commit is contained in:
Mark Fisher
2008-04-11 19:13:45 +00:00
parent b7517caaba
commit 88ac6af364
14 changed files with 79 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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.
@@ -24,8 +24,7 @@ import java.lang.annotation.Target;
/**
* Indicates that a method is capable of splitting a single message or message
* payload to produce multiple messages which are then sent to the channel whose
* name is provided with the 'channel' attribute.
* payload to produce multiple messages or payloads.
*
* @author Mark Fisher
*/
@@ -35,6 +34,4 @@ import java.lang.annotation.Target;
@Handler
public @interface Splitter {
String channel();
}

6
spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorParser.java
View File

@@ -16,6 +16,9 @@
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
@@ -27,8 +30,6 @@ import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.AggregatorAdapter;
import org.springframework.integration.router.CompletionStrategyAdapter;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;
/**
* Parser for the <em>aggregator</em> element of the integration namespace.
@@ -80,6 +81,7 @@ public class AggregatorParser implements BeanDefinitionParser {
public static final String COMPLETION_STRATEGY_ELEMENT = "completion-strategy";
public BeanDefinition parse(Element element, ParserContext parserContext) {
return parseAggregatorElement(element, parserContext, true);
}

28
spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java
View File

@@ -26,6 +26,7 @@ import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
@@ -52,6 +53,7 @@ import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.handler.config.DefaultMessageHandlerCreator;
@@ -83,11 +85,13 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
private final MessageBus messageBus;
public MessageEndpointAnnotationPostProcessor(MessageBus messageBus) {
Assert.notNull(messageBus, "'messageBus' must not be null");
this.messageBus = messageBus;
}
public void setCustomHandlerCreators(Map<Class<? extends Annotation>, MessageHandlerCreator> customHandlerCreators) {
for (Map.Entry<Class<? extends Annotation>, MessageHandlerCreator> entry : customHandlerCreators.entrySet()) {
this.handlerCreators.put(entry.getKey(), entry.getValue());
@@ -114,10 +118,16 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
if (bean instanceof ChannelRegistryAware) {
((ChannelRegistryAware) bean).setChannelRegistry(this.messageBus);
}
MessageHandlerChain handlerChain = this.createHandlerChain(bean);
String defaultOutputChannelName = endpointAnnotation.defaultOutput();
MessageHandlerChain handlerChain = this.createHandlerChain(bean, defaultOutputChannelName);
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handlerChain);
this.configureInput(bean, beanName, endpointAnnotation, endpoint);
this.configureDefaultOutput(bean, beanName, endpointAnnotation, endpoint);
if (StringUtils.hasText(defaultOutputChannelName)) {
endpoint.setDefaultOutputChannelName(defaultOutputChannelName);
}
else {
this.configureDefaultOutput(bean, beanName, endpoint);
}
Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(beanClass, Concurrency.class);
if (concurrencyAnnotation != null) {
ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(),
@@ -173,16 +183,9 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
});
}
private void configureDefaultOutput(final Object bean, final String beanName, final MessageEndpoint annotation,
final DefaultMessageEndpoint endpoint) {
String channelName = annotation.defaultOutput();
if (StringUtils.hasText(channelName)) {
endpoint.setDefaultOutputChannelName(channelName);
return;
}
private void configureDefaultOutput(final Object bean, final String beanName, final DefaultMessageEndpoint endpoint) {
ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() {
boolean foundDefaultOutput = false;
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, DefaultOutput.class);
if (annotation != null) {
@@ -246,14 +249,15 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
}
@SuppressWarnings("unchecked")
private MessageHandlerChain createHandlerChain(final Object bean) {
private MessageHandlerChain createHandlerChain(final Object bean, final String defaultOutputChannelName) {
final List<MessageHandler> handlers = new ArrayList<MessageHandler>();
ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation[] annotations = AnnotationUtils.getAnnotations(method);
for (Annotation annotation : annotations) {
if (isHandlerAnnotation(annotation)) {
Map<String, ?> attributes = AnnotationUtils.getAnnotationAttributes(annotation);
Map<String, Object> attributes = AnnotationUtils.getAnnotationAttributes(annotation);
attributes.put(AbstractMessageHandlerAdapter.DEFAULT_OUTPUT_CHANNEL_NAME_KEY, defaultOutputChannelName);
MessageHandlerCreator handlerCreator = handlerCreators.get(annotation.annotationType());
if (handlerCreator == null) {
if (logger.isWarnEnabled()) {

6
spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -240,7 +240,7 @@
<xsd:attribute name="ref" type="xsd:string" use="required"/>
<xsd:attribute name="method" type="xsd:string" use="optional"/>
<xsd:attribute name="completion-strategy" type="xsd:string" use="optional"/>
<xsd:attribute name="completion-strategy-method" type="xsd:string" use="optional"/>
<xsd:attribute name="completion-strategy-method" type="xsd:string" use="optional"/>
<xsd:attribute name="default-reply-channel" type="xsd:string" use="optional"/>
<xsd:attribute name="discard-channel" type="xsd:string" use="optional"/>
<xsd:attribute name="send-timeout" type="xsd:long" use="optional"/>
@@ -252,12 +252,12 @@
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="completion-strategy">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a completion strategy.
Defines a completion strategy.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="ref" type="xsd:string" use="required"/>

3
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java
View File

@@ -39,6 +39,9 @@ import org.springframework.util.Assert;
*/
public abstract class AbstractMessageHandlerAdapter<T> implements MessageHandler, Ordered, InitializingBean {
public static final String DEFAULT_OUTPUT_CHANNEL_NAME_KEY = "defaultOutputChannelName";
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile T object;

1
spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java
View File

@@ -33,6 +33,7 @@ public class MessageHandlerChain implements MessageHandler {
private final List<MessageHandler> handlers = new CopyOnWriteArrayList<MessageHandler>();
/**
* Add a handler to the end of the chain.
*/

5
spring-integration-core/src/main/java/org/springframework/integration/router/AggregatorAdapter.java
View File

@@ -38,7 +38,7 @@ import org.springframework.util.ReflectionUtils;
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class AggregatorAdapter extends MessageListMethodAdapter implements Aggregator {
public class AggregatorAdapter extends MessageListMethodAdapter implements Aggregator {
public AggregatorAdapter(Object object, Method method) {
super(object, method);
@@ -48,6 +48,7 @@ public class AggregatorAdapter extends MessageListMethodAdapter implements Aggr
super(object, methodName);
}
public Message<?> aggregate(List<Message<?>> messages) {
Object returnedValue = this.executeMethod(messages);
if (returnedValue == null) {
@@ -59,6 +60,4 @@ public class AggregatorAdapter extends MessageListMethodAdapter implements Aggr
return new GenericMessage<Object>(returnedValue);
}
}

17
spring-integration-core/src/main/java/org/springframework/integration/router/CompletionStrategyAdapter.java
View File

@@ -23,31 +23,30 @@ import org.springframework.integration.ConfigurationException;
import org.springframework.integration.message.Message;
/**
* Aggregator adapter for methods annotated with
* {@link org.springframework.integration.annotation.CompletionStrategy @CompletionStrategy}
* and for '<code>aggregator</code>' elements that include a '<code>method</code>'
* attribute (e.g. &lt;aggregator ref="beanReference" method="methodName"/&gt;).
* Adapter for methods annotated with {@link org.springframework.integration.annotation.CompletionStrategy @CompletionStrategy}
* and for '<code>completion-strategy</code>' elements that include a '<code>method</code>'
* attribute (e.g. &lt;completion-strategy ref="beanReference" method="methodName"/&gt;).
*
* @author Marius Bogoevici
*/
public class CompletionStrategyAdapter extends MessageListMethodAdapter implements CompletionStrategy {
public CompletionStrategyAdapter(Object object, Method method) {
super(object, method);
assertMethodReturnsBoolean();
this.assertMethodReturnsBoolean();
}
public CompletionStrategyAdapter(Object object, String methodName) {
super(object, methodName);
assertMethodReturnsBoolean();
this.assertMethodReturnsBoolean();
}
private void assertMethodReturnsBoolean() {
if (!Boolean.class.equals(this.getMethod().getReturnType())
&& !boolean.class.equals(this.getMethod().getReturnType())) {
throw new ConfigurationException("Method " + getMethod().getName()
+ " does not return a boolean value");
throw new ConfigurationException("Method '" + getMethod().getName()
+ "' does not return a boolean value");
}
}

23
spring-integration-core/src/main/java/org/springframework/integration/router/MessageListMethodAdapter.java
View File

@@ -24,7 +24,6 @@ import java.util.List;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.handler.HandlerMethodInvoker;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
@@ -39,7 +38,8 @@ public abstract class MessageListMethodAdapter {
private final HandlerMethodInvoker<Object> invoker;
protected Method method;
protected volatile Method method;
public MessageListMethodAdapter(Object object, String methodName) {
Assert.notNull(object, "'object' must not be null");
@@ -57,26 +57,23 @@ public abstract class MessageListMethodAdapter {
Assert.notNull(method, "'method' must not be null");
if (method.getParameterTypes().length != 1 || !method.getParameterTypes()[0].equals(List.class)) {
throw new ConfigurationException(
"Method must accept exactly one parameter, and it must be a Collection.");
"Method must accept exactly one parameter, and it must be a List.");
}
this.method = method;
this.invoker = new HandlerMethodInvoker<Object>(object, this.method.getName());
}
private static boolean isActualTypeParametrizedMessage(Method method) {
private static boolean isActualTypeParameterizedMessage(Method method) {
return getCollectionActualType(method) instanceof ParameterizedType
&& Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) getCollectionActualType(method))
.getRawType());
&& Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) getCollectionActualType(method)).getRawType());
}
protected final Object executeMethod(List<Message<?>> messages) {
if (isMethodParameterParametrized(this.method) && isHavingActualTypeArguments(this.method)
&& (isActualTypeRawMessage(this.method) || isActualTypeParametrizedMessage(this.method))) {
if (isMethodParameterParameterized(this.method) && isHavingActualTypeArguments(this.method)
&& (isActualTypeRawMessage(this.method) || isActualTypeParameterizedMessage(this.method))) {
return this.invoker.invokeMethod(messages);
}
else {
return this.invoker.invokeMethod(extractPayloadsFromMessages(messages));
}
return this.invoker.invokeMethod(extractPayloadsFromMessages(messages));
}
private List<?> extractPayloadsFromMessages(List<Message<?>> messages) {
@@ -99,7 +96,7 @@ public abstract class MessageListMethodAdapter {
return ((ParameterizedType) method.getGenericParameterTypes()[0]).getActualTypeArguments().length == 1;
}
private static boolean isMethodParameterParametrized(Method method) {
private static boolean isMethodParameterParameterized(Method method) {
return method.getGenericParameterTypes().length == 1
&& method.getGenericParameterTypes()[0] instanceof ParameterizedType;
}
@@ -112,4 +109,4 @@ public abstract class MessageListMethodAdapter {
this.method = method;
}
}
}

16
spring-integration-core/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java
View File

@@ -39,12 +39,9 @@ import org.springframework.util.Assert;
*/
public class SplitterMessageHandlerAdapter<T> extends AbstractMessageHandlerAdapter<T> implements ChannelRegistryAware {
public static final String CHANNEL_KEY = "channel";
private final Method method;
private final Map<String, ?> attributes;
private final String outputChannelName;
private volatile ChannelRegistry channelRegistry;
@@ -54,12 +51,12 @@ public class SplitterMessageHandlerAdapter<T> extends AbstractMessageHandlerAdap
public SplitterMessageHandlerAdapter(T object, Method method, Map<String, ?> attributes) {
Assert.notNull(object, "'object' must not be null");
Assert.notNull(method, "'method' must not be null");
Assert.isTrue(attributes != null && attributes.get(CHANNEL_KEY) != null,
"the 'channel' attribute is required");
Assert.isTrue(attributes != null && attributes.get(DEFAULT_OUTPUT_CHANNEL_NAME_KEY) != null,
"The '" + DEFAULT_OUTPUT_CHANNEL_NAME_KEY + "' attribute is required.");
this.setObject(object);
this.setMethodName(method.getName());
this.method = method;
this.attributes = attributes;
this.outputChannelName = (String) attributes.get(DEFAULT_OUTPUT_CHANNEL_NAME_KEY);
}
public void setChannelRegistry(ChannelRegistry channelRegistry) {
@@ -77,7 +74,6 @@ public class SplitterMessageHandlerAdapter<T> extends AbstractMessageHandlerAdap
throw new ConfigurationException(
"Splitter method must accept exactly one parameter");
}
String channelName = (String) attributes.get(CHANNEL_KEY);
Object retval = null;
Class<?> type = method.getParameterTypes()[0];
if (type.equals(Message.class)) {
@@ -100,7 +96,7 @@ public class SplitterMessageHandlerAdapter<T> extends AbstractMessageHandlerAdap
Message<?> splitMessage = (item instanceof Message<?>) ? (Message<?>) item :
this.createReplyMessage(item, originalMessageHeader);
this.prepareMessage(splitMessage, message.getId(), ++sequenceNumber, sequenceSize);
this.sendMessage(splitMessage, channelName);
this.sendMessage(splitMessage, this.outputChannelName);
}
}
else if (retval.getClass().isArray()) {
@@ -111,7 +107,7 @@ public class SplitterMessageHandlerAdapter<T> extends AbstractMessageHandlerAdap
Message<?> splitMessage = (item instanceof Message<?>) ? (Message<?>) item :
this.createReplyMessage(item, originalMessageHeader);
this.prepareMessage(splitMessage, message.getId(), ++sequenceNumber, sequenceSize);
this.sendMessage(splitMessage, channelName);
this.sendMessage(splitMessage, this.outputChannelName);
}
}
else {

22
spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java
View File

@@ -1,15 +1,31 @@
/*
* Copyright 2002-2008 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.util.List;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.ConcurrentHandler;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.router.AggregatingMessageHandler;
@@ -34,8 +50,8 @@ public class CompletionStrategyAnnotationTests {
@Test(expected=BeanCreationException.class)
public void testInvalidAnnotation() {
ApplicationContext context = new ClassPathXmlApplicationContext(
new String[] { "classpath:/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml" });
new ClassPathXmlApplicationContext(new String[] {
"classpath:/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml" });
}
@SuppressWarnings("unchecked")

2
spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java
View File

@@ -127,7 +127,7 @@ public class CorrelationIdTests {
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel("testChannel", testChannel);
Map<String, String> attributes = new HashMap<String, String>();
attributes.put("channel", "testChannel");
attributes.put(AbstractMessageHandlerAdapter.DEFAULT_OUTPUT_CHANNEL_NAME_KEY, "testChannel");
SplitterMessageHandlerAdapter<TestBean> splitter = new SplitterMessageHandlerAdapter<TestBean>(
new TestBean(), TestBean.class.getMethod("split", String.class), attributes);
splitter.setChannelRegistry(channelRegistry);

3
spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java
View File

@@ -32,6 +32,7 @@ import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.DefaultChannelRegistry;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
@@ -51,7 +52,7 @@ public class SplitterMessageHandlerAdapterTests {
public SplitterMessageHandlerAdapterTests() {
this.channelRegistry.registerChannel("testChannel", testChannel);
this.attribs.put(SplitterMessageHandlerAdapter.CHANNEL_KEY, "testChannel");
this.attribs.put(AbstractMessageHandlerAdapter.DEFAULT_OUTPUT_CHANNEL_NAME_KEY, "testChannel");
}

4
spring-integration-samples/src/main/java/org/springframework/integration/samples/cafe/OrderSplitter.java
View File

@@ -24,10 +24,10 @@ import org.springframework.integration.annotation.Splitter;
/**
* @author Mark Fisher
*/
@MessageEndpoint(input="orders")
@MessageEndpoint(input="orders", defaultOutput="drinks")
public class OrderSplitter {
@Splitter(channel="drinks")
@Splitter
public List<Drink> split(DrinkOrder order) {
return order.getDrinks();
}