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

Removing usage of the Subscription object, and resolved issue with 'auto-create' channels for SourceEndpoints (INT-235).

Browse Source
This commit is contained in:
Mark Fisher
2008-07-04 21:31:25 +00:00
parent 0aa00b969b
commit 96b22fd01b
22 changed files with 137 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -180,7 +180,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
Map<String, MessageEndpoint> endpointBeans = (Map<String, MessageEndpoint>) context
.getBeansOfType(MessageEndpoint.class);
for (Map.Entry<String, MessageEndpoint> entry : endpointBeans.entrySet()) {
this.registerEndpoint(entry.getKey(), entry.getValue());
this.registerEndpoint(entry.getValue());
}
}
@@ -241,34 +241,34 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
public void registerHandler(String name, MessageHandler handler, Subscription subscription) {
Assert.notNull(handler, "'handler' must not be null");
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
this.doRegisterEndpoint(name, endpoint, subscription);
endpoint.setName(name);
endpoint.setInputChannelName(subscription.getChannelName());
endpoint.setSchedule(subscription.getSchedule());
this.registerEndpoint(endpoint);
}
public void registerTarget(String name, MessageTarget target, Subscription subscription) {
Assert.notNull(target, "'target' must not be null");
TargetEndpoint endpoint = new TargetEndpoint(target);
this.doRegisterEndpoint(name, endpoint, subscription);
}
private void doRegisterEndpoint(String name, TargetEndpoint endpoint, Subscription subscription) {
endpoint.setName(name);
endpoint.setSubscription(subscription);
this.registerEndpoint(name, endpoint);
endpoint.setInputChannelName(subscription.getChannelName());
endpoint.setSchedule(subscription.getSchedule());
this.registerEndpoint(endpoint);
}
public void registerEndpoint(String name, MessageEndpoint endpoint) {
public void registerEndpoint(MessageEndpoint endpoint) {
if (!this.initialized) {
this.initialize();
}
if (endpoint instanceof ChannelRegistryAware) {
((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry);
}
this.endpointRegistry.registerEndpoint(name, endpoint);
this.endpointRegistry.registerEndpoint(endpoint);
if (this.isRunning()) {
this.activateEndpoint(endpoint);
}
if (logger.isInfoEnabled()) {
logger.info("registered endpoint '" + name + "'");
logger.info("registered endpoint '" + endpoint + "'");
}
}
@@ -310,26 +310,19 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry,
catch (Exception e) {
throw new ConfigurationException("failed to initialize endpoint", e);
}
Schedule schedule = null;
Subscription subscription = endpoint.getSubscription();
if (subscription != null) {
schedule = subscription.getSchedule();
MessageChannel channel = subscription.getChannel();
if (channel == null) {
channel = this.lookupOrCreateChannel(subscription.getChannelName());
}
if (channel != null && channel instanceof Subscribable) {
((Subscribable) channel).subscribe(endpoint);
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '"
+ channel.getName() + "' for endpoint '" + endpoint + "'");
}
return;
}
MessageChannel channel = endpoint.getInputChannel();
if (channel == null) {
channel = this.lookupOrCreateChannel(endpoint.getInputChannelName());
}
if (schedule == null) {
schedule = endpoint.getSchedule();
if (channel != null && channel instanceof Subscribable) {
((Subscribable) channel).subscribe(endpoint);
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '"
+ channel.getName() + "' for endpoint '" + endpoint + "'");
}
return;
}
Schedule schedule = endpoint.getSchedule();
EndpointTrigger trigger = endpoint.getTrigger();
if (trigger == null) {
trigger = new EndpointTrigger(schedule != null ? schedule : this.defaultPollerSchedule);

19
org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractTargetEndpointParser.java
View File

@@ -31,7 +31,6 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.StringUtils;
/**
@@ -43,8 +42,6 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef
private static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
private static final String SUBSCRIPTION_PROPERTY = "subscription";
private static final String SELECTOR_ATTRIBUTE = "selector";
private static final String SELECTOR_PROPERTY = "messageSelector";
@@ -79,7 +76,7 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
this.parseTarget(element, this.getTargetAttributeName(), parserContext, builder);
String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
String inputChannelName = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
Schedule schedule = null;
NodeList childNodes = element.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
@@ -97,15 +94,11 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef
}
}
}
if (StringUtils.hasText(inputChannel)) {
RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class);
subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(inputChannel);
if (schedule != null) {
subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(schedule);
}
String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName));
builder.addPropertyReference(SUBSCRIPTION_PROPERTY, subscriptionBeanName);
if (StringUtils.hasText(inputChannelName)) {
builder.addPropertyValue("inputChannelName", inputChannelName);
}
if (schedule != null) {
builder.addPropertyValue("schedule", schedule);
}
String selectorRef = element.getAttribute(SELECTOR_ATTRIBUTE);
if (StringUtils.hasText(selectorRef)) {

13
org.springframework.integration/src/main/java/org/springframework/integration/config/SourceEndpointParser.java
View File

@@ -63,12 +63,11 @@ public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser {
throw new ConfigurationException("'channel' is required");
}
builder.addConstructorArgReference(source);
builder.addConstructorArgReference(output);
Element scheduleElement = this.getScheduleElement(element);
if (scheduleElement == null) {
throw new ConfigurationException("The <schedule/> sub-element is required for a <source-endpoint/>.");
builder.addPropertyValue("outputChannelName", output);
Element scheduleElement = DomUtils.getChildElementByTagName(element, "schedule");
if (scheduleElement != null) {
builder.addPropertyValue("schedule", this.parseSchedule(scheduleElement));
}
builder.addPropertyValue("schedule", this.parseSchedule(scheduleElement));
Element interceptorsElement = DomUtils.getChildElementByTagName(element, "interceptors");
if (interceptorsElement != null) {
EndpointInterceptorParser parser = new EndpointInterceptorParser();
@@ -90,8 +89,4 @@ public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser {
return schedule;
}
private Element getScheduleElement(Element element) {
return DomUtils.getChildElementByTagName(element, "schedule");
}
}

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

@@ -140,7 +140,8 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost
Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class);
Subscription subscription = this.createSubscription(bean, beanName, endpointAnnotation, polledAnnotation);
if (subscription != null) {
endpoint.setSubscription(subscription);
endpoint.setSchedule(subscription.getSchedule());
endpoint.setInputChannelName(subscription.getChannelName());
}
Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class);
if (concurrencyAnnotation != null) {

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

@@ -84,7 +84,8 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
org.springframework.integration.endpoint.MessageEndpoint endpoint =
postProcessor.createEndpoint(bean, beanName, beanClass, endpointAnnotation);
if (endpoint != null) {
this.messageBus.registerEndpoint(beanName + "." + entry.getKey().getSimpleName() + ".endpoint", endpoint);
endpoint.setName(beanName + "." + entry.getKey().getSimpleName() + ".endpoint");
this.messageBus.registerEndpoint(endpoint);
}
}
}

18
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java
View File

@@ -61,6 +61,7 @@ public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostP
public MessageEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) {
SourceEndpoint endpoint = new SourceEndpoint((MessageSource<?>) bean);
Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class);
int period = polledAnnotation.period();
long initialDelay = polledAnnotation.initialDelay();
@@ -68,15 +69,16 @@ public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostP
PollingSchedule schedule = new PollingSchedule(period);
schedule.setInitialDelay(initialDelay);
schedule.setFixedRate(fixedRate);
String outputChannelName = endpointAnnotation.output();
MessageChannel outputChannel = (StringUtils.hasText(outputChannelName)) ?
this.getMessageBus().lookupChannel(outputChannelName) : null;
if (outputChannel == null) {
outputChannel = new DirectChannel();
this.getMessageBus().registerChannel(beanName + ".output", outputChannel);
}
SourceEndpoint endpoint = new SourceEndpoint((MessageSource<?>) bean, outputChannel);
endpoint.setSchedule(schedule);
String outputChannelName = endpointAnnotation.output();
if (!StringUtils.hasText(outputChannelName)) {
MessageChannel outputChannel = new DirectChannel();
this.getMessageBus().registerChannel(beanName + ".output", outputChannel);
endpoint.setOutputChannel(outputChannel);
}
else {
endpoint.setOutputChannelName(outputChannelName);
}
return endpoint;
}

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

@@ -64,7 +64,8 @@ public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostP
TargetEndpoint endpoint = new TargetEndpoint((MessageTarget) bean);
Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class);
Subscription subscription = this.createSubscription(bean, beanName, endpointAnnotation, polledAnnotation);
endpoint.setSubscription(subscription);
endpoint.setSchedule(subscription.getSchedule());
endpoint.setInputChannelName(subscription.getChannelName());
Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class);
if (concurrencyAnnotation != null) {
ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(),

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

@@ -33,7 +33,6 @@ import org.springframework.integration.handler.MessageHandlerNotRunningException
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Subscription;
/**
* Base class for {@link MessageEndpoint} implementations.
@@ -46,22 +45,26 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
private volatile String name;
private volatile String inputChannelName;
private MessageChannel inputChannel;
private volatile String outputChannelName;
private MessageChannel outputChannel;
private final List<Advice> interceptors = new ArrayList<Advice>();
private volatile boolean autoStartup = true;
private volatile boolean running;
private volatile Schedule schedule;
private volatile EndpointTrigger trigger;
private volatile Subscription subscription;
private volatile String outputChannelName;
private volatile ChannelRegistry channelRegistry;
private volatile boolean autoStartup = true;
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
@@ -93,12 +96,25 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
return this.trigger;
}
public Subscription getSubscription() {
return this.subscription;
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
public void setSubscription(Subscription subscription) {
this.subscription = subscription;
public String getInputChannelName() {
return this.inputChannelName;
}
public void setInputChannel(MessageChannel channel) {
this.inputChannel = channel;
this.inputChannelName = channel.getName();
}
public MessageChannel getInputChannel() {
if (this.inputChannel == null &&
(this.inputChannelName != null && this.channelRegistry != null)) {
this.inputChannel = this.channelRegistry.lookupChannel(this.inputChannelName);
}
return this.inputChannel;
}
/**
@@ -109,17 +125,23 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
this.outputChannelName = outputChannelName;
}
public MessageChannel getOutputChannel() {
if (this.outputChannelName != null && this.channelRegistry != null) {
return this.channelRegistry.lookupChannel(this.outputChannelName);
}
return null;
}
public String getOutputChannelName() {
return this.outputChannelName;
}
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
this.outputChannelName = outputChannel.getName();
}
public MessageChannel getOutputChannel() {
if (this.outputChannel == null &&
(this.outputChannelName != null && this.channelRegistry != null)) {
this.outputChannel = this.channelRegistry.lookupChannel(this.outputChannelName);
}
return this.outputChannel;
}
/**
* Set the channel registry to use for looking up channels by name.
*/

5
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpointRegistry.java
View File

@@ -36,10 +36,9 @@ public class DefaultEndpointRegistry implements EndpointRegistry {
return this.endpoints.get(endpointName);
}
public void registerEndpoint(String name, MessageEndpoint endpoint) {
Assert.notNull(name, "'name' must not be null");
public void registerEndpoint(MessageEndpoint endpoint) {
Assert.notNull(endpoint, "'endpoint' must not be null");
this.endpoints.put(name, endpoint);
this.endpoints.put(endpoint.getName(), endpoint);
}
public MessageEndpoint unregisterEndpoint(String name) {

2
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointRegistry.java
View File

@@ -25,7 +25,7 @@ import java.util.Set;
*/
public interface EndpointRegistry {
void registerEndpoint(String name, MessageEndpoint endpoint);
void registerEndpoint(MessageEndpoint endpoint);
MessageEndpoint unregisterEndpoint(String name);

7
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

@@ -21,7 +21,6 @@ import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Subscription;
/**
* Base interface for message endpoints.
@@ -30,13 +29,17 @@ import org.springframework.integration.scheduling.Subscription;
*/
public interface MessageEndpoint extends MessageTarget, ChannelRegistryAware, InitializingBean {
void setName(String name);
String getName();
Schedule getSchedule();
EndpointTrigger getTrigger();
Subscription getSubscription();
String getInputChannelName();
MessageChannel getInputChannel();
String getOutputChannelName();

17
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java
View File

@@ -16,14 +16,13 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.PollCommand;
import org.springframework.util.Assert;
/**
@@ -36,14 +35,10 @@ public class SourceEndpoint extends AbstractEndpoint {
private final MessageSource<?> source;
private final SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy());
public SourceEndpoint(MessageSource<?> source, MessageChannel channel) {
public SourceEndpoint(MessageSource<?> source) {
Assert.notNull(source, "source must not be null");
Assert.notNull(channel, "channel must not be null");
this.source = source;
this.dispatcher.addTarget(channel);
}
@@ -52,11 +47,15 @@ public class SourceEndpoint extends AbstractEndpoint {
}
public final boolean doInvoke(Message<?> pollCommandMessage) {
if (this.getOutputChannel() == null) {
throw new ConfigurationException(
"no output channel has been configured for source endpoint '" + this.getName() + "'");
}
Message<?> message = this.source.receive();
if (message == null) {
return false;
}
boolean sent = this.dispatcher.send(message);
boolean sent = this.getOutputChannel().send(message);
if (this.source instanceof MessageDeliveryAware) {
if (sent) {
((MessageDeliveryAware) this.source).onSend(message);

19
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java
View File

@@ -22,7 +22,6 @@ import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollCommand;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.Assert;
/**
@@ -34,8 +33,6 @@ public class TargetEndpoint extends AbstractEndpoint {
private volatile MessageTarget target;
private volatile Subscription subscription;
private volatile MessageSelector selector;
private volatile boolean initialized;
@@ -65,14 +62,6 @@ public class TargetEndpoint extends AbstractEndpoint {
this.selector = selector;
}
public Subscription getSubscription() {
return this.subscription;
}
public void setSubscription(Subscription subscription) {
this.subscription = subscription;
}
protected void initialize() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
@@ -88,10 +77,7 @@ public class TargetEndpoint extends AbstractEndpoint {
@Override
protected final boolean doInvoke(Message<?> message) {
if (message.getPayload() instanceof PollCommand) {
MessageChannel channel = this.getSubscription().getChannel();
if (channel == null && this.getSubscription().getChannelName() != null) {
channel = this.getChannelRegistry().lookupChannel(this.getSubscription().getChannelName());
}
MessageChannel channel = this.getInputChannel();
if (channel != null) {
Message<?> receivedMessage = channel.receive(5000);
if (receivedMessage != null) {
@@ -99,8 +85,7 @@ public class TargetEndpoint extends AbstractEndpoint {
}
}
else if (logger.isDebugEnabled()) {
logger.debug("TargetEndpoint unable to resolve channel '"
+ this.getSubscription().getChannelName() + "'");
logger.debug("TargetEndpoint unable to resolve channel '" + this.getInputChannelName() + "'");
}
return false;
}

6
org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java
View File

@@ -32,7 +32,6 @@ import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.scheduling.Subscription;
/**
* A template that facilitates the implementation of request-reply usage
@@ -215,8 +214,9 @@ public class RequestReplyTemplate implements MessageBusAware {
}
ReplyMessageCorrelator correlator = new ReplyMessageCorrelator(10);
HandlerEndpoint endpoint = new HandlerEndpoint(correlator);
endpoint.setSubscription(new Subscription(this.replyChannel));
this.endpointRegistry.registerEndpoint("internal.correlator." + this, endpoint);
endpoint.setInputChannel(this.replyChannel);
endpoint.setName("internal.correlator." + this);
this.endpointRegistry.registerEndpoint(endpoint);
this.replyMessageCorrelator = correlator;
}
}

11
org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java
View File

@@ -33,7 +33,6 @@ import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.Subscription;
/**
* @author Mark Fisher
@@ -57,9 +56,10 @@ public class DirectChannelSubscriptionTests {
@Test
public void testSendAndReceiveForRegisteredEndpoint() {
HandlerEndpoint endpoint = new HandlerEndpoint(new TestHandler());
endpoint.setSubscription(new Subscription("sourceChannel"));
endpoint.setInputChannelName("sourceChannel");
endpoint.setOutputChannelName("targetChannel");
bus.registerEndpoint("testEndpoint", endpoint);
endpoint.setName("testEndpoint");
bus.registerEndpoint(endpoint);
bus.start();
this.sourceChannel.send(new StringMessage("foo"));
Message<?> response = this.targetChannel.receive();
@@ -89,9 +89,10 @@ public class DirectChannelSubscriptionTests {
throw new RuntimeException("intentional test failure");
}
});
endpoint.setSubscription(new Subscription("sourceChannel"));
endpoint.setInputChannelName("sourceChannel");
endpoint.setOutputChannelName("targetChannel");
bus.registerEndpoint("testEndpoint", endpoint);
endpoint.setName("testEndpoint");
bus.registerEndpoint(endpoint);
bus.start();
this.sourceChannel.send(new StringMessage("foo"));
}

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

@@ -28,7 +28,6 @@ import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -169,9 +168,11 @@ public class MessageBusTests {
public void testErrorChannelWithFailedDispatch() throws InterruptedException {
MessageBus bus = new MessageBus();
CountDownLatch latch = new CountDownLatch(1);
SourceEndpoint sourceEndpoint = new SourceEndpoint(new FailingSource(latch), new QueueChannel());
SourceEndpoint sourceEndpoint = new SourceEndpoint(new FailingSource(latch));
sourceEndpoint.setOutputChannel(new QueueChannel());
sourceEndpoint.setSchedule(new PollingSchedule(1000));
bus.registerEndpoint("testEndpoint", sourceEndpoint);
sourceEndpoint.setName("testEndpoint");
bus.registerEndpoint(sourceEndpoint);
bus.start();
latch.await(2000, TimeUnit.MILLISECONDS);
Message<?> message = bus.getErrorChannel().receive(100);

6
org.springframework.integration/src/test/java/org/springframework/integration/bus/messageBusTests.xml
View File

@@ -12,11 +12,7 @@
<bean id="endpoint" class="org.springframework.integration.endpoint.HandlerEndpoint">
<constructor-arg ref="handler"/>
<property name="subscription">
<bean class="org.springframework.integration.scheduling.Subscription">
<constructor-arg ref="sourceChannel"/>
</bean>
</property>
<property name="inputChannel" ref="sourceChannel"/>
<property name="outputChannelName" value="targetChannel"/>
</bean>

2
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -361,7 +361,7 @@ public class MessagingAnnotationPostProcessorTests {
AnnotatedEndpointWithPolledAnnotation endpoint = new AnnotatedEndpointWithPolledAnnotation();
postProcessor.postProcessAfterInitialization(endpoint, "testBean");
HandlerEndpoint processedEndpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint");
Schedule schedule = processedEndpoint.getSubscription().getSchedule();
Schedule schedule = processedEndpoint.getSchedule();
assertEquals(PollingSchedule.class, schedule.getClass());
PollingSchedule pollingSchedule = (PollingSchedule) schedule;
assertEquals(1234, pollingSchedule.getPeriod());

6
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java
View File

@@ -40,7 +40,8 @@ public class SourceEndpointTests {
public void testPolledSourceSendsToChannel() {
TestSource source = new TestSource("testing", 1);
QueueChannel channel = new QueueChannel();
SourceEndpoint endpoint = new SourceEndpoint(source, channel);
SourceEndpoint endpoint = new SourceEndpoint(source);
endpoint.setOutputChannel(channel);
endpoint.afterPropertiesSet();
endpoint.send(new CommandMessage(new PollCommand()));
Message<?> message = channel.receive(1000);
@@ -52,7 +53,8 @@ public class SourceEndpointTests {
public void testAutoStartupDisabled() {
TestSource source = new TestSource("testing", 1);
QueueChannel channel = new QueueChannel();
SourceEndpoint endpoint = new SourceEndpoint(source, channel);
SourceEndpoint endpoint = new SourceEndpoint(source);
endpoint.setOutputChannel(channel);
endpoint.setAutoStartup(false);
endpoint.afterPropertiesSet();
endpoint.send(new CommandMessage(new PollCommand()));

8
org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml
View File

@@ -19,7 +19,7 @@
<property name="methodName" value="foo"/>
</bean>
</constructor-arg>
<constructor-arg ref="channel"/>
<property name="outputChannel" ref="channel"/>
<property name="schedule">
<bean class="org.springframework.integration.scheduling.PollingSchedule">
<constructor-arg value="1000"/>
@@ -34,11 +34,7 @@
<bean id="targetEndpoint" class="org.springframework.integration.endpoint.TargetEndpoint">
<constructor-arg ref="target"/>
<property name="subscription">
<bean class="org.springframework.integration.scheduling.Subscription">
<constructor-arg ref="channel"/>
</bean>
</property>
<property name="inputChannel" ref="channel"/>
</bean>
<bean id="sink" class="org.springframework.integration.handler.TestSink"/>