Added RmiSourceAdapter, RmiTargetAdapter, RmiSourceAdapterParser, and RmiTargetAdapterParser (INT-59, INT-60, and INT-146).

spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers

@@ -2,4 +2,6 @@ file-source=org.springframework.integration.adapter.file.config.FileSourceAdapte
 file-target=org.springframework.integration.adapter.file.config.FileTargetAdapterParser
 jms-source=org.springframework.integration.adapter.jms.config.JmsSourceAdapterParser
 jms-target=org.springframework.integration.adapter.jms.config.JmsTargetAdapterParser
+rmi-source=org.springframework.integration.adapter.rmi.config.RmiSourceAdapterParser
+rmi-target=org.springframework.integration.adapter.rmi.config.RmiTargetAdapterParser
 mail-target=org.springframework.integration.adapter.mail.config.MailTargetAdapterParser

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

@@ -93,6 +93,33 @@
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="rmi-source">
+		<xsd:complexType>
+			<xsd:annotation>
+				<xsd:documentation>
+	Defines an rmi-based source channel adapter.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:attribute name="id" type="xsd:string"/>
+			<xsd:attribute name="expect-reply" type="xsd:boolean" default="true"/>
+			<xsd:attribute name="channel" type="xsd:string" use="required"/>
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="rmi-target">
+		<xsd:complexType>
+			<xsd:annotation>
+				<xsd:documentation>
+	Defines an rmi-based target channel adapter.
+				</xsd:documentation>
+			</xsd:annotation>
+			<xsd:attribute name="id" type="xsd:string"/>
+			<xsd:attribute name="host" type="xsd:string" use="required"/>
+			<xsd:attribute name="local-channel" type="xsd:string" use="required"/>
+			<xsd:attribute name="remote-channel" type="xsd:string" use="required"/>
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:element name="mail-target">
 		<xsd:complexType>
 			<xsd:annotation>

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

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2002-2007 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;
+
+import java.rmi.RemoteException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.MessagingConfigurationException;
+import org.springframework.integration.adapter.SourceAdapter;
+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.remoting.rmi.RmiServiceExporter;
+
+/**
+ * A source channel adapter for RMI-based remoting.
+ * 
+ * @author Mark Fisher
+ */
+public class RmiSourceAdapter implements SourceAdapter, MessageHandler, InitializingBean {
+
+	public static final String SERVICE_NAME_PREFIX = "internal.rmiSourceAdapter.";
+
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private volatile MessageChannel channel;
+
+	private volatile RequestReplyTemplate requestReplyTemplate;
+
+	private volatile boolean expectReply = true;
+
+	private volatile long sendTimeout = -1;
+
+	private volatile long receiveTimeout = -1;
+
+
+	public void setChannel(MessageChannel channel) {
+		this.channel = channel;
+	}
+
+	/**
+	 * Specify whether the handle method should be expected to return a reply.
+	 * The default is '<code>true</code>'.
+	 */
+	public void setExpectReply(boolean expectReply) {
+		this.expectReply = expectReply;
+	}
+
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
+
+	public void setReceiveTimeout(long receiveTimeout) {
+		this.receiveTimeout = receiveTimeout;
+	}
+
+	public void afterPropertiesSet() throws RemoteException {
+		String channelName = this.channel.getName();
+		if (channelName == null) {
+			throw new MessagingConfigurationException("RmiSourceAdapter's MessageChannel must have a 'name'");
+		}
+		this.requestReplyTemplate = new RequestReplyTemplate(this.channel);
+		this.requestReplyTemplate.setDefaultSendTimeout(this.sendTimeout);
+		this.requestReplyTemplate.setDefaultReceiveTimeout(this.receiveTimeout);
+		RmiServiceExporter exporter = new RmiServiceExporter();
+		exporter.setService(this);
+		exporter.setServiceInterface(MessageHandler.class);
+		exporter.setServiceName(SERVICE_NAME_PREFIX + channelName);
+		exporter.afterPropertiesSet();
+	}
+
+	public Message<?> handle(Message<?> message) {
+		if (this.requestReplyTemplate == null) {
+			try {
+				this.afterPropertiesSet();
+			}
+			catch (RemoteException e) {
+				throw new MessagingConfigurationException("unable to initialize RmiSourceAdapter", e);
+			}
+		}
+		if (!this.expectReply) {
+			if (!this.channel.send(message, this.sendTimeout) && logger.isWarnEnabled()) {
+				logger.warn("failed to send message to channel '" + channel +
+						"' within timeout of " + this.sendTimeout + " milliseconds");
+			}
+			return null;
+		}
+		return this.requestReplyTemplate.request(message);
+	}
+
+}

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiTargetAdapter.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2002-2007 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;
+
+import java.io.Serializable;
+
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.remoting.RemoteAccessException;
+import org.springframework.remoting.rmi.RmiProxyFactoryBean;
+
+/**
+ * A target channel adapter for RMI-based remoting.
+ * 
+ * @author Mark Fisher
+ */
+public class RmiTargetAdapter implements MessageHandler {
+
+	private final MessageHandler handlerProxy;
+
+
+	public RmiTargetAdapter(String url) {
+		RmiProxyFactoryBean proxyFactory = new RmiProxyFactoryBean();
+		proxyFactory.setServiceInterface(MessageHandler.class);
+		proxyFactory.setServiceUrl(url);
+		proxyFactory.setLookupStubOnStartup(false);
+		proxyFactory.setRefreshStubOnConnectFailure(true);
+		proxyFactory.afterPropertiesSet();
+		this.handlerProxy = (MessageHandler) proxyFactory.getObject();
+	}
+
+
+	public Message<?> handle(Message<?> message) {
+		this.verifySerializability(message);
+		try {
+			return this.handlerProxy.handle(message);
+		}
+		catch (RemoteAccessException e) {
+			throw new MessageHandlingException("unable to handle message remotely", e);
+		}
+	}
+
+	private void verifySerializability(Message<?> message) {
+		if (!(message.getPayload() instanceof Serializable)) {
+			throw new MessageHandlingException(message,
+					"RmiTargetAdapter expects a Serializable payload type " +
+					"but encountered '" + message.getPayload().getClass().getName() + "'");
+		}
+		for (String attributeName : message.getHeader().getAttributeNames()) {
+			Object attribute = message.getHeader().getAttribute(attributeName);
+			if (!(attribute instanceof Serializable)) {
+				throw new MessageHandlingException(message,
+						"RmiTargetAdapter expects Serializable attribute types " +
+						"but encountered '" + attribute.getClass().getName() + "' for the attribute '" +
+						attributeName + "'");				
+			}
+		}
+	}
+
+}

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2002-2007 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.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.integration.MessagingConfigurationException;
+import org.springframework.integration.adapter.rmi.RmiSourceAdapter;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <rmi-source/> element. 
+ * 
+ * @author Mark Fisher
+ */
+public class RmiSourceAdapterParser extends AbstractSingleBeanDefinitionParser {
+
+	protected Class<?> getBeanClass(Element element) {
+		return RmiSourceAdapter.class;
+	}
+
+	protected boolean shouldGenerateId() {
+		return false;
+	}
+
+	protected boolean shouldGenerateIdAsFallback() {
+		return true;
+	}
+
+	protected void doParse(Element element, BeanDefinitionBuilder builder) {
+		String channelRef = element.getAttribute("channel");
+		if (!StringUtils.hasText(channelRef)) {
+			throw new MessagingConfigurationException("a 'channel' reference is required");
+		}
+		builder.addPropertyReference("channel", channelRef);
+		builder.addPropertyValue("expectReply", element.getAttribute("expect-reply").equals("true"));
+	}
+
+}

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

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2002-2007 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.w3c.dom.Element;
+
+import org.springframework.beans.factory.parsing.BeanComponentDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.MessagingConfigurationException;
+import org.springframework.integration.adapter.rmi.RmiSourceAdapter;
+import org.springframework.integration.adapter.rmi.RmiTargetAdapter;
+import org.springframework.integration.endpoint.DefaultMessageEndpoint;
+import org.springframework.integration.scheduling.Subscription;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the <rmi-target/> element. 
+ * 
+ * @author Mark Fisher
+ */
+public class RmiTargetAdapterParser extends AbstractSingleBeanDefinitionParser {
+
+	protected Class<?> getBeanClass(Element element) {
+		return DefaultMessageEndpoint.class;
+	}
+
+	protected boolean shouldGenerateId() {
+		return false;
+	}
+
+	protected boolean shouldGenerateIdAsFallback() {
+		return true;
+	}
+
+	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+		RootBeanDefinition adapterDef = new RootBeanDefinition(RmiTargetAdapter.class);
+		String host = element.getAttribute("host");
+		String localChannel = element.getAttribute("local-channel");
+		String remoteChannel = element.getAttribute("remote-channel");
+		if (!(StringUtils.hasText(host) && StringUtils.hasText(localChannel) && StringUtils.hasText(remoteChannel))) {
+			throw new MessagingConfigurationException(
+					"The 'host', 'local-channel', and 'remote-channel' attributes are all required");
+		}
+		String url = "rmi://" + host + "/" + RmiSourceAdapter.SERVICE_NAME_PREFIX + remoteChannel;
+		adapterDef.getConstructorArgumentValues().addGenericArgumentValue(url);
+		String adapterBeanName = parserContext.getReaderContext().generateBeanName(adapterDef);
+		parserContext.registerBeanComponent(new BeanComponentDefinition(adapterDef, adapterBeanName));
+		builder.addPropertyReference("handler", adapterBeanName);
+		Subscription subscription = new Subscription(localChannel);
+		builder.addPropertyValue("subscription", subscription);
+	}
+
+}

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/RmiTargetAdapterTests.java

@@ -0,0 +1,146 @@
+/*
+ * Copyright 2002-2007 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.rmi.RemoteException;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.remoting.RemoteLookupFailureException;
+import org.springframework.remoting.rmi.RmiServiceExporter;
+
+/**
+ * @author Mark Fisher
+ */
+public class RmiTargetAdapterTests {
+
+	private final RmiTargetAdapter adapter = new RmiTargetAdapter("rmi://localhost:1099/testRemoteHandler");
+
+
+	@Before
+	public void createExporter() throws RemoteException {
+		RmiServiceExporter exporter = new RmiServiceExporter();
+		exporter.setService(new TestHandler());
+		exporter.setServiceInterface(MessageHandler.class);
+		exporter.setServiceName("testRemoteHandler");
+		exporter.afterPropertiesSet();
+	}
+
+	@Test
+	public void testSerializablePayload() throws RemoteException {
+		Message<?> replyMessage = adapter.handle(new StringMessage("test"));
+		assertNotNull(replyMessage);
+		assertEquals("TEST", replyMessage.getPayload());
+	}
+
+	@Test
+	public void testSerializableAttribute() throws RemoteException {
+		Message<?> requestMessage = new StringMessage("test");
+		requestMessage.getHeader().setAttribute("testAttribute", "foo");
+		Message<?> replyMessage = adapter.handle(requestMessage);
+		assertNotNull(replyMessage);
+		assertEquals("foo", replyMessage.getHeader().getAttribute("testAttribute"));
+	}
+
+	@Test
+	public void testProperty() throws RemoteException {
+		Message<?> requestMessage = new StringMessage("test");
+		requestMessage.getHeader().setProperty("testProperty", "bar");
+		Message<?> replyMessage = adapter.handle(requestMessage);
+		assertNotNull(replyMessage);
+		assertEquals("bar", replyMessage.getHeader().getProperty("testProperty"));
+	}
+
+	@Test(expected=MessageHandlingException.class)
+	public void testNonSerializablePayload() throws RemoteException {
+		NonSerializableTestObject payload = new NonSerializableTestObject();
+		Message<?> requestMessage = new GenericMessage<NonSerializableTestObject>(payload);
+		adapter.handle(requestMessage);
+	}
+
+	@Test(expected=MessageHandlingException.class)
+	public void testNonSerializableAttribute() throws RemoteException {
+		Message<?> requestMessage = new StringMessage("test");
+		requestMessage.getHeader().setAttribute("testAttribute", new NonSerializableTestObject());
+		adapter.handle(requestMessage);
+	}
+
+	@Test
+	public void testInvalidServiceName() throws RemoteException {
+		RmiTargetAdapter adapter = new RmiTargetAdapter("rmi://localhost:1099/noSuchService");
+		boolean exceptionThrown = false;
+		try {
+			adapter.handle(new StringMessage("test"));
+		}
+		catch (MessageHandlingException e) {
+			assertEquals(RemoteLookupFailureException.class, e.getCause().getClass());
+			exceptionThrown = true;
+		}
+		assertTrue(exceptionThrown);
+	}
+
+	@Test
+	public void testInvalidHost() {
+		RmiTargetAdapter adapter = new RmiTargetAdapter("rmi://noSuchHost:1099/testRemoteHandler");
+		boolean exceptionThrown = false;
+		try {
+			adapter.handle(new StringMessage("test"));
+		}
+		catch (MessageHandlingException e) {
+			assertEquals(RemoteLookupFailureException.class, e.getCause().getClass());
+			exceptionThrown = true;
+		}
+		assertTrue(exceptionThrown);
+	}
+
+	@Test
+	public void testInvalidUrl() throws RemoteException {
+		RmiTargetAdapter adapter = new RmiTargetAdapter("invalid");
+		boolean exceptionThrown = false;
+		try {
+			adapter.handle(new StringMessage("test"));
+		}
+		catch (MessageHandlingException e) {
+			assertEquals(RemoteLookupFailureException.class, e.getCause().getClass());
+			exceptionThrown = true;
+		}
+		assertTrue(exceptionThrown);
+	}
+
+
+	private static class TestHandler implements MessageHandler {
+
+		public Message<?> handle(Message<?> message) {
+			return new GenericMessage<String>(message.getPayload().toString().toUpperCase(), message.getHeader());
+		}
+	}
+
+
+	private static class NonSerializableTestObject {
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java

@@ -36,7 +36,7 @@ public class GenericMessage<T> implements Message<T> {
 
 	private final T payload;
 
-	private final IdGenerator defaultIdGenerator = new RandomUuidGenerator();
+	private transient final IdGenerator defaultIdGenerator = new RandomUuidGenerator();
 
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/message/Message.java

@@ -16,12 +16,14 @@
 
 package org.springframework.integration.message;
 
+import java.io.Serializable;
+
 /**
  * The central interface that any Message type must implement.
  * 
  * @author Mark Fisher
  */
-public interface Message<T> {
+public interface Message<T> extends Serializable {
 
 	Object getId();
 

spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.message;
 
+import java.io.Serializable;
 import java.util.Date;
 import java.util.HashSet;
 import java.util.Properties;
@@ -32,7 +33,7 @@ import java.util.concurrent.ConcurrentMap;
  * 
  * @author Mark Fisher
  */
-public class MessageHeader {
+public class MessageHeader implements Serializable {
 
 	private final Date timestamp = new Date();
 
@@ -40,7 +41,7 @@ public class MessageHeader {
 
 	private volatile Object correlationId;
 
-	private volatile Object returnAddress;
+	private transient volatile Object returnAddress;
 
 	private volatile int sequenceNumber = 1;