Added MessagingGateway as a subclass of RequestReplyTemplate that is also a superclass for the GatewayProxyFactoryBean.

This commit is contained in:
Mark Fisher
2008-05-01 13:09:11 +00:00
parent 15ead1c3ad
commit 78905d9bcb
6 changed files with 306 additions and 216 deletions

View File

@@ -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;
}

View File

@@ -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 <code>null</code>
* @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 <code>null</code>
*/
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<Message<?>> clear() {
return null;
}
public DispatcherPolicy getDispatcherPolicy() {
return null;
}
public String getName() {
return null;
}
public List<Message<?>> 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);
}
}
}

View File

@@ -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;
}
}

View File

@@ -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;
}
}

View File

@@ -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();

View File

@@ -11,7 +11,7 @@
<channel id="requestChannel"/>
<channel id="responseChannel">
<channel id="replyChannel">
<interceptor ref="interceptor"/>
</channel>
@@ -20,7 +20,7 @@
<beans:bean id="proxy" class="org.springframework.integration.gateway.GatewayProxyFactoryBean">
<beans:property name="serviceInterface" value="org.springframework.integration.gateway.TestService"/>
<beans:property name="requestChannel" ref="requestChannel"/>
<beans:property name="responseChannel" ref="responseChannel"/>
<beans:property name="replyChannel" ref="replyChannel"/>
</beans:bean>
<beans:bean id="handler" class="org.springframework.integration.gateway.TestHandler"/>