diff --git a/org.springframework.integration.security/src/main/java/org/springframework/integration/security/endpoint/SecurityEndpointInterceptor.java b/org.springframework.integration.security/src/main/java/org/springframework/integration/security/endpoint/SecurityEndpointInterceptor.java index 49d18b4448..d047fd4aa5 100644 --- a/org.springframework.integration.security/src/main/java/org/springframework/integration/security/endpoint/SecurityEndpointInterceptor.java +++ b/org.springframework.integration.security/src/main/java/org/springframework/integration/security/endpoint/SecurityEndpointInterceptor.java @@ -18,7 +18,6 @@ package org.springframework.integration.security.endpoint; import org.springframework.integration.endpoint.interceptor.EndpointInterceptorAdapter; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; import org.springframework.integration.security.SecurityContextUtils; import org.springframework.security.AccessDecisionManager; import org.springframework.security.ConfigAttributeDefinition; @@ -42,7 +41,7 @@ public class SecurityEndpointInterceptor extends EndpointInterceptorAdapter { } @Override - public boolean aroundSend(Message message, MessageTarget endpoint) { + public Message preHandle(Message message) { SecurityContext securityContext = null; if (message != null) { securityContext = SecurityContextUtils.getSecurityContextFromHeader(message); @@ -51,8 +50,8 @@ public class SecurityEndpointInterceptor extends EndpointInterceptorAdapter { try { SecurityContextHolder.setContext(securityContext); this.accessDecisionManager.decide(SecurityContextHolder.getContext().getAuthentication(), - endpoint, this.targetSecurityAttributes); - return endpoint.send(message); + message, this.targetSecurityAttributes); + return message; } finally { SecurityContextHolder.clearContext(); @@ -60,8 +59,8 @@ public class SecurityEndpointInterceptor extends EndpointInterceptorAdapter { } else { this.accessDecisionManager.decide(SecurityContextHolder.getContext().getAuthentication(), - endpoint, this.targetSecurityAttributes); - return endpoint.send(message); + message, this.targetSecurityAttributes); + return message; } } diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest.java b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest.java index 654df01168..b3c1217ed1 100644 --- a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest.java +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/EndpointSecurityIntegrationTest.java @@ -24,6 +24,7 @@ import org.junit.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.StringMessage; import org.springframework.integration.security.SecurityTestUtil; import org.springframework.security.AccessDeniedException; @@ -67,9 +68,14 @@ public class EndpointSecurityIntegrationTest extends AbstractJUnit4SpringContext @Test(expected = AccessDeniedException.class) @DirtiesContext - public void testWithoutPermision() { + public void testWithoutPermision() throws Throwable { login("bob", "bobspassword", "ROLE_USER"); - input.send(new StringMessage("test")); + try { + input.send(new StringMessage("test")); + } + catch (MessageDeliveryException e) { + throw e.getCause().getCause(); + } } diff --git a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/SecurityEndpointInterceptorTests.java b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/SecurityEndpointInterceptorTests.java index ccee2b4f37..7a0e0c37ef 100644 --- a/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/SecurityEndpointInterceptorTests.java +++ b/org.springframework.integration.security/src/test/java/org/springframework/integration/security/endpoint/SecurityEndpointInterceptorTests.java @@ -22,12 +22,8 @@ import static org.easymock.EasyMock.replay; import static org.easymock.EasyMock.verify; import static org.junit.Assert.assertNull; -import org.junit.Before; import org.junit.Test; -import org.springframework.integration.endpoint.HandlerEndpoint; -import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.integration.security.SecurityContextUtils; @@ -44,32 +40,18 @@ import org.springframework.security.context.SecurityContextHolder; */ public class SecurityEndpointInterceptorTests { - private MessageEndpoint endpoint; - - - @Before - public void createEndpoint() throws Exception { - HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { - public Message handle(Message message) { - return null; - } - }); - endpoint.afterPropertiesSet(); - this.endpoint = endpoint; - } - - @Test(expected = AccessDeniedException.class) public void testUnauthenticatedAccessToSecuredEndpointWithNullMessage() throws Throwable { try { + Message message = null; ConfigAttributeDefinition attDefintion = new ConfigAttributeDefinition("ROLE_ADMIN"); AccessDecisionManager adm = createMock(AccessDecisionManager.class); - adm.decide(null, endpoint, attDefintion); + adm.decide(null, message, attDefintion); expectLastCall().andThrow(new AccessDeniedException("nope")); replay(adm); SecurityEndpointInterceptor interceptor = new SecurityEndpointInterceptor(attDefintion, adm); - interceptor.aroundSend(null, endpoint); + interceptor.preHandle(message); verify(adm); } finally { @@ -81,14 +63,15 @@ public class SecurityEndpointInterceptorTests { @Test(expected = AccessDeniedException.class) public void testUnauthenticatedAccessToSecuredEndpointWithNoSecurityContext() throws Throwable { try { + Message message = this.createMessageWithoutContext(); ConfigAttributeDefinition attDefintion = new ConfigAttributeDefinition("ROLE_ADMIN"); AccessDecisionManager adm = createMock(AccessDecisionManager.class); - adm.decide(null, endpoint, attDefintion); + adm.decide(null, message, attDefintion); expectLastCall().andThrow(new AccessDeniedException("nope")); replay(adm); SecurityEndpointInterceptor interceptor = new SecurityEndpointInterceptor(attDefintion, adm); - interceptor.aroundSend(this.createMessageWithoutContext(), endpoint); + interceptor.preHandle(message); verify(adm); } finally { @@ -103,14 +86,15 @@ public class SecurityEndpointInterceptorTests { SecurityContext context = SecurityTestUtil.createContext("bob", "bobspassword", new String[] { "ROLE_ADMIN" }); ConfigAttributeDefinition attDefintion = new ConfigAttributeDefinition("ROLE_ADMIN"); + Message message = this.createMessageWithContext(context); AccessDecisionManager adm = createMock(AccessDecisionManager.class); - adm.decide(context.getAuthentication(), endpoint, attDefintion); + adm.decide(context.getAuthentication(), message, attDefintion); expectLastCall().andThrow(new AccessDeniedException("nope")); replay(adm); SecurityEndpointInterceptor interceptor = new SecurityEndpointInterceptor(attDefintion, adm); - interceptor.aroundSend(this.createMessageWithContext(context), endpoint); + interceptor.preHandle(message); verify(adm); } finally { @@ -125,13 +109,14 @@ public class SecurityEndpointInterceptorTests { SecurityContext context = SecurityTestUtil.createContext("bob", "bobspassword", new String[] { "ROLE_ADMIN" }); ConfigAttributeDefinition attDefintion = new ConfigAttributeDefinition("ROLE_ADMIN"); + Message message = this.createMessageWithContext(context); AccessDecisionManager adm = createMock(AccessDecisionManager.class); - adm.decide(context.getAuthentication(), endpoint, attDefintion); + adm.decide(context.getAuthentication(), message, attDefintion); expectLastCall(); replay(adm); SecurityEndpointInterceptor interceptor = new SecurityEndpointInterceptor(attDefintion, adm); - interceptor.aroundSend(this.createMessageWithContext(context), endpoint); + interceptor.preHandle(message); verify(adm); } finally { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java index e7bea07ad5..be9bf65c3f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java @@ -332,8 +332,15 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A private void activateEndpoint(MessageEndpoint endpoint) { Assert.notNull(endpoint, "'endpoint' must not be null"); - if (endpoint.getTarget() == null) { - this.lookupOrCreateChannel(endpoint.getOutputChannelName()); + if (endpoint instanceof ChannelRegistryAware) { + ((ChannelRegistryAware) endpoint).setChannelRegistry(this); + } + MessageTarget target = endpoint.getTarget(); + if (target == null) { + target = this.lookupOrCreateChannel(endpoint.getOutputChannelName()); + if (target != null) { + endpoint.setTarget(target); + } } MessageSource source = endpoint.getSource(); if (source == null) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java index bccbe09c07..3c15582375 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java @@ -28,6 +28,7 @@ import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.ConfigurationException; import org.springframework.integration.endpoint.HandlerEndpoint; +import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; @@ -58,7 +59,7 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe @Override protected Class getBeanClass(Element element) { - return HandlerEndpoint.class; + return this.getEndpointClass(); } @Override @@ -116,8 +117,19 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe this.postProcessEndpointBean(builder, element, parserContext); } + /** + * Subclasses must override this to return the MessageHandler class. + * @return + */ protected abstract Class getHandlerAdapterClass(); + /** + * Subclasses may override this to return a specific MessageEndpoint class. + */ + protected Class getEndpointClass() { + return HandlerEndpoint.class; + } + protected void postProcessEndpointBean(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) { } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java index c9e089b90c..b15b722ce1 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java @@ -16,8 +16,10 @@ package org.springframework.integration.config; -import org.springframework.integration.handler.DefaultMessageHandlerAdapter; +import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.SimpleEndpoint; import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.handler.DefaultMessageHandler; /** * Parser for the <service-activator> element. @@ -26,9 +28,14 @@ import org.springframework.integration.handler.MessageHandler; */ public class ServiceActivatorParser extends AbstractHandlerEndpointParser { + @Override + protected Class getEndpointClass() { + return SimpleEndpoint.class; + } + @Override protected Class getHandlerAdapterClass() { - return DefaultMessageHandlerAdapter.class; + return DefaultMessageHandler.class; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index 103e132810..2b2895e812 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -27,6 +27,7 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.message.MessageRejectedException; @@ -208,45 +209,41 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist if (logger.isDebugEnabled()) { logger.debug("endpoint '" + this + "' handling message: " + message); } - return this.send(message, 0); + this.handleMessage(message, 0); + return true; } - private boolean send(final Message message, final int index) { + private Message handleMessage(Message message, final int index) { if (index == 0) { for (EndpointInterceptor interceptor : interceptors) { - if (!interceptor.preSend(message)) { - return false; + message = interceptor.preHandle(message); + if (message == null) { + return null; } } } if (index == interceptors.size()) { - boolean result = this.doSend(message); + if (!this.supports(message)) { + throw new MessageRejectedException(message, "unsupported message"); + } + Message reply = this.handleMessage(message); for (int i = index - 1; i >= 0; i--) { EndpointInterceptor interceptor = this.interceptors.get(i); - interceptor.postSend(message, result); + reply = interceptor.postHandle(message, reply); } - return result; + if (reply != null) { + this.getMessageExchangeTemplate().send(message, this.target); + } + return null; } EndpointInterceptor nextInterceptor = interceptors.get(index); - return nextInterceptor.aroundSend(message, new MessageTarget() { - @SuppressWarnings("unchecked") - public boolean send(Message message) { - return AbstractEndpoint.this.send(message, index + 1); + return nextInterceptor.aroundHandle(message, new MessageHandler() { + public Message handle(Message message) { + return AbstractEndpoint.this.handleMessage(message, index + 1); } }); } - private boolean doSend(Message message) { - if (!this.supports(message)) { - throw new MessageRejectedException(message, "unsupported message"); - } - Message result = this.handleMessage(message); - if (result != null) { - return this.getMessageExchangeTemplate().send(message, this.target); - } - return true; - } - protected boolean supports(Message message) { if (this.selector != null && !this.selector.accept(message)) { if (logger.isDebugEnabled()) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java index 2bcca6a5e4..027f2026a3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointInterceptor.java @@ -16,18 +16,18 @@ package org.springframework.integration.endpoint; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; /** * @author Mark Fisher */ public interface EndpointInterceptor { - boolean preSend(Message message); + Message preHandle(Message requestMessage); - boolean aroundSend(Message message, MessageTarget endpoint); + Message aroundHandle(Message message, MessageHandler handler); - void postSend(Message message, boolean result); + Message postHandle(Message requestMessage, Message replyMessage); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java index c2a7eaf91f..ab1cd214ba 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java @@ -17,7 +17,9 @@ package org.springframework.integration.endpoint; import java.lang.reflect.Array; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -29,11 +31,16 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.CompositeMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageHeaders; +import org.springframework.integration.message.MessageRejectedException; +import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.MessagingException; +import org.springframework.integration.message.selector.MessageSelector; +import org.springframework.integration.scheduling.Schedule; import org.springframework.integration.util.ErrorHandler; import org.springframework.util.Assert; @@ -59,7 +66,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class SimpleEndpoint implements MessageTarget, BeanNameAware { +public class SimpleEndpoint implements MessageEndpoint, ChannelRegistryAware, BeanNameAware { private final Log logger = LogFactory.getLog(this.getClass()); @@ -75,6 +82,10 @@ public class SimpleEndpoint implements MessageTarget, private volatile boolean requiresReply = false; + private volatile MessageSelector selector; + + private final List interceptors = new ArrayList(); + private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate(); @@ -120,6 +131,21 @@ public class SimpleEndpoint implements MessageTarget, this.errorHandler = errorHandler; } + public void setSelector(MessageSelector selector) { + this.selector = selector; + } + + public void addInterceptor(EndpointInterceptor interceptor) { + this.interceptors.add(interceptor); + } + + public void setInterceptors(List interceptors) { + this.interceptors.clear(); + for (EndpointInterceptor interceptor : interceptors) { + this.addInterceptor(interceptor); + } + } + public void setChannelRegistry(ChannelRegistry channelRegistry) { if (this.handler instanceof ChannelRegistryAware) { ((ChannelRegistryAware) this.handler).setChannelRegistry(channelRegistry); @@ -149,9 +175,15 @@ public class SimpleEndpoint implements MessageTarget, this.messageExchangeTemplate.setSendTimeout(replyTimeout); } - public boolean send(Message requestMessage) { + public final boolean send(Message requestMessage) { + if (requestMessage == null || requestMessage.getPayload() == null) { + throw new IllegalArgumentException("Message and its payload must not be null"); + } + if (this.logger.isDebugEnabled()) { + this.logger.debug("endpoint '" + this + "' handling message: " + requestMessage); + } try { - Message replyMessage = this.handler.handle(requestMessage); + Message replyMessage = this.handleMessage(requestMessage, 0); if (!this.isValidReply(replyMessage)) { if (this.requiresReply) { throw new MessageHandlingException(requestMessage, @@ -180,6 +212,48 @@ public class SimpleEndpoint implements MessageTarget, } } + private Message handleMessage(Message requestMessage, final int index) { + if (requestMessage == null || requestMessage.getPayload() == null) { + return null; + } + if (index == 0) { + for (EndpointInterceptor interceptor : this.interceptors) { + requestMessage = interceptor.preHandle(requestMessage); + if (requestMessage == null) { + return null; + } + } + } + if (index == this.interceptors.size()) { + if (!this.supports(requestMessage)) { + throw new MessageRejectedException(requestMessage, "unsupported message"); + } + Message replyMessage = this.handler.handle(requestMessage); + for (int i = index - 1; i >= 0; i--) { + EndpointInterceptor interceptor = this.interceptors.get(i); + replyMessage = interceptor.postHandle(requestMessage, replyMessage); + } + return replyMessage; + } + EndpointInterceptor nextInterceptor = this.interceptors.get(index); + return nextInterceptor.aroundHandle(requestMessage, new MessageHandler() { + @SuppressWarnings("unchecked") + public Message handle(Message message) { + return SimpleEndpoint.this.handleMessage(message, index + 1); + } + }); + } + + protected boolean supports(Message message) { + if (this.selector != null && !this.selector.accept(message)) { + if (logger.isDebugEnabled()) { + logger.debug("selector for endpoint '" + this + "' rejected message: " + message); + } + return false; + } + return true; + } + private void sendReplyMessage(Message replyMessage, Message requestMessage) { if (replyMessage == null) { throw new MessageHandlingException(requestMessage, "reply message must not be null"); @@ -189,6 +263,9 @@ public class SimpleEndpoint implements MessageTarget, throw new MessageEndpointReplyException(replyMessage, requestMessage, "unable to resolve reply target"); } + replyMessage = MessageBuilder.fromMessage(replyMessage) + .copyHeadersIfAbsent(requestMessage.getHeaders()) + .setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, requestMessage.getHeaders().getId()).build(); if (!this.messageExchangeTemplate.send(replyMessage, replyTarget)) { throw new MessageEndpointReplyException(replyMessage, requestMessage, "failed to send reply to '" + replyTarget + "'"); @@ -249,4 +326,53 @@ public class SimpleEndpoint implements MessageTarget, return replyTarget; } + public String toString() { + return (this.name != null) ? this.name : super.toString(); + } + + /* TODO: remove the following methods after they are removed from the MessageEndpoint interface. */ + + private String inputChannelName; + private String outputChannelName; + private MessageSource source; + + public String getInputChannelName() { + return this.inputChannelName; + } + + public void setInputChannelName(String inputChannelName) { + this.inputChannelName = inputChannelName; + } + + public String getOutputChannelName() { + return this.outputChannelName; + } + + public void setOutputChannelName(String outputChannelName) { + this.outputChannelName = outputChannelName; + } + + public void setReturnAddressOverrides(boolean b) { + } + + public Schedule getSchedule() { + return null; + } + + public MessageSource getSource() { + return this.source; + } + + public MessageTarget getTarget() { + return this.outputChannel; + } + + public void setSource(MessageSource source) { + this.source = source; + } + + public void setTarget(MessageTarget target) { + this.outputChannel = (MessageChannel) target; + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java index 9fd63503ab..9cc5d86fe9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java @@ -17,8 +17,10 @@ package org.springframework.integration.endpoint.interceptor; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; +import java.util.concurrent.FutureTask; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; @@ -36,9 +38,10 @@ import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.EndpointInterceptor; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerRejectedExecutionException; +import org.springframework.integration.message.AsyncMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; import org.springframework.integration.scheduling.MessagePublishingErrorHandler; import org.springframework.integration.util.ErrorHandler; import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; @@ -118,30 +121,36 @@ public class ConcurrencyInterceptor extends EndpointInterceptorAdapter } @Override - public boolean aroundSend(final Message message, final MessageTarget endpoint) { + @SuppressWarnings("unchecked") + public Message aroundHandle(final Message requestMessage, final MessageHandler handler) { try { - this.executor.execute(new Runnable() { - public void run() { + FutureTask> task = new FutureTask>(new Callable>() { + public Message call() throws Exception { try { - endpoint.send(message); + return handler.handle(requestMessage); } - catch (Throwable t) { + catch (Exception e) { if (logger.isDebugEnabled()) { - logger.debug("error occurred in handler execution", t); + logger.debug("error occurred in handler execution", e); } if (errorHandler != null) { - errorHandler.handle(t); + errorHandler.handle(e); } - else if (logger.isWarnEnabled() && !logger.isDebugEnabled()) { - logger.warn("error occurred in handler execution", t); + else { + if (logger.isWarnEnabled() && !logger.isDebugEnabled()) { + logger.warn("error occurred in handler execution", e); + } + throw e; } } + return null; } - }); - return true; + }); + this.executor.execute(task); + return new AsyncMessage(task); } catch (RuntimeException e) { - throw new MessageHandlerRejectedExecutionException(message, e); + throw new MessageHandlerRejectedExecutionException(requestMessage, e); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java index a2987f1f00..67ade9d92c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/EndpointInterceptorAdapter.java @@ -17,8 +17,8 @@ package org.springframework.integration.endpoint.interceptor; import org.springframework.integration.endpoint.EndpointInterceptor; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; /** * A convenience base class for implementing {@link EndpointInterceptor EndpointInterceptors}. @@ -27,15 +27,16 @@ import org.springframework.integration.message.MessageTarget; */ public class EndpointInterceptorAdapter implements EndpointInterceptor { - public boolean preSend(Message message) { - return true; + public Message preHandle(Message requestMessage) { + return requestMessage; } - public boolean aroundSend(Message message, MessageTarget endpoint) { - return endpoint.send(message); + public Message aroundHandle(Message message, MessageHandler handler) { + return handler.handle(message); } - public void postSend(Message message, boolean result) { + public Message postHandle(Message requestMessage, Message replyMessage) { + return replyMessage; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptor.java index cab6772b18..3108e52052 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptor.java @@ -22,8 +22,8 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.ConfigurationException; import org.springframework.integration.endpoint.EndpointInterceptor; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; @@ -83,20 +83,18 @@ public class TransactionInterceptor extends EndpointInterceptorAdapter implement } @Override - public boolean aroundSend(final Message message, final MessageTarget endpoint) { + public Message aroundHandle(final Message message, final MessageHandler handler) { if (this.transactionTemplate == null) { throw new ConfigurationException("TransactionInterceptor has not been initialized"); } - this.transactionTemplate.execute(new TransactionCallback() { + return (Message) this.transactionTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { if (logger.isDebugEnabled()) { - logger.debug("Executing endpoint '" + endpoint + "' within transaction [" + status + "]"); + logger.debug("Invoking handler '" + handler + "' within transaction [" + status + "]"); } - endpoint.send(message); - return null; + return handler.handle(message); } }); - return true; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java new file mode 100644 index 0000000000..c37defceaf --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -0,0 +1,333 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.core.GenericTypeResolver; +import org.springframework.core.LocalVariableTableParameterNameDiscoverer; +import org.springframework.core.MethodParameter; +import org.springframework.core.ParameterNameDiscoverer; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.handler.annotation.Header; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.message.MessageHeaders; +import org.springframework.integration.util.DefaultMethodInvoker; +import org.springframework.integration.util.MethodInvoker; +import org.springframework.integration.util.NameResolvingMethodInvoker; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.StringUtils; + +/** + * A base class for any {@link MessageHandler} that may act as an adapter + * by invoking a "plain" (not Message-aware) method for a given target object. + * When used as an adapter, the target Object is mandatory and either a + * {@link Method} reference or a 'methodName' must be provided. If no Object + * and Method are provided, the handler will simply process the request + * Message's payload. + * + * @author Mark Fisher + */ +public abstract class AbstractMessageHandler implements MessageHandler, InitializingBean { + + private static final Log logger = LogFactory.getLog(AbstractMessageHandler.class); + + private volatile boolean methodExpectsMessage; + + private volatile Object object; + + private volatile Method method; + + private volatile String methodName; + + private volatile MethodInvoker invoker; + + private volatile MethodParameterMetadata[] parameterMetadata; + + private final ParameterNameDiscoverer parameterNameDiscoverer = new LocalVariableTableParameterNameDiscoverer(); + + private volatile boolean initialized; + + private final Object initializationMonitor = new Object(); + + + public AbstractMessageHandler(Object object, Method method) { + Assert.notNull(object, "object must not be null"); + Assert.notNull(method, "method must not be null"); + this.object = object; + this.method = method; + this.methodName = method.getName(); + } + + public AbstractMessageHandler(Object object, String methodName) { + Assert.notNull(object, "object must not be null"); + Assert.notNull(methodName, "methodName must not be null"); + this.object = object; + this.methodName = methodName; + } + + public AbstractMessageHandler() { + } + + + public void setObject(Object object) { + this.object = object; + } + + public void setMethod(Method method) { + Assert.notNull(method, "method must not be null"); + this.method = method; + this.methodName = method.getName(); + } + + public void setMethodName(String methodName) { + Assert.notNull(methodName, "methodName must not be null"); + if (this.method != null && !this.method.getName().equals(methodName)) { + this.method = null; + } + this.methodName = methodName; + } + + public void afterPropertiesSet() { + synchronized (this.initializationMonitor) { + if (this.initialized) { + return; + } + if (this.object != null) { + if (this.method == null) { + if (this.methodName == null) { + throw new IllegalStateException("either the 'method' or 'methodName' must be provided when the 'object' property has been set"); + } + final List candidates = new ArrayList(); + ReflectionUtils.doWithMethods(this.object.getClass(), new ReflectionUtils.MethodCallback() { + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + if (method.getName().equals(AbstractMessageHandler.this.methodName)) { + candidates.add(method); + } + } + }); + if (candidates.size() == 0) { + throw new ConfigurationException("no such method '" + this.methodName + + "' on target class [" + this.object.getClass() + "]"); + } + if (candidates.size() == 1) { + this.method = candidates.get(0); + } + } + if (this.method != null) { + this.invoker = new DefaultMethodInvoker(this.object, this.method); + } + else { + // TODO: resolve the candidate method and/or create a dynamic resolver + this.invoker = new NameResolvingMethodInvoker(this.object, this.methodName); + } + this.configureParameterMetadata(); + } + this.initialized = true; + } + } + + private void configureParameterMetadata() { + if (this.method == null) { + return; + } + Class[] paramTypes = this.method.getParameterTypes(); + this.parameterMetadata = new MethodParameterMetadata[paramTypes.length]; + for (int i = 0; i < parameterMetadata.length; i++) { + MethodParameter methodParam = new MethodParameter(this.method, i); + methodParam.initParameterNameDiscovery(this.parameterNameDiscoverer); + GenericTypeResolver.resolveParameterType(methodParam, this.method.getDeclaringClass()); + Object[] paramAnnotations = methodParam.getParameterAnnotations(); + String headerName = null; + for (int j = 0; j < paramAnnotations.length; j++) { + if (Header.class.isInstance(paramAnnotations[j])) { + Header headerAnnotation = (Header) paramAnnotations[j]; + headerName = this.resolveParameterNameIfNecessary(headerAnnotation.value(), methodParam); + parameterMetadata[i] = new MethodParameterMetadata(Header.class, headerName, headerAnnotation.required()); + } + } + if (headerName == null) { + parameterMetadata[i] = new MethodParameterMetadata(methodParam.getParameterType(), null, false); + } + } + } + + public Message handle(Message message) { + if (message == null || message.getPayload() == null) { + if (logger.isDebugEnabled()) { + logger.debug("message handler received a null message"); + } + return null; + } + if (!this.initialized) { + this.afterPropertiesSet(); + } + Object result = (this.invoker != null) ? this.invokeHandlerMethod(message) : message.getPayload(); + if (result == null) { + return null; + } + return this.createReplyMessage(result, message.getHeaders()); + } + + /** + * Subclasses must implement this method to generate the reply Message. + * + * @param result the return value from an adapter method, or the Message payload if not acting as an adapter + * @param requestHeaders the MessageHeaders of the original request Message + * @return the Message to be sent to the reply MessageTarget + */ + protected abstract Message createReplyMessage(Object result, MessageHeaders requestHeaders); + + + private Object invokeHandlerMethod(Message message) { + if (this.invoker == null) { + throw new IllegalStateException("cannot invoke method, invoker is null"); + } + Object args[] = null; + Object mappingResult = this.methodExpectsMessage ? message + : this.mapMessageToMethodArguments(message); + if (mappingResult != null && mappingResult.getClass().isArray() + && (Object.class.isAssignableFrom(mappingResult.getClass().getComponentType()))) { + args = (Object[]) mappingResult; + } + else { + args = new Object[] { mappingResult }; + } + try { + Object result = null; + try { + result = this.invoker.invokeMethod(args); + } + catch (NoSuchMethodException e) { + try { + result = this.invoker.invokeMethod(args); + this.methodExpectsMessage = true; + } + catch (NoSuchMethodException e2) { + throw new MessageHandlingException(message, "unable to determine method match"); + } + } + if (result == null) { + return null; + } + return result; + } + catch (InvocationTargetException e) { + throw new MessageHandlingException(message, "Handler method '" + + this.methodName + "' threw an Exception.", e.getTargetException()); + } + catch (Throwable e) { + throw new MessageHandlingException(message, "Failed to invoke handler method '" + + this.methodName + "' with arguments: " + ObjectUtils.nullSafeToString(args), e); + } + } + + + private Object[] mapMessageToMethodArguments(Message message) { + if (message == null) { + return null; + } + if (message.getPayload() == null) { + throw new IllegalArgumentException("Message payload must not be null."); + } + if (ObjectUtils.isEmpty(this.parameterMetadata)) { + return new Object[] { message.getPayload() }; + } + Object[] args = new Object[this.parameterMetadata.length]; + for (int i = 0; i < this.parameterMetadata.length; i++) { + MethodParameterMetadata metadata = this.parameterMetadata[i]; + Class expectedType = metadata.type; + if (expectedType.equals(Header.class)) { + Object value = message.getHeaders().get(metadata.key); + if (value == null && metadata.required) { + throw new MessageHandlingException(message, + "required header '" + metadata.key + "' not available"); + } + args[i] = value; + } + else if (expectedType.isAssignableFrom(message.getClass())) { + args[i] = message; + } + else if (expectedType.isAssignableFrom(message.getPayload().getClass())) { + args[i] = message.getPayload(); + } + else if (expectedType.equals(Map.class)) { + args[i] = message.getHeaders(); + } + else if (expectedType.equals(Properties.class)) { + args[i] = this.getStringTypedHeaders(message); + } + else { + args[i] = message.getPayload(); + } + } + return args; + } + + private Properties getStringTypedHeaders(Message message) { + Properties properties = new Properties(); + MessageHeaders headers = message.getHeaders(); + for (String key : headers.keySet()) { + Object value = headers.get(key); + if (value instanceof String) { + properties.setProperty(key, (String) value); + } + } + return properties; + } + + private String resolveParameterNameIfNecessary(String paramName, MethodParameter methodParam) { + if (!StringUtils.hasText(paramName)) { + paramName = methodParam.getParameterName(); + if (paramName == null) { + throw new IllegalStateException("No parameter name specified and not available in class file."); + } + } + return paramName; + } + + + private static class MethodParameterMetadata { + + private final Class type; + + private final String key; + + private final boolean required; + + + MethodParameterMetadata(Class type, String key, boolean required) { + this.type = type; + this.key = key; + this.required = required; + } + + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/DefaultMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/DefaultMessageHandler.java new file mode 100644 index 0000000000..bf7778ddf7 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/DefaultMessageHandler.java @@ -0,0 +1,37 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageHeaders; + +/** + * The default MessageHandler implementation. Creates a Message for the reply payload. + * The request Message's headers are copied and the request Message's ID is set as this + * reply Message's correlationId. + * + * @author Mark Fisher + */ +public class DefaultMessageHandler extends AbstractMessageHandler { + + @Override + protected Message createReplyMessage(Object result, MessageHeaders requestHeaders) { + return MessageBuilder.fromPayload(result).copyHeaders(requestHeaders).setCorrelationId(requestHeaders.getId()).build(); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAroundSendEndpointInterceptor.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAroundSendEndpointInterceptor.java index 1022bdd7eb..ac8d355a7e 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAroundSendEndpointInterceptor.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestAroundSendEndpointInterceptor.java @@ -19,8 +19,8 @@ package org.springframework.integration.config; import java.util.concurrent.atomic.AtomicInteger; import org.springframework.integration.endpoint.interceptor.EndpointInterceptorAdapter; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; /** * @author Mark Fisher @@ -35,11 +35,11 @@ public class TestAroundSendEndpointInterceptor extends EndpointInterceptorAdapte } @Override - public boolean aroundSend(Message message, MessageTarget endpoint) { + public Message aroundHandle(Message message, MessageHandler handler) { this.counter.incrementAndGet(); - boolean result = endpoint.send(message); + Message reply = handler.handle(message); this.counter.incrementAndGet(); - return result; + return reply; } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestPreSendInterceptor.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestPreSendInterceptor.java index 839d3eb78f..60e6873a91 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestPreSendInterceptor.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestPreSendInterceptor.java @@ -34,9 +34,9 @@ public class TestPreSendInterceptor extends EndpointInterceptorAdapter { } @Override - public boolean preSend(Message message) { + public Message preHandle(Message message) { this.counter.incrementAndGet(); - return true; + return 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 a80251c2d3..75daa0566b 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 @@ -18,6 +18,7 @@ package org.springframework.integration.endpoint; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import org.junit.Test; @@ -34,17 +35,19 @@ import org.springframework.integration.message.StringMessage; public class ReturnAddressTests { @Test - public void testReturnAddressOverrides() { + public void testNextTargetOverrides() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "returnAddressTests.xml", this.getClass()); MessageChannel channel1 = (MessageChannel) context.getBean("channel1WithOverride"); PollableChannel replyChannel = (PollableChannel) context.getBean("replyChannel"); context.start(); Message message = MessageBuilder.fromPayload("*") - .setReturnAddress("replyChannel").build(); + .setNextTarget("replyChannel").build(); channel1.send(message); Message response = replyChannel.receive(3000); assertNotNull(response); + PollableChannel outputChannel = (PollableChannel) context.getBean("channel2"); + assertNull(outputChannel.receive(0)); assertEquals("**", response.getPayload()); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java index 4a3e0bae62..c2eac923be 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/TransactionInterceptorTests.java @@ -27,6 +27,7 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.StringMessage; import org.springframework.transaction.IllegalTransactionStateException; import org.springframework.transaction.TransactionStatus; @@ -156,11 +157,16 @@ public class TransactionInterceptorTests { } @Test(expected = IllegalTransactionStateException.class) - public void testPropagationMandatoryCalledWithoutTransaction() { + public void testPropagationMandatoryCalledWithoutTransaction() throws Throwable { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "transactionInterceptorPropagationTests.xml", this.getClass()); final MessageEndpoint endpoint = (MessageEndpoint) context.getBean("mandatory"); - endpoint.send(new StringMessage("test")); + try { + endpoint.send(new StringMessage("test")); + } + catch (MessageHandlingException e) { + throw e.getCause(); + } } }