diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java index 732b40436c..4b9dbc34d3 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java @@ -105,8 +105,8 @@ public class MessageHandlingSourceAdapter implements MessageHandler, Initializin private RequestReplyTemplate createRequestReplyTemplate() { RequestReplyTemplate template = new RequestReplyTemplate(this.channel); - template.setDefaultSendTimeout(this.sendTimeout); - template.setDefaultReceiveTimeout(this.receiveTimeout); + template.setRequestTimeout(this.sendTimeout); + template.setReplyTimeout(this.receiveTimeout); return template; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java index 1be20c09d7..9e7d5129e6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java @@ -16,13 +16,21 @@ package org.springframework.integration.channel; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.List; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.config.MessageBusParser; +import org.springframework.integration.endpoint.EndpointRegistry; +import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.handler.ReplyHandler; +import org.springframework.integration.handler.ResponseCorrelator; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHeader; -import org.springframework.util.Assert; +import org.springframework.integration.message.MessagingException; +import org.springframework.integration.message.selector.MessageSelector; +import org.springframework.integration.scheduling.Subscription; /** * A template that facilitates the implementation of request-reply usage @@ -30,121 +38,234 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class RequestReplyTemplate { +public class RequestReplyTemplate implements ApplicationContextAware { - private final MessageChannel requestChannel; + private MessageChannel requestChannel; - private final ExecutorService executor; + private MessageChannel replyChannel; - private volatile long defaultSendTimeout = -1; + private volatile long requestTimeout = -1; - private volatile long defaultReceiveTimeout = -1; + private volatile long replyTimeout = -1; + + private ResponseCorrelator responseCorrelator; + + private EndpointRegistry endpointRegistry; + + private final Object responseCorrelatorMonitor = new Object(); /** * Create a RequestReplyTemplate. * * @param requestChannel the channel to which request messages will be sent - * @param executor the executor that will handle asynchronous (non-blocking) - * requests + * @param replyChannel the channel from which reply messages will be received */ - public RequestReplyTemplate(MessageChannel requestChannel, ExecutorService executor) { - Assert.notNull(requestChannel, "'requestChannel' must not be null"); - Assert.notNull(executor, "'executor' must not be null"); + public RequestReplyTemplate(MessageChannel requestChannel, MessageChannel replyChannel) { this.requestChannel = requestChannel; - this.executor = executor; + this.replyChannel = replyChannel; } /** - * Create a RequestReplyTemplate with a default single-threaded executor. + * Create a RequestReplyTemplate that will use anonymous temporary channels for replies. * * @param requestChannel the channel to which request messages will be sent */ public RequestReplyTemplate(MessageChannel requestChannel) { - this(requestChannel, Executors.newSingleThreadExecutor()); + this(requestChannel, null); + } + + public RequestReplyTemplate() { } /** - * Set the default timeout value for sending request messages. If not - * explicitly configured, the default will be an indefinite timeout. + * Set the request channel. * - * @param defaultSendTimeout the timeout value in milliseconds + * @param requestChannel the channel to which request messages will be sent */ - public void setDefaultSendTimeout(long defaultSendTimeout) { - this.defaultSendTimeout = defaultSendTimeout; + public void setRequestChannel(MessageChannel requestChannel) { + this.requestChannel = requestChannel; } /** - * Set the default timeout value for receiving reply messages. If not - * explicitly configured, the default will be an indefinite timeout. + * Set the reply channel. If no reply channel is provided, this template will + * always use an anonymous, temporary channel for handling replies. * - * @param defaultReceiveTimeout the timeout value in milliseconds + * @param replyChannel the channel from which reply messages will be received */ - public void setDefaultReceiveTimeout(long defaultReceiveTimeout) { - this.defaultReceiveTimeout = defaultReceiveTimeout; + public void setReplyChannel(MessageChannel replyChannel) { + this.replyChannel = replyChannel; } /** - * Send a request message and wait for a reply message using the provided - * timeout values. + * Set the timeout value for sending request messages. If not + * explicitly configured, the default is an indefinite timeout. * - * @param requestMessage the request message to send - * @param sendTimeout the timeout value for sending the request message - * @param receiveTimeout the timeout value for receiving a reply message - * - * @return the reply message or null + * @param requestTimeout the timeout value in milliseconds */ - public Message request(Message requestMessage, long sendTimeout, long receiveTimeout) { - RendezvousChannel replyChannel = new RendezvousChannel(); - requestMessage.getHeader().setReturnAddress(replyChannel); - this.requestChannel.send(requestMessage, sendTimeout); - return replyChannel.receive(receiveTimeout); + public void setRequestTimeout(long requestTimeout) { + this.requestTimeout = requestTimeout; } /** - * Send a request message and wait for a reply message using the default + * Set the timeout value for receiving reply messages. If not + * explicitly configured, the default is an indefinite timeout. + * + * @param replyTimeout the timeout value in milliseconds + */ + public void setReplyTimeout(long replyTimeout) { + this.replyTimeout = replyTimeout; + } + + public void setEndpointRegistry(EndpointRegistry endpointRegistry) { + this.endpointRegistry = endpointRegistry; + } + + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + if (applicationContext.containsBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME)) { + this.setEndpointRegistry((EndpointRegistry) applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); + } + } + + public boolean send(Message message) { + if (message == null) { + throw new MessagingException("Message must not be null."); + } + if (this.requestChannel == null) { + throw new MessagingException("No request channel has been configured. Cannot send message."); + } + 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."); + } + return true; + } + + public Message receive() { + if (this.replyChannel == null) { + throw new MessagingException("No reply channel has been configured. Cannot perform receive only operation."); + } + return this.receiveResponse(this.replyChannel); + } + + /** + * Send a request message whose reply should be handled be the provided callback. + */ + public boolean request(Message message, ReplyHandler replyHandler) { + MessageChannel replyChannelAdapter = new ReplyHandlingChannelAdapter(message, replyHandler); + message.getHeader().setReturnAddress(replyChannelAdapter); + return this.send(message); + } + + /** + * Send a request message and wait for a reply message using the configured * timeout values. * * @param requestMessage the request message to send * * @return the reply message or null */ - public Message request(Message requestMessage) { - return this.request(requestMessage, this.defaultSendTimeout, this.defaultReceiveTimeout); + public Message request(Message message) { + if (this.requestChannel == null) { + throw new MessagingException("No request channel available. Cannot send request message."); + } + if (this.replyChannel != null) { + return this.sendAndReceiveWithResponseCorrelator(message); + } + else { + return this.sendAndReceiveWithTemporaryChannel(message); + } } - /** - * Send a request message asynchronously to be handled by the provided - * {@link ReplyHandler}. The provided values will be used for the send - * and receive timeouts. - * - * @param requestMessage the request message to send - * @param replyHandler the callback that will handle a reply message - * @param sendTimeout the timeout value for sending the request message - * @param receiveTimeout the timeout value for receiving the reply message - */ - public void request(final Message requestMessage, final ReplyHandler replyHandler, final long sendTimeout, - final long receiveTimeout) { - final MessageHeader header = requestMessage.getHeader(); - this.executor.submit(new Runnable() { - public void run() { - Message reply = request(requestMessage, sendTimeout, receiveTimeout); - replyHandler.handle(reply, header); + private Message sendAndReceiveWithResponseCorrelator(Message message) { + if (this.responseCorrelator == null) { + this.registerResponseCorrelator(); + } + message.getHeader().setReturnAddress(this.replyChannel); + this.send(message); + return (this.replyTimeout >= 0) ? this.responseCorrelator.getResponse(message.getId(), this.replyTimeout) : + this.responseCorrelator.getResponse(message.getId()); + } + + private Message sendAndReceiveWithTemporaryChannel(Message message) { + RendezvousChannel temporaryChannel = new RendezvousChannel(); + message.getHeader().setReturnAddress(temporaryChannel); + this.send(message); + return this.receiveResponse(temporaryChannel); + } + + private Message receiveResponse(MessageChannel channel) { + return (this.replyTimeout >= 0) ? channel.receive(this.replyTimeout) : 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."); + } + ResponseCorrelator correlator = new ResponseCorrelator(10); + HandlerEndpoint endpoint = new HandlerEndpoint(correlator); + endpoint.setSubscription(new Subscription(this.replyChannel)); + this.endpointRegistry.registerEndpoint("internal.correlator." + this, endpoint); + this.responseCorrelator = correlator; + } } - /** - * Send a request message asynchronously to be handled by the provided - * {@link ReplyHandler}. Default values will be used for the send and - * receive timeouts. - * - * @param requestMessage the request message to send - * @param replyHandler the callback that will handle a reply message - */ - public void request(final Message requestMessage, final ReplyHandler replyHandler) { - this.request(requestMessage, replyHandler, this.defaultSendTimeout, this.defaultReceiveTimeout); + + private static class ReplyHandlingChannelAdapter implements MessageChannel { + + private final Message originalMessage; + + private final ReplyHandler replyHandler; + + + ReplyHandlingChannelAdapter(Message originalMessage, ReplyHandler replyHandler) { + this.originalMessage = originalMessage; + this.replyHandler = replyHandler; + } + + + public List> clear() { + return null; + } + + public DispatcherPolicy getDispatcherPolicy() { + return null; + } + + public String getName() { + return null; + } + + public List> purge(MessageSelector selector) { + return null; + } + + public void setName(String name) { + } + + public Message receive() { + return null; + } + + public Message receive(long timeout) { + return null; + } + + public boolean send(Message message) { + this.replyHandler.handle(message, originalMessage.getHeader()); + return true; + } + + public boolean send(Message message, long timeout) { + return this.send(message); + } + } } 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 7cf365c5d7..b919c8a9aa 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 @@ -20,28 +20,12 @@ import org.aopalliance.intercept.MethodInterceptor; 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; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.RendezvousChannel; -import org.springframework.integration.config.MessageBusParser; -import org.springframework.integration.endpoint.EndpointRegistry; -import org.springframework.integration.endpoint.HandlerEndpoint; -import org.springframework.integration.handler.ResponseCorrelator; -import org.springframework.integration.message.DefaultMessageCreator; -import org.springframework.integration.message.DefaultMessageMapper; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageCreator; -import org.springframework.integration.message.MessageMapper; -import org.springframework.integration.message.MessagingException; -import org.springframework.integration.scheduling.Subscription; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -51,76 +35,26 @@ import org.springframework.util.ClassUtils; * * @author Mark Fisher */ -public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, InitializingBean, ApplicationContextAware, BeanClassLoaderAware { +public class GatewayProxyFactoryBean extends MessagingGateway implements FactoryBean, MethodInterceptor, InitializingBean, BeanClassLoaderAware { private Class serviceInterface; - private MessageChannel requestChannel; - - 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 final Object responseCorrelatorMonitor = new Object(); - public void setServiceInterface(Class serviceInterface) { this.serviceInterface = serviceInterface; } - public void setRequestChannel(MessageChannel requestChannel) { - this.requestChannel = requestChannel; - } - - public void setResponseChannel(MessageChannel responseChannel) { - 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; - } - - public void setMessageMapper(MessageMapper messageMapper) { - Assert.notNull(messageMapper, "messageMapper must not be null"); - 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); - } - } - public void setBeanClassLoader(ClassLoader beanClassLoader) { this.beanClassLoader = beanClassLoader; } @@ -143,85 +77,24 @@ public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, public Object invoke(MethodInvocation invocation) throws Throwable { Class returnType = invocation.getMethod().getReturnType(); - int params = invocation.getMethod().getParameterTypes().length; - Message response = null; - if (params == 0) { - if (this.responseChannel == null) { - throw new MessagingException("No response channel available. Cannot support methods with no arguments."); + boolean shouldReturnMessage = Message.class.isAssignableFrom(returnType); + int paramCount = invocation.getMethod().getParameterTypes().length; + Object response = null; + if (paramCount == 0) { + if (shouldReturnMessage) { + return this.receive(); } - response = this.receiveResponse(this.responseChannel); + response = this.invoke(); } else { - 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); + Object payload = (paramCount == 1) ? invocation.getArguments()[0] : invocation.getArguments(); if (returnType.equals(void.class)) { - this.sendRequest(message); + this.send(payload); return null; } - if (this.responseChannel != null) { - response = this.sendAndReceiveWithResponseCorrelator(message); - } - else { - response = this.sendAndReceiveWithTemporaryChannel(message); - } - } - if (returnType.isAssignableFrom(response.getClass())) { - return response; - } - Object responseObject = (response != null) ? this.messageMapper.mapMessage(response) : null; - return this.typeConverter.convertIfNecessary(responseObject, returnType); - } - - 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."); - } - ResponseCorrelator correlator = new ResponseCorrelator(10); - HandlerEndpoint endpoint = new HandlerEndpoint(correlator); - endpoint.setSubscription(new Subscription(this.responseChannel)); - this.endpointRegistry.registerEndpoint( - this.serviceInterface.getName() + "-" + this.responseChannel + "-correlator", endpoint); - this.responseCorrelator = correlator; + response = this.invoke(payload, !shouldReturnMessage); } + return (response != null) ? this.typeConverter.convertIfNecessary(response, returnType) : null; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGateway.java new file mode 100644 index 0000000000..d61b4b7003 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGateway.java @@ -0,0 +1,96 @@ +/* + * 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.gateway; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.RequestReplyTemplate; +import org.springframework.integration.message.DefaultMessageCreator; +import org.springframework.integration.message.DefaultMessageMapper; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageCreator; +import org.springframework.integration.message.MessageMapper; +import org.springframework.util.Assert; + +/** + * A general purpose class that supports a variety of message exchanges. Useful for connecting application code to + * {@link MessageChannel MessageChannels} for sending, receiving, or request-reply operations. May be used as a base + * class for framework components so that the details of messaging are well-encapsulated and hidden from application + * code. For example, see {@link GatewayProxyFactoryBean}. + * + * @author Mark Fisher + */ +public class MessagingGateway extends RequestReplyTemplate { + + private MessageCreator messageCreator = new DefaultMessageCreator(); + + private MessageMapper messageMapper = new DefaultMessageMapper(); + + + public MessagingGateway(MessageChannel requestChannel, MessageChannel replyChannel) { + super(requestChannel, replyChannel); + } + + public MessagingGateway(MessageChannel requestChannel) { + super(requestChannel); + } + + public MessagingGateway() { + super(); + } + + + public void setMessageCreator(MessageCreator messageCreator) { + Assert.notNull(messageCreator, "messageCreator must not be null"); + this.messageCreator = messageCreator; + } + + public void setMessageMapper(MessageMapper messageMapper) { + Assert.notNull(messageMapper, "messageMapper must not be null"); + this.messageMapper = messageMapper; + } + + public void send(Object object) { + Message message = (object instanceof Message) ? (Message) object : + this.messageCreator.createMessage(object); + if (message != null) { + this.send(message); + } + } + + public Object invoke() { + Message message = this.receive(); + return (message != null) ? this.messageMapper.mapMessage(message) : null; + } + + public Object invoke(Object object) { + return this.invoke(object, true); + } + + public Object invoke(Object object, boolean shouldMapMessage) { + Message request = (object instanceof Message) ? (Message) object : + this.messageCreator.createMessage(object); + if (request == null) { + return null; + } + Message reply = this.request(request); + if (!shouldMapMessage) { + return reply; + } + return (reply != null) ? this.messageMapper.mapMessage(reply) : 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 b4775dee7e..a28f15399e 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 @@ -61,11 +61,11 @@ public class GatewayProxyFactoryBeanTests { @Test public void testSolicitResponse() throws Exception { - MessageChannel responseChannel = new QueueChannel(); - responseChannel.send(new StringMessage("foo")); + MessageChannel replyChannel = new QueueChannel(); + replyChannel.send(new StringMessage("foo")); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); proxyFactory.setServiceInterface(TestService.class); - proxyFactory.setResponseChannel(responseChannel); + proxyFactory.setReplyChannel(replyChannel); proxyFactory.afterPropertiesSet(); TestService service = (TestService) proxyFactory.getObject(); String result = service.solicitResponse(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml index 3ac1a0c8d3..2ec396c67c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/gatewayWithResponseCorrelator.xml @@ -11,7 +11,7 @@ - + @@ -20,7 +20,7 @@ - +