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

Removed the registerHandler() method from MessageBus.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-13 01:16:32 +00:00
parent f56334f06b
commit 5c9ac92f6b
8 changed files with 68 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -46,13 +46,10 @@ import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.factory.ChannelFactory;
import org.springframework.integration.channel.factory.QueueChannelFactory;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.AbstractEndpoint;
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.MessagingGateway;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollableSource;
@@ -258,27 +255,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
return this.channelRegistry.unregisterChannel(name);
}
public void registerHandler(String name, MessageHandler handler, Object input, Schedule schedule) {
Assert.notNull(handler, "'handler' must not be null");
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
this.configureEndpoint(endpoint, name, input, schedule);
this.registerEndpoint(endpoint);
}
private void configureEndpoint(AbstractEndpoint endpoint, String name, Object input, Schedule schedule) {
endpoint.setName(name);
if (input instanceof MessageChannel) {
endpoint.setSource((MessageChannel) input);
}
else if (input instanceof String) {
endpoint.setInputChannelName((String) input);
}
else {
throw new ConfigurationException("'input' must be a MessageChannel or String");
}
endpoint.setSchedule(schedule);
}
public void registerEndpoint(MessageEndpoint endpoint) {
if (!this.initialized) {
this.initialize();

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

@@ -22,8 +22,6 @@ import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.factory.ChannelFactory;
import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.scheduling.Schedule;
/**
* The message bus interface.
@@ -36,6 +34,4 @@ public interface MessageBus extends ChannelRegistry, EndpointRegistry, Lifecycle
ChannelFactory getChannelFactory();
void registerHandler(String name, MessageHandler handler, Object input, Schedule schedule);
}

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

@@ -28,7 +28,8 @@ import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.Subscriber;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.handler.DefaultMessageHandlerAdapter;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.handler.DefaultMessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
@@ -85,13 +86,16 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor {
Annotation annotation = method.getAnnotation(subscriberAnnotationType);
if (annotation != null) {
String channelName = (String) AnnotationUtils.getValue(annotation, channelNameAttribute);
DefaultMessageHandlerAdapter adapter = new DefaultMessageHandlerAdapter();
adapter.setMethodName(method.getName());
adapter.setObject(bean);
adapter.afterPropertiesSet();
String adapterName = ClassUtils.getShortNameAsProperty(targetClass) +
"-" + method.getName() + "-endpoint";
messageBus.registerHandler(adapterName, adapter, channelName, null);
DefaultMessageHandler handler = new DefaultMessageHandler();
handler.setObject(bean);
handler.setMethod(method);
handler.afterPropertiesSet();
String endpointName = ClassUtils.getShortNameAsProperty(targetClass) +
"." + method.getName() + ".endpoint";
SimpleEndpoint<DefaultMessageHandler> endpoint = new SimpleEndpoint<DefaultMessageHandler>(handler);
endpoint.setBeanName(endpointName);
endpoint.setInputChannelName(channelName);
messageBus.registerEndpoint(endpoint);
}
}
});

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

@@ -38,6 +38,7 @@ public class DefaultEndpointRegistry implements EndpointRegistry {
public void registerEndpoint(MessageEndpoint endpoint) {
Assert.notNull(endpoint, "'endpoint' must not be null");
Assert.notNull(endpoint.getName(), "endpoint name must not be null");
this.endpoints.put(endpoint.getName(), endpoint);
}

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

@@ -34,6 +34,7 @@ import org.springframework.integration.channel.PollableChannelAdapter;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dispatcher.PublishSubscribeChannel;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
@@ -63,7 +64,10 @@ public class DefaultMessageBusTests {
return message;
}
};
bus.registerHandler("handler", handler, sourceChannel, null);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(handler);
endpoint.setBeanName("testEndpoint");
endpoint.setSource(sourceChannel);
bus.registerEndpoint(endpoint);
bus.start();
Message<?> result = targetChannel.receive(3000);
assertEquals("test", result.getPayload());
@@ -85,7 +89,10 @@ public class DefaultMessageBusTests {
return message;
}
};
bus.registerHandler("handler", handler, "sourceChannel", null);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(handler);
endpoint.setBeanName("testEndpoint");
endpoint.setInputChannelName("sourceChannel");
bus.registerEndpoint(endpoint);
bus.start();
Message<?> result = targetChannel.receive(3000);
assertEquals("test", result.getPayload());
@@ -127,21 +134,27 @@ public class DefaultMessageBusTests {
MessageHandler handler1 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
return MessageBuilder.fromMessage(message)
.setReturnAddress("output1").build();
.setNextTarget("output1").build();
}
};
MessageHandler handler2 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
return MessageBuilder.fromMessage(message)
.setReturnAddress("output2").build();
.setNextTarget("output2").build();
}
};
MessageBus bus = new DefaultMessageBus();
bus.registerChannel("input", inputChannel);
bus.registerChannel("output1", outputChannel1);
bus.registerChannel("output2", outputChannel2);
bus.registerHandler("handler1", handler1, inputChannel, null);
bus.registerHandler("handler2", handler2, inputChannel, null);
SimpleEndpoint<MessageHandler> endpoint1 = new SimpleEndpoint<MessageHandler>(handler1);
endpoint1.setBeanName("testEndpoint1");
endpoint1.setSource(inputChannel);
SimpleEndpoint<MessageHandler> endpoint2 = new SimpleEndpoint<MessageHandler>(handler2);
endpoint2.setBeanName("testEndpoint2");
endpoint2.setSource(inputChannel);
bus.registerEndpoint(endpoint1);
bus.registerEndpoint(endpoint2);
bus.start();
inputChannel.send(new StringMessage("testing"));
Message<?> message1 = outputChannel1.receive(500);
@@ -159,7 +172,7 @@ public class DefaultMessageBusTests {
MessageHandler handler1 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
Message<?> reply = MessageBuilder.fromMessage(message)
.setReturnAddress("output1").build();
.setNextTarget("output1").build();
latch.countDown();
return reply;
}
@@ -167,7 +180,7 @@ public class DefaultMessageBusTests {
MessageHandler handler2 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
Message<?> reply = MessageBuilder.fromMessage(message)
.setReturnAddress("output2").build();
.setNextTarget("output2").build();
latch.countDown();
return reply;
}
@@ -176,8 +189,14 @@ public class DefaultMessageBusTests {
bus.registerChannel("input", inputChannel);
bus.registerChannel("output1", outputChannel1);
bus.registerChannel("output2", outputChannel2);
bus.registerHandler("handler1", handler1, inputChannel, null);
bus.registerHandler("handler2", handler2, inputChannel, null);
SimpleEndpoint<MessageHandler> endpoint1 = new SimpleEndpoint<MessageHandler>(handler1);
endpoint1.setBeanName("testEndpoint1");
endpoint1.setSource(inputChannel);
SimpleEndpoint<MessageHandler> endpoint2 = new SimpleEndpoint<MessageHandler>(handler2);
endpoint2.setBeanName("testEndpoint2");
endpoint2.setSource(inputChannel);
bus.registerEndpoint(endpoint1);
bus.registerEndpoint(endpoint2);
bus.start();
inputChannel.send(new StringMessage("testing"));
latch.await(500, TimeUnit.MILLISECONDS);
@@ -240,7 +259,10 @@ public class DefaultMessageBusTests {
return null;
}
};
bus.registerHandler("testHandler", handler, MessageBus.ERROR_CHANNEL_NAME, null);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(handler);
endpoint.setBeanName("testEndpoint");
endpoint.setInputChannelName(MessageBus.ERROR_CHANNEL_NAME);
bus.registerEndpoint(endpoint);
bus.start();
errorChannel.send(new ErrorMessage(new RuntimeException("test-exception")));
latch.await(1000, TimeUnit.MILLISECONDS);

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

@@ -41,6 +41,8 @@ import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.config.ChannelParserTests;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.TestHandlers;
import org.springframework.integration.scheduling.spi.ProviderTaskScheduler;
@@ -73,7 +75,10 @@ public class MessageBusParserTests {
ApplicationContext context = new ClassPathXmlApplicationContext(
"messageBusWithDefaults.xml", this.getClass());
MessageBus bus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
bus.registerHandler("handler", TestHandlers.nullHandler(), "unknownChannel", null);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(TestHandlers.nullHandler());
endpoint.setBeanName("testEndpoint");
endpoint.setInputChannelName("unknownChannel");
bus.registerEndpoint(endpoint);
}
@Test
@@ -81,7 +86,10 @@ public class MessageBusParserTests {
ApplicationContext context = new ClassPathXmlApplicationContext(
"messageBusWithAutoCreateChannels.xml", this.getClass());
MessageBus bus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
bus.registerHandler("handler", TestHandlers.nullHandler(), "channelToCreate", null);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(TestHandlers.nullHandler());
endpoint.setBeanName("testEndpoint");
endpoint.setInputChannelName("channelToCreate");
bus.registerEndpoint(endpoint);
bus.start();
assertNotNull(bus.lookupChannel("channelToCreate"));
bus.stop();

9
org.springframework.integration/src/test/java/org/springframework/integration/handler/MethodInvokingTargetTests.java
View File

@@ -31,6 +31,8 @@ import org.springframework.integration.ConfigurationException;
import org.springframework.integration.bus.DefaultMessageBus;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
@@ -97,9 +99,12 @@ public class MethodInvokingTargetTests {
assertNull(queue.poll());
MessageBus bus = new DefaultMessageBus();
bus.registerChannel("channel", channel);
bus.registerHandler("targetAdapter", target, channel, null);
MessageEndpoint endpoint = new SimpleEndpoint<MethodInvokingTarget>(target);
endpoint.setBeanName("testEndpoint");
endpoint.setSource(channel);
bus.registerEndpoint(endpoint);
bus.start();
String result = queue.poll(500, TimeUnit.MILLISECONDS);
String result = queue.poll(1000, TimeUnit.MILLISECONDS);
assertNotNull(result);
assertEquals("testing", result);
bus.stop();

6
org.springframework.integration/src/test/java/org/springframework/integration/message/MessageExchangeTemplateTests.java
View File

@@ -29,6 +29,7 @@ import org.junit.Test;
import org.springframework.integration.bus.DefaultMessageBus;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
@@ -52,7 +53,10 @@ public class MessageExchangeTemplateTests {
};
MessageBus bus = new DefaultMessageBus();
bus.registerChannel("requestChannel", requestChannel);
bus.registerHandler("testHandler", testHandler, requestChannel, null);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>(testHandler);
endpoint.setBeanName("testEndpoint");
endpoint.setSource(requestChannel);
bus.registerEndpoint(endpoint);
bus.start();
}