LIBS-808 - Add support for prefetch count configuration (#16)

* LIBS-808 - Add support for prefetch count configuration

* LIBS-808 - improvements

* LIBS-808 - fix

* LIBS-808 - improve test method naming

* LIBS-808 - fix

Author: fegbers

README.md

@@ -12,7 +12,7 @@
 <dependency>
 	<groupId>com.avides.spring</groupId>
 	<artifactId>spring-rabbit</artifactId>
-	<version>2.5.0</version>
+	<version>2.6.0</version>
 </dependency>
 ```
 
@@ -143,6 +143,7 @@ spring.rabbitmq.queues[0].exchange.name=com.example.exchange.zero
 spring.rabbitmq.queues[0].exchange.type=DIRECT
 spring.rabbitmq.queues[0].limit=500000
 spring.rabbitmq.queues[0].listener.bean-name=myListenerZero
+spring.rabbitmq.queues[0].listener.prefetch-count=55
 spring.rabbitmq.queues[0].listener.max-concurrent-consumers=2
 spring.rabbitmq.queues[0].rabbit-admin.bean-name=myRabbitAdminZero
 spring.rabbitmq.queues[0].rabbit-admin.connection-factory.bean-name=myConnectionFactoryZero
@@ -203,6 +204,17 @@ See [ExchangeProperties](#exchangeproperties)
 
 See [MessageConverterProperties](#messageconverterproperties)
 
+### spring.rabbitmq.prefetch-count
+
+@Min(1)
+
+Default 500. Could be overridden by each listener.
+
+``` ini
+spring.rabbitmq.prefetch-count=50
+```
+
+
 ### spring.rabbitmq.max-concurrent-consumers
 
 @Min(1)
@@ -453,6 +465,15 @@ spring.rabbitmq.queues[0].listener.bean-name=myListener
 
 See [MessageConverterProperties](#messageconverterproperties)
 
+#### .prefetch-count
+
+Could be null or @Min(1)
+
+
+``` ini
+spring.rabbitmq.queues[0].listener.prefetch-count=30
+```
+
 #### .max-concurrent-consumers
 
 Could be null or @Min(1)

pom.xml

@@ -3,7 +3,7 @@
 
   <groupId>com.avides.spring</groupId>
   <artifactId>spring-rabbit</artifactId>
-  <version>2.5.0</version>
+  <version>2.6.0</version>
 
   <name>spring-rabbit</name>
   <description>Makes configuring RabbitMQ for Spring Boot applications more comfortable</description>

src/main/java/com/avides/spring/rabbit/configuration/SpringRabbitAutoConfiguration.java

@@ -46,7 +46,6 @@
 import com.avides.spring.rabbit.configuration.domain.ListenerProperties;
 import com.avides.spring.rabbit.configuration.domain.MessageConverterProperties;
 import com.avides.spring.rabbit.configuration.domain.QueueProperties;
-import com.avides.spring.rabbit.configuration.domain.RabbitAdminProperties;
 import com.avides.spring.rabbit.configuration.domain.RabbitTemplateProperties;
 import com.avides.spring.rabbit.configuration.provider.ConnectionFactoryProvider;
 import com.avides.spring.rabbit.configuration.util.DefaultValueResolver;
@@ -119,6 +118,10 @@ public class SpringRabbitAutoConfiguration implements InitializingBean
     @NestedConfigurationProperty
     private MessageConverterProperties messageConverter;
 
+    @NotNull
+    @Min(1)
+    private Integer prefetchCount = Integer.valueOf(500);
+
     @NotNull
     @Min(1)
     private Integer maxConcurrentConsumers = Integer.valueOf(1);
@@ -227,45 +230,46 @@ private void configureQueues()
         {
             if (queueProperties.isCreationEnabled())
             {
-                String queueName = queueProperties.getName();
-                RabbitAdminProperties rabbitAdminProperties = queueProperties.getRabbitAdmin();
-                String rabbitAdminBeanName = rabbitAdminProperties.getBeanName();
-                BeanReferenceConnectionFactoryProperties customConnectionFactoryProperties = rabbitAdminProperties.getConnectionFactory();
+                var queueName = queueProperties.getName();
+                var rabbitAdminProperties = queueProperties.getRabbitAdmin();
+                var rabbitAdminBeanName = rabbitAdminProperties.getBeanName();
+                var customConnectionFactoryProperties = rabbitAdminProperties.getConnectionFactory();
 
-                ConnectionFactory resolvedConnectionFactory = DefaultValueResolver
+                var resolvedConnectionFactory = DefaultValueResolver
                         .resolveConnectionFactory(customConnectionFactoryProperties, CONNECTION_FACTORY_BEAN_NAME, applicationContext);
 
-                RabbitAdmin rabbitAdmin = new RabbitAdminCreator(resolvedConnectionFactory).createInstance();
+                var rabbitAdmin = new RabbitAdminCreator(resolvedConnectionFactory).createInstance();
                 if (!applicationContext.containsBean(rabbitAdminBeanName))
                 {
                     applicationContext.registerBean(rabbitAdminBeanName, RabbitAdmin.class, () -> rabbitAdmin, beanDefinition -> beanDefinition
                             .setScope(BeanDefinition.SCOPE_SINGLETON));
                     log.info("RabbitAdmin build with bean-name '{}'", rabbitAdminBeanName);
                 }
 
-                Creator<Queue> queueCreator = new QueueCreator(queueProperties, rabbitAdmin, DefaultValueResolver
+                var queueCreator = new QueueCreator(queueProperties, rabbitAdmin, DefaultValueResolver
                         .resolveExchange(queueProperties.getExchange(), exchange));
                 addToContext(queueName, resolvedConnectionFactory, queueCreator.createInstance());
 
-                Creator<Queue> dlxQueueCreator = new DlxQueueCreator(rabbitAdmin, queueProperties);
+                var dlxQueueCreator = new DlxQueueCreator(rabbitAdmin, queueProperties);
                 addToContext(queueName + ".dlx", resolvedConnectionFactory, dlxQueueCreator.createInstance());
 
                 if (queueProperties.getListener() != null)
                 {
-                    ListenerProperties listenerProperties = queueProperties.getListener();
+                    var listenerProperties = queueProperties.getListener();
 
                     if (listenerProperties.isCreationEnabled())
                     {
-                        String listenerBeanName = listenerProperties.getBeanName();
-                        int resolvedMaxConcurrentConsumers = DefaultValueResolver
-                                .resolveMaxConcurrentConsumers(listenerProperties.getMaxConcurrentConsumers(), maxConcurrentConsumers);
-                        MessageConverter resolvedMessageConverter = DefaultValueResolver.resolveMessageConverter(listenerProperties
+                        var listenerBeanName = listenerProperties.getBeanName();
+                        var resolvedPrefetchCount = DefaultValueResolver.resolveValue(listenerProperties.getPrefetchCount(), prefetchCount);
+                        var resolvedMaxConcurrentConsumers = DefaultValueResolver
+                                .resolveValue(listenerProperties.getMaxConcurrentConsumers(), maxConcurrentConsumers);
+                        var resolvedMessageConverter = DefaultValueResolver.resolveMessageConverter(listenerProperties
                                 .getMessageConverter(), messageConverter, applicationContext, existingMessageConverterList);
-                        Object listener = applicationContext.getBean(listenerBeanName);
+                        var listener = applicationContext.getBean(listenerBeanName);
 
-                        Creator<DefaultMessageListenerContainer<Object>> listenerCreator = new ListenerCreator(resolvedConnectionFactory, queueName, resolvedMaxConcurrentConsumers, resolvedMessageConverter, listener);
+                        var listenerCreator = new ListenerCreator(resolvedConnectionFactory, queueName, resolvedPrefetchCount, resolvedMaxConcurrentConsumers, resolvedMessageConverter, listener);
 
-                        String beanName = createListenerContainerBeanName(listenerProperties, queueName, customConnectionFactoryProperties);
+                        var beanName = createListenerContainerBeanName(listenerProperties, queueName, customConnectionFactoryProperties);
 
                         applicationContext
                                 .registerBean(beanName, DefaultMessageListenerContainer.class, listenerCreator::createInstance, beanDefinition -> beanDefinition

src/main/java/com/avides/spring/rabbit/configuration/creator/ListenerCreator.java

@@ -16,6 +16,8 @@ public class ListenerCreator implements Creator<DefaultMessageListenerContainer<
 
     private final String queueName;
 
+    private final int prefetchCount;
+
     private final int maxConcurrentConsumers;
 
     private final MessageConverter messageConverter;
@@ -27,6 +29,7 @@ public DefaultMessageListenerContainer<Object> createInstance()
     {
         DefaultMessageListenerContainer<Object> container = new DefaultMessageListenerContainer<>(connectionFactory);
         container.setQueueNames(queueName);
+        container.setPrefetchCount(prefetchCount);
         container.setMaxConcurrentConsumers(maxConcurrentConsumers);
         apppendListenerToContainer(container, messageConverter);
         return container;

src/main/java/com/avides/spring/rabbit/configuration/domain/ListenerProperties.java

@@ -27,6 +27,9 @@ public class ListenerProperties
     @Valid
     private MessageConverterProperties messageConverter;
 
+    @Min(1)
+    private Integer prefetchCount;
+
     @Min(1)
     private Integer maxConcurrentConsumers;
 }

src/main/java/com/avides/spring/rabbit/configuration/util/DefaultValueResolver.java

@@ -23,10 +23,8 @@ public class DefaultValueResolver
     /**
      * Resolves and creates a {@link Exchange} for a custom and a default exchange configuration considering the override of a possible custom configuration.
      *
-     * @param customProperties
-     *            custom exchange configuration that overrides the default
-     * @param defaultProperties
-     *            default exchange configuration that is used if no custom configuration exists
+     * @param customProperties custom exchange configuration that overrides the default
+     * @param defaultProperties default exchange configuration that is used if no custom configuration exists
      * @return resolved {@link Exchange}
      */
     public Exchange resolveExchange(ExchangeProperties customProperties, ExchangeProperties defaultProperties)
@@ -46,12 +44,10 @@ else if (defaultProperties != null)
     /**
      * Resolves a {@link ConnectionFactory} for multiple connection factories considering the override of a possible custom connection factory.
      *
-     * @param customProperties
-     *            {@link BeanReferenceConnectionFactoryProperties} containing the configuration for a custom connection factory that overrides the default
-     * @param defaultBeanName
-     *            bean name for the default connection factory (created by spring and renamed by the configuration)
-     * @param applicationContext
-     *            context containing all spring beans
+     * @param customProperties {@link BeanReferenceConnectionFactoryProperties} containing the configuration for a custom connection factory that overrides the
+     *        default
+     * @param defaultBeanName bean name for the default connection factory (created by spring and renamed by the configuration)
+     * @param applicationContext context containing all spring beans
      * @return resolved {@link ConnectionFactory}
      */
     public ConnectionFactory resolveConnectionFactory(BeanReferenceConnectionFactoryProperties customProperties, String defaultBeanName,
@@ -77,39 +73,40 @@ public ConnectionFactory resolveConnectionFactory(BeanReferenceConnectionFactory
     /**
      * Resolves a bean name for multiple connection factories considering the override of a possible custom connection factory.
      *
-     * @param customProperties
-     *            {@link BeanReferenceConnectionFactoryProperties} containing the bean name for a custom connection factory that overrides the default
-     * @param defaultBeanName
-     *            bean name for the default connection factory (created by spring and renamed by the configuration) that is used if no custom
+     * @param customProperties {@link BeanReferenceConnectionFactoryProperties} containing the bean name for a custom connection factory that overrides the
+     *        default
+     * @param defaultBeanName bean name for the default connection factory (created by spring and renamed by the configuration) that is used if no custom
      * @return beanName for the resolved connectionFactory
      */
     public String resolveConnectionFactoryBeanName(BeanReferenceConnectionFactoryProperties customProperties, String defaultBeanName)
     {
-        if (customProperties != null)
-        {
-            return customProperties.getBeanName();
-        }
-
-        return defaultBeanName;
+        return customProperties != null ? customProperties.getBeanName() : defaultBeanName;
     }
 
     /**
      * Resolves the max concurrent consumers for a {@link DefaultMessageListenerContainer} considering the override of a possible custom configuration.
      *
-     * @param customMax
-     *            custom quantity to override the default
-     * @param defaultMax
-     *            quantity which is used if no custom configuration exists
+     * @deprecated Please use {@link #resolveValue(Integer, Integer)} instead
+     * @param customMax custom quantity to override the default
+     * @param defaultMax quantity which is used if no custom configuration exists
      * @return resolved max concurrent consumers for a {@link DefaultMessageListenerContainer}
      */
+    @Deprecated(forRemoval = true, since = "2.6.0")
     public int resolveMaxConcurrentConsumers(Integer customMax, Integer defaultMax)
     {
-        if (customMax != null)
-        {
-            return customMax.intValue();
-        }
+        return resolveValue(customMax, defaultMax);
+    }
 
-        return defaultMax.intValue();
+    /**
+     * Resolves the value between a custom and a default value
+     *
+     * @param customValue custom value to override the default
+     * @param defaultValue value which is used if no custom configuration exists
+     * @return resolved max
+     */
+    public int resolveValue(Integer customValue, Integer defaultValue)
+    {
+        return customValue != null ? customValue.intValue() : defaultValue.intValue();
     }
 
     /**
@@ -122,14 +119,10 @@ public int resolveMaxConcurrentConsumers(Integer customMax, Integer defaultMax)
      * <li>existing if only one {@link MessageConverter} exists</li>
      * </ol>
      *
-     * @param customProperties
-     *            custom configuration to override the default
-     * @param defaultProperties
-     *            default configuration which is used if no custom configuration exists
-     * @param applicationContext
-     *            context containing all spring beans
-     * @param existingMessageConverters
-     *            list of existing {@link MessageConverter}
+     * @param customProperties custom configuration to override the default
+     * @param defaultProperties default configuration which is used if no custom configuration exists
+     * @param applicationContext context containing all spring beans
+     * @param existingMessageConverters list of existing {@link MessageConverter}
      * @return resolved {@link MessageConverter}
      */
     public MessageConverter resolveMessageConverter(MessageConverterProperties customProperties, MessageConverterProperties defaultProperties,

src/test/java/com/avides/spring/rabbit/configuration/SpringRabbitAutoConfigurationForDifferentPrefetchCountIT.java

@@ -0,0 +1,50 @@
+package com.avides.spring.rabbit.configuration;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.List;
+
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.ActiveProfiles;
+
+import com.avides.spring.rabbit.listener.container.DefaultMessageListenerContainer;
+import com.avides.spring.rabbit.test.support.AbstractIT;
+import com.avides.spring.rabbit.test.support.DummyListenerOne;
+import com.avides.spring.rabbit.test.support.DummyListenerZero;
+
+@ActiveProfiles({ "prefetchCount" })
+@SpringBootTest(classes = { SpringRabbitAutoConfigurationForDifferentPrefetchCountIT.TestConfiguration.class, DummyListenerZero.class, DummyListenerOne.class })
+public class SpringRabbitAutoConfigurationForDifferentPrefetchCountIT extends AbstractIT
+{
+    @Autowired
+    private List<DefaultMessageListenerContainer<Object>> listenerContainer;
+
+    @Qualifier("[email protected]@firstConnectionFactory")
+    @Autowired
+    private DefaultMessageListenerContainer<Object> dummyListenerZeroContainer;
+
+    @Qualifier("[email protected]@secondConnectionFactory")
+    @Autowired
+    private DefaultMessageListenerContainer<Object> dummyListenerOneContainer;
+
+    @Test
+    public void testPrefetchCount()
+    {
+        assertEquals(2, listenerContainer.size());
+        assertEquals(20, getPrefetchCount(dummyListenerZeroContainer));
+        assertEquals(4, getPrefetchCount(dummyListenerOneContainer));
+    }
+
+    @EnableAutoConfiguration(exclude = RabbitAutoConfiguration.class)
+    @Configuration
+    static class TestConfiguration extends AbstractIT.TestConfiguration
+    {
+        // nothing, just to exclude RabbitAutoConfiguration
+    }
+}

src/test/java/com/avides/spring/rabbit/configuration/SpringRabbitAutoConfigurationForMultipleConnectionFactoriesIT.java

@@ -92,7 +92,8 @@ private void checkListenerContainer()
         assertEquals("guest", dummyListenerZeroContainer.getConnectionFactory().getUsername());
         assertEquals("/", dummyListenerZeroContainer.getConnectionFactory().getVirtualHost());
         assertEquals(host, dummyListenerZeroContainer.getConnectionFactory().getHost());
-        assertTrue(dummyListenerOneContainer.isActive());
+        assertTrue(dummyListenerZeroContainer.isActive());
+        assertEquals(500, getPrefetchCount(dummyListenerZeroContainer));
 
         assertEquals(AcknowledgeMode.AUTO, dummyListenerOneContainer.getAcknowledgeMode());
         assertEquals(1, dummyListenerOneContainer.getQueueNames().length);
@@ -101,6 +102,7 @@ private void checkListenerContainer()
         assertEquals("/", dummyListenerOneContainer.getConnectionFactory().getVirtualHost());
         assertEquals(host, dummyListenerOneContainer.getConnectionFactory().getHost());
         assertTrue(dummyListenerOneContainer.isActive());
+        assertEquals(500, getPrefetchCount(dummyListenerOneContainer));
     }
 
     private void checkQueues()