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

Removed the "NEXT_TARGET" header, since it is no longer used by the Router implementations. Also, the MessagingAnnotationPostProcessor now considers any component annotated with a "stereotype" as a candidate for post-processing.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-28 20:49:45 +00:00
parent ba9c086aea
commit 37b4ee5bd7
8 changed files with 128 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -138,14 +138,12 @@ public class DefaultMessageBusTests {
QueueChannel outputChannel2 = new QueueChannel();
MessageHandler handler1 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
return MessageBuilder.fromMessage(message)
.setNextTarget("output1").build();
return MessageBuilder.fromMessage(message).build();
}
};
MessageHandler handler2 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
return MessageBuilder.fromMessage(message)
.setNextTarget("output2").build();
return MessageBuilder.fromMessage(message).build();
}
};
MessageBus bus = new DefaultMessageBus();
@@ -158,9 +156,11 @@ public class DefaultMessageBusTests {
DefaultEndpoint<MessageHandler> endpoint1 = new DefaultEndpoint<MessageHandler>(handler1);
endpoint1.setBeanName("testEndpoint1");
endpoint1.setSource(inputChannel);
endpoint1.setOutputChannel(outputChannel1);
DefaultEndpoint<MessageHandler> endpoint2 = new DefaultEndpoint<MessageHandler>(handler2);
endpoint2.setBeanName("testEndpoint2");
endpoint2.setSource(inputChannel);
endpoint2.setOutputChannel(outputChannel2);
bus.registerEndpoint(endpoint1);
bus.registerEndpoint(endpoint2);
bus.start();
@@ -179,16 +179,14 @@ public class DefaultMessageBusTests {
final CountDownLatch latch = new CountDownLatch(2);
MessageHandler handler1 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
Message<?> reply = MessageBuilder.fromMessage(message)
.setNextTarget("output1").build();
Message<?> reply = MessageBuilder.fromMessage(message).build();
latch.countDown();
return reply;
}
};
MessageHandler handler2 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
Message<?> reply = MessageBuilder.fromMessage(message)
.setNextTarget("output2").build();
Message<?> reply = MessageBuilder.fromMessage(message).build();
latch.countDown();
return reply;
}
@@ -203,9 +201,11 @@ public class DefaultMessageBusTests {
DefaultEndpoint<MessageHandler> endpoint1 = new DefaultEndpoint<MessageHandler>(handler1);
endpoint1.setBeanName("testEndpoint1");
endpoint1.setSource(inputChannel);
endpoint1.setOutputChannel(outputChannel1);
DefaultEndpoint<MessageHandler> endpoint2 = new DefaultEndpoint<MessageHandler>(handler2);
endpoint2.setBeanName("testEndpoint2");
endpoint2.setSource(inputChannel);
endpoint2.setOutputChannel(outputChannel2);
bus.registerEndpoint(endpoint1);
bus.registerEndpoint(endpoint2);
bus.start();

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

@@ -38,7 +38,6 @@ import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.message.selector.MessageSelectorChain;
@@ -62,6 +61,21 @@ public class DefaultEndpointTests {
assertEquals("FOO", reply.getPayload());
}
@Test
public void outputChannelTakesPrecedence() {
QueueChannel channel1 = new QueueChannel(1);
QueueChannel channel2 = new QueueChannel(1);
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(new TestHandler());
endpoint.setOutputChannel(channel1);
Message<?> message = MessageBuilder.fromPayload("foo").setReturnAddress(channel2).build();
endpoint.send(message);
Message<?> reply1 = channel1.receive(0);
assertNotNull(reply1);
assertEquals("FOO", reply1.getPayload());
Message<?> reply2 = channel2.receive(0);
assertNull(reply2);
}
@Test
public void returnAddressHeader() {
QueueChannel channel = new QueueChannel(1);
@@ -74,25 +88,6 @@ public class DefaultEndpointTests {
assertEquals("FOO", reply.getPayload());
}
@Test
public void nextTargetHeaderTakesPrecedence() {
QueueChannel channel1 = new QueueChannel(1);
QueueChannel channel2 = new QueueChannel(1);
QueueChannel channel3 = new QueueChannel(1);
MessageHandler handler = new TestNextTargetSettingHandler(channel1);
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(handler);
endpoint.setOutputChannel(channel2);
Message<?> message = MessageBuilder.fromPayload("foo").setReturnAddress(channel3).build();
endpoint.send(message);
Message<?> reply1 = channel1.receive(0);
assertNotNull(reply1);
assertEquals("foo", reply1.getPayload());
Message<?> reply2 = channel2.receive(0);
assertNull(reply2);
Message<?> reply3 = channel3.receive(0);
assertNull(reply3);
}
@Test
public void returnAddressHeaderWithChannelName() {
QueueChannel channel = new QueueChannel(1);
@@ -109,29 +104,6 @@ public class DefaultEndpointTests {
assertEquals("FOO", reply.getPayload());
}
@Test
public void nextTargetHeaderWithChannelName() {
QueueChannel channel1 = new QueueChannel(1);
QueueChannel channel2 = new QueueChannel(1);
QueueChannel channel3 = new QueueChannel(1);
channel1.setBeanName("testChannel");
ChannelRegistry channelRegistry = new DefaultMessageBus();
channelRegistry.registerChannel(channel1);
MessageHandler handler = new TestNextTargetSettingHandler("testChannel");
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(handler);
endpoint.setChannelRegistry(channelRegistry);
endpoint.setOutputChannel(channel2);
Message<?> message = MessageBuilder.fromPayload("foo").setReturnAddress(channel3).build();
endpoint.send(message);
Message<?> reply1 = channel1.receive(0);
assertNotNull(reply1);
assertEquals("foo", reply1.getPayload());
Message<?> reply2 = channel2.receive(0);
assertNull(reply2);
Message<?> reply3 = channel3.receive(0);
assertNull(reply3);
}
@Test
public void dynamicReplyChannel() throws Exception {
final QueueChannel replyChannel1 = new QueueChannel();
@@ -176,19 +148,6 @@ public class DefaultEndpointTests {
assertEquals("FOO", reply.getPayload());
}
@Test
public void unknownNextTargetChannelFallsBackToOutputChannel() {
QueueChannel channel = new QueueChannel(1);
MessageHandler handler = new TestHandler();
DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(handler);
endpoint.setOutputChannel(channel);
Message<?> message = MessageBuilder.fromPayload("foo").setNextTarget("unknown").build();
endpoint.send(message);
Message<?> reply = channel.receive(0);
assertNotNull(reply);
assertEquals("FOO", reply.getPayload());
}
@Test(expected = MessageEndpointReplyException.class)
public void noReplyTarget() {
MessageHandler handler = new TestHandler();
@@ -360,32 +319,6 @@ public class DefaultEndpointTests {
}
@Test
public void nextTargetNotPropagatedPastCurrentEndpoint() {
final QueueChannel intermediateItemChannel = new QueueChannel(1);
final QueueChannel finalChannel = new QueueChannel(1);
DefaultEndpoint<MessageHandler> primaryEndpoint = new DefaultEndpoint<MessageHandler>(new MessageHandler() {
public Message<?> handle(Message<?> message) {
return MessageBuilder.fromMessage(message).setNextTarget(intermediateItemChannel).build();
}
});
DefaultEndpoint<MessageHandler> secondaryEndpoint = new DefaultEndpoint<MessageHandler>(new MessageHandler() {
public Message<?> handle(Message<?> message) {
return message;
}
});
secondaryEndpoint.setOutputChannel(finalChannel);
Message<String> message = MessageBuilder.fromPayload("test").build();
primaryEndpoint.send(message);
Message<?> reply = intermediateItemChannel.receive(500);
secondaryEndpoint.send(reply);
Message<?> replyOnIntermediateChannel = intermediateItemChannel.receive(500);
assertNull(replyOnIntermediateChannel);
Message<?> replyOnFinalChannel = finalChannel.receive(500);
assertNotNull(replyOnFinalChannel);
}
private static class TestHandler implements MessageHandler {
public Message<?> handle(Message<?> message) {
@@ -394,25 +327,6 @@ public class DefaultEndpointTests {
}
private static class TestNextTargetSettingHandler implements MessageHandler {
private final Object nextTarget;
TestNextTargetSettingHandler(Object nextTarget) {
this.nextTarget = nextTarget;
}
public Message<?> handle(Message<?> message) {
if (nextTarget instanceof MessageTarget) {
return MessageBuilder.fromPayload(message.getPayload())
.setNextTarget((MessageTarget) nextTarget).build();
}
return MessageBuilder.fromPayload(message.getPayload())
.setNextTarget((String) nextTarget).build();
}
}
private static class TestNullReplyHandler implements MessageHandler {
public Message<?> handle(Message<?> message) {

96
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java
View File

@@ -35,24 +35,54 @@ import org.springframework.integration.message.StringMessage;
public class ReturnAddressTests {
@Test
public void testNextTargetOverrides() {
public void returnAddressFallbackWithChannelReference() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"returnAddressTests.xml", this.getClass());
MessageChannel channel1 = (MessageChannel) context.getBean("channel1WithOverride");
PollableChannel replyChannel = (PollableChannel) context.getBean("replyChannel");
MessageChannel channel3 = (MessageChannel) context.getBean("channel3");
PollableChannel channel5 = (PollableChannel) context.getBean("channel5");
context.start();
Message<String> message = MessageBuilder.fromPayload("*")
.setNextTarget("replyChannel").build();
channel1.send(message);
Message<?> response = replyChannel.receive(3000);
.setReturnAddress(channel5).build();
channel3.send(message);
Message<?> response = channel5.receive(3000);
assertNotNull(response);
PollableChannel outputChannel = (PollableChannel) context.getBean("channel2");
assertNull(outputChannel.receive(0));
assertEquals("**", response.getPayload());
}
@Test
public void testOutputTakesPrecedenceByDefault() {
public void returnAddressFallbackWithChannelName() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"returnAddressTests.xml", this.getClass());
MessageChannel channel3 = (MessageChannel) context.getBean("channel3");
PollableChannel channel5 = (PollableChannel) context.getBean("channel5");
context.start();
Message<String> message = MessageBuilder.fromPayload("*")
.setReturnAddress("channel5").build();
channel3.send(message);
Message<?> response = channel5.receive(3000);
assertNotNull(response);
assertEquals("**", response.getPayload());
}
@Test
public void returnAddressWithChannelReferenceAfterMultipleEndpoints() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"returnAddressTests.xml", this.getClass());
MessageChannel channel1 = (MessageChannel) context.getBean("channel1");
PollableChannel replyChannel = (PollableChannel) context.getBean("replyChannel");
context.start();
Message<String> message = MessageBuilder.fromPayload("*")
.setReturnAddress(replyChannel).build();
channel1.send(message);
Message<?> response = replyChannel.receive(3000);
assertNotNull(response);
assertEquals("********", response.getPayload());
PollableChannel channel2 = (PollableChannel) context.getBean("channel2");
assertNull(channel2.receive(0));
}
@Test
public void returnAddressWithChannelNameAfterMultipleEndpoints() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"returnAddressTests.xml", this.getClass());
MessageChannel channel1 = (MessageChannel) context.getBean("channel1");
@@ -64,10 +94,25 @@ public class ReturnAddressTests {
Message<?> response = replyChannel.receive(3000);
assertNotNull(response);
assertEquals("********", response.getPayload());
PollableChannel channel2 = (PollableChannel) context.getBean("channel2");
assertNull(channel2.receive(0));
}
@Test
public void testOutputTakesPrecedenceAndNoReturnAddress() {
public void returnAddressFallbackButNotAvailable() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"returnAddressTests.xml", this.getClass());
MessageChannel channel3 = (MessageChannel) context.getBean("channel3");
PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel");
context.start();
StringMessage message = new StringMessage("*");
channel3.send(message);
Message<?> errorMessage = errorChannel.receive(3000);
assertNotNull(errorMessage.getPayload());
}
@Test
public void outputChannelWithNoReturnAddress() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"returnAddressTests.xml", this.getClass());
MessageChannel channel4 = (MessageChannel) context.getBean("channel4");
@@ -81,29 +126,20 @@ public class ReturnAddressTests {
}
@Test
public void testReturnAddressFallbackButNotAvailable() {
public void outputChannelTakesPrecedence() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"returnAddressTests.xml", this.getClass());
MessageChannel channel3 = (MessageChannel) context.getBean("channel3");
PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel");
MessageChannel channel4 = (MessageChannel) context.getBean("channel4");
PollableChannel replyChannel = (PollableChannel) context.getBean("replyChannel");
context.start();
StringMessage message = new StringMessage("*");
channel3.send(message);
Message<?> errorMessage = errorChannel.receive(3000);
assertNotNull(errorMessage.getPayload());
}
@Test
public void testOutputFallbackButNotAvailable() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"returnAddressTests.xml", this.getClass());
MessageChannel channel3 = (MessageChannel) context.getBean("channel3WithOverride");
PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel");
context.start();
StringMessage message = new StringMessage("*");
channel3.send(message);
Message<?> errorMessage = errorChannel.receive(3000);
assertNotNull(errorMessage.getPayload());
Message<String> message = MessageBuilder.fromPayload("*")
.setReturnAddress("channel5").build();
channel4.send(message);
Message<?> response = replyChannel.receive(3000);
assertNotNull(response);
assertEquals("**", response.getPayload());
PollableChannel channel5 = (PollableChannel) context.getBean("channel5");
assertNull(channel5.receive(0));
}
}

5
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/returnAddressTests.xml
View File

@@ -13,13 +13,10 @@
<si:channel id="channel2"/>
<si:channel id="channel3"/>
<si:channel id="channel4"/>
<si:channel id="channel1WithOverride"/>
<si:channel id="channel3WithOverride"/>
<si:channel id="channel5"/>
<si:channel id="replyChannel"/>
<si:channel id="errorChannel"/>
<si:service-activator input-channel="channel1WithOverride" ref="testBean" method="duplicate" output-channel="channel2"/>
<si:service-activator input-channel="channel3WithOverride" ref="testBean" method="duplicate"/>
<si:service-activator input-channel="channel1" ref="testBean" method="duplicate" output-channel="channel2"/>
<si:service-activator input-channel="channel2" ref="testBean" method="duplicate" output-channel="channel3"/>
<si:service-activator input-channel="channel3" ref="testBean" method="duplicate"/>