diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/AbstractMessageHandlingSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java similarity index 76% rename from spring-integration-adapters/src/main/java/org/springframework/integration/adapter/AbstractMessageHandlingSourceAdapter.java rename to spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java index 4f1e321f6a..160d57eb5c 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/AbstractMessageHandlingSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java @@ -25,13 +25,15 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.RequestReplyTemplate; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; +import org.springframework.util.Assert; /** - * Abstract base class for source adapters that handle request Messages. + * A source adapter that implements the {@link MessageHandler} interface. It may + * be used as a base class for source adapters with request-reply behavior. * * @author Mark Fisher */ -public abstract class AbstractMessageHandlingSourceAdapter implements SourceAdapter, MessageHandler, InitializingBean { +public class MessageHandlingSourceAdapter implements SourceAdapter, MessageHandler, InitializingBean { private final Log logger = LogFactory.getLog(this.getClass()); @@ -50,6 +52,28 @@ public abstract class AbstractMessageHandlingSourceAdapter implements SourceAdap private volatile boolean initialized; + /** + * Create an adapter that sends to the provided channel. + * + * @param channel the channel where messages will be sent, must not be + * null. + */ + public MessageHandlingSourceAdapter(MessageChannel channel) { + Assert.notNull(channel, "'channel' must not be null"); + this.channel = channel; + } + + /** + * No-arg constructor for configuration via setters. Note that upon + * initialization, this adapter will throw an exception if a + * {@link MessageChannel} has not been provided. + * + * @see #setChannel(MessageChannel) + */ + public MessageHandlingSourceAdapter() { + } + + public void setChannel(MessageChannel channel) { this.channel = channel; } @@ -76,8 +100,8 @@ public abstract class AbstractMessageHandlingSourceAdapter implements SourceAdap public final void afterPropertiesSet() throws Exception { if (this.channel == null) { - throw new MessagingConfigurationException("The 'channel' property of '" + - this.getClass().getName() + "' must not be null."); + throw new MessagingConfigurationException("The 'channel' property of '" + this.getClass().getName() + + "' must not be null."); } synchronized (this.lifecycleMonitor) { if (this.initialized) { @@ -115,8 +139,8 @@ public abstract class AbstractMessageHandlingSourceAdapter implements SourceAdap } if (!this.expectReply) { if (!this.channel.send(message, this.sendTimeout) && logger.isWarnEnabled()) { - logger.warn("failed to send message to channel '" + this.channel + - "' within timeout of " + this.sendTimeout + " milliseconds"); + logger.warn("failed to send message to channel '" + this.channel + "' within timeout of " + + this.sendTimeout + " milliseconds"); } return null; } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java index c29018af85..741f84f484 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java @@ -21,7 +21,7 @@ import org.w3c.dom.Element; import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.MessagingConfigurationException; import org.springframework.util.StringUtils; @@ -31,7 +31,7 @@ import org.springframework.util.StringUtils; * * @author Mark Fisher */ -public abstract class AbstractRequestReplySourceAdapterParser extends AbstractSingleBeanDefinitionParser { +public abstract class AbstractRequestReplySourceAdapterParser extends AbstractSimpleBeanDefinitionParser { protected abstract Class getBeanClass(Element element); @@ -48,7 +48,13 @@ public abstract class AbstractRequestReplySourceAdapterParser extends AbstractSi return id; } - protected void doParse(Element element, BeanDefinitionBuilder builder) { + @Override + protected boolean isEligibleAttribute(String attributeName) { + return !attributeName.equals("name") && super.isEligibleAttribute(attributeName); + } + + @Override + protected void postProcess(BeanDefinitionBuilder builder, Element element) { String channelRef = element.getAttribute("channel"); if (!StringUtils.hasText(channelRef)) { throw new MessagingConfigurationException("a 'channel' reference is required"); @@ -63,6 +69,13 @@ public abstract class AbstractRequestReplySourceAdapterParser extends AbstractSi if (StringUtils.hasText(receiveTimeout)) { builder.addPropertyValue("receiveTimeout", Long.parseLong(receiveTimeout)); } + this.doPostProcess(builder, element); + } + + /** + * Subclasses may add to the bean definition by overriding this method. + */ + protected void doPostProcess(BeanDefinitionBuilder builder, Element element) { } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd index 169c26cb5c..7f09130a6e 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd @@ -93,12 +93,21 @@ - - - + + + + Defines an rmi-based source channel adapter. - - + + + + + + + + + + @@ -110,6 +119,7 @@ + diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapter.java index e23e38e6bb..51b0ad4bca 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapter.java @@ -22,7 +22,7 @@ import javax.servlet.ServletException; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; -import org.springframework.integration.adapter.AbstractMessageHandlingSourceAdapter; +import org.springframework.integration.adapter.MessageHandlingSourceAdapter; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.MessageHandlingException; @@ -59,7 +59,7 @@ import org.springframework.web.HttpRequestHandler; * * @author Mark Fisher */ -public class HttpInvokerSourceAdapter extends AbstractMessageHandlingSourceAdapter implements HttpRequestHandler { +public class HttpInvokerSourceAdapter extends MessageHandlingSourceAdapter implements HttpRequestHandler { private volatile HttpInvokerServiceExporter exporter; diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java index 1f9c90d5e1..3c95df43d5 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java @@ -17,23 +17,32 @@ package org.springframework.integration.adapter.rmi; import java.rmi.RemoteException; +import java.rmi.registry.Registry; import org.springframework.integration.MessagingConfigurationException; -import org.springframework.integration.adapter.AbstractMessageHandlingSourceAdapter; +import org.springframework.integration.adapter.MessageHandlingSourceAdapter; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.remoting.rmi.RmiServiceExporter; +import org.springframework.remoting.support.RemoteInvocationExecutor; /** * A source channel adapter for RMI-based remoting. * * @author Mark Fisher */ -public class RmiSourceAdapter extends AbstractMessageHandlingSourceAdapter { +public class RmiSourceAdapter extends MessageHandlingSourceAdapter { public static final String SERVICE_NAME_PREFIX = "internal.rmiSourceAdapter."; + private volatile String registryHost; + + private volatile int registryPort = Registry.REGISTRY_PORT; + + private volatile RemoteInvocationExecutor remoteInvocationExecutor; + + public RmiSourceAdapter() { super(); } @@ -44,12 +53,31 @@ public class RmiSourceAdapter extends AbstractMessageHandlingSourceAdapter { } + public void setRegistryHost(String registryHost) { + this.registryHost = registryHost; + } + + public void setRegistryPort(int registryPort) { + this.registryPort = registryPort; + } + + public void setRemoteInvocationExecutor(RemoteInvocationExecutor remoteInvocationExecutor) { + this.remoteInvocationExecutor = remoteInvocationExecutor; + } + public void initialize() throws RemoteException { String channelName = this.getChannel().getName(); if (channelName == null) { throw new MessagingConfigurationException("RmiSourceAdapter's MessageChannel must have a 'name'"); } RmiServiceExporter exporter = new RmiServiceExporter(); + if (this.registryHost != null) { + exporter.setRegistryHost(this.registryHost); + } + exporter.setRegistryPort(this.registryPort); + if (this.remoteInvocationExecutor != null) { + exporter.setRemoteInvocationExecutor(this.remoteInvocationExecutor); + } exporter.setService(this); exporter.setServiceInterface(MessageHandler.class); exporter.setServiceName(SERVICE_NAME_PREFIX + channelName); diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParser.java index 2d92973143..fc59f283fc 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParser.java @@ -18,8 +18,10 @@ package org.springframework.integration.adapter.rmi.config; import org.w3c.dom.Element; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.integration.adapter.config.AbstractRequestReplySourceAdapterParser; import org.springframework.integration.adapter.rmi.RmiSourceAdapter; +import org.springframework.util.StringUtils; /** * Parser for the <rmi-source/> element. @@ -28,9 +30,26 @@ import org.springframework.integration.adapter.rmi.RmiSourceAdapter; */ public class RmiSourceAdapterParser extends AbstractRequestReplySourceAdapterParser { + private static final String REMOTE_INVOCATION_EXECUTOR_ATTRIBUTE = "remote-invocation-executor"; + + @Override protected Class getBeanClass(Element element) { return RmiSourceAdapter.class; } + @Override + protected boolean isEligibleAttribute(String attributeName) { + return !attributeName.equals(REMOTE_INVOCATION_EXECUTOR_ATTRIBUTE) + && super.isEligibleAttribute(attributeName); + } + + @Override + protected void doPostProcess(BeanDefinitionBuilder builder, Element element) { + String executorRef = element.getAttribute(REMOTE_INVOCATION_EXECUTOR_ATTRIBUTE); + if (StringUtils.hasText(executorRef)) { + builder.addPropertyReference("remoteInvocationExecutor", executorRef); + } + } + } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParser.java index 486ca578ee..236f3bc68b 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParser.java @@ -58,7 +58,9 @@ public class RmiTargetAdapterParser extends AbstractSingleBeanDefinitionParser { throw new MessagingConfigurationException( "The 'host', 'local-channel', and 'remote-channel' attributes are all required"); } - String url = "rmi://" + host + "/" + RmiSourceAdapter.SERVICE_NAME_PREFIX + remoteChannel; + String portAttribute = element.getAttribute("port"); + String port = StringUtils.hasText(portAttribute) ? portAttribute : "1099"; + String url = "rmi://" + host + ":" + port + "/" + RmiSourceAdapter.SERVICE_NAME_PREFIX + remoteChannel; adapterDef.getConstructorArgumentValues().addGenericArgumentValue(url); String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDef); parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, adapterBeanName)); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java index 56d6e82159..e13ba3ffb5 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.adapter.rmi.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import org.junit.Test; @@ -57,4 +58,33 @@ public class RmiSourceAdapterParserTests { assertEquals(456L, accessor.getPropertyValue("receiveTimeout")); } + @Test + public void testAdapterWithHost() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "rmiSourceAdapterParserTests.xml", this.getClass()); + RmiSourceAdapter adapter = (RmiSourceAdapter) context.getBean("adapterWithHost"); + DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); + assertEquals("localhost", accessor.getPropertyValue("registryHost")); + } + + @Test + public void testAdapterWithPort() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "rmiSourceAdapterParserTests.xml", this.getClass()); + RmiSourceAdapter adapter = (RmiSourceAdapter) context.getBean("adapterWithPort"); + DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); + assertEquals(1234, accessor.getPropertyValue("registryPort")); + } + + @Test + public void testAdapterWithRemoteInvocationExecutorReference() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "rmiSourceAdapterParserTests.xml", this.getClass()); + RmiSourceAdapter adapter = (RmiSourceAdapter) context.getBean("adapterWithExecutorRef"); + DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); + Object remoteInvocationExecutor = accessor.getPropertyValue("remoteInvocationExecutor"); + assertNotNull(remoteInvocationExecutor); + assertEquals(StubRemoteInvocationExecutor.class, remoteInvocationExecutor.getClass()); + } + } diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/StubRemoteInvocationExecutor.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/StubRemoteInvocationExecutor.java new file mode 100644 index 0000000000..9e254df922 --- /dev/null +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/StubRemoteInvocationExecutor.java @@ -0,0 +1,26 @@ +/* + * 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.adapter.rmi.config; + +import org.springframework.remoting.support.DefaultRemoteInvocationExecutor; + +/** + * @author Mark Fisher + */ +public class StubRemoteInvocationExecutor extends DefaultRemoteInvocationExecutor { + +} diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml index f0d373c7b9..5c9b00404d 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml @@ -16,4 +16,12 @@ + + + + + + + + \ No newline at end of file