AMQP-451: Support Flushing Batch

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

- Add `flush()` to support flushing any partial batch.
- Implement `Lifecycle` and invoke `flush()` when stopped.

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

@@ -23,6 +23,7 @@ import org.springframework.amqp.core.Message;
 import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
 import org.springframework.amqp.rabbit.core.support.MessageBatch;
 import org.springframework.amqp.rabbit.support.CorrelationData;
+import org.springframework.context.Lifecycle;
 import org.springframework.scheduling.TaskScheduler;
 
 /**
@@ -38,7 +39,7 @@ import org.springframework.scheduling.TaskScheduler;
  * @since 1.4.1
  *
  */
-public class BatchingRabbitTemplate extends RabbitTemplate {
+public class BatchingRabbitTemplate extends RabbitTemplate implements Lifecycle {
 
 	private final BatchingStrategy batchingStrategy;
 
@@ -85,6 +86,13 @@ public class BatchingRabbitTemplate extends RabbitTemplate {
 		}
 	}
 
+	/**
+	 * Flush any partial in-progress batches.
+	 */
+	public void flush() {
+		releaseBatches();
+	}
+
 	private synchronized void releaseBatches() {
 		MessageBatch batch;
 		while ((batch = this.batchingStrategy.releaseBatch()) != null) {
@@ -92,4 +100,18 @@ public class BatchingRabbitTemplate extends RabbitTemplate {
 		}
 	}
 
+	@Override
+	public void start() {
+	}
+
+	@Override
+	public void stop() {
+		flush();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return true;
+	}
+
 }