diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 9dc6d7835..4933a22a0 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/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 containerCustomizer() { + return (container, dest, group) -> container.setAdviceChain(advice1, advice2); +} + +@Bean +public MessageSourceCustomizer 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)_] - 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 294ece82c..ebc75d709 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 @@ -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 containerCustomizer; + private MessageSourceCustomizer sourceCustomizer; + private ApplicationEventPublisher applicationEventPublisher; @Autowired(required = false) @@ -129,12 +133,23 @@ public abstract class AbstractMessageChannelBinder 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) this.containerCustomizer; } + @SuppressWarnings("unchecked") + protected MessageSourceCustomizer getMessageSourceCustomizer() { + return (MessageSourceCustomizer) 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 {@link MessageSource} type + * @author Gary Russell + * @since 2.2 + */ +@FunctionalInterface +public interface MessageSourceCustomizer { + + /** + * 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); + +}