From dd6eaba70f9822d6a6b877c22463bc7d6967f412 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 26 Jun 2018 12:05:14 -0400 Subject: [PATCH] GH-139 polishing based on changes in cire GH-1399 Resolves #139 Resolves #145 --- .../rabbit/RabbitMessageChannelBinder.java | 12 +++--- .../config/ListenerContainerCustomizer.java | 43 ------------------- ...bbitMessageChannelBinderConfiguration.java | 10 ++--- .../integration/RabbitBinderModuleTests.java | 16 +++---- 4 files changed, 18 insertions(+), 63 deletions(-) delete mode 100644 spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index b508849a7..d8980563a 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -34,6 +34,7 @@ import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.support.BatchingStrategy; import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException; import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer; @@ -55,12 +56,12 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.HeaderMode; -import org.springframework.cloud.stream.binder.rabbit.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; +import org.springframework.cloud.stream.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; import org.springframework.core.task.SimpleAsyncTaskExecutor; @@ -132,8 +133,6 @@ public class RabbitMessageChannelBinder private final RabbitProperties rabbitProperties; - private final ListenerContainerCustomizer containerCustomizer; - private boolean destroyConnectionFactory; private ConnectionFactory connectionFactory; @@ -157,13 +156,12 @@ public class RabbitMessageChannelBinder public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties, RabbitExchangeQueueProvisioner provisioningProvider, - ListenerContainerCustomizer containerCustomizer) { - super(new String[0], provisioningProvider); + ListenerContainerCustomizer containerCustomizer) { + super(new String[0], provisioningProvider, containerCustomizer); Assert.notNull(connectionFactory, "connectionFactory must not be null"); Assert.notNull(rabbitProperties, "rabbitProperties must not be null"); this.connectionFactory = connectionFactory; this.rabbitProperties = rabbitProperties; - this.containerCustomizer = containerCustomizer == null ? (c, q, g) -> { } : containerCustomizer; } /** @@ -399,7 +397,7 @@ public class RabbitMessageChannelBinder else if (getApplicationContext() != null) { listenerContainer.setApplicationEventPublisher(getApplicationContext()); } - this.containerCustomizer.configure(listenerContainer, consumerDestination.getName(), group); + this.getContainerCustomizer().configure(listenerContainer, consumerDestination.getName(), group); listenerContainer.afterPropertiesSet(); AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer); diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java deleted file mode 100644 index abedb9f16..000000000 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Copyright 2018 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 - * - * http://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.binder.rabbit.config; - -import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; - -/** - * If a single bean of this type is in the application context, listener - * containers created by the binder can be further customized after all - * the properties are set. For example, to configure less-common - * properties. - * - * @author Gary Russell - * @since 2.1 - * - */ -@FunctionalInterface -public interface ListenerContainerCustomizer { - - /** - * Configure the container that is being created for the supplied queue name and - * consumer group. - * @param container the container. - * @param queueName the destination name. - * @param group the consumer group. - */ - void configure(AbstractMessageListenerContainer container, String queueName, String group); - -} diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index 432920978..8fd0ab6cf 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -22,6 +22,7 @@ import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; import org.springframework.beans.factory.annotation.Autowired; @@ -34,9 +35,11 @@ import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder import org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties; import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; +import org.springframework.cloud.stream.config.ListenerContainerCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; +import org.springframework.lang.Nullable; /** * Configuration class for RabbitMQ message channel binder. @@ -64,13 +67,10 @@ public class RabbitMessageChannelBinderConfiguration { @Autowired private RabbitExtendedBindingProperties rabbitExtendedBindingProperties; - @Autowired(required = false) - private ListenerContainerCustomizer containerCustomizer; - @Bean - RabbitMessageChannelBinder rabbitMessageChannelBinder() throws Exception { + RabbitMessageChannelBinder rabbitMessageChannelBinder(@Nullable ListenerContainerCustomizer listenerContainerCustomizer) throws Exception { RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(this.rabbitConnectionFactory, - this.rabbitProperties, provisioningProvider(), this.containerCustomizer); + this.rabbitProperties, provisioningProvider(), listenerContainerCustomizer); binder.setAdminAddresses(this.rabbitBinderConfigurationProperties.getAdminAddresses()); binder.setCompressingPostProcessor(gZipPostProcessor()); binder.setDecompressingPostProcessor(deCompressingPostProcessor()); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index b62a60941..c062a7e3e 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -16,6 +16,11 @@ package org.springframework.cloud.stream.binder.rabbit.integration; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -25,11 +30,11 @@ import org.junit.After; import org.junit.ClassRule; import org.junit.Test; import org.mockito.Mockito; - import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; @@ -47,11 +52,11 @@ import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; -import org.springframework.cloud.stream.binder.rabbit.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; import org.springframework.cloud.stream.binding.BindingService; +import org.springframework.cloud.stream.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; @@ -62,11 +67,6 @@ import org.springframework.retry.backoff.ExponentialBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.BDDMockito.willReturn; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; - /** * @author Marius Bogoevici * @author Gary Russell @@ -291,7 +291,7 @@ public class RabbitBinderModuleTests { public static class SimpleProcessor { @Bean - public ListenerContainerCustomizer containerCustomizer() { + public ListenerContainerCustomizer containerCustomizer() { return (c, q, g) -> c.setBeanName("setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g)); }