diff --git a/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpClientInterceptor.java b/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpClientInterceptor.java new file mode 100644 index 00000000..10c4bcf0 --- /dev/null +++ b/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpClientInterceptor.java @@ -0,0 +1,112 @@ +/* + * Copyright 2002-2013 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.amqp.remoting.client; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter; +import org.springframework.remoting.RemoteProxyFailureException; +import org.springframework.remoting.support.DefaultRemoteInvocationFactory; +import org.springframework.remoting.support.RemoteAccessor; +import org.springframework.remoting.support.RemoteInvocation; +import org.springframework.remoting.support.RemoteInvocationFactory; +import org.springframework.remoting.support.RemoteInvocationResult; + +/** + * {@link org.aopalliance.intercept.MethodInterceptor} for accessing RMI-style AMQP services. + * + * @author David Bilge + * @author Gary Russell + * @since 1.2 + * @see AmqpInvokerServiceExporter + * @see AmqpProxyFactoryBean + * @see org.springframework.remoting.RemoteAccessException + */ +public class AmqpClientInterceptor extends RemoteAccessor implements MethodInterceptor { + + private AmqpTemplate amqpTemplate; + + private String routingKey = null; + + private RemoteInvocationFactory remoteInvocationFactory = new DefaultRemoteInvocationFactory(); + + @Override + public Object invoke(MethodInvocation invocation) throws Throwable { + RemoteInvocation remoteInvocation = getRemoteInvocationFactory().createRemoteInvocation(invocation); + + Object rawResult; + if (getRoutingKey() == null) { + // Use the template's default routing key + rawResult = amqpTemplate.convertSendAndReceive(remoteInvocation); + } + else { + rawResult = amqpTemplate.convertSendAndReceive(routingKey, remoteInvocation); + } + + if (rawResult == null) { + throw new RemoteProxyFailureException("No reply received - perhaps a timeout in the template?", null); + } + else if (!(rawResult instanceof RemoteInvocationResult)) { + throw new RemoteProxyFailureException("Expected a result of type " + + RemoteInvocationResult.class.getCanonicalName() + " but found " + + rawResult.getClass().getCanonicalName(), null); + } + + RemoteInvocationResult result = (RemoteInvocationResult) rawResult; + return result.recreate(); + } + + public AmqpTemplate getAmqpTemplate() { + return amqpTemplate; + } + + /** + * The AMQP template to be used for sending messages and receiving results. This class is using "Request/Reply" for + * sending messages as described in the Spring-AMQP + * documentation. + */ + public void setAmqpTemplate(AmqpTemplate amqpTemplate) { + this.amqpTemplate = amqpTemplate; + } + + public String getRoutingKey() { + return routingKey; + } + + /** + * The routing key to send calls to the service with. Use this to route the messages to a specific queue on the + * broker. If not set, the {@link AmqpTemplate}'s default routing key will be used. + *

+ * This property is useful if you want to use the same AmqpTemplate to talk to multiple services. + */ + public void setRoutingKey(String routingKey) { + this.routingKey = routingKey; + } + + public RemoteInvocationFactory getRemoteInvocationFactory() { + return remoteInvocationFactory; + } + + /** + * Set the RemoteInvocationFactory to use for this accessor. Default is a {@link DefaultRemoteInvocationFactory}. + *

+ * A custom invocation factory can add further context information to the invocation, for example user credentials. + */ + public void setRemoteInvocationFactory(RemoteInvocationFactory remoteInvocationFactory) { + this.remoteInvocationFactory = remoteInvocationFactory; + } + +} diff --git a/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpProxyFactoryBean.java b/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpProxyFactoryBean.java new file mode 100644 index 00000000..c5f89fb9 --- /dev/null +++ b/spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpProxyFactoryBean.java @@ -0,0 +1,70 @@ +/* + * Copyright 2002-2013 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.amqp.remoting.client; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.remoting.rmi.RmiServiceExporter; + +/** + * {@link FactoryBean} for AMQP proxies. Exposes the proxied service for use as a bean reference, using the specified + * service interface. Proxies will throw Spring's unchecked RemoteAccessException on remote invocation failure. + * + *

+ * This is intended for an "RMI-style" (i.e. synchroneous) usage of the AMQP protocol. Obviously, AMQP allows for a much + * broader scope of execution styles, which are not the scope of the mechanism at hand. + *

+ * Calling a method on the proxy will cause an AMQP message being sent according to the configured {@link AmqpTemplate}. + * This can be received and answered by an {@link AmqpInvokerServiceExporter}. + * + * @author David Bilge + * @since 1.2 + * @see #setServiceInterface + * @see AmqpClientInterceptor + * @see RmiServiceExporter + * @see org.springframework.remoting.RemoteAccessException + */ +public class AmqpProxyFactoryBean extends AmqpClientInterceptor implements FactoryBean, BeanClassLoaderAware, + InitializingBean { + + private Object serviceProxy; + + @Override + public void afterPropertiesSet() { + if (getServiceInterface() == null) { + throw new IllegalArgumentException("Property 'serviceInterface' is required"); + } + this.serviceProxy = new ProxyFactory(getServiceInterface(), this).getProxy(getBeanClassLoader()); + } + + @Override + public Object getObject() throws Exception { + return this.serviceProxy; + } + + @Override + public Class getObjectType() { + return getServiceInterface(); + } + + @Override + public boolean isSingleton() { + return true; + } + +} diff --git a/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java b/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java new file mode 100644 index 00000000..96bf920e --- /dev/null +++ b/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java @@ -0,0 +1,120 @@ +/* + * Copyright 2002-2013 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.amqp.remoting.service; + +import org.springframework.amqp.AmqpRejectAndDontRequeueException; +import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.remoting.client.AmqpProxyFactoryBean; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.remoting.support.RemoteInvocation; +import org.springframework.remoting.support.RemoteInvocationBasedExporter; +import org.springframework.remoting.support.RemoteInvocationResult; + +/** + * This message listener exposes a plain java service via AMQP. Such services can be accessed via plain AMQP or via + * {@link AmqpProxyFactoryBean}. + * + * To configure this message listener so that it actually receives method calls via AMQP, it needs to be put into a + * listener container. See {@link MessageListener}. + * + *

+ * When receiving a message, a service method is called according to the contained {@link RemoteInvocation}. The result + * of that invocation is returned as a {@link RemoteInvocationResult} contained in a message that is sent according to + * the ReplyToAddress of the received message. + * + *

+ * Please note that this exporter does not use the {@link MessageConverter} of the injected {@link AmqpTemplate} to + * convert incoming calls and their results. Instead you have to directly inject the MessageConverter into + * this class. + * + *

+ * This listener responds to "Request/Reply"-style messages as described here. + * + * @author David Bilge + * @since 1.2 + */ +public class AmqpInvokerServiceExporter extends RemoteInvocationBasedExporter implements MessageListener { + + private AmqpTemplate amqpTemplate; + + private MessageConverter messageConverter = new SimpleMessageConverter(); + + @Override + public void onMessage(Message message) { + Address replyToAddress = message.getMessageProperties().getReplyToAddress(); + if (replyToAddress == null) { + throw new AmqpRejectAndDontRequeueException("No replyToAddress in inbound AMQP Message"); + } + + Object invocationRaw = messageConverter.fromMessage(message); + if (invocationRaw == null || !(invocationRaw instanceof RemoteInvocation)) { + send(new RuntimeException("The message does not contain a RemoteInvocation payload"), replyToAddress); + return; + } + RemoteInvocation invocation = (RemoteInvocation) invocationRaw; + + RemoteInvocationResult remoteInvocationResult = invokeAndCreateResult(invocation, getService()); + send(remoteInvocationResult, replyToAddress); + } + + private void send(Object object, Address replyToAddress) { + Message message = messageConverter.toMessage(object, new MessageProperties()); + + getAmqpTemplate().send(replyToAddress.getExchangeName(), replyToAddress.getRoutingKey(), message); + } + + public AmqpTemplate getAmqpTemplate() { + return amqpTemplate; + } + + /** + * The AMQP template to use for sending the return value. + * + *

+ * Note that the exchange and routing key parameters on this template are ignored for these return messages. Instead + * of those the respective parameters from the original message's returnAddress are being used. + *

+ * Also, the templates {@link MessageConverter} is not used for the reply. + * @see {@link AmqpInvokerServiceExporter#setMessageConverter(MessageConverter)} + */ + public void setAmqpTemplate(AmqpTemplate amqpTemplate) { + this.amqpTemplate = amqpTemplate; + } + + public MessageConverter getMessageConverter() { + return messageConverter; + } + + /** + * Set the message converter for this remote service. Used to deserialize remote method calls and to serialize their + * return values. + *

+ * The default converter is a SimpleMessageConverter, which is able to handle byte arrays, Strings, and Serializable + * Objects depending on the message content type header. + *

+ * Note that this class never uses the message converter of the underlying {@link AmqpTemplate}! + * + * @see org.springframework.amqp.support.converter.SimpleMessageConverter + */ + public void setMessageConverter(MessageConverter messageConverter) { + this.messageConverter = messageConverter; + } + +} diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/SimpleMessageConverter.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/SimpleMessageConverter.java index 8876c896..3beaef94 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/SimpleMessageConverter.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/SimpleMessageConverter.java @@ -28,7 +28,6 @@ import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.utils.SerializationUtils; import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.remoting.rmi.CodebaseAwareObjectInputStream; -import java.rmi.server.RMIClassLoader; import org.springframework.util.ClassUtils; /** @@ -78,6 +77,7 @@ public class SimpleMessageConverter extends AbstractMessageConverter implements /** * Converts from a AMQP Message to an Object. */ + @Override public Object fromMessage(Message message) throws MessageConversionException { Object content = null; MessageProperties properties = message.getMessageProperties(); @@ -118,6 +118,7 @@ public class SimpleMessageConverter extends AbstractMessageConverter implements /** * Creates an AMQP Message from the provided Object. */ + @Override protected Message createMessage(Object object, MessageProperties messageProperties) throws MessageConversionException { byte[] bytes = null; if (object instanceof byte[]) { diff --git a/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java b/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java new file mode 100644 index 00000000..e3228089 --- /dev/null +++ b/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java @@ -0,0 +1,103 @@ +/* + * Copyright 2002-2013 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.amqp.remoting; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.remoting.client.AmqpProxyFactoryBean; +import org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter; +import org.springframework.amqp.remoting.testhelper.AbstractAmqpTemplate; +import org.springframework.amqp.remoting.testhelper.SentSavingTemplate; +import org.springframework.amqp.remoting.testservice.GeneralException; +import org.springframework.amqp.remoting.testservice.SpecialException; +import org.springframework.amqp.remoting.testservice.TestServiceImpl; +import org.springframework.amqp.remoting.testservice.TestServiceInterface; +import org.springframework.amqp.support.converter.MessageConverter; + +/** + * @author David Bilge + * @since 1.2 + */ +public class RemotingTest { + + private TestServiceInterface riggedProxy; + + /** + * Set up a rig of directly wired-up proxy and service listener so that both can be tested together without needing + * a running rabbit. + */ + @Before + public void initializeTestRig() throws Exception { + // Set up the service + TestServiceInterface testService = new TestServiceImpl(); + final AmqpInvokerServiceExporter serviceExporter = new AmqpInvokerServiceExporter(); + final SentSavingTemplate sentSavingTemplate = new SentSavingTemplate(); + serviceExporter.setAmqpTemplate(sentSavingTemplate); + serviceExporter.setService(testService); + serviceExporter.setServiceInterface(TestServiceInterface.class); + + // Set up the client + AmqpProxyFactoryBean amqpProxyFactoryBean = new AmqpProxyFactoryBean(); + amqpProxyFactoryBean.setServiceInterface(TestServiceInterface.class); + AmqpTemplate directForwardingTemplate = new AbstractAmqpTemplate() { + @Override + public Object convertSendAndReceive(Object payload) throws AmqpException { + MessageConverter messageConverter = serviceExporter.getMessageConverter(); + + Address replyTo = new Address("fakeExchange", "fakeExchangeName", "fakeRoutingKey"); + MessageProperties messageProperties = new MessageProperties(); + messageProperties.setReplyToAddress(replyTo); + Message message = messageConverter.toMessage(payload, messageProperties); + + serviceExporter.onMessage(message); + + Message resultMessage = sentSavingTemplate.getLastMessage(); + return messageConverter.fromMessage(resultMessage); + } + }; + amqpProxyFactoryBean.setAmqpTemplate(directForwardingTemplate); + amqpProxyFactoryBean.afterPropertiesSet(); + Object rawProxy = amqpProxyFactoryBean.getObject(); + riggedProxy = (TestServiceInterface) rawProxy; + } + + @Test + public void testEcho() { + Assert.assertEquals("Echo Test", riggedProxy.simpleStringReturningTestMethod("Test")); + } + + @Test(expected = RuntimeException.class) + public void testExceptionPropagation() { + riggedProxy.exceptionThrowingMethod(); + } + + @Test(expected = GeneralException.class) + public void testExceptionReturningMethod() { + riggedProxy.notReallyExceptionReturningMethod(); + } + + @Test + public void testActuallyExceptionReturningMethod() { + SpecialException returnedException = riggedProxy.actuallyExceptionReturningMethod(); + + Assert.assertNotNull(returnedException); + Assert.assertTrue(returnedException instanceof SpecialException); + } +} diff --git a/spring-amqp/src/test/java/org/springframework/amqp/remoting/testhelper/AbstractAmqpTemplate.java b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testhelper/AbstractAmqpTemplate.java new file mode 100644 index 00000000..3e55ea80 --- /dev/null +++ b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testhelper/AbstractAmqpTemplate.java @@ -0,0 +1,142 @@ +/* + * Copyright 2002-2013 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.amqp.remoting.testhelper; + +import org.apache.commons.lang.NotImplementedException; +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessagePostProcessor; + +/** + * @author David Bilge + * @since 1.2 + */ +public abstract class AbstractAmqpTemplate implements AmqpTemplate { + + @Override + public void send(Message message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public void send(String routingKey, Message message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public void send(String exchange, String routingKey, Message message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public void convertAndSend(Object message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public void convertAndSend(String routingKey, Object message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public void convertAndSend(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor) + throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public void convertAndSend(String exchange, String routingKey, Object message, + MessagePostProcessor messagePostProcessor) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Message receive() throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Message receive(String queueName) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Object receiveAndConvert() throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Object receiveAndConvert(String queueName) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Message sendAndReceive(Message message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Message sendAndReceive(String routingKey, Message message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Message sendAndReceive(String exchange, String routingKey, Message message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Object convertSendAndReceive(Object message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Object convertSendAndReceive(String routingKey, Object message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Object convertSendAndReceive(String exchange, String routingKey, Object message) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Object convertSendAndReceive(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Object convertSendAndReceive(String routingKey, Object message, MessagePostProcessor messagePostProcessor) + throws AmqpException { + throw new NotImplementedException(); + } + + @Override + public Object convertSendAndReceive(String exchange, String routingKey, Object message, + MessagePostProcessor messagePostProcessor) throws AmqpException { + throw new NotImplementedException(); + } + +} diff --git a/spring-amqp/src/test/java/org/springframework/amqp/remoting/testhelper/SentSavingTemplate.java b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testhelper/SentSavingTemplate.java new file mode 100644 index 00000000..4f320d6f --- /dev/null +++ b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testhelper/SentSavingTemplate.java @@ -0,0 +1,46 @@ +/* + * Copyright 2002-2013 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.amqp.remoting.testhelper; + +import org.springframework.amqp.AmqpException; +import org.springframework.amqp.core.Message; + +/** + * @author David Bilge + * @since 1.2 + */ +public class SentSavingTemplate extends AbstractAmqpTemplate { + private Message lastMessage = null; + private String lastExchange = null; + private String lastRoutingKey = null; + + @Override + public void send(String exchange, String routingKey, Message message) throws AmqpException { + this.lastExchange = exchange; + this.lastRoutingKey = routingKey; + this.lastMessage = message; + } + + public Message getLastMessage() { + return lastMessage; + } + + public String getLastExchange() { + return lastExchange; + } + + public String getLastRoutingKey() { + return lastRoutingKey; + } +} diff --git a/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/GeneralException.java b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/GeneralException.java new file mode 100644 index 00000000..79294469 --- /dev/null +++ b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/GeneralException.java @@ -0,0 +1,31 @@ +/* + * Copyright 2002-2013 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.amqp.remoting.testservice; + +/** + * @author David Bilge + * @since 1.2 + */ +public class GeneralException extends RuntimeException { + private static final long serialVersionUID = 1763252570120227426L; + + public GeneralException(String message, Throwable cause) { + super(message, cause); + } + + public GeneralException(String message) { + super(message); + } + +} diff --git a/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/SpecialException.java b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/SpecialException.java new file mode 100644 index 00000000..b9e92571 --- /dev/null +++ b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/SpecialException.java @@ -0,0 +1,31 @@ +/* + * Copyright 2002-2013 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.amqp.remoting.testservice; + +/** + * @author David Bilge + * @since 1.2 + */ +public class SpecialException extends RuntimeException { + private static final long serialVersionUID = 7254934411128057730L; + + public SpecialException(String message, Throwable cause) { + super(message, cause); + } + + public SpecialException(String message) { + super(message); + } + +} diff --git a/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/TestServiceImpl.java b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/TestServiceImpl.java new file mode 100644 index 00000000..526c5089 --- /dev/null +++ b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/TestServiceImpl.java @@ -0,0 +1,50 @@ +/* + * Copyright 2002-2013 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.amqp.remoting.testservice; + +/** + * @author David Bilge + * @since 1.2 + */ +public class TestServiceImpl implements TestServiceInterface { + @Override + public void simpleTestMethod() { + // Do nothing + } + + @Override + public String simpleStringReturningTestMethod(String string) { + return "Echo " + string; + } + + @Override + public void exceptionThrowingMethod() { + throw new RuntimeException("This is an exception"); + } + + @Override + public Object echo(Object o) { + return o; + } + + @Override + public SpecialException notReallyExceptionReturningMethod() { + throw new GeneralException("This exception should not be interpreted as a return type but be thrown instead."); + } + + @Override + public SpecialException actuallyExceptionReturningMethod() { + return new SpecialException("This exception should not be thrown on the client side but just be returned!"); + } +} \ No newline at end of file diff --git a/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/TestServiceInterface.java b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/TestServiceInterface.java new file mode 100644 index 00000000..83af870d --- /dev/null +++ b/spring-amqp/src/test/java/org/springframework/amqp/remoting/testservice/TestServiceInterface.java @@ -0,0 +1,32 @@ +/* + * Copyright 2002-2013 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.amqp.remoting.testservice; + +/** + * @author David Bilge + * @since 1.2 + */ +public interface TestServiceInterface { + void simpleTestMethod(); + + String simpleStringReturningTestMethod(String string); + + void exceptionThrowingMethod(); + + Object echo(Object o); + + SpecialException notReallyExceptionReturningMethod(); + + SpecialException actuallyExceptionReturningMethod(); +} \ No newline at end of file diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/remoting/RemotingTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/remoting/RemotingTests.java new file mode 100644 index 00000000..148b0a94 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/remoting/RemotingTests.java @@ -0,0 +1,117 @@ +/* + * Copyright 2002-2013 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.amqp.rabbit.remoting; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.test.BrokerRunning; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.remoting.RemoteProxyFailureException; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 1.2 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class RemotingTests { + + @Rule + public BrokerRunning brokerRunning = BrokerRunning.isRunning(); + + @Autowired + private ServiceInterface client; + + @Autowired + private RabbitTemplate template; + + private static CountDownLatch latch; + + private static String receivedMessage; + + @Test + public void testEcho() throws Exception { + String reply = client.echo("foo"); + assertEquals("echo:foo", reply); + } + + @Test + public void testNoAnswer() throws Exception { + latch = new CountDownLatch(1); + client.noAnswer("foo"); + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertEquals("received:foo", receivedMessage); + } + + @Test + public void testTimeout() { + try { + client.suspend(); + fail("Exception expected"); + } + catch (RemoteProxyFailureException e) { + assertTrue("No reply received - perhaps a timeout in the template?".equals(e.getMessage())); + } + } + + public interface ServiceInterface { + + String echo(String message); + + void noAnswer(String message); + + void suspend(); + + } + + public static class ServiceImpl implements ServiceInterface { + + @Override + public String echo(String message) { + return "echo:" + message; + } + + @Override + public void noAnswer(String message) { + receivedMessage = "received:" + message; + latch.countDown(); + } + + @Override + public void suspend() { + try { + Thread.sleep(3000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + + } +} diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/remoting/RemotingTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/remoting/RemotingTests-context.xml new file mode 100644 index 00000000..43adab42 --- /dev/null +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/remoting/RemotingTests-context.xml @@ -0,0 +1,39 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index db9bae90..1315ca20 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -873,6 +873,87 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]>spring_reply_correlation. +

+ Spring Remoting with AMQP + + The Spring Framework has a general remoting capability, allowing + + Remote Procedure Calls (RPC) using various transports. + Spring-AMQP supports a similar mechanism with a AmqpProxyFactoryBean on the client + and a AmqpInvokerServiceExporter on the server. This provides RPC over AMQP. + On the client side, a RabbitTemplate is used as described above; on the server side, + the invoker (configured as a MessageListener) receives the message, invokes + the configured service, and returns the reply using the inbound message's replyTo information. + + + The client factory bean can be injected into any bean (using its serviceInterface); the client + can then invoke methods on the proxy, resulting in remote execution over AMQP. + + + + With the default MessageConverters, the method paramters and returned + value must be instances of Serializable. + + + On the server side, the AmqpInvokerServiceExporter has + both AmqpTemplate and MessageConverter + properties. Currently, the template's MessageConverter is not + used. If you need to supply a custom message converter, then you should provide it using + the messageConverter property. On the client side, a custom message converter + can be added to the AmqpTemplate which is provided to the + AmqpProxyFactoryBean using its amqpTemplate property. + + + + Sample client and server configurations are shown below. + + + + + + + + + + + + + + + + + + +]]> + + + + + + + + + + + + + + + + +]]> + + The AmqpInvokerServiceExporter can only process properly + formed messages, such as those sent from the AmqpProxyFactoryBean. + If it receives a message that it cannot interpret, a serialized + RuntimeException will be sent as a reply. If the message has + no replyToAddress property, the message will be rejected and permanently lost if no + Dead Letter Exchange has been configured. + +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 5dab722f..e6e23b72 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -57,6 +57,13 @@ with the existing converter that uses Jackson 1.x.
+
+ AMQP Remoting + + Facilities are now provided for using Spring Remoting techniques, using AMQP + as the transport for the RPC calls. For more information see + +