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

Moved input/output channel configuration to Method-level annotations. Also, the @Poller annotation is now expected at Method-level instead of Class-level. The @MessageEndpoint is now strictly a stereotype. Removed the @MessageTarget and @Pollable annotations. The @ChannelAdapter annotation post-processor now handles both inbound and outbound channel adapters based on the Method signature.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-03 17:31:51 +00:00
parent 82b92ecf05
commit 05fc8263f5
44 changed files with 961 additions and 936 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorMessageHandlerCreator.java
View File

@@ -1,128 +0,0 @@
/*
* 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.aggregator;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.aop.support.AopUtils;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.config.AbstractMessageHandlerCreator;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* Creates an {@link AggregatorAdapter AggregatorAdapter} for methods that aggregate messages.
*
* @author Marius Bogoevici
*/
public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreator {
private static final String DISCARD_CHANNEL = "discardChannel";
private static final String SEND_TIMEOUT = "sendTimeout";
private static final String SEND_PARTIAL_RESULTS_ON_TIMEOUT = "sendPartialResultsOnTimeout";
private static final String REAPER_INTERVAL = "reaperInterval";
private static final String TIMEOUT = "timeout";
private static final String TRACKED_CORRELATION_ID_CAPACITY = "trackedCorrelationIdCapacity";
private final ChannelRegistry channelRegistry;
public AggregatorMessageHandlerCreator(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
public MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> attributes) {
AggregatingMessageHandler messageHandler = new AggregatingMessageHandler(new AggregatorAdapter(object, method));
this.configureDefaultReplyChannel(messageHandler, object);
String discardChannelName = this.getAttribute(attributes, DISCARD_CHANNEL, String.class);
if (discardChannelName != null) {
messageHandler.setDiscardChannel(this.channelRegistry.lookupChannel(discardChannelName));
}
Long sendTimeout = this.getAttribute(attributes, SEND_TIMEOUT, Long.class);
if (sendTimeout != null) {
messageHandler.setSendTimeout(sendTimeout);
}
Boolean sendPartialResultOnTimeout = this.getAttribute(
attributes, SEND_PARTIAL_RESULTS_ON_TIMEOUT, Boolean.class);
if (sendPartialResultOnTimeout != null) {
messageHandler.setSendPartialResultOnTimeout(sendPartialResultOnTimeout);
}
Long reaperInterval = this.getAttribute(attributes, REAPER_INTERVAL, Long.class);
if (reaperInterval != null) {
messageHandler.setReaperInterval(reaperInterval);
}
Long timeout = this.getAttribute(attributes, TIMEOUT, Long.class);
if (timeout != null) {
messageHandler.setTimeout(timeout);
}
Integer trackedCorrelationIdCapacity = this.getAttribute(
attributes, TRACKED_CORRELATION_ID_CAPACITY, Integer.class);
if (trackedCorrelationIdCapacity != null) {
messageHandler.setTrackedCorrelationIdCapacity(trackedCorrelationIdCapacity);
}
this.configureCompletionStrategy(object, messageHandler);
return messageHandler;
}
private void configureDefaultReplyChannel(AggregatingMessageHandler handler, Object originalObject) {
MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(
AopUtils.getTargetClass(originalObject), MessageEndpoint.class);
if (endpointAnnotation != null) {
String outputChannelName = endpointAnnotation.output();
if (StringUtils.hasText(outputChannelName)) {
handler.setOutputChannel(this.channelRegistry.lookupChannel(outputChannelName));
}
}
}
private void configureCompletionStrategy(final Object object, final AggregatingMessageHandler handler) {
ReflectionUtils.doWithMethods(object.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class);
if (annotation != null) {
handler.setCompletionStrategy(new CompletionStrategyAdapter(object, method));
}
}
});
}
@SuppressWarnings("unchecked")
private <T> T getAttribute(Map<String, ?> attributes, String name, Class<T> expectedType) {
Object value = attributes.get(name);
if (value == null || !expectedType.isAssignableFrom(value.getClass())) {
return null;
}
if (value instanceof String && !StringUtils.hasText((String) value)) {
return null;
}
return (T) value;
}
}

10
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java
View File

@@ -39,6 +39,16 @@ import org.springframework.integration.aggregator.AggregatingMessageHandler;
@Handler
public @interface Aggregator {
/**
* channel name for receiving messages to be aggregated
*/
String inputChannel() default "";
/**
* channel name for sending aggregated result messages
*/
String outputChannel() default "";
/**
* channel name for sending discarded messages (due to a timeout)
*/

13
org.springframework.integration/src/main/java/org/springframework/integration/annotation/ChannelAdapter.java
View File

@@ -26,17 +26,26 @@ import java.lang.annotation.Target;
import org.springframework.stereotype.Component;
/**
* Indicates that a class is capable of serving as a message channel.
* Indicates that a method is capable of serving as a message channel.
* If the method accepts no arguments but does define a non-void return
* type, an inbound Channel Adapter will be created. If the method does
* accept an argument and has a void return, an outbound Channel Adapter
* will be created. If the method does not conform to either contract,
* an Exception will be thrown.
*
* @author Mark Fisher
*/
@Target(ElementType.TYPE)
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
@Component
public @interface ChannelAdapter {
/**
* The name of the channel being adapted. If the channel
* name is not resolvable, a new channel will be created.
*/
String value();
}

4
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Handler.java
View File

@@ -45,4 +45,8 @@ import java.lang.annotation.Target;
@Documented
public @interface Handler {
String inputChannel() default "";
String outputChannel() default "";
}

13
org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java
View File

@@ -26,7 +26,8 @@ import java.lang.annotation.Target;
import org.springframework.stereotype.Component;
/**
* Indicates that a class is capable of serving as a message endpoint.
* Stereotype annotation indicating that a class is capable of serving as a
* Message Endpoint.
*
* @author Mark Fisher
*/
@@ -37,8 +38,12 @@ import org.springframework.stereotype.Component;
@Component
public @interface MessageEndpoint {
String input() default "";
String output() default "";
/**
* The value may indicate a suggestion for a logical component name,
* to be turned into a Spring bean in case of an autodetected component.
*
* @return the suggested component name, if any
*/
String value() default "";
}

42
org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageTarget.java
View File

@@ -1,42 +0,0 @@
/*
* 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.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 org.springframework.integration.message.Message;
/**
* Indicates that a method is capable of consuming messages. The method must
* accept a single parameter that is either a {@link Message} or an Object of
* the expected message payload type. The method itself should define a void
* return, and the enclosing class may also be annotated with
* {@link MessageEndpoint @MessageEndpoint}.
*
* @author Mark Fisher
*/
@java.lang.annotation.Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface MessageTarget {
}

6
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java
View File

@@ -27,13 +27,13 @@ import java.util.concurrent.TimeUnit;
import org.springframework.integration.scheduling.PollingSchedule;
/**
* Annotation that can be specified at class-level alongside a
* {@link MessageEndpoint @MessageEndpoint} annotation in order to provide the
* Annotation that can be specified at method-level alongside a Message Endpoint
* annotation (e.g. @Splitter, @ChannelAdapter, etc.) in order to provide the
* polling metadata and scheduling information for that endpoint.
*
* @author Mark Fisher
*/
@Target(ElementType.TYPE)
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented

4
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Transformer.java
View File

@@ -36,4 +36,8 @@ import java.lang.annotation.Target;
@Handler
public @interface Transformer {
String inputChannel() default "";
String outputChannel() default "";
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java
View File

@@ -16,9 +16,9 @@
package org.springframework.integration.config;
import org.springframework.integration.endpoint.DefaultServiceInvoker;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
import org.springframework.integration.message.MessageMappingMethodInvoker;
/**
* Parser for the &lt;service-activator&gt; element.
@@ -34,7 +34,7 @@ public class ServiceActivatorParser extends AbstractEndpointParser {
@Override
protected Class<?> getMethodInvokingAdapterClass() {
return MessageMappingMethodInvoker.class;
return DefaultServiceInvoker.class;
}
}

102
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractAnnotationMethodPostProcessor.java
View File

@@ -1,102 +0,0 @@
/*
* 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.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.DelegatingIntroductionInterceptor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.bus.MessageBus;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
/**
* Base class for post-processing annotated methods.
*
* @author Mark Fisher
*/
public abstract class AbstractAnnotationMethodPostProcessor<T> implements AnnotationMethodPostProcessor {
protected final Log logger = LogFactory.getLog(this.getClass());
private final Class<? extends Annotation> annotationType;
private final MessageBus messageBus;
private final ClassLoader beanClassLoader;
public AbstractAnnotationMethodPostProcessor(Class<? extends Annotation> annotationType, MessageBus messageBus, ClassLoader beanClassLoader) {
Assert.notNull(annotationType, "Annotation type must not be null.");
Assert.notNull(messageBus, "MessageBus must not be null.");
this.annotationType = annotationType;
this.messageBus = messageBus;
this.beanClassLoader = (beanClassLoader != null) ? beanClassLoader : ClassUtils.getDefaultClassLoader();
}
protected MessageBus getMessageBus() {
return this.messageBus;
}
public Object postProcess(final Object bean, final String beanName, final Class<?> originalBeanClass) {
final List<T> results = new ArrayList<T>();
ReflectionUtils.doWithMethods(originalBeanClass, new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = getAnnotation(method);
if (annotation != null) {
T result = processMethod(bean, method, annotation);
if (result != null) {
results.add(result);
}
}
}
});
T postProcessedBean = (results.size() > 0) ? this.processResults(results) : null;
if (postProcessedBean == null) {
return bean;
}
ProxyFactory proxyFactory = new ProxyFactory(bean);
proxyFactory.addAdvice(new DelegatingIntroductionInterceptor(postProcessedBean));
return proxyFactory.getProxy(this.beanClassLoader);
}
private Annotation getAnnotation(Method method) {
Annotation[] annotations = AnnotationUtils.getAnnotations(method);
for (Annotation annotation : annotations) {
if (annotation.annotationType().equals(this.annotationType)
|| annotation.annotationType().isAnnotationPresent(this.annotationType)) {
return annotation;
}
}
return null;
}
protected abstract T processMethod(Object bean, Method method, Annotation annotation);
protected abstract T processResults(List<T> results);
}

131
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,131 @@
/*
* 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.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
/**
* Base class for Method-level annotation post-processors.
*
* @author Mark Fisher
*/
public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation> implements MethodAnnotationPostProcessor<T> {
private static final String INPUT_CHANNEL_ATTRIBUTE = "inputChannel";
private static final String OUTPUT_CHANNEL_ATTRIBUTE = "outputChannel";
private final MessageBus messageBus;
public AbstractMethodAnnotationPostProcessor(MessageBus messageBus) {
Assert.notNull(messageBus, "MessageBus must not be null");
this.messageBus = messageBus;
}
protected ChannelRegistry getChannelRegistry() {
return this.messageBus;
}
public Object postProcess(Object bean, String beanName, Method method, T annotation) {
Object adapter = this.createMethodInvokingAdapter(bean, method, annotation);
if (adapter != null && this.shouldCreateEndpoint(annotation)) {
AbstractEndpoint endpoint = this.createEndpoint(adapter);
if (endpoint != null) {
Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
this.configureEndpoint(endpoint, annotation, pollerAnnotation);
if (endpoint.getName() == null) {
endpoint.setBeanName(this.generateEndpointName(beanName, annotation));
}
return endpoint;
}
}
return adapter;
}
protected boolean shouldCreateEndpoint(T annotation) {
return (StringUtils.hasText((String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE)));
}
protected void configureEndpoint(AbstractEndpoint endpoint, T annotation, Poller pollerAnnotation) {
String inputChannelName = (String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(inputChannelName)) {
MessageChannel inputChannel = this.messageBus.lookupChannel(inputChannelName);
if (inputChannel == null) {
throw new ConfigurationException("unable to resolve inputChannel '" + inputChannelName + "'");
}
if (pollerAnnotation != null) {
if (inputChannel instanceof PollableChannel) {
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
PollingDispatcher poller = new PollingDispatcher((PollableChannel) inputChannel, schedule);
poller.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
endpoint.setSource(poller);
}
else {
throw new ConfigurationException("The @Poller annotation should only be provided for a PollableSource");
}
}
else {
endpoint.setSource(inputChannel);
}
String outputChannelName = (String) AnnotationUtils.getValue(annotation, OUTPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(outputChannelName)) {
MessageChannel outputChannel = this.messageBus.lookupChannel(outputChannelName);
if (outputChannel == null) {
throw new ConfigurationException("unable to resolve outputChannel '" + outputChannelName + "'");
}
endpoint.setTarget(outputChannel);
}
}
}
private String generateEndpointName(String beanName, T annotation) {
String endpointName = beanName + "." + ClassUtils.getShortNameAsProperty(annotation.annotationType());
String id = endpointName;
int counter = 0;
while (this.messageBus.lookupEndpoint(id) != null) {
id = endpointName + "#" + counter;
counter++;
}
return id;
}
protected abstract Object createMethodInvokingAdapter(Object bean, Method method, T annotation);
protected abstract AbstractEndpoint createEndpoint(Object adapter);
}

88
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,88 @@
/*
* 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.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.AggregatorAdapter;
import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.DefaultEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* Post-processor for the {@link Aggregator @Aggregator} annotation.
*
* @author Mark Fisher
*/
public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Aggregator> {
public AggregatorAnnotationPostProcessor(MessageBus messageBus) {
super(messageBus);
}
@Override
protected Object createMethodInvokingAdapter(Object bean, Method method, Aggregator annotation) {
Aggregator aggregatorAnnotation = (Aggregator) annotation;
AggregatingMessageHandler messageHandler = new AggregatingMessageHandler(new AggregatorAdapter(bean, method));
String outputChannelName = aggregatorAnnotation.outputChannel();
if (StringUtils.hasText(outputChannelName)) {
messageHandler.setOutputChannel(this.getChannelRegistry().lookupChannel(outputChannelName));
}
String discardChannelName = aggregatorAnnotation.discardChannel();
if (StringUtils.hasText(discardChannelName)) {
messageHandler.setDiscardChannel(this.getChannelRegistry().lookupChannel(discardChannelName));
}
messageHandler.setSendTimeout(aggregatorAnnotation.sendTimeout());
messageHandler.setSendPartialResultOnTimeout(aggregatorAnnotation.sendPartialResultsOnTimeout());
messageHandler.setReaperInterval(aggregatorAnnotation.reaperInterval());
messageHandler.setTimeout(aggregatorAnnotation.timeout());
messageHandler.setTrackedCorrelationIdCapacity(aggregatorAnnotation.trackedCorrelationIdCapacity());
this.configureCompletionStrategy(bean, messageHandler);
messageHandler.afterPropertiesSet();
return messageHandler;
}
@Override
protected AbstractEndpoint createEndpoint(Object adapter) {
if (adapter instanceof MessageHandler) {
return new DefaultEndpoint<MessageHandler>((MessageHandler) adapter);
}
return null;
}
private void configureCompletionStrategy(final Object object, final AggregatingMessageHandler handler) {
ReflectionUtils.doWithMethods(object.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class);
if (annotation != null) {
handler.setCompletionStrategy(new CompletionStrategyAdapter(object, method));
}
}
});
}
}

145
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,145 @@
/*
* 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.annotation;
import java.lang.reflect.Method;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.ChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.InboundChannelAdapter;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.OutboundChannelAdapter;
import org.springframework.integration.handler.MethodInvokingTarget;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.util.Assert;
/**
* Post-processor for methods annotated with {@link ChannelAdapter @ChannelAdapter}.
*
* @author Mark Fisher
*/
public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPostProcessor<ChannelAdapter> {
private final MessageBus messageBus;
public ChannelAdapterAnnotationPostProcessor(MessageBus messageBus) {
Assert.notNull(messageBus, "MessageBus must not be null");
this.messageBus = messageBus;
}
public Object postProcess(Object bean, String beanName, Method method, ChannelAdapter annotation) {
MessageEndpoint endpoint = null;
String channelName = annotation.value();
MessageChannel channel = this.messageBus.lookupChannel(channelName);
if (channel == null) {
DirectChannel directChannel = new DirectChannel();
directChannel.setBeanName(channelName);
this.messageBus.registerChannel(directChannel);
channel = directChannel;
}
Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
if (method.getParameterTypes().length == 0 && hasReturnValue(method)) {
MethodInvokingSource source = new MethodInvokingSource();
source.setObject(bean);
source.setMethod(method);
endpoint = this.createInboundChannelAdapter(source, channel, pollerAnnotation);
}
else if (method.getParameterTypes().length > 0 && !hasReturnValue(method)) {
MethodInvokingTarget target = new MethodInvokingTarget(bean, method);
endpoint = this.createOutboundChannelAdapter(target, channel, pollerAnnotation);
}
else {
throw new ConfigurationException("The @ChannelAdapter can only be applied to methods that accept no arguments but have"
+ " a return value (inbound) or methods that have no return value but do accept arguments (outbound)");
}
if (endpoint != null) {
this.messageBus.registerEndpoint(endpoint);
}
return bean;
}
private InboundChannelAdapter createInboundChannelAdapter(MethodInvokingSource source, MessageChannel channel, Poller pollerAnnotation) {
if (pollerAnnotation == null) {
throw new ConfigurationException("The @Poller annotation is required (at method-level) "
+ "when using the @ChannelAdapter annotation with a no-arg method.");
}
PollingDispatcher poller = this.createPoller(source, pollerAnnotation);
InboundChannelAdapter adapter = new InboundChannelAdapter();
adapter.setSource(poller);
adapter.setTarget(channel);
adapter.setBeanName(this.generateUniqueName(channel.getName() + ".inboundAdapter"));
return adapter;
}
private OutboundChannelAdapter createOutboundChannelAdapter(MethodInvokingTarget target, MessageChannel channel, Poller pollerAnnotation) {
OutboundChannelAdapter adapter = new OutboundChannelAdapter();
adapter.setTarget(target);
if (channel instanceof PollableChannel) {
PollingDispatcher poller = (pollerAnnotation != null)
? this.createPoller((PollableChannel) channel, pollerAnnotation)
: new PollingDispatcher((PollableSource<?>) channel, new PollingSchedule(0));
adapter.setSource(poller);
}
else {
adapter.setSource(channel);
}
adapter.setBeanName(this.generateUniqueName(channel.getName() + ".outboundAdapter"));
return adapter;
}
private PollingDispatcher createPoller(PollableSource<?> source, Poller pollerAnnotation) {
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
PollingDispatcher poller = new PollingDispatcher((PollableSource<?>) source, schedule);
int maxMessagesPerPoll = pollerAnnotation.maxMessagesPerPoll();
if (maxMessagesPerPoll == -1) {
// the default is 1 since a MethodInvokingSource might return a non-null value
// every time it is invoked, thus producing an infinite number of messages per poll
maxMessagesPerPoll = 1;
}
poller.setMaxMessagesPerPoll(maxMessagesPerPoll);
return poller;
}
private boolean hasReturnValue(Method method) {
return !method.getReturnType().equals(void.class);
}
private String generateUniqueName(String name) {
int counter = 0;
String id = name;
while (this.messageBus.lookupEndpoint(id) != null) {
id = name + "#" + counter;
counter++;
}
return id;
}
}

134
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java
View File

@@ -1,134 +0,0 @@
/*
* 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.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.factory.InitializingBean;
import org.springframework.core.OrderComparator;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.Order;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.aggregator.AggregatorMessageHandlerCreator;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.DefaultEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.handler.config.DefaultMessageHandlerCreator;
import org.springframework.integration.handler.config.MessageHandlerCreator;
import org.springframework.integration.router.RouterMessageHandlerCreator;
import org.springframework.integration.splitter.SplitterMessageHandlerCreator;
import org.springframework.integration.transformer.config.TransformerMessageHandlerCreator;
/**
* Post-processor for the {@link Handler @Handler} annotation.
*
* @author Mark Fisher
*/
public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor<MessageHandler> {
private final Map<Class<? extends Annotation>, MessageHandlerCreator> handlerCreators =
new ConcurrentHashMap<Class<? extends Annotation>, MessageHandlerCreator>();
private final MessageHandlerCreator defaultHandlerCreator = new DefaultMessageHandlerCreator();
public HandlerAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) {
super(Handler.class, messageBus, beanClassLoader);
this.handlerCreators.put(Router.class, new RouterMessageHandlerCreator());
this.handlerCreators.put(Splitter.class, new SplitterMessageHandlerCreator());
this.handlerCreators.put(Aggregator.class, new AggregatorMessageHandlerCreator(messageBus));
this.handlerCreators.put(Transformer.class, new TransformerMessageHandlerCreator());
}
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());
}
}
protected MessageHandler processMethod(Object bean, Method method, Annotation annotation) {
MessageHandlerCreator handlerCreator = this.handlerCreators.get(annotation.annotationType());
if (handlerCreator == null) {
handlerCreator = this.defaultHandlerCreator;
if (logger.isDebugEnabled()) {
logger.debug("No handler creator has been registered for handler annotation '"
+ annotation.annotationType() + "', using DefaultMessageHandlerCreator.");
}
}
Map<String, Object> attributes = AnnotationUtils.getAnnotationAttributes(annotation);
Order order = AnnotationUtils.findAnnotation(method, Order.class);
if (order != null) {
attributes.put("order", order.value());
}
MessageHandler handler = handlerCreator.createHandler(bean, method, attributes);
if (handler != null) {
if (handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) handler).setChannelRegistry(this.getMessageBus());
}
if (handler instanceof InitializingBean) {
try {
((InitializingBean) handler).afterPropertiesSet();
}
catch (Exception e) {
throw new ConfigurationException("failed to initialize handler", e);
}
}
}
return handler;
}
@SuppressWarnings("unchecked")
protected MessageHandler processResults(List<MessageHandler> results) {
MessageHandlerChain handlerChain = new MessageHandlerChain();
for (MessageHandler handler : results) {
handlerChain.add(handler);
}
if (handlerChain.getHandlers().size() == 0) {
return null;
}
if (handlerChain.getHandlers().size() == 1) {
return handlerChain.getHandlers().get(0);
}
List<MessageHandler> handlers = new ArrayList<MessageHandler>(handlerChain.getHandlers());
Collections.sort(handlers, new OrderComparator());
handlerChain.setHandlers(handlers);
return handlerChain;
}
public AbstractEndpoint createEndpoint(Object bean) {
if (bean instanceof MessageHandler) {
return new DefaultEndpoint<MessageHandler>((MessageHandler) bean);
}
return null;
}
}

136
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java
View File

@@ -17,40 +17,40 @@
package org.springframework.integration.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.aop.support.DelegatingIntroductionInterceptor;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.ChannelAdapter;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
/**
* A {@link BeanPostProcessor} implementation that processes method-level
* messaging annotations such as @Handler, @MessageSource, and @MessageTarget.
* It also generates endpoints for classes annotated with the class-level
* {@link MessageEndpoint @MessageEndpoint} annotation.
* messaging annotations such as @Transformer, @Splitter, and @Router.
*
* @author Mark Fisher
* @author Marius Bogoevici
@@ -59,10 +59,10 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
private final MessageBus messageBus;
private volatile ClassLoader beanClassLoader;
private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
private final Map<Class<?>, AnnotationMethodPostProcessor> postProcessors =
new HashMap<Class<?>, AnnotationMethodPostProcessor>();
private final Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> postProcessors =
new HashMap<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>>();
public MessagingAnnotationPostProcessor(MessageBus messageBus) {
@@ -76,77 +76,69 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
}
public void afterPropertiesSet() {
this.postProcessors.put(MessageHandler.class, new HandlerAnnotationPostProcessor(this.messageBus, this.beanClassLoader));
this.postProcessors.put(MessageSource.class, new PollableAnnotationPostProcessor(this.messageBus, this.beanClassLoader));
this.postProcessors.put(MessageTarget.class, new TargetAnnotationPostProcessor(this.messageBus, this.beanClassLoader));
postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.messageBus));
postProcessors.put(ChannelAdapter.class, new ChannelAdapterAnnotationPostProcessor(this.messageBus));
postProcessors.put(Handler.class, new ServiceActivatorAnnotationPostProcessor(this.messageBus));
postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.messageBus));
postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.messageBus));
postProcessors.put(Transformer.class, new TransformerAnnotationPostProcessor(this.messageBus));
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
Object originalBean = bean;
Class<?> beanClass = this.getBeanClass(bean);
public Object postProcessAfterInitialization(Object bean, final String beanName) throws BeansException {
final Object originalBean = bean;
final Class<?> beanClass = this.getBeanClass(bean);
if (!this.isStereotype(beanClass)) {
// we only post-process stereotype components
return bean;
}
MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(beanClass, MessageEndpoint.class);
for (Map.Entry<Class<?>, AnnotationMethodPostProcessor> entry : this.postProcessors.entrySet()) {
AnnotationMethodPostProcessor postProcessor = entry.getValue();
bean = postProcessor.postProcess(bean, beanName, beanClass);
if (endpointAnnotation != null && entry.getKey().isAssignableFrom(bean.getClass())) {
AbstractEndpoint endpoint = postProcessor.createEndpoint(bean);
if (endpoint != null) {
endpoint.setBeanName(beanName + "." + entry.getKey().getSimpleName() + ".endpoint");
String inputChannelName = endpointAnnotation.input();
if (!StringUtils.hasText(inputChannelName)) {
continue;
}
MessageChannel inputChannel = this.messageBus.lookupChannel(inputChannelName);
if (inputChannel == null) {
throw new ConfigurationException("unable to resolve input channel '" + inputChannelName + "'");
}
Poller pollerAnnotation = AnnotationUtils.findAnnotation(beanClass, Poller.class);
if (pollerAnnotation != null) {
if (inputChannel instanceof PollableChannel) {
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
PollingDispatcher poller = new PollingDispatcher((PollableChannel) inputChannel, schedule);
poller.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
endpoint.setSource(poller);
}
else {
throw new ConfigurationException("The @Poller annotation should only be provided for a PollableSource");
final ProxyFactory proxyFactory = new ProxyFactory(bean);
final AtomicBoolean isProxy = new AtomicBoolean(false);
ReflectionUtils.doWithMethods(beanClass, new ReflectionUtils.MethodCallback() {
@SuppressWarnings("unchecked")
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation[] annotations = AnnotationUtils.getAnnotations(method);
for (Annotation annotation : annotations) {
MethodAnnotationPostProcessor postProcessor = postProcessors.get(annotation.annotationType());
if (postProcessor != null) {
Object result = postProcessor.postProcess(originalBean, beanName, method, annotation);
if (result != null) {
if (result instanceof MessageEndpoint) {
messageBus.registerEndpoint((MessageEndpoint) result);
}
else {
boolean shouldProxy = false;
Class<?>[] interfaces = ClassUtils.getAllInterfaces(result);
for (Class<?> iface : interfaces) {
if (!iface.getPackage().getName().startsWith("org.springframework.integration")) {
continue;
}
if (proxyFactory.isInterfaceProxied(iface)) {
throw new ConfigurationException("interface [" + iface + "] is already proxied");
}
shouldProxy = true;
}
if (result instanceof ChannelRegistryAware) {
((ChannelRegistryAware) result).setChannelRegistry(messageBus);
}
if (shouldProxy) {
proxyFactory.addAdvice(new DelegatingIntroductionInterceptor(result));
isProxy.set(true);
}
}
}
}
else {
endpoint.setSource(inputChannel);
}
String outputChannelName = endpointAnnotation.output();
if (StringUtils.hasText(outputChannelName)) {
MessageChannel outputChannel = this.messageBus.lookupChannel(outputChannelName);
if (outputChannel == null) {
throw new ConfigurationException("unable to resolve output channel '" + outputChannelName + "'");
}
endpoint.setTarget(outputChannel);
}
this.messageBus.registerEndpoint(endpoint);
}
}
}
});
if (bean instanceof ChannelRegistryAware) {
((ChannelRegistryAware) bean).setChannelRegistry(this.messageBus);
((ChannelRegistryAware) bean).setChannelRegistry(messageBus);
}
if (!bean.equals(originalBean) && originalBean instanceof ChannelRegistryAware) {
((ChannelRegistryAware) originalBean).setChannelRegistry(this.messageBus);
}
if (endpointAnnotation != null && bean.equals(originalBean)) {
throw new ConfigurationException("Class [" + beanClass.getName()
+ "] is annotated with @MessageEndpoint but contains no source, target, or handler method annotations.");
if (isProxy.get()) {
return proxyFactory.getProxy(this.beanClassLoader);
}
return bean;
}

9
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationMethodPostProcessor.java → org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MethodAnnotationPostProcessor.java
View File

@@ -16,17 +16,16 @@
package org.springframework.integration.config.annotation;
import org.springframework.integration.endpoint.AbstractEndpoint;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
/**
* Strategy interface for post-processing annotated methods.
*
* @author Mark Fisher
*/
public interface AnnotationMethodPostProcessor {
public interface MethodAnnotationPostProcessor<T extends Annotation> {
Object postProcess(Object bean, String beanName, Class<?> originalBeanClass);
AbstractEndpoint createEndpoint(Object bean);
Object postProcess(Object bean, String beanName, Method method, T annotation);
}

105
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java
View File

@@ -1,105 +0,0 @@
/*
* 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.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.List;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.ChannelAdapter;
import org.springframework.integration.annotation.Pollable;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.InboundChannelAdapter;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.PollingSchedule;
/**
* Post-processor for methods annotated with {@link Pollable @Pollable}.
*
* @author Mark Fisher
*/
public class PollableAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor<MessageSource<?>> {
public PollableAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) {
super(Pollable.class, messageBus, beanClassLoader);
}
protected MessageSource<?> processMethod(Object bean, Method method, Annotation annotation) {
MethodInvokingSource source = new MethodInvokingSource();
source.setObject(bean);
source.setMethod(method);
ChannelAdapter channelAdapterAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), ChannelAdapter.class);
if (channelAdapterAnnotation != null) {
Poller pollerAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), Poller.class);
if (pollerAnnotation == null) {
throw new ConfigurationException("The @Poller annotation is required (at class-level) "
+ "when using the @ChannelAdapter annotation with a @Pollable method annotation.");
}
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
PollingDispatcher poller = new PollingDispatcher((PollableSource<?>) source, schedule);
int maxMessagesPerPoll = pollerAnnotation.maxMessagesPerPoll();
if (maxMessagesPerPoll == -1) {
// the default is 1 since a MethodInvokingSource might return a non-null value
// every time it is invoked, thus producing an infinite number of messages per poll
maxMessagesPerPoll = 1;
}
poller.setMaxMessagesPerPoll(maxMessagesPerPoll);
InboundChannelAdapter adapter = new InboundChannelAdapter();
adapter.setSource(poller);
String channelName = channelAdapterAnnotation.value();
MessageChannel channel = this.getMessageBus().lookupChannel(channelName);
if (channel == null) {
adapter.setBeanName(channelName + ".adapter");
DirectChannel directChannel = new DirectChannel();
directChannel.setBeanName(channelName);
this.getMessageBus().registerChannel(directChannel);
channel = directChannel;
}
else {
adapter.setBeanName(channelName);
}
adapter.setTarget(channel);
this.getMessageBus().registerEndpoint(adapter);
}
return source;
}
protected MessageSource<?> processResults(List<MessageSource<?>> results) {
if (results.size() > 1) {
throw new ConfigurationException("At most one @Pollable annotation is allowed per class.");
}
return (results.size() == 1) ? results.get(0) : null;
}
public AbstractEndpoint createEndpoint(Object bean) {
return null;
}
}

69
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,69 @@
/*
* 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.annotation;
import java.lang.reflect.Method;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.router.MethodInvokingRouter;
import org.springframework.integration.router.RouterEndpoint;
import org.springframework.util.StringUtils;
/**
* Post-processor for Methods annotated with {@link Router @Router}.
*
* @author Mark Fisher
*/
public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Router> {
public RouterAnnotationPostProcessor(MessageBus messageBus) {
super(messageBus);
}
@Override
protected Object createMethodInvokingAdapter(Object bean, Method method, Router annotation) {
return new MethodInvokingRouter(bean, method);
}
@Override
protected AbstractEndpoint createEndpoint(Object adapter) {
if (adapter instanceof MethodInvokingRouter) {
return new RouterEndpoint((MethodInvokingRouter) adapter);
}
return null;
}
@Override
protected void configureEndpoint(AbstractEndpoint endpoint, Router annotation, Poller pollerAnnotation) {
super.configureEndpoint(endpoint, annotation, pollerAnnotation);
String defaultOutputChannelName = annotation.defaultOutputChannel();
if (StringUtils.hasText(defaultOutputChannelName)) {
MessageChannel defaultOutputChannel = this.getChannelRegistry().lookupChannel(defaultOutputChannelName);
if (defaultOutputChannel == null) {
throw new ConfigurationException("unable to resolve defaultOutputChannel '" + defaultOutputChannelName + "'");
}
((RouterEndpoint) endpoint).setDefaultOutputChannel(defaultOutputChannel);
}
}
}

51
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* 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.annotation;
import java.lang.reflect.Method;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.DefaultServiceInvoker;
import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
import org.springframework.integration.endpoint.ServiceInvoker;
/**
* @author Mark Fisher
*/
public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Handler> {
public ServiceActivatorAnnotationPostProcessor(MessageBus messageBus) {
super(messageBus);
}
@Override
protected Object createMethodInvokingAdapter(Object bean, Method method, Handler annotation) {
return new DefaultServiceInvoker(bean, method);
}
@Override
protected AbstractEndpoint createEndpoint(Object adapter) {
if (adapter instanceof ServiceInvoker) {
return new ServiceActivatorEndpoint((ServiceInvoker) adapter);
}
return null;
}
}

52
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* 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.annotation;
import java.lang.reflect.Method;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.splitter.MethodInvokingSplitter;
import org.springframework.integration.splitter.SplitterEndpoint;
/**
* Post-processor for Methods annotated with {@link Splitter @Splitter}.
*
* @author Mark Fisher
*/
public class SplitterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Splitter> {
public SplitterAnnotationPostProcessor(MessageBus messageBus) {
super(messageBus);
}
@Override
protected Object createMethodInvokingAdapter(Object bean, Method method, Splitter annotation) {
return new MethodInvokingSplitter(bean, method);
}
@Override
protected AbstractEndpoint createEndpoint(Object adapter) {
if (adapter instanceof MethodInvokingSplitter) {
return new SplitterEndpoint((MethodInvokingSplitter) adapter);
}
return null;
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java
View File

@@ -30,8 +30,8 @@ import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Subscriber;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.DefaultServiceInvoker;
import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
import org.springframework.integration.message.MessageMappingMethodInvoker;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
@@ -92,7 +92,7 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor {
if (!StringUtils.hasText(channelName)) {
throw new ConfigurationException("no channel name provided for subscriber");
}
MessageMappingMethodInvoker invoker = new MessageMappingMethodInvoker(bean, method);
DefaultServiceInvoker invoker = new DefaultServiceInvoker(bean, method);
invoker.afterPropertiesSet();
String endpointName = ClassUtils.getShortNameAsProperty(targetClass) +
"." + method.getName() + ".endpoint";

91
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java
View File

@@ -1,91 +0,0 @@
/*
* 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.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.List;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.ChannelAdapter;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.OutboundChannelAdapter;
import org.springframework.integration.handler.MethodInvokingTarget;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.PollingSchedule;
/**
* Post-processor for classes annotated with {@link MessageTarget @MessageTarget}.
*
* @author Mark Fisher
*/
public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor<MessageTarget> {
public TargetAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) {
super(org.springframework.integration.annotation.MessageTarget.class, messageBus, beanClassLoader);
}
protected MessageTarget processMethod(Object bean, Method method, Annotation annotation) {
MethodInvokingTarget target = new MethodInvokingTarget(bean, method);
ChannelAdapter channelAdapterAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), ChannelAdapter.class);
if (channelAdapterAnnotation != null) {
OutboundChannelAdapter adapter = new OutboundChannelAdapter();
String channelName = channelAdapterAnnotation.value();
MessageChannel channel = this.getMessageBus().lookupChannel(channelName);
if (channel == null) {
adapter.setBeanName(channelName + ".adapter");
DirectChannel directChannel = new DirectChannel();
directChannel.setBeanName(channelName);
this.getMessageBus().registerChannel(directChannel);
channel = directChannel;
}
else {
adapter.setBeanName(channelName);
}
if (channel instanceof PollableSource) {
// TODO: add poller config if period, etc is provided (add to @Pollable)
PollingDispatcher poller = new PollingDispatcher((PollableSource<?>) channel, new PollingSchedule(0));
adapter.setSource(poller);
}
else {
adapter.setSource(channel);
}
adapter.setTarget(target);
this.getMessageBus().registerEndpoint(adapter);
}
return target;
}
protected MessageTarget processResults(List<MessageTarget> results) {
if (results.size() > 1) {
throw new ConfigurationException("At most one @MessageTarget annotation is allowed per class.");
}
return (results.size() == 1) ? results.get(0) : null;
}
public AbstractEndpoint createEndpoint(Object bean) {
return null;
}
}

52
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* 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.annotation;
import java.lang.reflect.Method;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.transformer.MethodInvokingTransformer;
import org.springframework.integration.transformer.TransformerEndpoint;
/**
* Post-processor for Methods annotated with {@link Transformer @Transformer}.
*
* @author Mark Fisher
*/
public class TransformerAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Transformer> {
public TransformerAnnotationPostProcessor(MessageBus messageBus) {
super(messageBus);
}
@Override
protected Object createMethodInvokingAdapter(Object bean, Method method, Transformer annotation) {
return new MethodInvokingTransformer(bean, method);
}
@Override
protected AbstractEndpoint createEndpoint(Object adapter) {
if (adapter instanceof MethodInvokingTransformer) {
return new TransformerEndpoint((MethodInvokingTransformer) adapter);
}
return null;
}
}

33
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Pollable.java → org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultServiceInvoker.java
View File

@@ -14,28 +14,29 @@
* limitations under the License.
*/
package org.springframework.integration.annotation;
package org.springframework.integration.endpoint;
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.reflect.Method;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMappingMethodInvoker;
/**
* Indicates that a method is capable of producing messages. The method must
* accept no parameters and return either a {@link Message} or an Object to
* be passed as the message payload. The enclosing class may also be annotated
* with {@link ChannelAdapter @ChannelAdapter}.
*
* @author Mark Fisher
*/
@java.lang.annotation.Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Pollable {
public class DefaultServiceInvoker extends MessageMappingMethodInvoker implements ServiceInvoker {
public DefaultServiceInvoker(Object object, Method method) {
super(object, method);
}
public DefaultServiceInvoker(Object object, String methodName) {
super(object, methodName);
}
public Object invoke(Message<?> message) {
return this.invokeMethod(message);
}
}

11
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceActivatorEndpoint.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.endpoint;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMappingMethodInvoker;
import org.springframework.util.Assert;
/**
@@ -27,12 +26,12 @@ import org.springframework.util.Assert;
*/
public class ServiceActivatorEndpoint extends AbstractInOutEndpoint implements InitializingBean {
private final MessageMappingMethodInvoker invoker;
private final ServiceInvoker invoker;
private final MessageHandler handler;
public ServiceActivatorEndpoint(MessageMappingMethodInvoker invoker) {
public ServiceActivatorEndpoint(ServiceInvoker invoker) {
Assert.notNull(invoker, "invoker must not be null");
this.invoker = invoker;
this.handler = null;
@@ -46,15 +45,15 @@ public class ServiceActivatorEndpoint extends AbstractInOutEndpoint implements I
public void afterPropertiesSet() throws Exception {
if (this.invoker != null) {
this.invoker.afterPropertiesSet();
if (this.invoker != null && (this.invoker instanceof InitializingBean)) {
((InitializingBean) this.invoker).afterPropertiesSet();
}
}
@Override
protected Object handle(Message<?> message) {
if (this.invoker != null) {
return this.invoker.invokeMethod(message);
return this.invoker.invoke(message);
}
return this.handler.handle(message);
}

28
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceInvoker.java Normal file
View File

@@ -0,0 +1,28 @@
/*
* 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.endpoint;
import org.springframework.integration.message.Message;
/**
* @author Mark Fisher
*/
public interface ServiceInvoker {
Object invoke(Message<?> message);
}

4
org.springframework.integration/src/main/java/org/springframework/integration/handler/MethodInvokingTarget.java
View File

@@ -18,8 +18,8 @@ package org.springframework.integration.handler;
import java.lang.reflect.Method;
import org.springframework.integration.endpoint.DefaultServiceInvoker;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMappingMethodInvoker;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.MessagingException;
@@ -28,7 +28,7 @@ import org.springframework.integration.message.MessagingException;
*
* @author Mark Fisher
*/
public class MethodInvokingTarget extends MessageMappingMethodInvoker implements MessageTarget {
public class MethodInvokingTarget extends DefaultServiceInvoker implements MessageTarget {
public MethodInvokingTarget(Object object, Method method) {
super(object, method);