AMQP-187 added integration tests for new template methods

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