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

ServiceActivatorEndpoint is now ServiceActivatingConsumer.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-14 15:06:18 +00:00
parent 10fd9304f0
commit 5f6e6277cd
11 changed files with 85 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,7 +31,7 @@ import org.springframework.integration.config.annotation.MessagingAnnotationPost
import org.springframework.integration.config.xml.MessageBusParser;
import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.endpoint.ReplyMessageHolder;
import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
import org.springframework.integration.endpoint.ServiceActivatingConsumer;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
@@ -67,7 +67,7 @@ public class DirectChannelSubscriptionTests {
@Test
public void sendAndReceiveForRegisteredEndpoint() {
GenericApplicationContext context = new GenericApplicationContext();
ServiceActivatorEndpoint serviceActivator = new ServiceActivatorEndpoint(new TestBean(), "handle");
ServiceActivatingConsumer serviceActivator = new ServiceActivatingConsumer(new TestBean(), "handle");
serviceActivator.setOutputChannel(targetChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(serviceActivator, sourceChannel);
context.getBeanFactory().registerSingleton("testEndpoint", endpoint);

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

@@ -22,7 +22,7 @@
<constructor-arg ref="sourceChannel"/>
</bean>
<bean id="serviceActivator" class="org.springframework.integration.endpoint.ServiceActivatorEndpoint">
<bean id="serviceActivator" class="org.springframework.integration.endpoint.ServiceActivatingConsumer">
<constructor-arg ref="handler"/>
<property name="outputChannel" ref="targetChannel"/>
</bean>

60
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java
View File

@@ -27,7 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
import org.springframework.integration.endpoint.ServiceActivatingConsumer;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageDeliveryException;
@@ -45,7 +45,7 @@ public class SimpleDispatcherTests {
public void singleMessage() throws InterruptedException {
SimpleDispatcher dispatcher = new SimpleDispatcher();
final CountDownLatch latch = new CountDownLatch(1);
dispatcher.addConsumer(createEndpoint(TestHandlers.countDownHandler(latch)));
dispatcher.addConsumer(createConsumer(TestHandlers.countDownHandler(latch)));
dispatcher.dispatch(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -57,8 +57,8 @@ public class SimpleDispatcherTests {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
dispatcher.addConsumer(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
dispatcher.addConsumer(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
dispatcher.addConsumer(createConsumer(TestHandlers.countingCountDownHandler(counter1, latch)));
dispatcher.addConsumer(createConsumer(TestHandlers.countingCountDownHandler(counter2, latch)));
dispatcher.dispatch(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -161,21 +161,21 @@ public class SimpleDispatcherTests {
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final AtomicInteger selectorCounter = new AtomicInteger();
AbstractReplyProducingMessageConsumer endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch));
AbstractReplyProducingMessageConsumer endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch));
AbstractReplyProducingMessageConsumer endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch));
endpoint1.setSelector(new TestMessageSelector(selectorCounter, false));
endpoint2.setSelector(new TestMessageSelector(selectorCounter, false));
endpoint3.setSelector(new TestMessageSelector(selectorCounter, true));
dispatcher.addConsumer(endpoint1);
dispatcher.addConsumer(endpoint2);
dispatcher.addConsumer(endpoint3);
AbstractReplyProducingMessageConsumer consumer1 = createConsumer(TestHandlers.countingCountDownHandler(counter1, latch));
AbstractReplyProducingMessageConsumer consumer2 = createConsumer(TestHandlers.countingCountDownHandler(counter2, latch));
AbstractReplyProducingMessageConsumer consumer3 = createConsumer(TestHandlers.countingCountDownHandler(counter3, latch));
consumer1.setSelector(new TestMessageSelector(selectorCounter, false));
consumer2.setSelector(new TestMessageSelector(selectorCounter, false));
consumer3.setSelector(new TestMessageSelector(selectorCounter, true));
dispatcher.addConsumer(consumer1);
dispatcher.addConsumer(consumer2);
dispatcher.addConsumer(consumer3);
dispatcher.dispatch(new StringMessage("test"));
assertEquals(0, latch.getCount());
assertEquals("selectors should have been invoked one time each", 3, selectorCounter.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter1.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter2.get());
assertEquals("handler with accepting selector should have received the message", 1, counter3.get());
assertEquals("consumer with rejecting selector should not have received the message", 0, counter1.get());
assertEquals("consumer with rejecting selector should not have received the message", 0, counter2.get());
assertEquals("consumer with accepting selector should have received the message", 1, counter3.get());
}
@Test
@@ -186,15 +186,15 @@ public class SimpleDispatcherTests {
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final AtomicInteger selectorCounter = new AtomicInteger();
AbstractReplyProducingMessageConsumer endpoint1 = createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch));
AbstractReplyProducingMessageConsumer endpoint2 = createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch));
AbstractReplyProducingMessageConsumer endpoint3 = createEndpoint(TestHandlers.countingCountDownHandler(counter3, latch));
endpoint1.setSelector(new TestMessageSelector(selectorCounter, false));
endpoint2.setSelector(new TestMessageSelector(selectorCounter, false));
endpoint3.setSelector(new TestMessageSelector(selectorCounter, false));
dispatcher.addConsumer(endpoint1);
dispatcher.addConsumer(endpoint2);
dispatcher.addConsumer(endpoint3);
AbstractReplyProducingMessageConsumer consumer1 = createConsumer(TestHandlers.countingCountDownHandler(counter1, latch));
AbstractReplyProducingMessageConsumer consumer2 = createConsumer(TestHandlers.countingCountDownHandler(counter2, latch));
AbstractReplyProducingMessageConsumer consumer3 = createConsumer(TestHandlers.countingCountDownHandler(counter3, latch));
consumer1.setSelector(new TestMessageSelector(selectorCounter, false));
consumer2.setSelector(new TestMessageSelector(selectorCounter, false));
consumer3.setSelector(new TestMessageSelector(selectorCounter, false));
dispatcher.addConsumer(consumer1);
dispatcher.addConsumer(consumer2);
dispatcher.addConsumer(consumer3);
boolean exceptionThrown = false;
try {
dispatcher.dispatch(new StringMessage("test"));
@@ -204,9 +204,9 @@ public class SimpleDispatcherTests {
}
assertTrue(exceptionThrown);
assertEquals("selectors should have been invoked one time each", 3, selectorCounter.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter1.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter2.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter3.get());
assertEquals("consumer with rejecting selector should not have received the message", 0, counter1.get());
assertEquals("consumer with rejecting selector should not have received the message", 0, counter2.get());
assertEquals("consumer with rejecting selector should not have received the message", 0, counter3.get());
}
@Test
@@ -257,8 +257,8 @@ public class SimpleDispatcherTests {
}
private static ServiceActivatorEndpoint createEndpoint(Object handler) {
return new ServiceActivatorEndpoint(handler);
private static ServiceActivatingConsumer createConsumer(Object object) {
return new ServiceActivatingConsumer(object);
}

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

@@ -41,7 +41,7 @@ public class CorrelationIdTests {
.setCorrelationId(correlationId).build();
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel(1);
ServiceActivatorEndpoint serviceActivator = new ServiceActivatorEndpoint(new TestBean(), "upperCase");
ServiceActivatingConsumer serviceActivator = new ServiceActivatingConsumer(new TestBean(), "upperCase");
serviceActivator.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(serviceActivator, inputChannel);
endpoint.start();
@@ -56,7 +56,7 @@ public class CorrelationIdTests {
.setCorrelationId("correlationId").build();
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel(1);
ServiceActivatorEndpoint serviceActivator = new ServiceActivatorEndpoint(new TestBean(), "upperCase");
ServiceActivatingConsumer serviceActivator = new ServiceActivatingConsumer(new TestBean(), "upperCase");
serviceActivator.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(serviceActivator, inputChannel);
endpoint.start();
@@ -73,7 +73,7 @@ public class CorrelationIdTests {
.setCorrelationId(correlationId).build();
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel(1);
ServiceActivatorEndpoint serviceActivator = new ServiceActivatorEndpoint(new TestBean(), "createMessage");
ServiceActivatingConsumer serviceActivator = new ServiceActivatingConsumer(new TestBean(), "createMessage");
serviceActivator.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(serviceActivator, inputChannel);
endpoint.start();
@@ -87,7 +87,7 @@ public class CorrelationIdTests {
Message<?> message = new StringMessage("test");
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel(1);
ServiceActivatorEndpoint serviceActivator = new ServiceActivatorEndpoint(new TestBean(), "createMessage");
ServiceActivatingConsumer serviceActivator = new ServiceActivatingConsumer(new TestBean(), "createMessage");
serviceActivator.setOutputChannel(outputChannel);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(serviceActivator, inputChannel);
endpoint.start();

36
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java
View File

@@ -49,7 +49,7 @@ public class ServiceActivatorEndpointTests {
@Test
public void outputChannel() {
QueueChannel channel = new QueueChannel(1);
ServiceActivatorEndpoint endpoint = this.createEndpoint();
ServiceActivatingConsumer endpoint = this.createEndpoint();
endpoint.setOutputChannel(channel);
Message<?> message = MessageBuilder.withPayload("foo").build();
endpoint.onMessage(message);
@@ -62,7 +62,7 @@ public class ServiceActivatorEndpointTests {
public void outputChannelTakesPrecedence() {
QueueChannel channel1 = new QueueChannel(1);
QueueChannel channel2 = new QueueChannel(1);
ServiceActivatorEndpoint endpoint = this.createEndpoint();
ServiceActivatingConsumer endpoint = this.createEndpoint();
endpoint.setOutputChannel(channel1);
Message<?> message = MessageBuilder.withPayload("foo").setReturnAddress(channel2).build();
endpoint.onMessage(message);
@@ -76,7 +76,7 @@ public class ServiceActivatorEndpointTests {
@Test
public void returnAddressHeader() {
QueueChannel channel = new QueueChannel(1);
ServiceActivatorEndpoint endpoint = this.createEndpoint();
ServiceActivatingConsumer endpoint = this.createEndpoint();
Message<?> message = MessageBuilder.withPayload("foo").setReturnAddress(channel).build();
endpoint.onMessage(message);
Message<?> reply = channel.receive(0);
@@ -90,7 +90,7 @@ public class ServiceActivatorEndpointTests {
channel.setBeanName("testChannel");
TestChannelResolver channelResolver = new TestChannelResolver();
channelResolver.addChannel(channel);
ServiceActivatorEndpoint endpoint = this.createEndpoint();
ServiceActivatingConsumer endpoint = this.createEndpoint();
endpoint.setChannelResolver(channelResolver);
Message<?> message = MessageBuilder.withPayload("foo").setReturnAddress("testChannel").build();
endpoint.onMessage(message);
@@ -110,7 +110,7 @@ public class ServiceActivatorEndpointTests {
return new StringMessage("foo" + message.getPayload());
}
};
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(handler, "handle");
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(handler, "handle");
TestChannelResolver channelResolver = new TestChannelResolver();
channelResolver.addChannel(replyChannel2);
endpoint.setChannelResolver(channelResolver);
@@ -135,7 +135,7 @@ public class ServiceActivatorEndpointTests {
@Test
public void noOutputChannelFallsBackToReturnAddress() {
QueueChannel channel = new QueueChannel(1);
ServiceActivatorEndpoint endpoint = this.createEndpoint();
ServiceActivatingConsumer endpoint = this.createEndpoint();
Message<?> message = MessageBuilder.withPayload("foo").setReturnAddress(channel).build();
endpoint.onMessage(message);
Message<?> reply = channel.receive(0);
@@ -145,7 +145,7 @@ public class ServiceActivatorEndpointTests {
@Test(expected = MessagingException.class)
public void noReplyTarget() {
ServiceActivatorEndpoint endpoint = this.createEndpoint();
ServiceActivatingConsumer endpoint = this.createEndpoint();
Message<?> message = MessageBuilder.withPayload("foo").build();
endpoint.onMessage(message);
}
@@ -153,7 +153,7 @@ public class ServiceActivatorEndpointTests {
@Test
public void noReplyMessage() {
QueueChannel channel = new QueueChannel(1);
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(
new TestNullReplyBean(), "handle");
endpoint.setOutputChannel(channel);
Message<?> message = MessageBuilder.withPayload("foo").build();
@@ -164,7 +164,7 @@ public class ServiceActivatorEndpointTests {
@Test(expected = MessageHandlingException.class)
public void noReplyMessageWithRequiresReply() {
QueueChannel channel = new QueueChannel(1);
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(
new TestNullReplyBean(), "handle");
endpoint.setRequiresReply(true);
endpoint.setOutputChannel(channel);
@@ -174,7 +174,7 @@ public class ServiceActivatorEndpointTests {
@Test(expected=MessageRejectedException.class)
public void endpointWithSelectorRejecting() {
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(
TestHandlers.nullHandler(), "handle");
endpoint.setSelector(new MessageSelector() {
public boolean accept(Message<?> message) {
@@ -187,7 +187,7 @@ public class ServiceActivatorEndpointTests {
@Test
public void endpointWithSelectorAccepting() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(
TestHandlers.countDownHandler(latch), "handle");
endpoint.setSelector(new MessageSelector() {
public boolean accept(Message<?> message) {
@@ -202,7 +202,7 @@ public class ServiceActivatorEndpointTests {
@Test
public void endpointWithMultipleSelectorsAndFirstRejects() {
final AtomicInteger counter = new AtomicInteger();
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(
TestHandlers.countingHandler(counter), "handle");
MessageSelectorChain selectorChain = new MessageSelectorChain();
selectorChain.add(new MessageSelector() {
@@ -233,7 +233,7 @@ public class ServiceActivatorEndpointTests {
public void endpointWithMultipleSelectorsAndFirstAccepts() {
final AtomicInteger selectorCounter = new AtomicInteger();
AtomicInteger handlerCounter = new AtomicInteger();
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(
TestHandlers.countingHandler(handlerCounter), "handle");
MessageSelectorChain selectorChain = new MessageSelectorChain();
selectorChain.add(new MessageSelector() {
@@ -264,7 +264,7 @@ public class ServiceActivatorEndpointTests {
@Test
public void endpointWithMultipleSelectorsAndBothAccept() {
final AtomicInteger counter = new AtomicInteger();
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(
TestHandlers.countingHandler(counter), "handle");
MessageSelectorChain selectorChain = new MessageSelectorChain();
selectorChain.add(new MessageSelector() {
@@ -287,7 +287,7 @@ public class ServiceActivatorEndpointTests {
@Test
public void correlationIdNotSetIfMessageIsReturnedUnaltered() {
QueueChannel replyChannel = new QueueChannel(1);
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(new Object() {
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(new Object() {
@SuppressWarnings("unused")
public Message<?> handle(Message<?> message) {
return message;
@@ -303,7 +303,7 @@ public class ServiceActivatorEndpointTests {
@Test
public void correlationIdSetByHandlerTakesPrecedence() {
QueueChannel replyChannel = new QueueChannel(1);
ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(new Object() {
ServiceActivatingConsumer endpoint = new ServiceActivatingConsumer(new Object() {
@SuppressWarnings("unused")
public Message<?> handle(Message<?> message) {
return MessageBuilder.fromMessage(message)
@@ -320,8 +320,8 @@ public class ServiceActivatorEndpointTests {
}
private ServiceActivatorEndpoint createEndpoint() {
return new ServiceActivatorEndpoint(new TestBean(), "handle");
private ServiceActivatingConsumer createEndpoint() {
return new ServiceActivatingConsumer(new TestBean(), "handle");
}

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

@@ -33,7 +33,7 @@ public class ServiceActivatorMethodResolutionTests {
@Test
public void singleAnnotationMatches() {
SingleAnnotationTestBean testBean = new SingleAnnotationTestBean();
ServiceActivatorEndpoint serviceActivator = new ServiceActivatorEndpoint(testBean);
ServiceActivatingConsumer serviceActivator = new ServiceActivatingConsumer(testBean);
QueueChannel outputChannel = new QueueChannel();
serviceActivator.setOutputChannel(outputChannel);
serviceActivator.onMessage(new StringMessage("foo"));
@@ -44,13 +44,13 @@ public class ServiceActivatorMethodResolutionTests {
@Test(expected = IllegalArgumentException.class)
public void multipleAnnotationFails() {
MultipleAnnotationTestBean testBean = new MultipleAnnotationTestBean();
new ServiceActivatorEndpoint(testBean);
new ServiceActivatingConsumer(testBean);
}
@Test
public void singlePublicMethodMatches() {
SinglePublicMethodTestBean testBean = new SinglePublicMethodTestBean();
ServiceActivatorEndpoint serviceActivator = new ServiceActivatorEndpoint(testBean);
ServiceActivatingConsumer serviceActivator = new ServiceActivatingConsumer(testBean);
QueueChannel outputChannel = new QueueChannel();
serviceActivator.setOutputChannel(outputChannel);
serviceActivator.onMessage(new StringMessage("foo"));
@@ -61,7 +61,7 @@ public class ServiceActivatorMethodResolutionTests {
@Test(expected = IllegalArgumentException.class)
public void multiplePublicMethodFails() {
MultiplePublicMethodTestBean testBean = new MultiplePublicMethodTestBean();
new ServiceActivatorEndpoint(testBean);
new ServiceActivatingConsumer(testBean);
}

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

@@ -32,7 +32,7 @@ import org.springframework.aop.framework.ProxyFactory;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
import org.springframework.integration.endpoint.ServiceActivatingConsumer;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.message.StringMessage;
@@ -82,7 +82,7 @@ public class DefaultMethodResolverTests {
ProxyFactory proxyFactory = new ProxyFactory(testBean);
proxyFactory.setProxyTargetClass(false);
testBean = (GreetingService) proxyFactory.getProxy();
ServiceActivatorEndpoint consumer = new ServiceActivatorEndpoint(testBean);
ServiceActivatingConsumer consumer = new ServiceActivatingConsumer(testBean);
consumer.setOutputChannel(output);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(consumer, input);
endpoint.start();
@@ -98,7 +98,7 @@ public class DefaultMethodResolverTests {
ProxyFactory proxyFactory = new ProxyFactory(testBean);
proxyFactory.setProxyTargetClass(true);
testBean = (GreetingService) proxyFactory.getProxy();
ServiceActivatorEndpoint consumer = new ServiceActivatorEndpoint(testBean);
ServiceActivatingConsumer consumer = new ServiceActivatingConsumer(testBean);
consumer.setOutputChannel(output);
SubscribingConsumerEndpoint endpoint = new SubscribingConsumerEndpoint(consumer, input);
endpoint.start();