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

AMQP-19 added send-and-receive methods to AmqpTemplate and RabbitTemplate

Browse Source
This commit is contained in:
Mark Fisher
2011-03-03 13:40:04 -05:00
parent 9c92b09556
commit 60f58ff91b
3 changed files with 217 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-amqp-core/src/main/java/org/springframework/amqp/core/AmqpTemplate.java
View File

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

32
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View File

@@ -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<Message> replyHandoff = new SynchronousQueue<Message>();
// 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();

176
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
View File

@@ -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<Message> received = executor.submit(new Callable<Message>() {
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<Message> received = executor.submit(new Callable<Message>() {
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<Message> received = executor.submit(new Callable<Message>() {
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<String> received = executor.submit(new Callable<String>() {
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<String> received = executor.submit(new Callable<String>() {
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<String> received = executor.submit(new Callable<String>() {
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() {