SGH-1616: Add MessageSourceCustomizer

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/1616
Resolves #1619

docs/src/main/asciidoc/spring-cloud-stream.adoc

@@ -1498,6 +1498,28 @@ See the producer property `useNativeEncoding`.
 +
 Default: `false`.
 
+==== Advanced Consumer Configuration
+
+For advanced configuration of the underlying message listener container for message-driven consumers, add a single `ListenerContainerCustomizer` bean to the application context.
+It will be invoked after the above properties have been applied and can be used to set additional properties.
+Similarly, for polled consumers, add a `MessageSourceCustomizer` bean.
+
+The following is an example for the RabbitMQ binder:
+
+====
+[source, java]
+----
+@Bean
+public ListenerContainerCustomizer<AbstractMessageListenerContainer> containerCustomizer() {
+    return (container, dest, group) -> container.setAdviceChain(advice1, advice2);
+}
+
+@Bean
+public MessageSourceCustomizer<AmqpMessageSource> sourceCustomizer() {
+    return (source, dest, group) -> source.setPropertiesConverter(customPropertiesConverter);
+}
+----
+====
 
 ==== Producer Properties
 
@@ -2756,4 +2778,3 @@ The following is the list of available binder implementations
 * https://github.com/spring-cloud/spring-cloud-gcp/tree/master/spring-cloud-gcp-pubsub-stream-binder[Google PubSub _(partner maintained)_]
 * https://github.com/SolaceProducts/spring-cloud-stream-binder-solace[Solace PubSub+ _(partner maintained)_]
 * https://github.com/Microsoft/spring-cloud-azure/tree/master/spring-cloud-azure-eventhub-stream-binder[Azure Event Hubs _(partner maintained)_]
-

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -31,6 +31,7 @@ import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry;
 import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
+import org.springframework.cloud.stream.config.MessageSourceCustomizer;
 import org.springframework.cloud.stream.function.IntegrationFlowFunctionSupport;
 import org.springframework.cloud.stream.function.StreamFunctionProperties;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
@@ -56,6 +57,7 @@ import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.handler.BridgeHandler;
 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
 import org.springframework.integration.support.ErrorMessageStrategy;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -114,6 +116,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 	private final ListenerContainerCustomizer<?> containerCustomizer;
 
+	private MessageSourceCustomizer<?> sourceCustomizer;
+
 	private ApplicationEventPublisher applicationEventPublisher;
 
 	@Autowired(required = false)
@@ -129,12 +133,23 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		this(headersToEmbed, provisioningProvider, null);
 	}
 
+	@Deprecated
 	public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider,
 			ListenerContainerCustomizer<?> containerCustomizer) {
+
+		this(headersToEmbed, provisioningProvider, containerCustomizer, null);
+	}
+
+	public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider,
+			@Nullable ListenerContainerCustomizer<?> containerCustomizer,
+			@Nullable MessageSourceCustomizer<?> sourceCustomizer) {
+
 		this.headersToEmbed = headersToEmbed == null ? new String[0] : headersToEmbed;
 		this.provisioningProvider = provisioningProvider;
 		this.containerCustomizer = containerCustomizer == null ? (c, q, g) -> {
 		} : containerCustomizer;
+		this.sourceCustomizer = sourceCustomizer == null ? (s, q, g) -> {
+		} : sourceCustomizer;
 	}
 
 	protected ApplicationEventPublisher getApplicationEventPublisher() {
@@ -152,6 +167,11 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		return (ListenerContainerCustomizer<L>) this.containerCustomizer;
 	}
 
+	@SuppressWarnings("unchecked")
+	protected <S> MessageSourceCustomizer<S> getMessageSourceCustomizer() {
+		return (MessageSourceCustomizer<S>) this.sourceCustomizer;
+	}
+
 	/**
 	 * Binds an outbound channel to a given destination. The implementation delegates to
 	 * {@link ProvisioningProvider#provisionProducerDestination(String, ProducerProperties)}
@@ -363,7 +383,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			// the function support for the inbound channel is only for Sink
 			if (shouldWireDunctionToChannel(false)) {
 				inputChannel = this.postProcessInboundChannelForFunction(inputChannel,
-						(ConsumerProperties) properties);
+						properties);
 			}
 			if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
 				enhanceMessageChannel(inputChannel);

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/MessageSourceCustomizer.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2018-2019 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.cloud.stream.config;
+
+import org.springframework.integration.core.MessageSource;
+
+/**
+ * If a single bean of this type is in the application context, pollable message sources
+ * created by the binder can be further customized after all the properties are set. For
+ * example, to configure less-common properties.
+ *
+ * @param <T> {@link MessageSource} type
+ * @author Gary Russell
+ * @since 2.2
+ */
+@FunctionalInterface
+public interface MessageSourceCustomizer<T> {
+
+	/**
+	 * Configure the container that is being created for the supplied queue name and
+	 * consumer group.
+	 * @param source the MessageSource.
+	 * @param destinationName the destination name.
+	 * @param group the consumer group.
+	 */
+	void configure(T source, String destinationName, String group);
+
+}