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

InboundChannelAdapter now extends AbstractMessageProducingEndpoint.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-17 23:10:16 +00:00
parent 6e56e8acbe
commit baa1474ef9
5 changed files with 10 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -75,10 +75,10 @@ public class ChannelAdapterParser extends AbstractBeanDefinitionParser {
adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(InboundChannelAdapter.class);
adapterBuilder.addPropertyReference("source", source);
if (StringUtils.hasText(channelName)) {
adapterBuilder.addPropertyReference("channel", channelName);
adapterBuilder.addPropertyReference("outputChannel", channelName);
}
else {
adapterBuilder.addPropertyReference("channel", this.createDirectChannel(element, parserContext));
adapterBuilder.addPropertyReference("outputChannel", this.createDirectChannel(element, parserContext));
}
if (pollerElement != null) {
IntegrationNamespaceUtils.configureSchedule(pollerElement, adapterBuilder);

2
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java
View File

@@ -90,7 +90,7 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
Schedule schedule = this.createSchedule(pollerAnnotation);
InboundChannelAdapter adapter = new InboundChannelAdapter();
adapter.setSource(source);
adapter.setChannel(channel);
adapter.setOutputChannel(channel);
adapter.setSchedule(schedule);
adapter.setBeanName(this.generateUniqueName(channel.getName() + ".inboundAdapter"));
return adapter;

4
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageProducingEndpoint.java
View File

@@ -32,6 +32,10 @@ public class AbstractMessageProducingEndpoint extends AbstractEndpoint {
this.outputChannel = outputChannel;
}
protected MessageChannel getOutputChannel() {
return this.outputChannel;
}
@Override
protected void initialize() {
Assert.notNull(this.outputChannel, "outputChannel is required");

10
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java
View File

@@ -30,12 +30,10 @@ import org.springframework.integration.scheduling.TaskScheduler;
*
* @author Mark Fisher
*/
public class InboundChannelAdapter extends AbstractEndpoint implements Lifecycle {
public class InboundChannelAdapter extends AbstractMessageProducingEndpoint implements Lifecycle {
private volatile PollableSource<?> source;
private volatile MessageChannel channel;
private volatile Schedule schedule;
private volatile SourcePoller poller;
@@ -51,10 +49,6 @@ public class InboundChannelAdapter extends AbstractEndpoint implements Lifecycle
this.source = source;
}
public void setChannel(MessageChannel channel) {
this.channel = channel;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
@@ -75,7 +69,7 @@ public class InboundChannelAdapter extends AbstractEndpoint implements Lifecycle
if (this.running) {
return;
}
this.poller = new SourcePoller(source, channel, schedule);
this.poller = new SourcePoller(source, this.getOutputChannel(), schedule);
if (maxMessagesPerPoll < 0 && source instanceof MethodInvokingSource) {
// 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

2
org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java
View File

@@ -197,7 +197,7 @@ public class DefaultMessageBusTests {
InboundChannelAdapter channelAdapter = new InboundChannelAdapter();
channelAdapter.setSource(new FailingSource(latch));
channelAdapter.setSchedule(new PollingSchedule(1000));
channelAdapter.setChannel(outputChannel);
channelAdapter.setOutputChannel(outputChannel);
channelAdapter.setBeanName("testChannel");
bus.registerEndpoint(channelAdapter);
bus.start();