INT-4371: Add MessageSource Acknowlegment support

JIRA: https://jira.spring.io/browse/INT-4371

Add an abstract message source that populates the message with a header containing
an abstraction that allows the application to acknowledge/reject/requeue the message.
(In amqp, for example, this would map to channel.basicAck(tag),
channel.basicReject(tag, false), and channel.basicReject(tag, true).

With some brokers (such as kafka), acknowledge and reject might be equivalent
since there is no automatic DLQ processing in kafka both would simply commit the offset.

Also include a strategy interface for a factory to provide the acknowledgment header.

Add MessageSourcePollableChannel

Fix typo

Polishing - PR Comments

- remove marker interface; just use presence of the header for auto-ack
- remove the MessageSourcePollableChannel - users can use a MethodInterceptor to enhance the message
- other PR comments

Save entire response to allow customization of the ack/nack actions.

Docs, Copyrights; add MessageSourcePollingTemplate

Polishing - PR Comments - support disabling auto-ack, when possible.

More Utilities

* Simple code style polishing

Author: garyrussell

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -22,12 +22,14 @@
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.amqp.inbound.AmqpMessageSource.AmqpAckCallbackFactory;
 
 /**
  * Factory class for AMQP components.
  *
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 5.0
  */
 public final class Amqp {
@@ -53,6 +55,7 @@ public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connec
 	 */
 	public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
 			String... queueNames) {
+
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueueNames(queueNames);
 		return inboundGateway(listenerContainer, amqpTemplate);
@@ -80,6 +83,7 @@ public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connec
 	 */
 	public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate,
 			Queue... queues) {
+
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueues(queues);
 		return inboundGateway(listenerContainer, amqpTemplate);
@@ -112,6 +116,7 @@ public static AmqpInboundGatewaySMLCSpec inboundGateway(SimpleMessageListenerCon
 	 */
 	public static AmqpInboundGatewaySMLCSpec inboundGateway(SimpleMessageListenerContainer listenerContainer,
 			AmqpTemplate amqpTemplate) {
+
 		return new AmqpInboundGatewaySMLCSpec(listenerContainer, amqpTemplate);
 	}
 
@@ -142,9 +147,37 @@ public static AmqpInboundGatewayDMLCSpec inboundGateway(DirectMessageListenerCon
 	 */
 	public static AmqpInboundGatewayDMLCSpec inboundGateway(DirectMessageListenerContainer listenerContainer,
 			AmqpTemplate amqpTemplate) {
+
 		return new AmqpInboundGatewayDMLCSpec(listenerContainer, amqpTemplate);
 	}
 
+	/**
+	 * Create an initial AmqpInboundPolledChannelAdapterSpec
+	 * @param connectionFactory the connectionFactory.
+	 * @param queue the queue.
+	 * @return the AmqpInboundPolledChannelAdapterSpec.
+	 * @since 5.0.1
+	 */
+	public static AmqpInboundPolledChannelAdapterSpec inboundPolledAdapter(ConnectionFactory connectionFactory,
+			String queue) {
+
+		return new AmqpInboundPolledChannelAdapterSpec(connectionFactory, queue);
+	}
+
+	/**
+	 * Create an initial AmqpInboundPolledChannelAdapterSpec
+	 * @param connectionFactory the connectionFactory.
+	 * @param ackCallbackFactory the ackCallbackFactory
+	 * @param queue the queue.
+	 * @return the AmqpInboundPolledChannelAdapterSpec.
+	 * @since 5.0.1
+	 */
+	public static AmqpInboundPolledChannelAdapterSpec inboundPolledAdapter(ConnectionFactory connectionFactory,
+			AmqpAckCallbackFactory ackCallbackFactory, String queue) {
+
+		return new AmqpInboundPolledChannelAdapterSpec(connectionFactory, ackCallbackFactory, queue);
+	}
+
 	/**
 	 * Create an initial AmqpInboundChannelAdapterSpec using a
 	 * {@link SimpleMessageListenerContainer}.
@@ -154,6 +187,7 @@ public static AmqpInboundGatewayDMLCSpec inboundGateway(DirectMessageListenerCon
 	 */
 	public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(ConnectionFactory connectionFactory,
 			String... queueNames) {
+
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueueNames(queueNames);
 		return new AmqpInboundChannelAdapterSMLCSpec(listenerContainer);
@@ -166,7 +200,9 @@ public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(ConnectionFactory
 	 * @param queues the queues.
 	 * @return the AmqpInboundChannelAdapterSpec.
 	 */
-	public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(ConnectionFactory connectionFactory, Queue... queues) {
+	public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(ConnectionFactory connectionFactory,
+			Queue... queues) {
+
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueues(queues);
 		return new AmqpInboundChannelAdapterSMLCSpec(listenerContainer);
@@ -235,6 +271,7 @@ public static AmqpAsyncOutboundGatewaySpec asyncOutboundGateway(AsyncRabbitTempl
 	 */
 	public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S> pollableChannel(
 			ConnectionFactory connectionFactory) {
+
 		return pollableChannel(null, connectionFactory);
 	}
 
@@ -247,6 +284,7 @@ public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageC
 	 */
 	public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S> pollableChannel(
 			String id, ConnectionFactory connectionFactory) {
+
 		return new AmqpPollableMessageChannelSpec<S>(connectionFactory).id(id);
 	}
 
@@ -258,6 +296,7 @@ public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageC
 	 */
 	public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(
 			ConnectionFactory connectionFactory) {
+
 		return channel(null, connectionFactory);
 	}
 
@@ -270,6 +309,7 @@ public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> ch
 	 */
 	public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(String id,
 			ConnectionFactory connectionFactory) {
+
 		return new AmqpMessageChannelSpec<S>(connectionFactory).id(id);
 	}
 
@@ -290,6 +330,7 @@ public static AmqpPublishSubscribeMessageChannelSpec publishSubscribeChannel(Con
 	 */
 	public static AmqpPublishSubscribeMessageChannelSpec publishSubscribeChannel(String id,
 			ConnectionFactory connectionFactory) {
+
 		return new AmqpPublishSubscribeMessageChannelSpec(connectionFactory).id(id);
 	}
 

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundPolledChannelAdapterSpec.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2018 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.integration.amqp.dsl;
+
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.integration.amqp.inbound.AmqpMessageSource;
+import org.springframework.integration.amqp.inbound.AmqpMessageSource.AmqpAckCallbackFactory;
+import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.dsl.MessageSourceSpec;
+
+/**
+ * Spec for a polled AMQP inbound channel adapter.
+ *
+ * @author Gary Russell
+ *
+ * @since 5.0.1
+ *
+ */
+public class AmqpInboundPolledChannelAdapterSpec
+		extends MessageSourceSpec<AmqpInboundPolledChannelAdapterSpec, AmqpMessageSource> {
+
+	AmqpInboundPolledChannelAdapterSpec(ConnectionFactory connectionFactory, String queue) {
+		this.target = new AmqpMessageSource(connectionFactory, queue);
+	}
+
+	AmqpInboundPolledChannelAdapterSpec(ConnectionFactory connectionFactory,
+			AmqpAckCallbackFactory ackCallbackFactory, String queue) {
+
+		this.target = new AmqpMessageSource(connectionFactory, ackCallbackFactory, queue);
+	}
+
+	public AmqpInboundPolledChannelAdapterSpec transacted(boolean transacted) {
+		this.target.setTransacted(transacted);
+		return this;
+	}
+
+	public AmqpInboundPolledChannelAdapterSpec setPropertiesConverter(MessagePropertiesConverter propertiesConverter) {
+		this.target.setPropertiesConverter(propertiesConverter);
+		return this;
+	}
+
+	public AmqpInboundPolledChannelAdapterSpec setHeaderMapper(AmqpHeaderMapper headerMapper) {
+		this.target.setHeaderMapper(headerMapper);
+		return this;
+	}
+
+	public AmqpInboundPolledChannelAdapterSpec setMessageConverter(MessageConverter messageConverter) {
+		this.target.setMessageConverter(messageConverter);
+		return this;
+	}
+
+}