GH-1841: Add ConsumerEndpointCustomizer

Fixes https://github.com/spring-cloud/spring-cloud-stream/issues/1841

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

@@ -31,6 +31,7 @@ import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry;
+import org.springframework.cloud.stream.config.ConsumerEndpointCustomizer;
 import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
 import org.springframework.cloud.stream.config.MessageSourceCustomizer;
 import org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer;
@@ -114,6 +115,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	private ProducerMessageHandlerCustomizer<MessageHandler> handlerCustomizer =
 		(handler, destination) -> { };
 
+	private ConsumerEndpointCustomizer<MessageProducer> consumerCustomizer =
+		(adapter, destination, group) -> { };
+
 	private ApplicationEventPublisher applicationEventPublisher;
 
 	public AbstractMessageChannelBinder(String[] headersToEmbed,
@@ -149,7 +153,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 	/**
 	 * Configure an optional {@link ProducerMessageHandlerCustomizer} for further
-	 * producer {@link MessageHandler} instances created by the binder.
+	 * configuration of producer {@link MessageHandler} instances created by the binder.
 	 * @param handlerCustomizer the {@link ProducerMessageHandlerCustomizer} to use.
 	 * @since 3.0
 	 */
@@ -163,6 +167,22 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				: (ProducerMessageHandlerCustomizer<MessageHandler>) handlerCustomizer;
 	}
 
+	/**
+	 * Configure an optional {@link ConsumerEndpointCustomizer} for further
+	 * configuration of consumer {@link MessageProducer} instances created by the binder.
+	 * @param endpointCustomizer the {@link ConsumerEndpointCustomizer} to use.
+	 * @since 3.0
+	 */
+	@SuppressWarnings("unchecked")
+	public void setConsumerEndpointCustomizer(
+		@Nullable ConsumerEndpointCustomizer<? extends MessageProducer> endpointCustomizer) {
+
+		this.consumerCustomizer =
+				endpointCustomizer == null
+				? (handler, destination, group) -> { }
+				: (ConsumerEndpointCustomizer<MessageProducer>) endpointCustomizer;
+	}
+
 	@SuppressWarnings("unchecked")
 	protected <L> ListenerContainerCustomizer<L> getContainerCustomizer() {
 		return (ListenerContainerCustomizer<L>) this.containerCustomizer;
@@ -402,6 +422,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			if (properties.isAutoStartup() && consumerEndpoint instanceof Lifecycle) {
 				((Lifecycle) consumerEndpoint).start();
 			}
+			this.consumerCustomizer.configure(consumerEndpoint, name, group);
 
 			Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(name,
 					group, inputChannel, consumerEndpoint instanceof Lifecycle

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2019-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
+ *
+ *      https://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.MessageProducer;
+
+/**
+ * If a single bean of this type is in the application context, an inbound channel adapter
+ * created by the binder can be further customized after all the properties are set. For
+ * example, to configure less-common properties.
+ *
+ * @param <E> {@link MessageProducer} type
+ *
+ * @author Gary Russell
+ *
+ * @since 3.0
+ */
+@FunctionalInterface
+public interface ConsumerEndpointCustomizer<E extends MessageProducer> {
+
+	/**
+	 * Configure a {@link MessageProducer} that is being created by the binder for the
+	 * provided destination name and group.
+	 * @param endpoint the {@link MessageProducer} from the binder.
+	 * @param destinationName the bound destination name.
+	 * @param group the group.
+	 */
+	void configure(E endpoint, String destinationName, String group);
+
+}

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

@@ -19,9 +19,9 @@ package org.springframework.cloud.stream.config;
 import org.springframework.messaging.MessageHandler;
 
 /**
- * If a single bean of this type is in the application context, producing message handler
- * created by the binder can be further customized after all the properties are set. For
- * example, to configure less-common properties.
+ * If a single bean of this type is in the application context, a producing message
+ * handler created by the binder can be further customized after all the properties are
+ * set. For example, to configure less-common properties.
  *
  * @param <H> {@link MessageHandler} type
  *
@@ -33,8 +33,8 @@ import org.springframework.messaging.MessageHandler;
 public interface ProducerMessageHandlerCustomizer<H extends MessageHandler> {
 
 	/**
-	 * Configure a provided {@link MessageHandler} by the binder
-	 * that is being created for the provided destination name.
+	 * Configure a {@link MessageHandler} that is being created by the binder for the
+	 * provided destination name.
 	 * @param handler the {@link MessageHandler} from the binder.
 	 * @param destinationName the bound destination name.
 	 */