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 688f0018..48542d3e 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 @@ -473,6 +473,128 @@ public class RabbitTemplateIntegrationTests { assertEquals(null, result); } + @Test + public void testAtomicSendAndReceiveWithConversionAndMessagePostProcessor() 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++) { + message = template.receive(); + if (message != null) { + break; + } + Thread.sleep(100L); + } + assertNotNull("No message received", message); + template.send(message.getMessageProperties().getReplyTo(), message); + return (String) template.getMessageConverter().fromMessage(message); + } + + }); + String result = (String) template.convertSendAndReceive((Object) "message", new MessagePostProcessor() { + public Message postProcessMessage(Message message) throws AmqpException { + try { + byte[] newBody = new String(message.getBody(), "UTF-8").toUpperCase().getBytes("UTF-8"); + return new Message(newBody, message.getMessageProperties()); + } + catch (Exception e) { + throw new AmqpException("unexpected failure in test", e); + } + } + }); + 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 testAtomicSendAndReceiveWithConversionAndMessagePostProcessorUsingRoutingKey() 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++) { + message = template.receive(ROUTE); + if (message != null) { + break; + } + Thread.sleep(100L); + } + assertNotNull("No message received", message); + template.send(message.getMessageProperties().getReplyTo(), message); + return (String) template.getMessageConverter().fromMessage(message); + } + + }); + String result = (String) template.convertSendAndReceive(ROUTE, (Object) "message", new MessagePostProcessor() { + public Message postProcessMessage(Message message) throws AmqpException { + try { + byte[] newBody = new String(message.getBody(), "UTF-8").toUpperCase().getBytes("UTF-8"); + return new Message(newBody, message.getMessageProperties()); + } + catch (Exception e) { + throw new AmqpException("unexpected failure in test", e); + } + } + }); + 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); + } + + @Test + public void testAtomicSendAndReceiveWithConversionAndMessagePostProcessorUsingExchangeAndRoutingKey() 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++) { + message = template.receive(ROUTE); + if (message != null) { + break; + } + Thread.sleep(100L); + } + assertNotNull("No message received", message); + template.send(message.getMessageProperties().getReplyTo(), message); + return (String) template.getMessageConverter().fromMessage(message); + } + + }); + String result = (String) template.convertSendAndReceive("", ROUTE, "message", new MessagePostProcessor() { + + public Message postProcessMessage(Message message) throws AmqpException { + try { + byte[] newBody = new String(message.getBody(), "UTF-8").toUpperCase().getBytes("UTF-8"); + return new Message(newBody, message.getMessageProperties()); + } + catch (Exception e) { + throw new AmqpException("unexpected failure in test", e); + } + } + }); + 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() {