AMQP-847: Close channel in RabbitTemplate.receive

JIRA: https://jira.spring.io/browse/AMQP-847

To avoid unacked messages race condition when client timeouts, but at
this moment the message becomes available in queue, physically close
a receive channel on the `TimeoutException` from the `Future.get()`

**Cherry-pick to 2.0.x & 1.7.x**

Author: artembilan

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

@@ -890,6 +890,7 @@ public Message doInRabbit(Channel channel) throws Exception {
 				}
 				channel.basicCancel(consumer.getConsumerTag());
 				if (delivery == null) {
+					RabbitUtils.setPhysicalCloseRequired(channel, true);
 					return null;
 				}
 				else {
@@ -1029,6 +1030,9 @@ else if (channelTransacted) {
 						}
 						receiveMessage = buildMessageFromDelivery(delivery);
 					}
+					else {
+						RabbitUtils.setPhysicalCloseRequired(channel, true);
+					}
 				}
 				if (receiveMessage != null) {
 					Object receive = receiveMessage;

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

@@ -47,6 +47,7 @@
 import java.lang.reflect.Field;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.Callable;
@@ -348,6 +349,14 @@ public void testReceiveBlockingNoTimeout() throws Exception {
 		assertNull(this.template.receive(ROUTE));
 	}
 
+	@Test
+	public void testReceiveTimeoutRequeue() {
+		assertNull(this.template.receiveAndConvert(ROUTE, 1));
+		assertEquals(0,
+				TestUtils.getPropertyValue(this.connectionFactory, "cachedChannelsNonTransactional", List.class)
+						.size());
+	}
+
 	@Test
 	public void testReceiveBlockingTx() throws Exception {
 		this.template.convertAndSend(ROUTE, "blockTX");