diff --git a/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers b/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers index fe3a26c73d..c0db7af08b 100644 --- a/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers +++ b/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 \ No newline at end of file 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 da8572d532..4de40072fb 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,6 +93,33 @@ + + + + + Defines an rmi-based source channel adapter. + + + + + + + + + + + + + Defines an rmi-based target channel adapter. + + + + + + + + + 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 new file mode 100644 index 0000000000..8427c37128 --- /dev/null +++ b/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 'true'. + */ + 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); + } + +} diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiTargetAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/rmi/RmiTargetAdapter.java new file mode 100644 index 0000000000..489c977ca8 --- /dev/null +++ b/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 + "'"); + } + } + } + +} 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 new file mode 100644 index 0000000000..b151f7b86a --- /dev/null +++ b/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")); + } + +} 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 new file mode 100644 index 0000000000..486ca578ee --- /dev/null +++ b/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); + } + +} diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/RmiTargetAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/RmiTargetAdapterTests.java new file mode 100644 index 0000000000..ab4eb9f8dc --- /dev/null +++ b/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(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(message.getPayload().toString().toUpperCase(), message.getHeader()); + } + } + + + private static class NonSerializableTestObject { + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java b/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java index 959087256d..efa1750030 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/GenericMessage.java @@ -36,7 +36,7 @@ public class GenericMessage implements Message { private final T payload; - private final IdGenerator defaultIdGenerator = new RandomUuidGenerator(); + private transient final IdGenerator defaultIdGenerator = new RandomUuidGenerator(); /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java b/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java index 5d83ded23f..5dc7b8ae5e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/Message.java +++ b/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 { +public interface Message extends Serializable { Object getId(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java index 93f457806e..4803c84634 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageHeader.java +++ b/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;