From 60f58ff91b4b3a05e4448ad53a940ef64a928acf Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 3 Mar 2011 13:40:04 -0500 Subject: [PATCH] AMQP-19 added send-and-receive methods to AmqpTemplate and RabbitTemplate --- .../amqp/core/AmqpTemplate.java | 22 +++ .../amqp/rabbit/core/RabbitTemplate.java | 32 ++-- .../core/RabbitTemplateIntegrationTests.java | 176 +++++++++++++++++- 3 files changed, 217 insertions(+), 13 deletions(-) diff --git a/spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpTemplate.java b/spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpTemplate.java index fcd3fdd0..00790f01 100644 --- a/spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpTemplate.java +++ b/spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpTemplate.java @@ -31,12 +31,16 @@ import org.springframework.amqp.AmqpException; */ public interface AmqpTemplate { + // send methods for messages + void send(Message message) throws AmqpException; void send(String routingKey, Message message) throws AmqpException; void send(String exchange, String routingKey, Message message) throws AmqpException; + // send methods with conversion + void convertAndSend(Object message) throws AmqpException; void convertAndSend(String routingKey, Object message) throws AmqpException; @@ -49,14 +53,32 @@ public interface AmqpTemplate { void convertAndSend(String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor) throws AmqpException; + // receive methods for messages + Message receive() throws AmqpException; Message receive(String queueName) throws AmqpException; + // receive methods with conversion + Object receiveAndConvert() throws AmqpException; Object receiveAndConvert(String queueName) throws AmqpException; + // send and receive methods for messages + + Message sendAndReceive(Message message) throws AmqpException; + + Message sendAndReceive(String routingKey, Message message) throws AmqpException; + + Message sendAndReceive(String exchange, String routingKey, Message message) throws AmqpException; + + // send and receive methods with conversion + Object convertSendAndReceive(Object message) throws AmqpException; + Object convertSendAndReceive(String routingKey, Object message) throws AmqpException; + + Object convertSendAndReceive(String exchange, String routingKey, Object message) throws AmqpException; + } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 81206167..b2bb92dd 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2011 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 @@ -214,6 +214,18 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { return null; } + public Message sendAndReceive(final Message message) throws AmqpException { + return this.doSendAndReceive(this.exchange, this.routingKey, message); + } + + public Message sendAndReceive(final String routingKey, final Message message) throws AmqpException { + return this.doSendAndReceive(this.exchange, routingKey, message); + } + + public Message sendAndReceive(final String exchange, final String routingKey, final Message message) throws AmqpException { + return this.doSendAndReceive(exchange, routingKey, message); + } + public Object convertSendAndReceive(final Object message) throws AmqpException { return this.convertSendAndReceive(this.exchange, this.routingKey, message); } @@ -222,8 +234,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { return this.convertSendAndReceive(this.exchange, routingKey, message); } - public Object convertSendAndReceive(final String exchange, final String routingKey, final Object message) - throws AmqpException { + public Object convertSendAndReceive(final String exchange, final String routingKey, final Object message) throws AmqpException { MessageProperties messageProperties = new MessageProperties(); Message requestMessage = getRequiredMessageConverter().toMessage(message, messageProperties); Message replyMessage = this.doSendAndReceive(exchange, routingKey, requestMessage); @@ -238,15 +249,12 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations { public Message doInRabbit(Channel channel) throws Exception { final SynchronousQueue replyHandoff = new SynchronousQueue(); - // TODO: extract this to a method - Address replyToAddress = message.getMessageProperties().getReplyTo(); - if (replyToAddress == null) { - // TODO: first check for a replyToAddress property on this - // template - DeclareOk queueDeclaration = channel.queueDeclare(); - replyToAddress = new Address(ExchangeTypes.DIRECT, DEFAULT_EXCHANGE, queueDeclaration.getQueue()); - message.getMessageProperties().setReplyTo(replyToAddress); - } + Assert.isNull(message.getMessageProperties().getReplyTo(), + "Send-and-receive methods can only be used if the Message does not already have a replyTo property."); + // TODO: first check for a replyToAddress property on this template + DeclareOk queueDeclaration = channel.queueDeclare(); + Address replyToAddress = new Address(ExchangeTypes.DIRECT, DEFAULT_EXCHANGE, queueDeclaration.getQueue()); + message.getMessageProperties().setReplyTo(replyToAddress); boolean noAck = false; String consumerTag = UUID.randomUUID().toString(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 200d00e5..c538c057 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -1,3 +1,16 @@ +/* + * Copyright 2010-2011 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.core; import static org.junit.Assert.assertEquals; @@ -232,8 +245,139 @@ public class RabbitTemplateIntegrationTests { @Test public void testAtomicSendAndReceive() throws Exception { + final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory()); + template.setRoutingKey(ROUTE); + template.setQueue(ROUTE); ExecutorService executor = Executors.newFixedThreadPool(1); - // Set up a consumer to respond to out producer + // Set up a consumer to respond to our producer + Future received = executor.submit(new Callable() { + + public Message call() throws Exception { + Message message = null; + for (int i = 0; i < 10; i++) { + // TODO: AMQP-71 add receive timeout + message = template.receive(); + if (message != null) { + break; + } + Thread.sleep(100L); + } + assertNotNull("No message received", message); + template.send(message.getMessageProperties().getReplyTo().getRoutingKey(), message); + return message; + } + + }); + Message message = new Message("test-message".getBytes(), new MessageProperties()); + Message reply = template.sendAndReceive(message); + assertEquals(new String(message.getBody()), new String(received.get(1000, TimeUnit.MILLISECONDS).getBody())); + assertEquals(new String(message.getBody()), new String(reply.getBody())); + // Message was consumed so nothing left on queue + reply = template.receive(); + assertEquals(null, reply); + } + + @Test + public void testAtomicSendAndReceiveWithRoutingKey() throws Exception { + final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory()); + ExecutorService executor = Executors.newFixedThreadPool(1); + // Set up a consumer to respond to our producer + Future received = executor.submit(new Callable() { + + public Message call() throws Exception { + Message message = null; + for (int i = 0; i < 10; i++) { + // TODO: AMQP-71 add receive timeout + message = template.receive(ROUTE); + if (message != null) { + break; + } + Thread.sleep(100L); + } + assertNotNull("No message received", message); + template.send(message.getMessageProperties().getReplyTo().getRoutingKey(), message); + return message; + } + + }); + Message message = new Message("test-message".getBytes(), new MessageProperties()); + Message reply = template.sendAndReceive(ROUTE, message); + assertEquals(new String(message.getBody()), new String(received.get(1000, TimeUnit.MILLISECONDS).getBody())); + assertEquals(new String(message.getBody()), new String(reply.getBody())); + // Message was consumed so nothing left on queue + reply = template.receive(ROUTE); + assertEquals(null, reply); + } + + @Test + public void testAtomicSendAndReceiveWithExchangeAndRoutingKey() throws Exception { + final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory()); + ExecutorService executor = Executors.newFixedThreadPool(1); + // Set up a consumer to respond to our producer + Future received = executor.submit(new Callable() { + + public Message call() throws Exception { + Message message = null; + for (int i = 0; i < 10; i++) { + // TODO: AMQP-71 add receive timeout + message = template.receive(ROUTE); + if (message != null) { + break; + } + Thread.sleep(100L); + } + assertNotNull("No message received", message); + template.send(message.getMessageProperties().getReplyTo().getRoutingKey(), message); + return message; + } + + }); + Message message = new Message("test-message".getBytes(), new MessageProperties()); + Message reply = template.sendAndReceive("", ROUTE, message); + assertEquals(new String(message.getBody()), new String(received.get(1000, TimeUnit.MILLISECONDS).getBody())); + assertEquals(new String(message.getBody()), new String(reply.getBody())); + // Message was consumed so nothing left on queue + reply = template.receive(ROUTE); + assertEquals(null, reply); + } + + @Test + public void testAtomicSendAndReceiveWithConversion() throws Exception { + final RabbitTemplate template = new RabbitTemplate(new CachingConnectionFactory()); + template.setRoutingKey(ROUTE); + template.setQueue(ROUTE); + ExecutorService executor = Executors.newFixedThreadPool(1); + // Set up a consumer to respond to our producer + Future received = executor.submit(new Callable() { + + public String call() throws Exception { + Message message = null; + for (int i = 0; i < 10; i++) { + // TODO: AMQP-71 add receive timeout + message = template.receive(); + if (message != null) { + break; + } + Thread.sleep(100L); + } + assertNotNull("No message received", message); + template.send(message.getMessageProperties().getReplyTo().getRoutingKey(), message); + return (String) template.getMessageConverter().fromMessage(message); + } + + }); + String result = (String) template.convertSendAndReceive("message"); + assertEquals("message", received.get(1000, TimeUnit.MILLISECONDS)); + assertEquals("message", result); + // Message was consumed so nothing left on queue + result = (String) template.receiveAndConvert(); + assertEquals(null, result); + } + + @Test + public void testAtomicSendAndReceiveWithConversionUsingRoutingKey() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(1); + // Set up a consumer to respond to our producer Future received = executor.submit(new Callable() { public String call() throws Exception { @@ -260,6 +404,36 @@ public class RabbitTemplateIntegrationTests { assertEquals(null, result); } + @Test + public void testAtomicSendAndReceiveWithConversionUsingExchangeAndRoutingKey() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(1); + // Set up a consumer to respond to our producer + Future received = executor.submit(new Callable() { + + public String call() throws Exception { + Message message = null; + for (int i = 0; i < 10; i++) { + // TODO: AMQP-71 add receive timeout + message = template.receive(ROUTE); + if (message != null) { + break; + } + Thread.sleep(100L); + } + assertNotNull("No message received", message); + template.send(message.getMessageProperties().getReplyTo().getRoutingKey(), message); + return (String) template.getMessageConverter().fromMessage(message); + } + + }); + String result = (String) template.convertSendAndReceive("", ROUTE, "message"); + assertEquals("message", received.get(1000, TimeUnit.MILLISECONDS)); + assertEquals("message", result); + // Message was consumed so nothing left on queue + result = (String) template.receiveAndConvert(ROUTE); + assertEquals(null, result); + } + @SuppressWarnings("serial") private class PlannedException extends RuntimeException { public PlannedException() {