diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java index 5258e5fa4c..1efaf5ba5f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java @@ -16,7 +16,11 @@ package org.springframework.integration.config.annotation; +import java.lang.annotation.Annotation; +import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; +import java.util.List; import java.util.Map; import org.springframework.aop.support.AopUtils; @@ -37,6 +41,7 @@ import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.stereotype.Component; import org.springframework.util.Assert; /** @@ -81,10 +86,11 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { Object originalBean = bean; Class beanClass = this.getBeanClass(bean); - MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(beanClass, MessageEndpoint.class); - if (endpointAnnotation == null) { + if (!this.isStereotype(beanClass)) { + // we only post-process stereotype components return bean; } + MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(beanClass, MessageEndpoint.class); for (Map.Entry, AnnotationMethodPostProcessor> entry : this.postProcessors.entrySet()) { AnnotationMethodPostProcessor postProcessor = entry.getValue(); bean = postProcessor.postProcess(bean, beanName, beanClass); @@ -138,4 +144,19 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init return (targetClass != null) ? targetClass : bean.getClass(); } + private boolean isStereotype(Class beanClass) { + List annotations = new ArrayList(Arrays.asList(beanClass.getAnnotations())); + Class[] interfaces = beanClass.getInterfaces(); + for (Class iface : interfaces) { + annotations.addAll(Arrays.asList(iface.getAnnotations())); + } + for (Annotation annotation : annotations) { + Class annotationType = annotation.annotationType(); + if (annotationType.equals(Component.class) || annotationType.isAnnotationPresent(Component.class)) { + return true; + } + } + return false; + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java index acd76e3cb0..8acd2f89cb 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java @@ -44,14 +44,12 @@ import org.springframework.util.Assert; * *

The reply target is resolved according to the following order: *

    - *
  1. the 'nextTarget' header value of the reply Message
    2. *
  2. the 'outputChannel' of this Message Endpoint
    3. *
  3. the 'returnAddress' header value of the request Message
    4. *
- * For the 'nextTarget' and 'returnAddress' values, either a - * {@link MessageTarget} instance or String is accepted. If the - * value is a String, then the endpoint will consult its - * {@link ChannelRegistry} (typically provided by the MessageBus). + * For the 'returnAddress' value, either a {@link MessageTarget} instance + * or String is accepted. If the value is a String, then the endpoint will + * consult its {@link ChannelRegistry} (typically provided by the MessageBus). * If no reply target can be determined for a non-null reply Message, * a {@link MessageEndpointReplyException} will be thrown. * @@ -150,7 +148,6 @@ public class DefaultEndpoint extends AbstractRequestRe replyMessage = MessageBuilder.fromMessage(replyMessage) .copyHeadersIfAbsent(requestMessage.getHeaders()) .setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, requestMessage.getHeaders().getId()) - .setNextTarget((String)null) .build(); if (!this.getMessageExchangeTemplate().send(replyMessage, replyTarget)) { throw new MessageEndpointReplyException(replyMessage, requestMessage, @@ -201,26 +198,18 @@ public class DefaultEndpoint extends AbstractRequestRe } private MessageTarget resolveReplyTarget(Message replyMessage, MessageHeaders requestHeaders) { - MessageTarget replyTarget = this.resolveTargetAttribute(replyMessage.getHeaders().getNextTarget()); + MessageTarget replyTarget = this.getTarget(); if (replyTarget == null) { - replyTarget = this.getTarget(); - } - if (replyTarget == null) { - replyTarget = this.resolveTargetAttribute(requestHeaders.getReturnAddress()); - } - return replyTarget; - } - - private MessageTarget resolveTargetAttribute(Object targetAttribute) { - MessageTarget replyTarget = null; - if (targetAttribute != null) { - if (targetAttribute instanceof MessageTarget) { - replyTarget = (MessageTarget) targetAttribute; - } - else if (targetAttribute instanceof String) { - ChannelRegistry registry = this.getChannelRegistry(); - if (registry != null) { - replyTarget = registry.lookupChannel((String) targetAttribute); + Object returnAddress = requestHeaders.getReturnAddress(); + if (returnAddress != null) { + if (returnAddress instanceof MessageTarget) { + replyTarget = (MessageTarget) returnAddress; + } + else if (returnAddress instanceof String) { + ChannelRegistry registry = this.getChannelRegistry(); + if (registry != null) { + replyTarget = registry.lookupChannel((String) returnAddress); + } } } } @@ -228,8 +217,7 @@ public class DefaultEndpoint extends AbstractRequestRe } /** - * Specify the channel where reply Messages should be sent if - * no 'nextTarget' header value is available on the reply Message. + * Specify the channel where reply Messages should be sent. */ public void setOutputChannel(MessageChannel outputChannel) { this.setTarget(outputChannel); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java index 31e1c731b9..a4192b3ed1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java @@ -146,14 +146,6 @@ public final class MessageBuilder { return this.setHeader(MessageHeaders.CORRELATION_ID, correlationId); } - public MessageBuilder setNextTarget(MessageTarget nextTarget) { - return this.setHeader(MessageHeaders.NEXT_TARGET, nextTarget); - } - - public MessageBuilder setNextTarget(String nextTarget) { - return this.setHeader(MessageHeaders.NEXT_TARGET, nextTarget); - } - public MessageBuilder setReturnAddress(MessageTarget returnAddress) { return this.setHeader(MessageHeaders.RETURN_ADDRESS, returnAddress); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java index ad51e2868b..9570d66f0a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java @@ -38,8 +38,6 @@ public final class MessageHeaders implements Map, Serializable { public static final String CORRELATION_ID = "internal.header.correlationId"; - public static final String NEXT_TARGET = "internal.header.nextTarget"; - public static final String RETURN_ADDRESS = "internal.header.returnAddress"; public static final String EXPIRATION_DATE = "internal.header.exprirationDate"; @@ -82,10 +80,6 @@ public final class MessageHeaders implements Map, Serializable { return this.get(RETURN_ADDRESS); } - public Object getNextTarget() { - return this.get(NEXT_TARGET); - } - public Integer getSequenceNumber() { Integer sequenceNumber = this.get(SEQUENCE_NUMBER, Integer.class); return (sequenceNumber != null ? sequenceNumber : 0); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java index efaeeaed03..9edef28c3c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java @@ -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 endpoint1 = new DefaultEndpoint(handler1); endpoint1.setBeanName("testEndpoint1"); endpoint1.setSource(inputChannel); + endpoint1.setOutputChannel(outputChannel1); DefaultEndpoint endpoint2 = new DefaultEndpoint(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 endpoint1 = new DefaultEndpoint(handler1); endpoint1.setBeanName("testEndpoint1"); endpoint1.setSource(inputChannel); + endpoint1.setOutputChannel(outputChannel1); DefaultEndpoint endpoint2 = new DefaultEndpoint(handler2); endpoint2.setBeanName("testEndpoint2"); endpoint2.setSource(inputChannel); + endpoint2.setOutputChannel(outputChannel2); bus.registerEndpoint(endpoint1); bus.registerEndpoint(endpoint2); bus.start(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/DefaultEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/DefaultEndpointTests.java index 2f09f35697..476c69f8ce 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/DefaultEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/DefaultEndpointTests.java @@ -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 endpoint = new DefaultEndpoint(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 endpoint = new DefaultEndpoint(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 endpoint = new DefaultEndpoint(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 endpoint = new DefaultEndpoint(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 primaryEndpoint = new DefaultEndpoint(new MessageHandler() { - public Message handle(Message message) { - return MessageBuilder.fromMessage(message).setNextTarget(intermediateItemChannel).build(); - } - }); - DefaultEndpoint secondaryEndpoint = new DefaultEndpoint(new MessageHandler() { - public Message handle(Message message) { - return message; - } - }); - secondaryEndpoint.setOutputChannel(finalChannel); - Message 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) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java index fe5fe48bc0..0a6e2a316a 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ReturnAddressTests.java @@ -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 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 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 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 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)); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/returnAddressTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/returnAddressTests.xml index acc2945633..a161de0042 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/returnAddressTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/returnAddressTests.xml @@ -13,13 +13,10 @@ - - + - -