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

Renamed SubscribingConsumerEndpoint to EventDrivenConsumer.

Browse Source
This commit is contained in:
Mark Fisher
2008-11-03 14:41:53 +00:00
parent 3e2d09e5d7
commit 515ed698a6
21 changed files with 78 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -28,7 +28,7 @@ import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.Trigger;
@@ -146,7 +146,7 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar
if (channel instanceof SubscribableChannel) {
Assert.isNull(trigger, "A trigger should not be specified for endpoint '" + this.beanName
+ "', since '" + this.inputChannelName + "' is a SubscribableChannel (not pollable).");
this.endpoint = new SubscribingConsumerEndpoint((SubscribableChannel) channel, this.handler);
this.endpoint = new EventDrivenConsumer((SubscribableChannel) channel, this.handler);
}
else if (channel instanceof PollableChannel) {
if (this.trigger == null) {

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

@@ -32,7 +32,7 @@ import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -100,7 +100,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
else if (inputChannel instanceof SubscribableChannel) {
Assert.isTrue(pollerAnnotation == null,
"The @Poller annotation should only be provided for a PollableChannel");
endpoint = new SubscribingConsumerEndpoint((SubscribableChannel) inputChannel, handler);
endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);
}
else {
throw new IllegalArgumentException("unsupported channel type: [" + inputChannel.getClass() + "]");

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

@@ -33,7 +33,7 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.Trigger;
@@ -120,7 +120,7 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
return endpoint;
}
if (channel instanceof SubscribableChannel) {
return new SubscribingConsumerEndpoint((SubscribableChannel) channel, handler);
return new EventDrivenConsumer((SubscribableChannel) channel, handler);
}
return null;
}

7
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SubscribingConsumerEndpoint.java → org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java
View File

@@ -22,9 +22,12 @@ import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
/**
* Message Endpoint that connects any {@link MessageHandler} implementation
* to a {@link SubscribableChannel}.
*
* @author Mark Fisher
*/
public class SubscribingConsumerEndpoint extends AbstractEndpoint implements Lifecycle {
public class EventDrivenConsumer extends AbstractEndpoint implements Lifecycle {
private final SubscribableChannel inputChannel;
@@ -35,7 +38,7 @@ public class SubscribingConsumerEndpoint extends AbstractEndpoint implements Lif
private final Object lifecycleMonitor = new Object();
public SubscribingConsumerEndpoint(SubscribableChannel inputChannel, MessageHandler handler) {
public EventDrivenConsumer(SubscribableChannel inputChannel, MessageHandler handler) {
Assert.notNull(inputChannel, "inputChannel must not be null");
Assert.notNull(handler, "handler must not be null");
this.inputChannel = inputChannel;

4
org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java
View File

@@ -28,7 +28,7 @@ import org.springframework.integration.core.MessagingException;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.MessagingGateway;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
@@ -185,7 +185,7 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
}
};
if (this.replyChannel instanceof SubscribableChannel) {
correlator = new SubscribingConsumerEndpoint(
correlator = new EventDrivenConsumer(
(SubscribableChannel) this.replyChannel, handler);
}
else if (this.replyChannel instanceof PollableChannel) {

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

@@ -42,7 +42,7 @@ import org.springframework.integration.consumer.ReplyMessageHolder;
import org.springframework.integration.core.Message;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.MessageBuilder;
@@ -195,8 +195,8 @@ public class ApplicationContextMessageBusTests {
context.getBeanFactory().registerSingleton("output2", outputChannel2);
handler1.setOutputChannel(outputChannel1);
handler2.setOutputChannel(outputChannel2);
SubscribingConsumerEndpoint endpoint1 = new SubscribingConsumerEndpoint(inputChannel, handler1);
SubscribingConsumerEndpoint endpoint2 = new SubscribingConsumerEndpoint(inputChannel, handler2);
EventDrivenConsumer endpoint1 = new EventDrivenConsumer(inputChannel, handler1);
EventDrivenConsumer endpoint2 = new EventDrivenConsumer(inputChannel, handler2);
context.getBeanFactory().registerSingleton("testEndpoint1", endpoint1);
context.getBeanFactory().registerSingleton("testEndpoint2", endpoint2);
ApplicationContextMessageBus bus = new ApplicationContextMessageBus();

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

@@ -34,7 +34,7 @@ import org.springframework.integration.consumer.ReplyMessageHolder;
import org.springframework.integration.consumer.ServiceActivatingHandler;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.util.TestUtils;
@@ -69,7 +69,7 @@ public class DirectChannelSubscriptionTests {
GenericApplicationContext context = new GenericApplicationContext();
ServiceActivatingHandler serviceActivator = new ServiceActivatingHandler(new TestBean(), "handle");
serviceActivator.setOutputChannel(targetChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(sourceChannel, serviceActivator);
EventDrivenConsumer endpoint = new EventDrivenConsumer(sourceChannel, serviceActivator);
context.getBeanFactory().registerSingleton("testEndpoint", endpoint);
bus.setApplicationContext(context);
context.refresh();
@@ -104,7 +104,7 @@ public class DirectChannelSubscriptionTests {
}
};
handler.setOutputChannel(targetChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(sourceChannel, handler);
EventDrivenConsumer endpoint = new EventDrivenConsumer(sourceChannel, handler);
context.getBeanFactory().registerSingleton("testEndpoint", endpoint);
bus.setApplicationContext(context);
context.refresh();

10
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
View File

@@ -34,7 +34,7 @@ import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.util.MethodInvoker;
import org.springframework.integration.util.TestUtils;
@@ -73,8 +73,8 @@ public class AggregatorParserTests {
@Test
public void testPropertyAssignment() throws Exception {
SubscribingConsumerEndpoint endpoint =
(SubscribingConsumerEndpoint) context.getBean("completelyDefinedAggregator");
EventDrivenConsumer endpoint =
(EventDrivenConsumer) context.getBean("completelyDefinedAggregator");
CompletionStrategy completionStrategy = (CompletionStrategy) context.getBean("completionStrategy");
MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel");
@@ -135,8 +135,8 @@ public class AggregatorParserTests {
@Test
public void testAggregatorWithPojoCompletionStrategy() {
MessageChannel input = (MessageChannel) context.getBean("aggregatorWithPojoCompletionStrategyInput");
SubscribingConsumerEndpoint endpoint =
(SubscribingConsumerEndpoint) context.getBean("aggregatorWithPojoCompletionStrategy");
EventDrivenConsumer endpoint =
(EventDrivenConsumer) context.getBean("aggregatorWithPojoCompletionStrategy");
CompletionStrategy completionStrategy = (CompletionStrategy) new DirectFieldAccessor(
new DirectFieldAccessor(endpoint).getPropertyValue("handler")).getPropertyValue("completionStrategy");
Assert.assertTrue(completionStrategy instanceof CompletionStrategyAdapter);

6
org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java
View File

@@ -32,7 +32,7 @@ import org.springframework.integration.config.xml.MessageBusParser;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.StringMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
@@ -54,7 +54,7 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests
assertNotNull(channelResolver.resolveChannelName(beanName));
Object adapter = this.applicationContext.getBean(beanName + ".adapter");
assertNotNull(adapter);
assertTrue(adapter instanceof SubscribingConsumerEndpoint);
assertTrue(adapter instanceof EventDrivenConsumer);
TestConsumer consumer = (TestConsumer) this.applicationContext.getBean("consumer");
assertNull(consumer.getLastMessage());
Message<?> message = new StringMessage("test");
@@ -75,7 +75,7 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests
assertNotNull(channelResolver.resolveChannelName(beanName));
Object adapter = this.applicationContext.getBean(beanName + ".adapter");
assertNotNull(adapter);
assertTrue(adapter instanceof SubscribingConsumerEndpoint);
assertTrue(adapter instanceof EventDrivenConsumer);
TestBean testBean = (TestBean) this.applicationContext.getBean("testBean");
assertNull(testBean.getMessage());
Message<?> message = new StringMessage("consumer test");

6
org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

@@ -34,7 +34,7 @@ import org.springframework.integration.aggregator.Resequencer;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.MessageBuilder;
/**
@@ -75,7 +75,7 @@ public class ResequencerParserTests {
@Test
public void testDefaultResequencerProperties() {
SubscribingConsumerEndpoint endpoint = (SubscribingConsumerEndpoint) context.getBean("defaultResequencer");
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("defaultResequencer");
Resequencer resequencer = (Resequencer) new DirectFieldAccessor(endpoint).getPropertyValue("handler");
assertNull(getPropertyValue(resequencer, "outputChannel"));
assertNull(getPropertyValue(resequencer, "discardChannel"));
@@ -95,7 +95,7 @@ public class ResequencerParserTests {
@Test
public void testPropertyAssignment() throws Exception {
SubscribingConsumerEndpoint endpoint = (SubscribingConsumerEndpoint) context.getBean("completelyDefinedResequencer");
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("completelyDefinedResequencer");
MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel");
Resequencer resequencer = (Resequencer) new DirectFieldAccessor(endpoint).getPropertyValue("handler");

4
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

@@ -36,7 +36,7 @@ import org.springframework.integration.aggregator.CompletionStrategyAdapter;
import org.springframework.integration.aggregator.SequenceSizeCompletionStrategy;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
/**
* @author Marius Bogoevici
@@ -102,7 +102,7 @@ public class AggregatorAnnotationTests {
private AbstractMessageAggregator getAggregator(ApplicationContext context, final String endpointName) {
SubscribingConsumerEndpoint endpoint = (SubscribingConsumerEndpoint) context.getBean(
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean(
endpointName + ".aggregatingMethod.aggregator");
return (AbstractMessageAggregator) new DirectFieldAccessor(endpoint).getPropertyValue("handler");
}

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

@@ -44,7 +44,7 @@ public class CorrelationIdTests {
QueueChannel outputChannel = new QueueChannel(1);
ServiceActivatingHandler serviceActivator = new ServiceActivatingHandler(new TestBean(), "upperCase");
serviceActivator.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(inputChannel, serviceActivator);
EventDrivenConsumer endpoint = new EventDrivenConsumer(inputChannel, serviceActivator);
endpoint.start();
assertTrue(inputChannel.send(message));
Message<?> reply = outputChannel.receive(0);
@@ -59,7 +59,7 @@ public class CorrelationIdTests {
QueueChannel outputChannel = new QueueChannel(1);
ServiceActivatingHandler serviceActivator = new ServiceActivatingHandler(new TestBean(), "upperCase");
serviceActivator.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(inputChannel, serviceActivator);
EventDrivenConsumer endpoint = new EventDrivenConsumer(inputChannel, serviceActivator);
endpoint.start();
assertTrue(inputChannel.send(message));
Message<?> reply = outputChannel.receive(0);
@@ -76,7 +76,7 @@ public class CorrelationIdTests {
QueueChannel outputChannel = new QueueChannel(1);
ServiceActivatingHandler serviceActivator = new ServiceActivatingHandler(new TestBean(), "createMessage");
serviceActivator.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(inputChannel, serviceActivator);
EventDrivenConsumer endpoint = new EventDrivenConsumer(inputChannel, serviceActivator);
endpoint.start();
assertTrue(inputChannel.send(message));
Message<?> reply = outputChannel.receive(0);
@@ -90,7 +90,7 @@ public class CorrelationIdTests {
QueueChannel outputChannel = new QueueChannel(1);
ServiceActivatingHandler serviceActivator = new ServiceActivatingHandler(new TestBean(), "createMessage");
serviceActivator.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(inputChannel, serviceActivator);
EventDrivenConsumer endpoint = new EventDrivenConsumer(inputChannel, serviceActivator);
endpoint.start();
assertTrue(inputChannel.send(message));
Message<?> reply = outputChannel.receive(0);

6
org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java
View File

@@ -26,7 +26,7 @@ import org.junit.Test;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.filter.MessageFilter;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.selector.MessageSelector;
@@ -73,7 +73,7 @@ public class MessageFilterTests {
}
});
filter.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(inputChannel, filter);
EventDrivenConsumer endpoint = new EventDrivenConsumer(inputChannel, filter);
endpoint.start();
Message<?> message = new StringMessage("test");
assertTrue(inputChannel.send(message));
@@ -92,7 +92,7 @@ public class MessageFilterTests {
}
});
filter.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(inputChannel, filter);
EventDrivenConsumer endpoint = new EventDrivenConsumer(inputChannel, filter);
endpoint.start();
Message<?> message = new StringMessage("test");
assertTrue(inputChannel.send(message));

4
org.springframework.integration/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java
View File

@@ -28,7 +28,7 @@ import org.junit.Test;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.MessageBuilder;
/**
@@ -85,7 +85,7 @@ public class DefaultSplitterTests {
QueueChannel outputChannel = new QueueChannel(1);
DefaultMessageSplitter splitter = new DefaultMessageSplitter();
splitter.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(inputChannel, splitter);
EventDrivenConsumer endpoint = new EventDrivenConsumer(inputChannel, splitter);
endpoint.start();
assertTrue(inputChannel.send(message));
Message<?> reply = outputChannel.receive(0);

6
org.springframework.integration/src/test/java/org/springframework/integration/util/DefaultMethodResolverTests.java
View File

@@ -33,7 +33,7 @@ import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.consumer.ServiceActivatingHandler;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.StringMessage;
/**
@@ -84,7 +84,7 @@ public class DefaultMethodResolverTests {
testBean = (GreetingService) proxyFactory.getProxy();
ServiceActivatingHandler handler = new ServiceActivatingHandler(testBean);
handler.setOutputChannel(output);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(input, handler);
EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler);
endpoint.start();
input.send(new StringMessage("proxy"));
assertEquals("hello proxy", output.receive(0).getPayload());;
@@ -100,7 +100,7 @@ public class DefaultMethodResolverTests {
testBean = (GreetingService) proxyFactory.getProxy();
ServiceActivatingHandler handler = new ServiceActivatingHandler(testBean);
handler.setOutputChannel(output);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(input, handler);
EventDrivenConsumer endpoint = new EventDrivenConsumer(input, handler);
endpoint.start();
input.send(new StringMessage("proxy"));
assertEquals("hello proxy", output.receive(0).getPayload());;