diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 8c9a4cbadc..7cf365c5d7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -21,6 +21,8 @@ import org.aopalliance.intercept.MethodInvocation; import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.BeansException; +import org.springframework.beans.SimpleTypeConverter; +import org.springframework.beans.TypeConverter; import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; @@ -57,17 +59,25 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, private MessageChannel responseChannel; + private long requestTimeout = 0; + + private long responseTimeout = 1000; + private ResponseCorrelator responseCorrelator; + private MessageCreator messageCreator = new DefaultMessageCreator(); + + private MessageMapper messageMapper = new DefaultMessageMapper(); + + private TypeConverter typeConverter = new SimpleTypeConverter(); + private EndpointRegistry endpointRegistry; private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader(); private Object serviceProxy; - private MessageCreator messageCreator = new DefaultMessageCreator(); - - private MessageMapper messageMapper = new DefaultMessageMapper(); + private final Object responseCorrelatorMonitor = new Object(); public void setServiceInterface(Class serviceInterface) { @@ -82,6 +92,14 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, this.responseChannel = responseChannel; } + public void setRequestTimeout(long requestTimeout) { + this.requestTimeout = requestTimeout; + } + + public void setResponseTimeout(long responseTimeout) { + this.responseTimeout = responseTimeout; + } + public void setMessageCreator(MessageCreator messageCreator) { Assert.notNull(messageCreator, "messageCreator must not be null"); this.messageCreator = messageCreator; @@ -92,6 +110,11 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, this.messageMapper = messageMapper; } + public void setTypeConverter(TypeConverter typeConverter) { + Assert.notNull(typeConverter, "typeConverter must not be null"); + this.typeConverter = typeConverter; + } + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { if (applicationContext.containsBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME)) { this.endpointRegistry = (EndpointRegistry) applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); @@ -103,7 +126,6 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, } public void afterPropertiesSet() { - this.registerResponseCorrelatorIfNecessary(); this.serviceProxy = new ProxyFactory(this.serviceInterface, this).getProxy(this.beanClassLoader); } @@ -120,38 +142,76 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, } public Object invoke(MethodInvocation invocation) throws Throwable { - boolean returnsVoid = invocation.getMethod().getReturnType().equals(void.class); + Class returnType = invocation.getMethod().getReturnType(); int params = invocation.getMethod().getParameterTypes().length; - if (params == 0) { - // TODO: add support for receive-only - throw new MessagingException("Method invocation contains no arguments. Cannot send a message."); - } - if (this.requestChannel == null) { - throw new MessagingException("No request channel available. Cannot invoke methods with arguments."); - } - Object payload = (params == 1) ? invocation.getArguments()[0] : invocation.getArguments(); - Message message = this.messageCreator.createMessage(payload); - if (returnsVoid) { - this.requestChannel.send(message); - return null; - } Message response = null; - if (this.responseCorrelator != null) { - message.getHeader().setReturnAddress(this.responseChannel); - this.requestChannel.send(message); - response = this.responseCorrelator.getResponse(message.getId()); + if (params == 0) { + if (this.responseChannel == null) { + throw new MessagingException("No response channel available. Cannot support methods with no arguments."); + } + response = this.receiveResponse(this.responseChannel); } else { - RendezvousChannel temporaryChannel = new RendezvousChannel(); - message.getHeader().setReturnAddress(temporaryChannel); - this.requestChannel.send(message); - response = temporaryChannel.receive(); + if (this.requestChannel == null) { + throw new MessagingException("No request channel available. Cannot support methods with arguments."); + } + Object payload = (params == 1) ? invocation.getArguments()[0] : invocation.getArguments(); + Message message = this.messageCreator.createMessage(payload); + if (returnType.equals(void.class)) { + this.sendRequest(message); + return null; + } + if (this.responseChannel != null) { + response = this.sendAndReceiveWithResponseCorrelator(message); + } + else { + response = this.sendAndReceiveWithTemporaryChannel(message); + } } - return (response != null) ? this.messageMapper.mapMessage(response) : null; + if (returnType.isAssignableFrom(response.getClass())) { + return response; + } + Object responseObject = (response != null) ? this.messageMapper.mapMessage(response) : null; + return this.typeConverter.convertIfNecessary(responseObject, returnType); } - private void registerResponseCorrelatorIfNecessary() { - if (this.responseChannel != null) { + private Message sendAndReceiveWithResponseCorrelator(Message message) { + if (this.responseCorrelator == null) { + this.registerResponseCorrelator(); + } + message.getHeader().setReturnAddress(this.responseChannel); + this.sendRequest(message); + return (this.responseTimeout >= 0) ? this.responseCorrelator.getResponse(message.getId(), this.responseTimeout) : + this.responseCorrelator.getResponse(message.getId()); + } + + private Message sendAndReceiveWithTemporaryChannel(Message message) { + RendezvousChannel temporaryChannel = new RendezvousChannel(); + message.getHeader().setReturnAddress(temporaryChannel); + this.sendRequest(message); + return this.receiveResponse(temporaryChannel); + } + + private void sendRequest(Message message) { + if (message == null) { + throw new MessagingException("Created Message is null, cannot be sent."); + } + boolean sent = (this.requestTimeout >= 0) ? + this.requestChannel.send(message, this.requestTimeout) : this.requestChannel.send(message); + if (!sent) { + throw new MessagingException("Failed to send request message."); + } + } + + private Message receiveResponse(MessageChannel channel) { + return (this.responseTimeout >= 0) ? channel.receive(this.responseTimeout) : channel.receive(); + } + + private void registerResponseCorrelator() { + synchronized (this.responseCorrelatorMonitor) { + if (this.responseCorrelator != null) { + return; + } if (this.endpointRegistry == null) { throw new ConfigurationException("No EndpointRegistry available. Cannot register ResponseCorrelator."); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageCreator.java index 8bee454a0e..05b609a635 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageCreator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageCreator.java @@ -25,6 +25,9 @@ package org.springframework.integration.message; public class DefaultMessageCreator implements MessageCreator { public Message createMessage(Object object) { + if (object instanceof Message) { + return (Message) object; + } return (object != null) ? new GenericMessage(object) : null; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java index b5db0318e5..b4775dee7e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java @@ -35,13 +35,7 @@ public class GatewayProxyFactoryBeanTests { @Test public void testRequestReplyWithAnonymousChannel() throws Exception { final MessageChannel requestChannel = new QueueChannel(); - new Thread(new Runnable() { - public void run() { - Message input = requestChannel.receive(); - StringMessage response = new StringMessage(input.getPayload() + "bar"); - ((MessageChannel) input.getHeader().getReturnAddress()).send(response); - } - }).start(); + startResponder(requestChannel); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); proxyFactory.setServiceInterface(TestService.class); proxyFactory.setRequestChannel(requestChannel); @@ -65,6 +59,39 @@ public class GatewayProxyFactoryBeanTests { assertEquals("test", message.getPayload()); } + @Test + public void testSolicitResponse() throws Exception { + MessageChannel responseChannel = new QueueChannel(); + responseChannel.send(new StringMessage("foo")); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setServiceInterface(TestService.class); + proxyFactory.setResponseChannel(responseChannel); + proxyFactory.afterPropertiesSet(); + TestService service = (TestService) proxyFactory.getObject(); + String result = service.solicitResponse(); + assertNotNull(result); + assertEquals("foo", result); + } + + @Test + public void testRequestReplyWithTypeConversion() throws Exception { + final MessageChannel requestChannel = new QueueChannel(); + new Thread(new Runnable() { + public void run() { + Message input = requestChannel.receive(); + StringMessage response = new StringMessage(input.getPayload() + "456"); + ((MessageChannel) input.getHeader().getReturnAddress()).send(response); + } + }).start(); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setServiceInterface(TestService.class); + proxyFactory.setRequestChannel(requestChannel); + proxyFactory.afterPropertiesSet(); + TestService service = (TestService) proxyFactory.getObject(); + Integer result = service.requestReplyWithIntegers(123); + assertEquals(new Integer(123456), result); + } + @Test public void testRequestReplyWithRendezvousChannelInApplicationContext() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( @@ -86,4 +113,47 @@ public class GatewayProxyFactoryBeanTests { assertEquals(1, interceptor.getReceivedCount()); } + @Test + public void testMessageAsMethodArgument() throws Exception { + final MessageChannel requestChannel = new QueueChannel(); + startResponder(requestChannel); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setServiceInterface(TestService.class); + proxyFactory.setRequestChannel(requestChannel); + proxyFactory.afterPropertiesSet(); + TestService service = (TestService) proxyFactory.getObject(); + String result = service.requestReplyWithMessageParameter(new StringMessage("foo")); + assertEquals("foobar", result); + } + + @Test + public void testMessageAsReturnValue() throws Exception { + final MessageChannel requestChannel = new QueueChannel(); + new Thread(new Runnable() { + public void run() { + Message input = requestChannel.receive(); + StringMessage response = new StringMessage(input.getPayload() + "bar"); + ((MessageChannel) input.getHeader().getReturnAddress()).send(response); + } + }).start(); + GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setServiceInterface(TestService.class); + proxyFactory.setRequestChannel(requestChannel); + proxyFactory.afterPropertiesSet(); + TestService service = (TestService) proxyFactory.getObject(); + Message result = service.requestReplyWithMessageReturnValue("foo"); + assertEquals("foobar", result.getPayload()); + } + + + private static void startResponder(final MessageChannel requestChannel) { + new Thread(new Runnable() { + public void run() { + Message input = requestChannel.receive(); + StringMessage response = new StringMessage(input.getPayload() + "bar"); + ((MessageChannel) input.getHeader().getReturnAddress()).send(response); + } + }).start(); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java index 3475e628e4..e668bfbb2c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java @@ -16,6 +16,8 @@ package org.springframework.integration.gateway; +import org.springframework.integration.message.Message; + /** * @author Mark Fisher */ @@ -25,4 +27,12 @@ public interface TestService { void oneWay(String input); + String solicitResponse(); + + Integer requestReplyWithIntegers(Integer input); + + String requestReplyWithMessageParameter(Message message); + + Message requestReplyWithMessageReturnValue(String input); + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageCreatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageCreatorTests.java index 9952217005..4a260fe10b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageCreatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageCreatorTests.java @@ -41,6 +41,14 @@ public class DefaultMessageCreatorTests { assertEquals(test, message.getPayload()); } + @Test + public void testMessage() { + DefaultMessageCreator creator = new DefaultMessageCreator(); + Message inputMessage = new GenericMessage("testing"); + Message message = creator.createMessage(inputMessage); + assertEquals(inputMessage, message); + } + @Test public void testNull() { DefaultMessageCreator creator = new DefaultMessageCreator();