diff --git a/org.springframework.integration.rmi/src/main/java/org/springframework/integration/rmi/RmiInboundGateway.java b/org.springframework.integration.rmi/src/main/java/org/springframework/integration/rmi/RmiInboundGateway.java
index ae068d8a9f..415b4bdb77 100644
--- a/org.springframework.integration.rmi/src/main/java/org/springframework/integration/rmi/RmiInboundGateway.java
+++ b/org.springframework.integration.rmi/src/main/java/org/springframework/integration/rmi/RmiInboundGateway.java
@@ -19,6 +19,7 @@ package org.springframework.integration.rmi;
import java.rmi.RemoteException;
import java.rmi.registry.Registry;
+import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.adapter.MessageHandler;
import org.springframework.integration.adapter.RemotingInboundGatewaySupport;
import org.springframework.integration.channel.MessageChannel;
@@ -32,7 +33,7 @@ import org.springframework.util.StringUtils;
*
* @author Mark Fisher
*/
-public class RmiInboundGateway extends RemotingInboundGatewaySupport {
+public class RmiInboundGateway extends RemotingInboundGatewaySupport implements InitializingBean {
public static final String SERVICE_NAME_PREFIX = "org.springframewok.integration.rmiGateway.";
@@ -71,7 +72,6 @@ public class RmiInboundGateway extends RemotingInboundGatewaySupport {
this.remoteInvocationExecutor = remoteInvocationExecutor;
}
- @Override
public void afterPropertiesSet() throws RemoteException {
RmiServiceExporter exporter = new RmiServiceExporter();
if (this.registryHost != null) {
diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/annotation/cafeDemo.xml b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/annotation/cafeDemo.xml
index 492a389ee4..aaed429cac 100644
--- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/annotation/cafeDemo.xml
+++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/annotation/cafeDemo.xml
@@ -18,7 +18,9 @@
-
+
diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/xml/cafeDemo.xml b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/xml/cafeDemo.xml
index f8f7885485..d786cdd93e 100644
--- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/xml/cafeDemo.xml
+++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/xml/cafeDemo.xml
@@ -12,7 +12,9 @@
-
+
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd
index d866d1a00b..d3d076a312 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd
@@ -154,11 +154,10 @@
-
-
-
-
-
+
+
+
+
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java
index b6a351f60d..0dbb9cd750 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java
@@ -16,8 +16,6 @@
package org.springframework.integration.gateway;
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.integration.ConfigurationException;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.bus.MessageBusAware;
import org.springframework.integration.channel.MessageChannel;
@@ -36,7 +34,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
-public abstract class AbstractMessagingGateway implements MessagingGateway, MessageBusAware, InitializingBean {
+public abstract class AbstractMessagingGateway implements MessagingGateway, MessageBusAware {
private volatile MessageChannel requestChannel;
@@ -95,15 +93,9 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
this.messageBus = messageBus;
}
- public void afterPropertiesSet() throws Exception {
- Assert.notNull(this.requestChannel, "requestChannel must not be null");
- }
-
public void send(Object object) {
- if (this.requestChannel == null) {
- throw new IllegalStateException(
- "send is not supported, because no request channel has been configured");
- }
+ Assert.state(this.requestChannel != null,
+ "send is not supported, because no request channel has been configured");
Message> message = this.toMessage(object);
Assert.notNull(message, "message must not be null");
if (!this.channelTemplate.send(message, this.requestChannel)) {
@@ -112,10 +104,8 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
}
public Object receive() {
- if (this.replyChannel == null || !(this.replyChannel instanceof PollableChannel)) {
- throw new IllegalStateException(
- "no-arg receive is not supported, because no pollable reply channel has been configured");
- }
+ Assert.state(this.replyChannel != null && (this.replyChannel instanceof PollableChannel),
+ "no-arg receive is not supported, because no pollable reply channel has been configured");
Message> message = this.channelTemplate.receive((PollableChannel) this.replyChannel);
return this.fromMessage(message);
}
@@ -154,9 +144,7 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
if (this.replyMessageCorrelator != null) {
return;
}
- if (this.messageBus == null) {
- throw new ConfigurationException("No MessageBus available. Cannot register ReplyMessageCorrelator.");
- }
+ Assert.state(this.messageBus != null, "No MessageBus available. Cannot register ReplyMessageCorrelator.");
ReplyMessageCorrelator correlator = new ReplyMessageCorrelator();
correlator.setBeanName("internal.correlator." + this);
correlator.setInputChannel(this.replyChannel);
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
index 1dfe9fbfa6..f4030949f4 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
@@ -17,9 +17,12 @@
package org.springframework.integration.gateway;
import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.Map;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
+
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.SimpleTypeConverter;
@@ -30,8 +33,12 @@ import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.config.MessageBusParser;
+import org.springframework.integration.endpoint.MessagingGateway;
import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MethodParameterMessageMapper;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -41,17 +48,28 @@ import org.springframework.util.ClassUtils;
*
* @author Mark Fisher
*/
-public class GatewayProxyFactoryBean extends SimpleMessagingGateway
- implements FactoryBean, MethodInterceptor, InitializingBean, BeanClassLoaderAware, BeanFactoryAware {
+public class GatewayProxyFactoryBean implements FactoryBean, MethodInterceptor, InitializingBean, BeanClassLoaderAware, BeanFactoryAware {
private volatile Class> serviceInterface;
+ private volatile MessageChannel defaultRequestChannel;
+
+ private volatile PollableChannel defaultReplyChannel;
+
+ private volatile long defaultRequestTimeout = -1;
+
+ private volatile long defaultReplyTimeout = -1;
+
private volatile TypeConverter typeConverter = new SimpleTypeConverter();
private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
private volatile Object serviceProxy;
+ private final Map gatewayMap = new HashMap();
+
+ private volatile MessageBus messageBus;
+
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
@@ -64,6 +82,48 @@ public class GatewayProxyFactoryBean extends SimpleMessagingGateway
this.serviceInterface = serviceInterface;
}
+ /**
+ * Set the default request channel.
+ *
+ * @param defaulRequestChannel the channel to which request messages will
+ * be sent if no request channel has been configured with an annotation
+ */
+ public void setDefaultRequestChannel(MessageChannel defaultRequestChannel) {
+ this.defaultRequestChannel = defaultRequestChannel;
+ }
+
+ /**
+ * Set the default reply channel. If no default reply channel is provided,
+ * and no reply channel is configured with annotations, an anonymous,
+ * temporary channel will be used for handling replies.
+ *
+ * @param replyChannel the channel from which reply messages will be
+ * received if no reply channel has been configured with an annotation
+ */
+ public void setDefaultReplyChannel(PollableChannel defaultReplyChannel) {
+ this.defaultReplyChannel = defaultReplyChannel;
+ }
+
+ /**
+ * Set the default timeout value for sending request messages. If not
+ * explicitly configured with an annotation, this value will be used.
+ *
+ * @param defaultRequestTimeout the timeout value in milliseconds
+ */
+ public void setDefaultRequestTimeout(long defaultRequestTimeout) {
+ this.defaultRequestTimeout = defaultRequestTimeout;
+ }
+
+ /**
+ * Set the default timeout value for receiving reply messages. If not
+ * explicitly configured with an annotation, this value will be used.
+ *
+ * @param defaultReplyTimeout the timeout value in milliseconds
+ */
+ public void setDefaultReplyTimeout(long defaultReplyTimeout) {
+ this.defaultReplyTimeout = defaultReplyTimeout;
+ }
+
public void setTypeConverter(TypeConverter typeConverter) {
Assert.notNull(typeConverter, "typeConverter must not be null");
this.typeConverter = typeConverter;
@@ -74,11 +134,10 @@ public class GatewayProxyFactoryBean extends SimpleMessagingGateway
}
public void setBeanFactory(BeanFactory beanFactory) {
- this.setMessageBus(
- (MessageBus) beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
+ this.messageBus = (MessageBus) beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
}
- public void afterPropertiesSet() {
+ public void afterPropertiesSet() throws Exception {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
@@ -86,6 +145,11 @@ public class GatewayProxyFactoryBean extends SimpleMessagingGateway
if (this.serviceInterface == null) {
throw new IllegalArgumentException("'serviceInterface' must not be null");
}
+ Method[] methods = this.serviceInterface.getDeclaredMethods();
+ for (Method method : methods) {
+ MessagingGateway gateway = this.createGatewayForMethod(method);
+ this.gatewayMap.put(method, gateway);
+ }
this.serviceProxy = new ProxyFactory(this.serviceInterface, this).getProxy(this.beanClassLoader);
this.initialized = true;
}
@@ -119,6 +183,7 @@ public class GatewayProxyFactoryBean extends SimpleMessagingGateway
this.afterPropertiesSet();
}
Method method = invocation.getMethod();
+ MessagingGateway gateway = this.gatewayMap.get(method);
Class> returnType = method.getReturnType();
boolean isReturnTypeMessage = Message.class.isAssignableFrom(returnType);
boolean shouldReply = returnType != void.class;
@@ -127,22 +192,34 @@ public class GatewayProxyFactoryBean extends SimpleMessagingGateway
if (paramCount == 0) {
if (shouldReply) {
if (isReturnTypeMessage) {
- return this.receive();
+ return gateway.receive();
}
- response = this.receive();
+ response = gateway.receive();
}
}
else {
- Object payload = (paramCount == 1) ? invocation.getArguments()[0] : invocation.getArguments();
+ Object[] args = invocation.getArguments();
if (shouldReply) {
- response = isReturnTypeMessage ? this.sendAndReceiveMessage(payload) : this.sendAndReceive(payload);
+ response = isReturnTypeMessage ? gateway.sendAndReceiveMessage(args) : gateway.sendAndReceive(args);
}
else {
- this.send(payload);
+ gateway.send(args);
response = null;
}
}
return (response != null) ? this.typeConverter.convertIfNecessary(response, returnType) : null;
}
+ private MessagingGateway createGatewayForMethod(Method method) throws Exception {
+ SimpleMessagingGateway gateway = new SimpleMessagingGateway(
+ new MethodParameterMessageMapper(method), new SimpleMessageMapper());
+ gateway.setMessageBus(this.messageBus);
+ //TODO: get request and reply channels from annotation, else fall back to these defaults
+ gateway.setRequestChannel(this.defaultRequestChannel);
+ gateway.setReplyChannel(this.defaultReplyChannel);
+ gateway.setRequestTimeout(this.defaultRequestTimeout);
+ gateway.setReplyTimeout(this.defaultReplyTimeout);
+ return gateway;
+ }
+
}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/DefaultMessageMapper.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessageMapper.java
similarity index 78%
rename from org.springframework.integration/src/main/java/org/springframework/integration/gateway/DefaultMessageMapper.java
rename to org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessageMapper.java
index eaa68c1dd5..dc8a3fc039 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/DefaultMessageMapper.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessageMapper.java
@@ -18,14 +18,17 @@ package org.springframework.integration.gateway;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
-import org.springframework.integration.message.MessageMapper;
+import org.springframework.integration.message.InboundMessageMapper;
+import org.springframework.integration.message.OutboundMessageMapper;
/**
- * A default implementation of the {@link MessageMapper} strategy interface.
+ * An implementation of the {@link InboundMessageMapper} and
+ * {@link OutboundMessageMapper} strategy interfaces that maps directly to and
+ * from the Message payload instance.
*
* @author Mark Fisher
*/
-public class DefaultMessageMapper implements MessageMapper