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

Added PollingSourceEndpoint, removed PollingSourceAdapter, and source-adapter parsers only parse the Source itself.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-23 23:47:02 +00:00
parent 9f3bc80949
commit 4d19a02f15
51 changed files with 568 additions and 554 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -36,7 +36,6 @@ import org.springframework.context.ApplicationListener;
import org.springframework.context.Lifecycle;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.adapter.SourceAdapter;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.DefaultChannelRegistry;
@@ -47,6 +46,7 @@ import org.springframework.integration.endpoint.DefaultEndpointRegistry;
import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.endpoint.TargetEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Target;
@@ -79,7 +79,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
private final Map<MessageChannel, SubscriptionManager> subscriptionManagers = new ConcurrentHashMap<MessageChannel, SubscriptionManager>();
private final List<Lifecycle> lifecycleSourceAdapters = new CopyOnWriteArrayList<Lifecycle>();
private final List<Lifecycle> lifecycleEndpoints = new CopyOnWriteArrayList<Lifecycle>();
private volatile MessagingTaskScheduler taskScheduler;
@@ -161,15 +161,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
}
}
@SuppressWarnings("unchecked")
private void registerSourceAdapters(ApplicationContext context) {
Map<String, SourceAdapter> sourceAdapterBeans =
(Map<String, SourceAdapter>) context.getBeansOfType(SourceAdapter.class);
for (Map.Entry<String, SourceAdapter> entry : sourceAdapterBeans.entrySet()) {
this.registerSourceAdapter(entry.getKey(), entry.getValue());
}
}
public void initialize() {
synchronized (this.lifecycleMonitor) {
if (this.initialized || this.initializing) {
@@ -259,12 +250,11 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
if (endpoint instanceof ChannelRegistryAware) {
((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry);
}
if (endpoint.getConcurrencyPolicy() == null && this.defaultConcurrencyPolicy != null
&& endpoint instanceof TargetEndpoint) {
((TargetEndpoint) endpoint).setConcurrencyPolicy(this.defaultConcurrencyPolicy);
}
if (endpoint instanceof TargetEndpoint) {
((TargetEndpoint) endpoint).afterPropertiesSet();
this.registerTargetEndpoint(name, (TargetEndpoint) endpoint);
}
else if (endpoint instanceof SourceEndpoint) {
this.registerSourceEndpoint(name, (SourceEndpoint) endpoint);
}
this.endpointRegistry.registerEndpoint(name, endpoint);
if (this.isRunning()) {
@@ -275,18 +265,27 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
}
}
private void registerTargetEndpoint(String name, TargetEndpoint endpoint) {
if (endpoint.getConcurrencyPolicy() == null && this.defaultConcurrencyPolicy != null) {
endpoint.setConcurrencyPolicy(this.defaultConcurrencyPolicy);
}
endpoint.afterPropertiesSet();
}
public MessageEndpoint unregisterEndpoint(String name) {
MessageEndpoint endpoint = this.endpointRegistry.unregisterEndpoint(name);
if (endpoint == null) {
return null;
}
Collection<SubscriptionManager> managers = this.subscriptionManagers.values();
boolean removed = false;
for (SubscriptionManager manager : managers) {
removed = (removed || manager.removeTarget(endpoint));
}
if (removed) {
return endpoint;
if (endpoint instanceof TargetEndpoint) {
Collection<SubscriptionManager> managers = this.subscriptionManagers.values();
boolean removed = false;
for (SubscriptionManager manager : managers) {
removed = (removed || manager.removeTarget((TargetEndpoint) endpoint));
}
if (removed) {
return endpoint;
}
}
return null;
}
@@ -310,6 +309,12 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
}
private void activateEndpoint(MessageEndpoint endpoint) {
if (endpoint instanceof TargetEndpoint) {
this.activateTargetEndpoint((TargetEndpoint) endpoint);
}
}
private void activateTargetEndpoint(TargetEndpoint endpoint) {
Subscription subscription = endpoint.getSubscription();
if (subscription == null) {
throw new ConfigurationException("Unable to register endpoint '" +
@@ -360,17 +365,17 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
}
}
public void registerSourceAdapter(String name, SourceAdapter adapter) {
private void registerSourceEndpoint(String name, SourceEndpoint endpoint) {
if (!this.initialized) {
this.initialize();
}
if (adapter instanceof MessagingTask) {
this.taskScheduler.schedule((MessagingTask) adapter);
if (endpoint instanceof MessagingTask) {
this.taskScheduler.schedule((MessagingTask) endpoint);
}
if (adapter instanceof Lifecycle) {
this.lifecycleSourceAdapters.add((Lifecycle) adapter);
if (endpoint instanceof Lifecycle) {
this.lifecycleEndpoints.add((Lifecycle) endpoint);
if (this.isRunning()) {
((Lifecycle) adapter).start();
((Lifecycle) endpoint).start();
}
}
if (logger.isInfoEnabled()) {
@@ -415,10 +420,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
logger.info("started subscription manager '" + manager + "'");
}
}
for (Lifecycle adapter : this.lifecycleSourceAdapters) {
adapter.start();
for (Lifecycle endpoint : this.lifecycleEndpoints) {
endpoint.start();
if (logger.isInfoEnabled()) {
logger.info("started source adapter '" + adapter + "'");
logger.info("started endpoint '" + endpoint + "'");
}
}
}
@@ -436,10 +441,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
synchronized (this.lifecycleMonitor) {
this.running = false;
this.taskScheduler.stop();
for (Lifecycle adapter : this.lifecycleSourceAdapters) {
adapter.stop();
for (Lifecycle endpoint : this.lifecycleEndpoints) {
endpoint.stop();
if (logger.isInfoEnabled()) {
logger.info("stopped source adapter '" + adapter + "'");
logger.info("stopped endpoint '" + endpoint + "'");
}
}
for (SubscriptionManager manager : this.subscriptionManagers.values()) {
@@ -458,7 +463,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
if (event instanceof ContextRefreshedEvent) {
ApplicationContext context = ((ContextRefreshedEvent) event).getApplicationContext();
this.registerEndpoints(context);
this.registerSourceAdapters(context);
if (this.autoStartup) {
this.start();
}

77
spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPublisher.java Normal file
View File

@@ -0,0 +1,77 @@
/*
* 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.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
/**
* Sends to a channel and provides a configurable timeout. Convenient for either
* subclassing or delegation from components that need to publish to a channel.
*
* @author Mark Fisher
*/
public class ChannelPublisher {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile MessageChannel channel;
private volatile long timeout = 0;
public ChannelPublisher() {
}
public ChannelPublisher(MessageChannel channel) {
this.setChannel(channel);
}
public void setChannel(MessageChannel channel) {
Assert.notNull(channel, "channel must not be null");
this.channel = channel;
}
public void setTimeout(long timeout) {
this.timeout = timeout;
}
protected MessageChannel getChannel() {
return this.channel;
}
public boolean publish(Message<?> message) {
if (this.channel == null) {
if (logger.isWarnEnabled()) {
logger.warn("unable to send message, no channel available");
}
return false;
}
if (message == null) {
if (logger.isWarnEnabled()) {
logger.warn("null messages are not supported");
}
return false;
}
return (this.timeout < 0) ? this.channel.send(message) : this.channel.send(message, this.timeout);
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java
View File

@@ -27,8 +27,8 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.adapter.MethodInvokingSource;
import org.springframework.integration.adapter.MethodInvokingTarget;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.StringUtils;
@@ -75,7 +75,7 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
RootBeanDefinition adapterDef = null;
RootBeanDefinition invokerDef = null;
if (this.isInbound) {
adapterDef = new RootBeanDefinition(PollingSourceAdapter.class);
adapterDef = new RootBeanDefinition(PollingSourceEndpoint.class);
invokerDef = new RootBeanDefinition(MethodInvokingSource.class);
invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref));
invokerDef.getPropertyValues().addPropertyValue("method", method);

1
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java
View File

@@ -51,6 +51,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
registerBeanDefinitionParser("priority-channel", new ChannelParser());
registerBeanDefinitionParser("source-adapter", new ChannelAdapterParser(true));
registerBeanDefinitionParser("target-adapter", new ChannelAdapterParser(false));
registerBeanDefinitionParser("source-endpoint", new SourceEndpointParser());
registerBeanDefinitionParser("endpoint", new EndpointParser());
registerBeanDefinitionParser("handler", new HandlerParser());
registerBeanDefinitionParser("handler-chain", new HandlerParser());

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

@@ -36,7 +36,6 @@ import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.adapter.MethodInvokingSource;
import org.springframework.integration.adapter.MethodInvokingTarget;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.annotation.Concurrency;
@@ -51,6 +50,7 @@ import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.dispatcher.SynchronousChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
@@ -163,10 +163,10 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
PollingSchedule schedule = new PollingSchedule(period);
schedule.setInitialDelay(initialDelay);
schedule.setFixedRate(fixedRate);
PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule);
PollingSourceEndpoint sourceEndpoint = new PollingSourceEndpoint(source, channel, schedule);
String channelName = beanName + "-inputChannel";
messageBus.registerChannel(channelName, channel);
messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter);
messageBus.registerEndpoint(beanName + "-sourceEndpoint", sourceEndpoint);
Subscription subscription = new Subscription(channel);
endpoint.setSubscription(subscription);
}

93
spring-integration-core/src/main/java/org/springframework/integration/config/SourceEndpointParser.java Normal file
View File

@@ -0,0 +1,93 @@
/*
* 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.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.endpoint.SimpleSourceEndpoint;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
* Parser for the <source-endpoint/> element.
*
* @author Mark Fisher
*/
public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser {
protected final Class<?> getBeanClass(Element element) {
if (this.getScheduleElement(element) != null) {
return PollingSourceEndpoint.class;
}
return SimpleSourceEndpoint.class;
}
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
protected boolean isEligibleAttribute(String name) {
return (!"source".equals(name) && !"channel".equals(name) && super.isEligibleAttribute(name));
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
String source = element.getAttribute("source");
if (!StringUtils.hasText(source)) {
throw new ConfigurationException("'source' is required");
}
String output = element.getAttribute("channel");
if (!StringUtils.hasText(output)) {
throw new ConfigurationException("'channel' is required");
}
builder.addConstructorArgReference(source);
builder.addConstructorArgReference(output);
Element scheduleElement = this.getScheduleElement(element);
if (scheduleElement != null) {
builder.addConstructorArgValue(this.parseSchedule(scheduleElement));
}
}
/**
* Subclasses may override this method to control the creation of the {@link Schedule}. The default
* implementation creates a {@link PollingSchedule} instance based on the provided "period" attribute.
*/
protected Schedule parseSchedule(Element element) {
String period = element.getAttribute("period");
if (!StringUtils.hasText(period)) {
throw new ConfigurationException("The 'period' attribute is required for the 'schedule' element.");
}
PollingSchedule schedule = new PollingSchedule(Long.valueOf(period));
return schedule;
}
private Element getScheduleElement(Element element) {
return DomUtils.getChildElementByTagName(element, "schedule");
}
}

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

@@ -88,6 +88,25 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="source-endpoint">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a source endpoint.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:sequence>
<xsd:element ref="schedule" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="source" type="xsd:string" use="required"/>
<xsd:attribute name="channel" type="xsd:string" use="required"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="source-adapter">
<xsd:complexType>
<xsd:annotation>

44
spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java → spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractSourceEndpoint.java
View File

@@ -14,41 +14,53 @@
* limitations under the License.
*/
package org.springframework.integration.adapter;
package org.springframework.integration.endpoint;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.Source;
import org.springframework.util.Assert;
/**
* Base class for {@link SourceEndpoint} implementations.
*
* @author Mark Fisher
*/
public abstract class AbstractSourceAdapter implements SourceAdapter {
public abstract class AbstractSourceEndpoint implements SourceEndpoint {
protected final Log logger = LogFactory.getLog(this.getClass());
private final Source source;
private final MessageChannel channel;
private volatile long sendTimeout = -1;
private volatile String name;
public AbstractSourceAdapter(MessageChannel channel) {
public AbstractSourceEndpoint(Source source, MessageChannel channel) {
Assert.notNull(source, "source must not be null");
Assert.notNull(channel, "channel must not be null");
this.source = source;
this.channel = channel;
}
public Source getSource() {
return this.source;
}
public void setBeanName(String name) {
this.name = name;
}
public String getName() {
return this.name;
}
protected MessageChannel getChannel() {
return this.channel;
}
public void setSendTimeout(long sendTimeout) {
this.sendTimeout = sendTimeout;
}
protected boolean sendToChannel(Message<?> message) {
if (message == null) {
throw new IllegalArgumentException("message must not be null");
}
return (this.sendTimeout < 0) ? this.channel.send(message) : this.channel.send(message, this.sendTimeout);
}
}

12
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

@@ -16,23 +16,15 @@
package org.springframework.integration.endpoint;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.message.Target;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.beans.factory.BeanNameAware;
/**
* Base interface for message endpoints.
*
* @author Mark Fisher
*/
public interface MessageEndpoint extends Target, ChannelRegistryAware, InitializingBean, Lifecycle {
public interface MessageEndpoint extends BeanNameAware {
String getName();
Subscription getSubscription();
ConcurrencyPolicy getConcurrencyPolicy();
}

51
spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java → spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java
View File

@@ -14,17 +14,12 @@
* limitations under the License.
*/
package org.springframework.integration.adapter;
package org.springframework.integration.endpoint;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.SynchronousChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessageDeliveryException;
@@ -40,31 +35,26 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class PollingSourceAdapter extends AbstractSourceAdapter implements MessagingTask, InitializingBean {
private final Log logger = LogFactory.getLog(this.getClass());
private final PollableSource<?> source;
public class PollingSourceEndpoint extends AbstractSourceEndpoint implements MessagingTask {
private final PollingSchedule schedule;
private volatile long sendTimeout = 0;
private volatile int maxMessagesPerTask = 1;
private volatile boolean initialized;
/**
* Create a new adapter for the given source.
*/
public PollingSourceAdapter(PollableSource<?> source, MessageChannel channel, PollingSchedule schedule) {
super(channel);
Assert.notNull(source, "source must not be null");
public PollingSourceEndpoint(PollableSource<?> source, MessageChannel channel, PollingSchedule schedule) {
super(source, channel);
Assert.notNull(schedule, "schedule must not be null");
this.source = source;
this.schedule = schedule;
}
public void setSendTimeout(long sendTimeout) {
this.sendTimeout = sendTimeout;
}
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least one");
this.maxMessagesPerTask = maxMessagesPerTask;
@@ -74,18 +64,11 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements Messa
return this.schedule;
}
public void afterPropertiesSet() {
if (this.getChannel() instanceof SynchronousChannel) {
((SynchronousChannel) this.getChannel()).setSource(this.source);
}
this.initialized = true;
}
public List<Message<?>> poll(int limit) {
List<Message<?>> results = new ArrayList<Message<?>>();
int count = 0;
while (count < limit) {
Message<?> message = this.source.receive();
Message<?> message = ((PollableSource<?>) this.getSource()).receive();
if (message == null) {
break;
}
@@ -96,16 +79,16 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements Messa
}
protected boolean sendMessage(Message<?> message) {
if (!this.initialized) {
this.afterPropertiesSet();
if (message == null) {
throw new IllegalArgumentException("message must not be null");
}
boolean sent = super.sendToChannel(message);
if (this.source instanceof MessageDeliveryAware) {
boolean sent = (this.sendTimeout < 0) ? this.getChannel().send(message) : this.getChannel().send(message, this.sendTimeout);
if (this.getSource() instanceof MessageDeliveryAware) {
if (sent) {
((MessageDeliveryAware) this.source).onSend(message);
((MessageDeliveryAware) this.getSource()).onSend(message);
}
else {
((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message"));
((MessageDeliveryAware) this.getSource()).onFailure(new MessageDeliveryException(message, "failed to send message"));
}
}
return sent;

32
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SimpleSourceEndpoint.java Normal file
View File

@@ -0,0 +1,32 @@
/*
* 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.SubscribableSource;
/**
* @author Mark Fisher
*/
public class SimpleSourceEndpoint extends AbstractSourceEndpoint {
public SimpleSourceEndpoint(SubscribableSource source, MessageChannel channel) {
super(source, channel);
source.subscribe(channel);
}
}

10
spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java → spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java
View File

@@ -14,13 +14,17 @@
* limitations under the License.
*/
package org.springframework.integration.adapter;
package org.springframework.integration.endpoint;
import org.springframework.integration.message.Source;
/**
* Base interface for source adapters.
* Base interface for source endpoints.
*
* @author Mark Fisher
*/
public interface SourceAdapter {
public interface SourceEndpoint extends MessageEndpoint {
Source getSource();
}

5
spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java
View File

@@ -28,8 +28,9 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
@@ -47,7 +48,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class TargetEndpoint implements MessageEndpoint, BeanNameAware {
public class TargetEndpoint implements Target, MessageEndpoint, ChannelRegistryAware, InitializingBean, Lifecycle {
protected final Log logger = LogFactory.getLog(this.getClass());

2
spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml
View File

@@ -10,7 +10,7 @@
<bean id="channel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="sourceAdapter" class="org.springframework.integration.adapter.PollingSourceAdapter">
<bean id="sourceEndpoint" class="org.springframework.integration.endpoint.PollingSourceEndpoint">
<constructor-arg>
<bean class="org.springframework.integration.adapter.MethodInvokingSource">
<property name="object">

7
spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
View File

@@ -29,13 +29,12 @@ import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.adapter.SourceAdapter;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.RendezvousChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
@@ -174,8 +173,8 @@ public class MessageBusTests {
public void testErrorChannelWithFailedDispatch() throws InterruptedException {
MessageBus bus = new MessageBus();
CountDownLatch latch = new CountDownLatch(1);
SourceAdapter sourceAdapter = new PollingSourceAdapter(new FailingSource(latch), new QueueChannel(), new PollingSchedule(1000));
bus.registerSourceAdapter("testAdapter", sourceAdapter);
PollingSourceEndpoint sourceEndpoint = new PollingSourceEndpoint(new FailingSource(latch), new QueueChannel(), new PollingSchedule(1000));
bus.registerEndpoint("testEndpoint", sourceEndpoint);
bus.start();
latch.await(1000, TimeUnit.MILLISECONDS);
Message<?> message = bus.getErrorChannel().receive(100);

7
spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java
View File

@@ -32,7 +32,6 @@ import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
import org.springframework.integration.handler.TestHandlers;
@@ -100,7 +99,7 @@ public class SubscriptionManagerTests {
MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
QueueChannel channel = new QueueChannel();
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
MessageEndpoint inactiveEndpoint = createEndpoint(handler1, true);
HandlerEndpoint inactiveEndpoint = createEndpoint(handler1, true);
manager.addTarget(inactiveEndpoint);
manager.addTarget(createEndpoint(handler2, true));
manager.addTarget(createEndpoint(handler3, true));
@@ -124,7 +123,7 @@ public class SubscriptionManagerTests {
MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true));
SubscriptionManager manager = new SubscriptionManager(channel, scheduler);
MessageEndpoint inactiveEndpoint = createEndpoint(handler2, true);
HandlerEndpoint inactiveEndpoint = createEndpoint(handler2, true);
manager.addTarget(createEndpoint(handler1, true));
manager.addTarget(inactiveEndpoint);
manager.addTarget(createEndpoint(handler3, true));
@@ -450,7 +449,7 @@ public class SubscriptionManagerTests {
}
private static MessageEndpoint createEndpoint(MessageHandler handler, boolean asynchronous) {
private static HandlerEndpoint createEndpoint(MessageHandler handler, boolean asynchronous) {
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
if (asynchronous) {
endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));

6
spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java
View File

@@ -28,7 +28,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.TargetEndpoint;
import org.springframework.integration.handler.TestHandlers;
import org.springframework.integration.scheduling.Subscription;
@@ -115,7 +115,7 @@ public class MessageBusParserTests {
public void testDefaultConcurrency() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"messageBusWithDefaultConcurrencyTests.xml", this.getClass());
MessageEndpoint endpoint1 = (MessageEndpoint) context.getBean("endpoint1");
TargetEndpoint endpoint1 = (TargetEndpoint) context.getBean("endpoint1");
assertEquals(4, endpoint1.getConcurrencyPolicy().getCoreSize());
assertEquals(7, endpoint1.getConcurrencyPolicy().getMaxSize());
}
@@ -124,7 +124,7 @@ public class MessageBusParserTests {
public void testExplicitConcurrencyTakesPrecedence() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"messageBusWithDefaultConcurrencyTests.xml", this.getClass());
MessageEndpoint endpoint2 = (MessageEndpoint) context.getBean("endpoint2");
TargetEndpoint endpoint2 = (TargetEndpoint) context.getBean("endpoint2");
assertEquals(14, endpoint2.getConcurrencyPolicy().getCoreSize());
assertEquals(17, endpoint2.getConcurrencyPolicy().getMaxSize());
}

22
spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java → spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.adapter;
package org.springframework.integration.endpoint;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@@ -33,15 +33,15 @@ import org.springframework.integration.scheduling.PollingSchedule;
/**
* @author Mark Fisher
*/
public class PollingSourceAdapterTests {
public class PollingSourceEndpointTests {
@Test
public void testPolledSourceSendsToChannel() {
TestSource source = new TestSource("testing", 1);
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(100);
PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule);
adapter.run();
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
endpoint.run();
Message<?> message = channel.receive(1000);
assertNotNull("message should not be null", message);
assertEquals("testing.1", message.getPayload());
@@ -53,16 +53,16 @@ public class PollingSourceAdapterTests {
QueueChannel channel = new QueueChannel(1);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule);
adapter.setSendTimeout(10);
adapter.run();
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
endpoint.setSendTimeout(10);
endpoint.run();
Message<?> message1 = channel.receive(1000);
assertNotNull("message should not be null", message1);
assertEquals("testing.1", message1.getPayload());
Message<?> message2 = channel.receive(0);
assertNull("second message should be null", message2);
source.resetCounter();
adapter.run();
endpoint.run();
Message<?> message3 = channel.receive(100);
assertNotNull("third message should not be null", message3);
assertEquals("testing.1", message3.getPayload());
@@ -74,9 +74,9 @@ public class PollingSourceAdapterTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule);
adapter.setMaxMessagesPerTask(5);
adapter.run();
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(5);
endpoint.run();
Message<?> message1 = channel.receive(0);
assertNotNull("message should not be null", message1);
assertEquals("testing.1", message1.getPayload());