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

MessageEndpoint now defines getSource() and getTarget() instead of getInputChannel() and getOutputChannel(). EndpointPoller now uses a MessageExchangeTemplate and MessageEndpoint no longer defines poll().

Browse Source
This commit is contained in:
Mark Fisher
2008-07-29 19:32:14 +00:00
parent df079b1206
commit 91226f8a34
6 changed files with 35 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
View File

@@ -54,6 +54,7 @@ import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.MessagingGateway;
import org.springframework.integration.endpoint.TargetEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
@@ -319,21 +320,21 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
private void activateEndpoint(MessageEndpoint endpoint) {
Assert.notNull(endpoint, "'endpoint' must not be null");
if (endpoint.getOutputChannel() == null) {
if (endpoint.getTarget() == null) {
this.lookupOrCreateChannel(endpoint.getOutputChannelName());
}
MessageChannel channel = endpoint.getInputChannel();
if (channel == null) {
channel = this.lookupOrCreateChannel(endpoint.getInputChannelName());
if (channel != null) {
endpoint.setSource(channel);
MessageSource<?> source = endpoint.getSource();
if (source == null) {
source = this.lookupOrCreateChannel(endpoint.getInputChannelName());
if (source != null) {
endpoint.setSource(source);
}
}
if (channel != null && channel instanceof Subscribable) {
((Subscribable) channel).subscribe(endpoint);
if (source != null && source instanceof Subscribable) {
((Subscribable) source).subscribe(endpoint);
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '"
+ channel.getName() + "' for endpoint '" + endpoint + "'");
+ source + "' for endpoint '" + endpoint + "'");
}
return;
}

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

@@ -105,20 +105,14 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
this.source = source;
}
public MessageChannel getInputChannel() {
if (this.source != null) {
if (this.source instanceof MessageChannel) {
return (MessageChannel) this.source;
}
}
else if (this.inputChannelName != null && this.channelRegistry != null) {
public MessageSource<?> getSource() {
if (this.source == null && this.inputChannelName != null && this.channelRegistry != null) {
MessageChannel inputChannel = this.channelRegistry.lookupChannel(this.inputChannelName);
if (inputChannel != null) {
this.source = inputChannel;
}
return inputChannel;
}
return null;
return this.source;
}
/**
@@ -144,20 +138,14 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
this.messageExchangeTemplate.setSendTimeout(sendTimeout);
}
public MessageChannel getOutputChannel() {
if (this.target != null) {
if (this.target instanceof MessageChannel) {
return (MessageChannel) this.target;
}
}
else if (this.outputChannelName != null && this.channelRegistry != null) {
public MessageTarget getTarget() {
if (this.target == null && this.outputChannelName != null && this.channelRegistry != null) {
MessageChannel outputChannel = this.channelRegistry.lookupChannel(this.outputChannelName);
if (outputChannel != null) {
this.target = outputChannel;
}
return outputChannel;
}
return null;
return this.target;
}
/**
@@ -199,7 +187,7 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
this.messageExchangeTemplate.afterPropertiesSet();
}
if (this.target == null) {
this.target = this.getOutputChannel();
this.target = this.getTarget();
}
if (this.target != null && this.target instanceof ChannelRegistryAware
&& this.channelRegistry != null) {

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

@@ -16,13 +16,22 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.message.MessageExchangeTemplate;
/**
* @author Mark Fisher
*/
public class EndpointPoller implements EndpointVisitor {
private final MessageExchangeTemplate template;
public EndpointPoller() {
this.template = new MessageExchangeTemplate();
this.template.setSendTimeout(0);
}
public void visitEndpoint(MessageEndpoint endpoint) {
endpoint.poll();
template.receiveAndForward(endpoint.getSource(), endpoint);
}
}

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

@@ -80,12 +80,12 @@ public class HandlerEndpoint extends AbstractEndpoint {
if (this.returnAddressOverrides) {
MessageTarget target = this.getReturnAddress(originalMessage);
if (target == null) {
target = this.getOutputChannel();
target = this.getTarget();
}
return target;
}
else {
MessageTarget target = this.getOutputChannel();
MessageTarget target = this.getTarget();
if (target == null) {
target = this.getReturnAddress(originalMessage);
}

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

@@ -16,7 +16,6 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.Schedule;
@@ -36,16 +35,14 @@ public interface MessageEndpoint extends MessageTarget {
void setSource(MessageSource<?> source);
MessageSource<?> getSource();
void setTarget(MessageTarget target);
MessageTarget getTarget();
String getInputChannelName();
MessageChannel getInputChannel();
String getOutputChannelName();
MessageChannel getOutputChannel();
boolean poll();
}

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

@@ -37,9 +37,9 @@ public class SourceEndpoint extends AbstractEndpoint {
@Override
public void initialize() {
if (this.getOutputChannel() == null) {
if (this.getTarget() == null) {
throw new ConfigurationException(
"no output channel has been configured for source endpoint '" + this.getName() + "'");
"no output has been configured for source endpoint '" + this.getName() + "'");
}
}