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

Removed channel and channel name properties from MessageEndpoint and its implementations. Now the MessageEndpoint only has the 'source' and 'target' properties.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-30 23:55:13 +00:00
parent f96354671c
commit ea16a5f48c
19 changed files with 46 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -50,7 +50,6 @@ import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.MessagingGateway;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.integration.scheduling.PollingSchedule;
@@ -268,29 +267,9 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
if (endpoint instanceof ChannelRegistryAware) {
((ChannelRegistryAware) endpoint).setChannelRegistry(this);
}
MessageTarget target = endpoint.getTarget();
if (target == null) {
String outputChannelName = endpoint.getOutputChannelName();
if (outputChannelName != null) {
target = this.lookupChannel(outputChannelName);
if (target == null) {
throw new ConfigurationException("cannot activate endpoint '" + endpoint +
"', unable to resolve output-channel '" + outputChannelName + "'");
}
endpoint.setTarget(target);
}
}
MessageSource<?> source = endpoint.getSource();
if (source == null) {
String inputChannelName = endpoint.getInputChannelName();
if (inputChannelName != null) {
source = this.lookupChannel(inputChannelName);
if (source == null) {
throw new ConfigurationException("cannot activate endpoint '" + endpoint +
"', unable to resolve input-channel '" + inputChannelName + "'");
}
endpoint.setSource(source);
}
throw new ConfigurationException("endpoint '" + endpoint + "' has no source");
}
if (source != null && source instanceof SubscribableSource) {
((SubscribableSource) source).subscribe(endpoint);

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

@@ -93,7 +93,7 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio
builder.addPropertyReference("source", pollerBeanName);
}
else {
builder.addPropertyValue("inputChannelName", inputChannel);
builder.addPropertyReference("source", inputChannel);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(
builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "target");

6
org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractMessageEndpointParser.java
View File

@@ -100,7 +100,7 @@ public abstract class AbstractMessageEndpointParser extends AbstractSingleBeanDe
builder.addPropertyReference("source", pollerBeanName);
}
else {
builder.addPropertyValue("inputChannelName", inputChannel);
builder.addPropertyReference("source", inputChannel);
}
Element interceptorsElement = DomUtils.getChildElementByTagName(element, INTERCEPTORS_ELEMENT);
if (interceptorsElement != null) {
@@ -108,8 +108,8 @@ public abstract class AbstractMessageEndpointParser extends AbstractSingleBeanDe
ManagedList interceptors = parser.parseInterceptors(interceptorsElement, parserContext);
builder.addPropertyValue("interceptors", interceptors);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(
builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "outputChannelName");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(
builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "target");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, SELECTOR_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, ERROR_HANDLER_ATTRIBUTE);
this.postProcessEndpointBean(builder, element, parserContext);

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

@@ -46,7 +46,6 @@ import org.springframework.integration.handler.config.MessageHandlerCreator;
import org.springframework.integration.router.RouterMessageHandlerCreator;
import org.springframework.integration.splitter.SplitterMessageHandlerCreator;
import org.springframework.integration.transformer.config.TransformerMessageHandlerCreator;
import org.springframework.util.StringUtils;
/**
* Post-processor for the {@link Handler @Handler} annotation.
@@ -127,16 +126,7 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost
public MessageEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) {
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>((MessageHandler) bean);
String outputChannelName = endpointAnnotation.output();
if (StringUtils.hasText(outputChannelName)) {
endpoint.setOutputChannelName(outputChannelName);
}
String inputChannelName = endpointAnnotation.input();
if (StringUtils.hasText(inputChannelName)) {
endpoint.setInputChannelName(inputChannelName);
}
return endpoint;
return new DefaultEndpoint<MessageHandler>((MessageHandler) bean);
}
}

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

@@ -123,6 +123,17 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
throw new ConfigurationException("The @Poller annotation should only be provided for a PollableSource");
}
}
else {
endpoint.setSource(inputChannel);
}
String outputChannelName = endpointAnnotation.output();
if (StringUtils.hasText(outputChannelName)) {
MessageChannel outputChannel = this.messageBus.lookupChannel(outputChannelName);
if (outputChannel == null) {
throw new ConfigurationException("unable to resolve output channel '" + outputChannelName + "'");
}
endpoint.setTarget(outputChannel);
}
this.messageBus.registerEndpoint(endpoint);
}
}

12
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java
View File

@@ -26,13 +26,16 @@ import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Subscriber;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.DefaultEndpoint;
import org.springframework.integration.handler.DefaultMessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* A {@link BeanPostProcessor} that creates a method-invoking handler adapter
@@ -86,6 +89,9 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor {
Annotation annotation = method.getAnnotation(subscriberAnnotationType);
if (annotation != null) {
String channelName = (String) AnnotationUtils.getValue(annotation, channelNameAttribute);
if (!StringUtils.hasText(channelName)) {
throw new ConfigurationException("no channel name provided for subscriber");
}
DefaultMessageHandler handler = new DefaultMessageHandler();
handler.setObject(bean);
handler.setMethod(method);
@@ -94,7 +100,11 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor {
"." + method.getName() + ".endpoint";
DefaultEndpoint<DefaultMessageHandler> endpoint = new DefaultEndpoint<DefaultMessageHandler>(handler);
endpoint.setBeanName(endpointName);
endpoint.setInputChannelName(channelName);
MessageChannel inputChannel = messageBus.lookupChannel(channelName);
if (inputChannel == null) {
throw new ConfigurationException("unable to resolve channel '" + channelName + "' for subscriber");
}
endpoint.setSource(inputChannel);
messageBus.registerEndpoint(endpoint);
}
}

20
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractInOutEndpoint.java
View File

@@ -21,7 +21,6 @@ import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.CompositeMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
@@ -159,23 +158,4 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint {
return replyTarget;
}
// TODO: remove these methods after refactoring
private volatile String inputChannelName;
public String getInputChannelName() {
return this.inputChannelName;
}
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
public String getOutputChannelName() {
if (this.getTarget() instanceof MessageChannel) {
return ((MessageChannel) this.getTarget()).getName();
}
return null;
}
}

39
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractRequestReplyEndpoint.java
View File

@@ -16,41 +16,18 @@
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.
@@ -85,20 +62,4 @@ public abstract class AbstractRequestReplyEndpoint extends AbstractEndpoint {
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);
}
}

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

@@ -23,7 +23,6 @@ import java.util.List;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.CompositeMessage;
import org.springframework.integration.message.Message;
@@ -203,11 +202,4 @@ public class DefaultEndpoint<T extends MessageHandler> extends AbstractRequestRe
return replyTarget;
}
/**
* Specify the channel where reply Messages should be sent.
*/
public void setOutputChannel(MessageChannel outputChannel) {
this.setTarget(outputChannel);
}
}

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

@@ -49,18 +49,4 @@ public class InboundChannelAdapter extends AbstractEndpoint {
}
}
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) {
}
}

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

@@ -37,10 +37,4 @@ public interface MessageEndpoint extends MessageTarget, BeanNameAware {
MessageTarget getTarget();
void setInputChannelName(String inputChannelName);
String getInputChannelName();
String getOutputChannelName();
}

14
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java
View File

@@ -32,18 +32,4 @@ public class OutboundChannelAdapter extends AbstractEndpoint {
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) {
}
}

17
org.springframework.integration/src/main/java/org/springframework/integration/router/RouterEndpoint.java
View File

@@ -87,21 +87,4 @@ public class RouterEndpoint extends AbstractEndpoint {
return sent;
}
// TODO: remove these methods after refactoring
private volatile String inputChannelName;
public String getInputChannelName() {
return this.inputChannelName;
}
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
public String getOutputChannelName() {
return null;
}
}

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

@@ -75,33 +75,6 @@ public class DefaultMessageBusTests {
bus.stop();
}
@Test
public void testRegistrationWithInputChannelName() {
MessageBus bus = new DefaultMessageBus();
QueueChannel sourceChannel = new QueueChannel();
QueueChannel targetChannel = new QueueChannel();
sourceChannel.setBeanName("sourceChannel");
targetChannel.setBeanName("targetChannel");
bus.registerChannel(sourceChannel);
Message<String> message = MessageBuilder.fromPayload("test")
.setReturnAddress("targetChannel").build();
sourceChannel.send(message);
bus.registerChannel(targetChannel);
MessageHandler handler = new MessageHandler() {
public Message<?> handle(Message<?> message) {
return message;
}
};
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(handler);
endpoint.setBeanName("testEndpoint");
endpoint.setInputChannelName("sourceChannel");
bus.registerEndpoint(endpoint);
bus.start();
Message<?> result = targetChannel.receive(3000);
assertEquals("test", result.getPayload());
bus.stop();
}
@Test
public void testChannelsWithoutHandlers() {
MessageBus bus = new DefaultMessageBus();
@@ -156,11 +129,11 @@ public class DefaultMessageBusTests {
DefaultEndpoint<MessageHandler> endpoint1 = new DefaultEndpoint<MessageHandler>(handler1);
endpoint1.setBeanName("testEndpoint1");
endpoint1.setSource(inputChannel);
endpoint1.setOutputChannel(outputChannel1);
endpoint1.setTarget(outputChannel1);
DefaultEndpoint<MessageHandler> endpoint2 = new DefaultEndpoint<MessageHandler>(handler2);
endpoint2.setBeanName("testEndpoint2");
endpoint2.setSource(inputChannel);
endpoint2.setOutputChannel(outputChannel2);
endpoint2.setTarget(outputChannel2);
bus.registerEndpoint(endpoint1);
bus.registerEndpoint(endpoint2);
bus.start();
@@ -201,11 +174,11 @@ public class DefaultMessageBusTests {
DefaultEndpoint<MessageHandler> endpoint1 = new DefaultEndpoint<MessageHandler>(handler1);
endpoint1.setBeanName("testEndpoint1");
endpoint1.setSource(inputChannel);
endpoint1.setOutputChannel(outputChannel1);
endpoint1.setTarget(outputChannel1);
DefaultEndpoint<MessageHandler> endpoint2 = new DefaultEndpoint<MessageHandler>(handler2);
endpoint2.setBeanName("testEndpoint2");
endpoint2.setSource(inputChannel);
endpoint2.setOutputChannel(outputChannel2);
endpoint2.setTarget(outputChannel2);
bus.registerEndpoint(endpoint1);
bus.registerEndpoint(endpoint2);
bus.start();
@@ -267,7 +240,7 @@ public class DefaultMessageBusTests {
};
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(handler);
endpoint.setBeanName("testEndpoint");
endpoint.setInputChannelName(MessageBus.ERROR_CHANNEL_NAME);
endpoint.setSource(errorChannel);
bus.registerEndpoint(endpoint);
bus.start();
errorChannel.send(new ErrorMessage(new RuntimeException("test-exception")));

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

@@ -58,8 +58,8 @@ public class DirectChannelSubscriptionTests {
@Test
public void testSendAndReceiveForRegisteredEndpoint() {
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(new TestHandler());
endpoint.setInputChannelName("sourceChannel");
endpoint.setOutputChannelName("targetChannel");
endpoint.setSource(sourceChannel);
endpoint.setTarget(targetChannel);
endpoint.setBeanName("testEndpoint");
bus.registerEndpoint(endpoint);
bus.start();
@@ -92,8 +92,8 @@ public class DirectChannelSubscriptionTests {
throw new RuntimeException("intentional test failure");
}
});
endpoint.setInputChannelName("sourceChannel");
endpoint.setOutputChannelName("targetChannel");
endpoint.setSource(sourceChannel);
endpoint.setTarget(targetChannel);
endpoint.setBeanName("testEndpoint");
bus.registerEndpoint(endpoint);
bus.start();

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

@@ -13,7 +13,7 @@
<bean id="endpoint" class="org.springframework.integration.endpoint.DefaultEndpoint">
<constructor-arg ref="handler"/>
<property name="source" ref="sourceChannel"/>
<property name="outputChannelName" value="targetChannel"/>
<property name="target" ref="targetChannel"/>
</bean>
<bean id="handler" class="org.springframework.integration.handler.TestHandlers" factory-method="echoHandler"/>
@@ -23,4 +23,5 @@
</bean>
<bean id= "messageBusAwareBean" class="org.springframework.integration.bus.TestMessageBusAwareImpl"/>
</beans>

14
org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java
View File

@@ -40,9 +40,6 @@ import org.springframework.integration.bus.TestMessageBusStopInterceptor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.DefaultEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.TestHandlers;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.spi.ProviderTaskScheduler;
@@ -70,17 +67,6 @@ public class MessageBusParserTests {
assertNotNull("bus should have created a default error channel", bus.getErrorChannel());
}
@Test(expected = ConfigurationException.class)
public void testAutoCreateChannelsDisabledByDefault() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"messageBusWithDefaults.xml", this.getClass());
MessageBus bus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(TestHandlers.nullHandler());
endpoint.setBeanName("testEndpoint");
endpoint.setInputChannelName("unknownChannel");
bus.registerEndpoint(endpoint);
}
@Test
public void testMultipleMessageBusElements() {
boolean exceptionThrown = false;

8
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/simpleAnnotatedEndpointTests.xml
View File

@@ -7,7 +7,9 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<bean id="bus" class="org.springframework.integration.bus.DefaultMessageBus"/>
<integration:message-bus/>
<integration:annotation-driven/>
<integration:channel id="inputChannel"/>
@@ -15,8 +17,4 @@
<bean id="endpoint" class="org.springframework.integration.config.annotation.SimpleAnnotatedEndpoint"/>
<bean class="org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor">
<constructor-arg ref="bus"/>
</bean>
</beans>

8
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/DefaultEndpointTests.java
View File

@@ -53,7 +53,7 @@ public class DefaultEndpointTests {
QueueChannel channel = new QueueChannel(1);
MessageHandler handler = new TestHandler();
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(handler);
endpoint.setOutputChannel(channel);
endpoint.setTarget(channel);
Message<?> message = MessageBuilder.fromPayload("foo").build();
endpoint.send(message);
Message<?> reply = channel.receive(0);
@@ -66,7 +66,7 @@ public class DefaultEndpointTests {
QueueChannel channel1 = new QueueChannel(1);
QueueChannel channel2 = new QueueChannel(1);
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(new TestHandler());
endpoint.setOutputChannel(channel1);
endpoint.setTarget(channel1);
Message<?> message = MessageBuilder.fromPayload("foo").setReturnAddress(channel2).build();
endpoint.send(message);
Message<?> reply1 = channel1.receive(0);
@@ -161,7 +161,7 @@ public class DefaultEndpointTests {
QueueChannel channel = new QueueChannel(1);
MessageHandler handler = new TestNullReplyHandler();
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(handler);
endpoint.setOutputChannel(channel);
endpoint.setTarget(channel);
Message<?> message = MessageBuilder.fromPayload("foo").build();
endpoint.send(message);
assertNull(channel.receive(0));
@@ -173,7 +173,7 @@ public class DefaultEndpointTests {
MessageHandler handler = new TestNullReplyHandler();
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(handler);
endpoint.setRequiresReply(true);
endpoint.setOutputChannel(channel);
endpoint.setTarget(channel);
Message<?> message = MessageBuilder.fromPayload("foo").build();
endpoint.send(message);
}