From a516f9382d89d09e3399d58e74b328da38dc16d1 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 22 Oct 2019 14:23:41 -0400 Subject: [PATCH] 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 --- .../binder/AbstractMessageChannelBinder.java | 29 ++++++++++++- .../ProducerMessageHandlerCustomizer.java | 43 +++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ProducerMessageHandlerCustomizer.java diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 1e0f2203b..5df27ccf5 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/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 the consumer properties type * @param

the producer properties type * @param 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 containerCustomizer; - private MessageSourceCustomizer sourceCustomizer; + private final MessageSourceCustomizer sourceCustomizer; + + private ProducerMessageHandlerCustomizer handlerCustomizer = + (handler, destination) -> { }; private ApplicationEventPublisher applicationEventPublisher; @@ -154,6 +160,22 @@ public abstract class AbstractMessageChannelBinder handlerCustomizer) { + + this.handlerCustomizer = + handlerCustomizer == null + ? (handler, destination) -> { } + : (ProducerMessageHandlerCustomizer) handlerCustomizer; + } + @SuppressWarnings("unchecked") protected ListenerContainerCustomizer getContainerCustomizer() { return (ListenerContainerCustomizer) this.containerCustomizer; @@ -195,6 +217,7 @@ public abstract class AbstractMessageChannelBinder {@link MessageHandler} type + * + * @author Artem Bilan + * + * @since 3.0 + */ +@FunctionalInterface +public interface ProducerMessageHandlerCustomizer { + + /** + * 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); + +}