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 85f947bca..32ac6d1e1 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.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 handlerCustomizer = (handler, destination) -> { }; + private ConsumerEndpointCustomizer consumerCustomizer = + (adapter, destination, group) -> { }; + private ApplicationEventPublisher applicationEventPublisher; public AbstractMessageChannelBinder(String[] headersToEmbed, @@ -149,7 +153,7 @@ public abstract class AbstractMessageChannelBinder) 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 endpointCustomizer) { + + this.consumerCustomizer = + endpointCustomizer == null + ? (handler, destination, group) -> { } + : (ConsumerEndpointCustomizer) endpointCustomizer; + } + @SuppressWarnings("unchecked") protected ListenerContainerCustomizer getContainerCustomizer() { return (ListenerContainerCustomizer) this.containerCustomizer; @@ -402,6 +422,7 @@ public abstract class AbstractMessageChannelBinder binding = new DefaultBinding(name, group, inputChannel, consumerEndpoint instanceof Lifecycle diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ConsumerEndpointCustomizer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ConsumerEndpointCustomizer.java new file mode 100644 index 000000000..4eeaa6432 --- /dev/null +++ b/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 {@link MessageProducer} type + * + * @author Gary Russell + * + * @since 3.0 + */ +@FunctionalInterface +public interface ConsumerEndpointCustomizer { + + /** + * 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); + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ProducerMessageHandlerCustomizer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ProducerMessageHandlerCustomizer.java index e6c1bba7b..548a48d40 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ProducerMessageHandlerCustomizer.java +++ b/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 {@link MessageHandler} type * @@ -33,8 +33,8 @@ import org.springframework.messaging.MessageHandler; public interface ProducerMessageHandlerCustomizer { /** - * 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. */