Add ProducerMessageHandlerCustomizer support

Related to https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/issues/107

In some cases there is not enough configuration properties to setup
a producing `MessageHandler` in the binder.
Or we just can't do that via properties because some real object is
required for particular MH option.
For example in the AWS Kinesis Binder end-user would like to provide
a custom `AsyncHandler` which definitely cannot be expressed via properties

* Introduce a `ProducerMessageHandlerCustomizer` similar to existing
`MessageSourceCustomizer` and `ListenerContainerCustomizer`
* Add an `AbstractMessageChannelBinder.setProducerMessageHandlerCustomizer()`
to avoid breaking changes with an additional constructor arg
* Use a provided `handlerCustomizer` in the `customizeProducerMessageHandler()`
called from the `doBindProducer()` on the current `MH` before its
`afterPropertiesSet()`
* It is a target binder configuration responsibility to inject a
`ProducerMessageHandlerCustomizer` into the binder bean

Resolves #1828

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

@@ -33,6 +33,7 @@ import org.springframework.beans.factory.InitializingBean;
 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.config.ProducerMessageHandlerCustomizer;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.cloud.stream.provisioning.ProvisioningException;
@@ -75,12 +76,14 @@ import org.springframework.util.Assert;
  * @param <C> the consumer properties type
  * @param <P> the producer properties type
  * @param <PP> the provisioning producer properties type
+ *
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
  * @author Soby Chacko
  * @author Oleg Zhurakousky
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 1.1
  */
 // @checkstyle:off
@@ -111,7 +114,10 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 	private final ListenerContainerCustomizer<?> containerCustomizer;
 
-	private MessageSourceCustomizer<?> sourceCustomizer;
+	private final MessageSourceCustomizer<?> sourceCustomizer;
+
+	private ProducerMessageHandlerCustomizer<MessageHandler> handlerCustomizer =
+		(handler, destination) -> { };
 
 	private ApplicationEventPublisher applicationEventPublisher;
 
@@ -154,6 +160,22 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		this.applicationEventPublisher = applicationEventPublisher;
 	}
 
+	/**
+	 * Configure an optional {@link ProducerMessageHandlerCustomizer} for further
+	 * producer {@link MessageHandler} instances created by the binder.
+	 * @param handlerCustomizer the {@link ProducerMessageHandlerCustomizer} to use.
+	 * @since 3.0
+	 */
+	@SuppressWarnings("unchecked")
+	public void setProducerMessageHandlerCustomizer(
+		@Nullable ProducerMessageHandlerCustomizer<? extends MessageHandler> handlerCustomizer) {
+
+		this.handlerCustomizer =
+			handlerCustomizer == null
+				? (handler, destination) -> { }
+				: (ProducerMessageHandlerCustomizer<MessageHandler>) handlerCustomizer;
+	}
+
 	@SuppressWarnings("unchecked")
 	protected <L> ListenerContainerCustomizer<L> getContainerCustomizer() {
 		return (ListenerContainerCustomizer<L>) this.containerCustomizer;
@@ -195,6 +217,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 					? registerErrorInfrastructure(producerDestination) : null;
 			producerMessageHandler = createProducerMessageHandler(producerDestination,
 					producerProperties, outputChannel, errorChannel);
+			customizeProducerMessageHandler(producerMessageHandler, producerDestination.getName());
 			if (producerMessageHandler instanceof InitializingBean) {
 				((InitializingBean) producerMessageHandler).afterPropertiesSet();
 			}
@@ -264,6 +287,10 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		return binding;
 	}
 
+	private void customizeProducerMessageHandler(MessageHandler producerMessageHandler, String destinationName) {
+		this.handlerCustomizer.configure(producerMessageHandler, destinationName);
+	}
+
 	/**
 	 * Whether the producer for the destination being created should be configured to use
 	 * native encoding which may, or may not, be determined from the properties. For

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

@@ -0,0 +1,43 @@
+/*
+ * 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.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.
+ *
+ * @param <H> {@link MessageHandler} type
+ *
+ * @author Artem Bilan
+ *
+ * @since 3.0
+ */
+@FunctionalInterface
+public interface ProducerMessageHandlerCustomizer<H extends MessageHandler> {
+
+	/**
+	 * Configure a provided {@link MessageHandler} by the binder
+	 * that is being created for the provided destination name.
+	 * @param handler the {@link MessageHandler} from the binder.
+	 * @param destinationName the bound destination name.
+	 */
+	void configure(H handler, String destinationName);
+
+}