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

Channel Adapters are now endpoints, but if no "channel" attribute if provided for a <channel-adapter/> element, a DirectChannel will be created automatically. The <poller/> sub-element now belongs within the <channel-adapter/> (not the consumer endpoint downstream). This enables support for multiple Channel Adapters to share a MessageChannel. Also, the @Poller annotation belongs at class-level along with @ChannelAdapter if a @Pollable method is being adapted via MethodInvokingSource.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-17 22:37:35 +00:00
parent c48fa5a361
commit 9b85225675
36 changed files with 701 additions and 802 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -39,7 +39,7 @@ import org.springframework.integration.scheduling.PollingSchedule;
@Documented
public @interface Poller {
int period() default 0;
int period();
long initialDelay() default PollingSchedule.DEFAULT_INITIAL_DELAY;
@@ -47,6 +47,6 @@ public @interface Poller {
TimeUnit timeUnit() default TimeUnit.MILLISECONDS;
int maxMessagesPerPoll() default 1;
int maxMessagesPerPoll() default -1;
}

51
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractChannelAdapter.java
View File

@@ -1,51 +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.channel;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.Assert;
/**
* The base class for Channel Adapters.
*
* @author Mark Fisher
*/
public class AbstractChannelAdapter extends AbstractMessageChannel {
private final MessageTarget target;
public AbstractChannelAdapter(String name, MessageTarget target) {
Assert.notNull(name, "name must not be null");
this.setBeanName(name);
this.target = target;
}
@Override
protected boolean doSend(Message<?> message, long timeout) {
if (this.target == null) {
return false;
}
return (timeout >= 0 && this.target instanceof BlockingTarget)
? ((BlockingTarget) this.target).send(message, timeout)
: this.target.send(message);
}
}

84
org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannelAdapter.java
View File

@@ -1,84 +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.channel;
import java.util.List;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.selector.MessageSelector;
/**
* Channel Adapter implementation for a {@link PollableSource}.
*
* @author Mark Fisher
*/
public class PollableChannelAdapter extends AbstractChannelAdapter implements PollableChannel, MessageDeliveryAware {
private final PollableSource<?> source;
public PollableChannelAdapter(String name, PollableSource<?> source, MessageTarget target) {
super(name, target);
this.source = source;
}
public Message<?> receive() {
return this.receive(-1);
}
public Message<?> receive(long timeout) {
if (this.source != null) {
return (timeout >= 0 && this.source instanceof BlockingSource)
? ((BlockingSource<?>) this.source).receive(timeout)
: this.source.receive();
}
return null;
}
public List<Message<?>> clear() {
if (this.source != null && this.source instanceof PollableChannel) {
return ((PollableChannel) this.source).clear();
}
return null;
}
public List<Message<?>> purge(MessageSelector selector) {
if (this.source != null && this.source instanceof PollableChannel) {
return ((PollableChannel) this.source).purge(selector);
}
return null;
}
public void onSend(Message<?> sentMessage) {
if (this.source != null && this.source instanceof MessageDeliveryAware) {
((MessageDeliveryAware) this.source).onSend(sentMessage);
}
}
public void onFailure(MessagingException exception) {
if (this.source != null && this.source instanceof MessageDeliveryAware) {
((MessageDeliveryAware) this.source).onFailure(exception);
}
}
}

48
org.springframework.integration/src/main/java/org/springframework/integration/channel/SubscribableChannelAdapter.java
View File

@@ -1,48 +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.channel;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.util.Assert;
/**
* Channel Adapter implementation for a {@link SubscribableSource}.
*
* @author Mark Fisher
*/
public class SubscribableChannelAdapter extends AbstractChannelAdapter implements SubscribableSource {
private final SubscribableSource source;
public SubscribableChannelAdapter(String name, SubscribableSource source, MessageTarget target) {
super(name, target);
Assert.notNull(source, "source must not be null");
this.source = source;
}
public boolean subscribe(MessageTarget target) {
return this.source.subscribe(target);
}
public boolean unsubscribe(MessageTarget target) {
return this.source.unsubscribe(target);
}
}

110
org.springframework.integration/src/main/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBean.java
View File

@@ -1,110 +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.channel.config;
import java.util.List;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannelAdapter;
import org.springframework.integration.channel.SubscribableChannelAdapter;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.SubscribableSource;
/**
* @author Mark Fisher
*/
public class ChannelAdapterFactoryBean implements FactoryBean, BeanNameAware {
private volatile String name;
private volatile MessageSource<?> source;
private volatile MessageTarget target;
private volatile AbstractMessageChannel channel;
private List<ChannelInterceptor> interceptors;
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
public void setBeanName(String name) {
this.name = name;
}
public void setSource(MessageSource<?> source) {
this.source = source;
}
public void setTarget(MessageTarget target) {
this.target = target;
}
public void setInterceptors(List<ChannelInterceptor> interceptors) {
this.interceptors = interceptors;
}
public Object getObject() throws Exception {
if (!this.initialized) {
this.initializeChannel();
}
return this.channel;
}
public Class<?> getObjectType() {
if (!this.initialized) {
return MessageChannel.class;
}
return this.channel.getClass();
}
public boolean isSingleton() {
return true;
}
private void initializeChannel() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
if (this.source == null || source instanceof PollableSource) {
this.channel = new PollableChannelAdapter(
this.name, (PollableSource<?>) this.source, this.target);
}
else if (this.source instanceof SubscribableSource) {
this.channel = new SubscribableChannelAdapter(
this.name, (SubscribableSource) this.source, this.target);
}
else {
throw new ConfigurationException("source must be either a PollableSource or SubscribableSource");
}
if (this.interceptors != null) {
this.channel.setInterceptors(this.interceptors);
}
}
}
}

90
org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java
View File

@@ -18,24 +18,98 @@ package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.integration.channel.config.ChannelAdapterFactoryBean;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.endpoint.InboundChannelAdapter;
import org.springframework.integration.endpoint.OutboundChannelAdapter;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
* Parser for the <channel-adapter/> element.
* Parser for the &lt;channel-adapter/&gt; element.
*
* @author Mark Fisher
*/
public class ChannelAdapterParser extends AbstractChannelParser {
public class ChannelAdapterParser extends AbstractBeanDefinitionParser {
protected final Class<?> getBeanClass(Element element) {
return ChannelAdapterFactoryBean.class;
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) throws BeanDefinitionStoreException {
String id = element.getAttribute("id");
if (!element.hasAttribute("channel")) {
// the created channel will get the 'id', so the adapter's bean name includes a suffix
id = id + ".adapter";
}
else if (!StringUtils.hasText(id)) {
id = parserContext.getReaderContext().generateBeanName(definition);
}
return id;
}
@Override
protected void postProcess(BeanDefinitionBuilder builder, Element element) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "source");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "target");
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
String source = element.getAttribute("source");
String target = element.getAttribute("target");
String channelName = element.getAttribute("channel");
Element pollerElement = DomUtils.getChildElementByTagName(element, "poller");
BeanDefinitionBuilder adapterBuilder = null;
if (StringUtils.hasText(source)) {
if (StringUtils.hasText(target)) {
throw new ConfigurationException("both 'source' and 'target' are not allowed, provide only one");
}
adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(InboundChannelAdapter.class);
if (pollerElement != null) {
String pollerBeanName = IntegrationNamespaceUtils.parsePoller(source, pollerElement, parserContext);
adapterBuilder.addPropertyReference("source", pollerBeanName);
}
else {
adapterBuilder.addPropertyReference("source", source);
}
if (StringUtils.hasText(channelName)) {
adapterBuilder.addPropertyReference("target", channelName);
}
else {
adapterBuilder.addPropertyReference("target",
this.createDirectChannel(element, parserContext));
}
}
else if (StringUtils.hasText(target)) {
adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(OutboundChannelAdapter.class);
adapterBuilder.addPropertyReference("target", target);
if (pollerElement != null) {
if (!StringUtils.hasText(channelName)) {
throw new ConfigurationException("outbound channel-adapter with a 'poller' requires a 'channel' to poll");
}
String pollerBeanName = IntegrationNamespaceUtils.parsePoller(channelName, pollerElement, parserContext);
adapterBuilder.addPropertyReference("source", pollerBeanName);
}
else {
adapterBuilder.addPropertyReference("source",
this.createDirectChannel(element, parserContext));
}
}
else {
throw new ConfigurationException("either 'source' or 'target' is required");
}
return adapterBuilder.getBeanDefinition();
}
private String createDirectChannel(Element element, ParserContext parserContext) {
String channelId = element.getAttribute("id");
if (!StringUtils.hasText(channelId)) {
throw new ConfigurationException("The channel-adapter's 'id' attribute is required when no 'channel' "
+ "reference has been provided, because that 'id' would be used for the created channel.");
}
BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelId);
BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry());
return channelId;
}
}

50
org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java
View File

@@ -25,12 +25,7 @@ import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.message.AsyncMessageExchangeTemplate;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -145,7 +140,7 @@ public abstract class IntegrationNamespaceUtils {
* @return the name of the poller bean definition
*/
public static String parsePoller(String sourceBeanName, Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PollingDispatcher.class);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PollingDispatcherFactoryBean.class);
Long period = Long.valueOf(element.getAttribute("period"));
PollingSchedule schedule = new PollingSchedule(period);
String initialDelay = element.getAttribute("initial-delay");
@@ -158,42 +153,21 @@ public abstract class IntegrationNamespaceUtils {
else {
schedule.setFixedRate(false);
}
String templateBeanName = parseMessageExhangeTemplate(element, parserContext);
builder.addConstructorArgReference(sourceBeanName);
builder.addConstructorArgValue(schedule);
builder.addConstructorArgValue(new SimpleDispatcher());
builder.addConstructorArgReference(templateBeanName);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
if (txElement != null) {
builder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
builder.addPropertyValue("propagationBehaviorName", txElement.getAttribute("propagation"));
builder.addPropertyValue("isolationLevelName", txElement.getAttribute("isolation"));
builder.addPropertyValue("transactionTimeout", txElement.getAttribute("timeout"));
builder.addPropertyValue("transactionReadOnly", txElement.getAttribute("read-only"));
}
builder.addPropertyReference("source", sourceBeanName);
builder.addPropertyValue("schedule", schedule);
setValueIfAttributeDefined(builder, element, "receive-timeout");
setValueIfAttributeDefined(builder, element, "send-timeout");
setValueIfAttributeDefined(builder, element, "max-messages-per-poll");
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
}
private static String parseMessageExhangeTemplate(Element element, ParserContext parserContext) {
String taskExecutorRef = element.getAttribute("task-executor");
Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
Class<?> beanClass = (StringUtils.hasText(taskExecutorRef)) ?
AsyncMessageExchangeTemplate.class : MessageExchangeTemplate.class;
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(beanClass);
if (StringUtils.hasText(taskExecutorRef)) {
builder.addConstructorArgReference(taskExecutorRef);
}
if (txElement != null) {
builder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
builder.addPropertyValue("propagationBehaviorName", DefaultTransactionDefinition.PREFIX_PROPAGATION + txElement.getAttribute("propagation"));
builder.addPropertyValue("isolationLevelName", DefaultTransactionDefinition.PREFIX_ISOLATION + txElement.getAttribute("isolation"));
builder.addPropertyValue("transactionTimeout", txElement.getAttribute("timeout"));
builder.addPropertyValue("transactionReadOnly", txElement.getAttribute("read-only"));
}
String receiveTimeout = element.getAttribute("receive-timeout");
if (StringUtils.hasText(receiveTimeout)) {
builder.addPropertyValue("receiveTimeout", Long.parseLong(receiveTimeout));
}
String sendTimeout = element.getAttribute("send-timeout");
if (StringUtils.hasText(sendTimeout)) {
builder.addPropertyValue("sendTimeout", Long.parseLong(sendTimeout));
}
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
}
}

174
org.springframework.integration/src/main/java/org/springframework/integration/config/PollingDispatcherFactoryBean.java Normal file
View File

@@ -0,0 +1,174 @@
/*
* 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 org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.message.AsyncMessageExchangeTemplate;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.DefaultTransactionDefinition;
/**
* @author Mark Fisher
*/
public class PollingDispatcherFactoryBean implements FactoryBean, InitializingBean {
private volatile PollingDispatcher poller;
private volatile MessageSource<?> source;
private volatile Schedule schedule;
private volatile long receiveTimeout = -1;
private volatile long sendTimeout = -1;
private volatile int maxMessagesPerPoll = -1;
private volatile TaskExecutor taskExecutor;
private volatile PlatformTransactionManager transactionManager;
private volatile String propagationBehaviorName;
private volatile String isolationLevelName;
private volatile int transactionTimeout;
private volatile boolean transactionReadOnly;
private volatile boolean validated;
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
public void setSource(MessageSource<?> source) {
this.source = source;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
public void setSendTimeout(long sendTimeout) {
this.sendTimeout = sendTimeout;
}
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
public void setPropagationBehaviorName(String propagationBehaviorName) {
this.propagationBehaviorName = propagationBehaviorName;
}
public void setIsolationLevelName(String isolationLevelName) {
this.isolationLevelName = isolationLevelName;
}
public void setTransactionTimeout(int transactionTimeout) {
this.transactionTimeout = transactionTimeout;
}
public void setTransactionReadOnly(boolean transactionReadOnly) {
this.transactionReadOnly = transactionReadOnly;
}
public void afterPropertiesSet() {
synchronized (this.initializationMonitor) {
if (this.source == null) {
throw new ConfigurationException("source is required");
}
if (!(this.source instanceof PollableSource)) {
throw new BeanCreationException("Poller requires a PollableSource, but actual type of '"
+ this.source + "' is [" + this.source.getClass() + "]");
}
this.validated = true;
}
}
public Object getObject() throws Exception {
if (!this.initialized) {
this.initPoller();
}
return this.poller;
}
public Class<?> getObjectType() {
return PollingDispatcher.class;
}
public boolean isSingleton() {
return true;
}
private void initPoller() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
if (!this.validated) {
this.afterPropertiesSet();
}
if (this.schedule == null) {
this.schedule = new PollingSchedule(0);
}
MessageExchangeTemplate template = this.createMessageExchangeTemplate();
this.poller = new PollingDispatcher((PollableSource<?>) this.source, this.schedule, new SimpleDispatcher(), template);
this.poller.setMaxMessagesPerPoll(this.maxMessagesPerPoll);
this.initialized = true;
}
}
private MessageExchangeTemplate createMessageExchangeTemplate() {
MessageExchangeTemplate template = (this.taskExecutor != null) ?
new AsyncMessageExchangeTemplate(this.taskExecutor) : new MessageExchangeTemplate();
template.setTransactionManager(this.transactionManager);
template.setPropagationBehaviorName(DefaultTransactionDefinition.PREFIX_PROPAGATION + this.propagationBehaviorName);
template.setIsolationLevelName(DefaultTransactionDefinition.PREFIX_ISOLATION + this.isolationLevelName);
template.setTransactionTimeout(this.transactionTimeout);
template.setTransactionReadOnly(this.transactionReadOnly);
template.setReceiveTimeout(this.receiveTimeout);
template.setSendTimeout(this.sendTimeout);
return template;
}
}

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

@@ -24,11 +24,17 @@ 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.PollableChannelAdapter;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.InboundChannelAdapter;
import org.springframework.integration.endpoint.MessageEndpoint;
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}.
@@ -48,9 +54,39 @@ public class PollableAnnotationPostProcessor extends AbstractAnnotationMethodPos
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();
PollableChannelAdapter adapter = new PollableChannelAdapter(channelName, source, null);
this.getMessageBus().registerChannel(adapter);
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;
}

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

@@ -24,10 +24,15 @@ 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.PollableChannelAdapter;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.MessageEndpoint;
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}.
@@ -47,9 +52,29 @@ public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostP
target.setMethod(method);
ChannelAdapter channelAdapterAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), ChannelAdapter.class);
if (channelAdapterAnnotation != null) {
OutboundChannelAdapter adapter = new OutboundChannelAdapter();
String channelName = channelAdapterAnnotation.value();
PollableChannelAdapter adapter = new PollableChannelAdapter(channelName, null, target);
this.getMessageBus().registerChannel(adapter);
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;
}

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

@@ -162,12 +162,13 @@
MessageTarget. Therefore, either "source" or "target" should be provided (but never both).
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="channelType">
<xsd:attribute name="source" type="xsd:string"/>
<xsd:attribute name="target" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
<xsd:sequence>
<xsd:element ref="poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="source" type="xsd:string"/>
<xsd:attribute name="target" type="xsd:string"/>
<xsd:attribute name="channel" type="xsd:string"/>
</xsd:complexType>
</xsd:element>

86
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -20,10 +20,13 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.integration.message.CompositeMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.util.ErrorHandler;
/**
@@ -37,15 +40,17 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
private volatile String name;
private MessageSource<?> source;
private MessageTarget target;
private volatile Schedule schedule;
private volatile ErrorHandler errorHandler;
private volatile boolean requiresReply = false;
private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate();
public void setBeanName(String name) {
this.name = name;
}
/**
* Return the name of this endpoint.
*/
@@ -53,12 +58,36 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
return this.name;
}
/**
* Specify whether this endpoint should throw an Exception when
* it returns an invalid reply Message after handling the request.
*/
public void setRequiresReply(boolean requiresReply) {
this.requiresReply = requiresReply;
public void setBeanName(String name) {
this.name = name;
}
public MessageSource<?> getSource() {
return this.source;
}
public void setSource(MessageSource<?> source) {
this.source = source;
}
public MessageTarget getTarget() {
return this.target;
}
public void setTarget(MessageTarget target) {
this.target = target;
}
public Schedule getSchedule() {
return this.schedule;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
protected MessageExchangeTemplate getMessageExchangeTemplate() {
return this.messageExchangeTemplate;
}
/**
@@ -71,49 +100,29 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
this.errorHandler = errorHandler;
}
public final boolean send(Message<?> requestMessage) {
if (requestMessage == null || requestMessage.getPayload() == null) {
public final boolean send(Message<?> message) {
if (message == null || message.getPayload() == null) {
throw new IllegalArgumentException("Message and its payload must not be null");
}
if (this.logger.isDebugEnabled()) {
this.logger.debug("endpoint '" + this + "' handling message: " + requestMessage);
this.logger.debug("endpoint '" + this + "' processing message: " + message);
}
try {
Message<?> replyMessage = this.handleRequestMessage(requestMessage);
if (!this.isValidReplyMessage(replyMessage)) {
if (this.requiresReply) {
throw new MessageHandlingException(requestMessage,
"endpoint requires reply but none was received");
}
}
else if (replyMessage instanceof CompositeMessage) {
for (Message<?> nextReply : (CompositeMessage) replyMessage) {
this.sendReplyMessage(nextReply, requestMessage);
}
}
else {
this.sendReplyMessage(replyMessage, requestMessage);
}
return true;
return this.sendInternal(message);
}
catch (Exception e) {
if (e instanceof MessagingException) {
this.handleException((MessagingException) e);
}
else {
this.handleException(new MessageHandlingException(requestMessage,
this.handleException(new MessageHandlingException(message,
"failure occurred in endpoint's send operation", e));
}
return false;
}
}
protected abstract Message<?> handleRequestMessage(Message<?> requestMessage);
protected abstract boolean isValidReplyMessage(Message<?> replyMessage);
protected abstract void sendReplyMessage(Message<?> replyMessage, Message<?> requestMessage);
protected abstract boolean sendInternal(Message<?> message);
private void handleException(MessagingException exception) {
if (this.errorHandler == null) {
@@ -125,7 +134,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
this.errorHandler.handle(exception);
}
public String toString() {
return (this.name != null) ? this.name : super.toString();
}

104
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractRequestReplyEndpoint.java Normal file
View File

@@ -0,0 +1,104 @@
/*
* 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.channel.MessageChannel;
import org.springframework.integration.message.CompositeMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
/**
* @author Mark Fisher
*/
public abstract class AbstractRequestReplyEndpoint extends AbstractEndpoint {
private volatile String inputChannelName;
private volatile String outputChannelName;
private volatile boolean requiresReply = false;
public String getInputChannelName() {
return this.inputChannelName;
}
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
public String getOutputChannelName() {
return this.outputChannelName;
}
public void setOutputChannelName(String outputChannelName) {
this.outputChannelName = outputChannelName;
}
/**
* Specify whether this endpoint should throw an Exception when
* it returns an invalid reply Message after handling the request.
*/
public void setRequiresReply(boolean requiresReply) {
this.requiresReply = requiresReply;
}
protected boolean sendInternal(Message<?> requestMessage) {
Message<?> replyMessage = this.handleRequestMessage(requestMessage);
if (!this.isValidReplyMessage(replyMessage)) {
if (this.requiresReply) {
throw new MessageHandlingException(requestMessage,
"endpoint requires reply but none was received");
}
}
else if (replyMessage instanceof CompositeMessage) {
for (Message<?> nextReply : (CompositeMessage) replyMessage) {
this.sendReplyMessage(nextReply, requestMessage);
}
}
else {
this.sendReplyMessage(replyMessage, requestMessage);
}
return true;
}
protected abstract Message<?> handleRequestMessage(Message<?> requestMessage);
protected abstract boolean isValidReplyMessage(Message<?> replyMessage);
protected abstract void sendReplyMessage(Message<?> replyMessage, Message<?> requestMessage);
@Override
public void setSource(MessageSource<?> source) {
if (source instanceof MessageChannel) {
this.setInputChannelName(((MessageChannel) source).getName());
}
super.setSource(source);
}
@Override
public void setTarget(MessageTarget target) {
if (target instanceof MessageChannel) {
this.setOutputChannelName(((MessageChannel) target).getName());
}
super.setTarget(target);
}
}

77
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java
View File

@@ -28,14 +28,11 @@ import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.CompositeMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageHeaders;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
@@ -60,20 +57,16 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class DefaultEndpoint<T extends MessageHandler> extends AbstractEndpoint implements ChannelRegistryAware {
public class DefaultEndpoint<T extends MessageHandler> extends AbstractRequestReplyEndpoint implements ChannelRegistryAware {
private final T handler;
private volatile MessageChannel outputChannel;
private volatile ChannelRegistry channelRegistry;
private volatile MessageSelector selector;
private final List<EndpointInterceptor> interceptors = new ArrayList<EndpointInterceptor>();
private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate();
/**
* Create an endpoint for the given handler.
@@ -83,15 +76,6 @@ public class DefaultEndpoint<T extends MessageHandler> extends AbstractEndpoint
this.handler = handler;
}
/**
* Specify the channel where reply Messages should be sent if
* no 'nextTarget' header value is available on the reply Message.
*/
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
protected T getHandler() {
return this.handler;
}
@@ -127,7 +111,7 @@ public class DefaultEndpoint<T extends MessageHandler> extends AbstractEndpoint
* target. The default value indicates an indefinite timeout.
*/
public void setReplyTimeout(long replyTimeout) {
this.messageExchangeTemplate.setSendTimeout(replyTimeout);
this.getMessageExchangeTemplate().setSendTimeout(replyTimeout);
}
@Override
@@ -170,7 +154,7 @@ public class DefaultEndpoint<T extends MessageHandler> extends AbstractEndpoint
replyMessage = MessageBuilder.fromMessage(replyMessage)
.copyHeadersIfAbsent(requestMessage.getHeaders())
.setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, requestMessage.getHeaders().getId()).build();
if (!this.messageExchangeTemplate.send(replyMessage, replyTarget)) {
if (!this.getMessageExchangeTemplate().send(replyMessage, replyTarget)) {
throw new MessageEndpointReplyException(replyMessage, requestMessage,
"failed to send reply to '" + replyTarget + "'");
}
@@ -221,7 +205,7 @@ public class DefaultEndpoint<T extends MessageHandler> extends AbstractEndpoint
private MessageTarget resolveReplyTarget(Message<?> replyMessage, MessageHeaders requestHeaders) {
MessageTarget replyTarget = this.resolveTargetAttribute(replyMessage.getHeaders().getNextTarget());
if (replyTarget == null) {
replyTarget = this.outputChannel;
replyTarget = this.getTarget();
}
if (replyTarget == null) {
replyTarget = this.resolveTargetAttribute(requestHeaders.getReturnAddress());
@@ -245,54 +229,17 @@ public class DefaultEndpoint<T extends MessageHandler> extends AbstractEndpoint
return replyTarget;
}
/* TODO: the following properties/methods are candidates for removal from the MessageEndpoint interface. */
// TODO: remove
private volatile String inputChannelName;
private volatile String outputChannelName;
private volatile MessageSource<?> source;
private volatile Schedule schedule;
public String getInputChannelName() {
return this.inputChannelName;
public void setReturnAddressOverrides(boolean returnAddressOverrides) {
}
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
public String getOutputChannelName() {
return this.outputChannelName;
}
public void setOutputChannelName(String outputChannelName) {
this.outputChannelName = outputChannelName;
}
public void setReturnAddressOverrides(boolean b) {
}
public Schedule getSchedule() {
return this.schedule;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
public MessageSource<?> getSource() {
return this.source;
}
public MessageTarget getTarget() {
return this.outputChannel;
}
public void setSource(MessageSource<?> source) {
this.source = source;
}
public void setTarget(MessageTarget target) {
this.outputChannel = (MessageChannel) target;
/**
* Specify the channel where reply Messages should be sent if
* no 'nextTarget' header value is available on the reply Message.
*/
public void setOutputChannel(MessageChannel outputChannel) {
this.setTarget(outputChannel);
}
}

67
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java Normal file
View File

@@ -0,0 +1,67 @@
/*
* 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.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessagingException;
/**
* A Channel Adapter implementation for connecting a
* {@link org.springframework.integration.message.MessageSource}
* to a {@link MessageChannel}.
*
* @author Mark Fisher
*/
public class InboundChannelAdapter extends AbstractEndpoint {
@Override
protected boolean sendInternal(Message<?> message) {
try {
boolean sent = this.getMessageExchangeTemplate().send(message, this.getTarget());
if (sent && this.getSource() instanceof MessageDeliveryAware) {
((MessageDeliveryAware) this.getSource()).onSend(message);
}
return sent;
}
catch (Exception e) {
MessagingException exception = (e instanceof MessagingException) ? (MessagingException) e
: new MessageDeliveryException(message, "channel-adapter failed to send message to target");
if (this.getSource() instanceof MessageDeliveryAware) {
((MessageDeliveryAware) this.getSource()).onFailure(exception);
}
return false;
}
}
public String getInputChannelName() {
return null;
}
public String getOutputChannelName() {
if (this.getTarget() instanceof MessageChannel) {
return ((MessageChannel) this.getTarget()).getName();
}
return null;
}
public void setInputChannelName(String inputChannelName) {
}
}

49
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* 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.channel.MessageChannel;
import org.springframework.integration.message.Message;
/**
* A Channel Adapter implementation for connecting a {@link MessageChannel}
* to a {@link org.springframework.integration.message.MessageTarget}.
*
* @author Mark Fisher
*/
public class OutboundChannelAdapter extends AbstractEndpoint {
@Override
protected boolean sendInternal(Message<?> message) {
return this.getMessageExchangeTemplate().send(message, this.getTarget());
}
public String getInputChannelName() {
if (this.getSource() instanceof MessageChannel) {
return ((MessageChannel) this.getSource()).getName();
}
return null;
}
public String getOutputChannelName() {
return null;
}
public void setInputChannelName(String inputChannelName) {
}
}

3
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java
View File

@@ -32,6 +32,7 @@ import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
/**
* This is the central class for invoking message exchange operations
@@ -192,6 +193,7 @@ public class MessageExchangeTemplate implements InitializingBean {
private boolean doSend(Message<?> message, MessageTarget target) {
Assert.notNull(target, "target must not be null");
long timeout = this.sendTimeout;
boolean sent = (timeout >= 0 && target instanceof BlockingTarget)
? ((BlockingTarget) target).send(message, timeout)
@@ -203,6 +205,7 @@ public class MessageExchangeTemplate implements InitializingBean {
}
private Message<?> doReceive(PollableSource<?> source) {
Assert.notNull(source, "source must not be null");
long timeout = this.receiveTimeout;
Message<?> message = (timeout >= 0 && source instanceof BlockingSource)
? ((BlockingSource<?>) source).receive(timeout)