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

Added 'registry-host', 'registry-port', and 'remote-invocation-executor' attributes to the 'rmi-source' element, and added 'port' attribute to the 'rmi-target' element.

Browse Source
This commit is contained in:
Mark Fisher
2008-03-29 20:40:48 +00:00
parent dd6635b86e
commit f59d9e1bb0
10 changed files with 179 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/AbstractMessageHandlingSourceAdapter.java → spring-integration-adapters/src/main/java/org/springframework/integration/adapter/MessageHandlingSourceAdapter.java
View File

@@ -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
* <code>null</code>.
*/
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;
}

19
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/AbstractRequestReplySourceAdapterParser.java
View File

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

20
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd
View File

@@ -93,12 +93,21 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="rmi-source" type="requestReplySource">
<xsd:annotation>
<xsd:documentation>
<xsd:element name="rmi-source">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an rmi-based source channel adapter.
</xsd:documentation>
</xsd:annotation>
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
<xsd:extension base="requestReplySource">
<xsd:attribute name="registry-host" type="xsd:string"/>
<xsd:attribute name="registry-port" type="xsd:integer"/>
<xsd:attribute name="remote-invocation-executor" type="xsd:string"/>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="rmi-target">
@@ -110,6 +119,7 @@
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="host" type="xsd:string" use="required"/>
<xsd:attribute name="port" type="xsd:integer"/>
<xsd:attribute name="local-channel" type="xsd:string" use="required"/>
<xsd:attribute name="remote-channel" type="xsd:string" use="required"/>
</xsd:complexType>

4
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapter.java
View File

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

32
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiSourceAdapter.java
View File

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

19
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParser.java
View File

@@ -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 &lt;rmi-source/&gt; 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);
}
}
}

4
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParser.java
View File

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

30
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiSourceAdapterParserTests.java
View File

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

26
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/StubRemoteInvocationExecutor.java Normal file
View File

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

8
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/rmiSourceAdapterParserTests.xml
View File

@@ -16,4 +16,12 @@
<rmi-source id="adapterWithCustomProperties" channel="testChannel"
expect-reply="false" send-timeout="123" receive-timeout="456"/>
<rmi-source id="adapterWithHost" channel="testChannel" registry-host="localhost"/>
<rmi-source id="adapterWithPort" channel="testChannel" registry-port="1234"/>
<rmi-source id="adapterWithExecutorRef" channel="testChannel" remote-invocation-executor="invocationExecutor"/>
<beans:bean id="invocationExecutor" class="org.springframework.integration.adapter.rmi.config.StubRemoteInvocationExecutor"/>
</beans:beans>