SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-139 polishing based on changes in cire GH-1399

Browse Source 
Resolves #139
Resolves #145
This commit is contained in:
Oleg Zhurakousky
2018-06-26 12:05:14 -04:00
parent f7f801d0ee
commit dd6eaba70f
4 changed files with 18 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -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<AbstractMessageListenerContainer> 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);

43
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ListenerContainerCustomizer.java
View File

@@ -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);
}

10
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java
View File

@@ -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<AbstractMessageListenerContainer> 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());

16
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java
View File

@@ -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<AbstractMessageListenerContainer> containerCustomizer() {
return (c, q, g) -> c.setBeanName("setByCustomizerForQueue:" + q +
(g == null ? "" : ",andGroup:" + g));
}