Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GatewayProxyFactoryBean now supports receive-only operations. Also added support for timeouts and type-conversion.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-25 19:14:14 +00:00
parent 24a15520ba
commit 64c498838c
5 changed files with 187 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -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.");
}

3
spring-integration-core/src/main/java/org/springframework/integration/message/DefaultMessageCreator.java
View File

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

84
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java
View File

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

10
spring-integration-core/src/test/java/org/springframework/integration/gateway/TestService.java
View File

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

8
spring-integration-core/src/test/java/org/springframework/integration/message/DefaultMessageCreatorTests.java
View File

@@ -41,6 +41,14 @@ public class DefaultMessageCreatorTests {
assertEquals(test, message.getPayload());
}
@Test
public void testMessage() {
DefaultMessageCreator creator = new DefaultMessageCreator();
Message<?> inputMessage = new GenericMessage<String>("testing");
Message<?> message = creator.createMessage(inputMessage);
assertEquals(inputMessage, message);
}
@Test
public void testNull() {
DefaultMessageCreator creator = new DefaultMessageCreator();