Adapt rabbitmq listener properties (#538)

* Adapt rabbitmq listener properties

Signed-off-by: SirWayne <dennis.melzer@bosch-si.com>

* Add default for spring.rabbitmq.requested-heartbeat

Signed-off-by: SirWayne <dennis.melzer@bosch-si.com>

hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java

@@ -39,11 +39,13 @@ import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.bus.ServiceMatcher;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.PropertySource;
 import org.springframework.retry.backoff.ExponentialBackOffPolicy;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.ErrorHandler;
@@ -57,6 +59,7 @@ import com.google.common.collect.Maps;
  */
 @EnableConfigurationProperties({ AmqpProperties.class, AmqpDeadletterProperties.class })
 @ConditionalOnProperty(prefix = "hawkbit.dmf.rabbitmq", name = "enabled", matchIfMissing = true)
+@PropertySource("classpath:/hawkbit-dmf-defaults.properties")
 public class AmqpConfiguration {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(AmqpConfiguration.class);
@@ -265,18 +268,14 @@ public class AmqpConfiguration {
         return new DefaultAmqpSenderService(rabbitTemplate());
     }
 
-    /**
+    @Bean
-     * Returns the Listener factory.
+    @ConditionalOnMissingBean(name = "listenerContainerFactory")
-     *
-     * @param errorHandler
-     *            the error hander
-     * @return the {@link SimpleMessageListenerContainer} that gets used receive
-     *         AMQP messages
-     */
-    @Bean(name = { "listenerContainerFactory" })
     public RabbitListenerContainerFactory<SimpleMessageListenerContainer> listenerContainerFactory(
-            final ErrorHandler errorHandler) {
+            final SimpleRabbitListenerContainerFactoryConfigurer configurer, final ErrorHandler errorHandler) {
-        return new ConfigurableRabbitListenerContainerFactory(amqpProperties, rabbitConnectionFactory, errorHandler);
+        final ConfigurableRabbitListenerContainerFactory factory = new ConfigurableRabbitListenerContainerFactory(
+                amqpProperties.isMissingQueuesFatal(), amqpProperties.getDeclarationRetries(), errorHandler);
+        configurer.configure(factory, rabbitConnectionFactory);
+        return factory;
     }
 
     /**

hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpProperties.java

@@ -8,8 +8,6 @@
  */
 package org.eclipse.hawkbit.amqp;
 
-import java.util.concurrent.TimeUnit;
-
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
 /**
@@ -20,16 +18,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 @ConfigurationProperties("hawkbit.dmf.rabbitmq")
 public class AmqpProperties {
 
-    private static final int ONE_MINUTE = 60;
-
     private static final int DEFAULT_QUEUE_DECLARATION_RETRIES = 50;
 
-    private static final int DEFAULT_INITIAL_CONSUMERS = 3;
-
-    private static final int DEFAULT_PREFETCH_COUNT = 10;
-
-    private static final int DEFAULT_MAX_CONSUMERS = 10;
-
     private static final long DEFAULT_REQUEUE_DELAY = 0;
 
     /**
@@ -63,28 +53,6 @@ public class AmqpProperties {
      */
     private boolean missingQueuesFatal;
 
-    /**
-     * Requested heartbeat interval from broker in {@link TimeUnit#SECONDS}.
-     */
-    private int requestedHeartBeat = (int) TimeUnit.SECONDS.toSeconds(ONE_MINUTE);
-
-    /**
-     * Sets an upper limit to the number of consumers.
-     */
-    private int maxConcurrentConsumers = DEFAULT_MAX_CONSUMERS;
-
-    /**
-     * Tells the broker how many messages to send to each consumer in a single
-     * request. Often this can be set quite high to improve throughput.
-     */
-    private int prefetchCount = DEFAULT_PREFETCH_COUNT;
-
-    /**
-     * Initial number of consumers. Is scaled up if necessary up to
-     * {@link #maxConcurrentConsumers}.
-     */
-    private int initialConcurrentConsumers = DEFAULT_INITIAL_CONSUMERS;
-
     /**
      * The number of retry attempts when passive queue declaration fails.
      * Passive queue declaration occurs when the consumer starts or, when
@@ -122,30 +90,6 @@ public class AmqpProperties {
         this.authenticationReceiverQueue = authenticationReceiverQueue;
     }
 
-    public int getPrefetchCount() {
-        return prefetchCount;
-    }
-
-    public void setPrefetchCount(final int prefetchCount) {
-        this.prefetchCount = prefetchCount;
-    }
-
-    public int getInitialConcurrentConsumers() {
-        return initialConcurrentConsumers;
-    }
-
-    public void setInitialConcurrentConsumers(final int initialConcurrentConsumers) {
-        this.initialConcurrentConsumers = initialConcurrentConsumers;
-    }
-
-    public int getMaxConcurrentConsumers() {
-        return maxConcurrentConsumers;
-    }
-
-    public void setMaxConcurrentConsumers(final int maxConcurrentConsumers) {
-        this.maxConcurrentConsumers = maxConcurrentConsumers;
-    }
-
     public boolean isMissingQueuesFatal() {
         return missingQueuesFatal;
     }
@@ -174,14 +118,6 @@ public class AmqpProperties {
         return receiverQueue;
     }
 
-    public int getRequestedHeartBeat() {
-        return requestedHeartBeat;
-    }
-
-    public void setRequestedHeartBeat(final int requestedHeartBeat) {
-        this.requestedHeartBeat = requestedHeartBeat;
-    }
-
     public void setReceiverQueue(final String receiverQueue) {
         this.receiverQueue = receiverQueue;
     }

hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/ConfigurableRabbitListenerContainerFactory.java

@@ -9,7 +9,6 @@
 package org.eclipse.hawkbit.amqp;
 
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.util.ErrorHandler;
@@ -20,29 +19,25 @@ import org.springframework.util.ErrorHandler;
  *
  */
 public class ConfigurableRabbitListenerContainerFactory extends SimpleRabbitListenerContainerFactory {
-    private final AmqpProperties amqpProperties;
+    private final int declarationRetries;
 
     /**
      * Constructor.
      * 
-     * @param rabbitConnectionFactory
+     * @param missingQueuesFatal
-     *            for the container factory
+     *            the missingQueuesFatal to set.
-     * @param amqpProperties
+     * @see SimpleMessageListenerContainer#setMissingQueuesFatal
-     *            to configure the container factory
+     * @param declarationRetries
+     *            The number of retries
      * @param errorHandler
      *            the error handler which should be use
      */
-    public ConfigurableRabbitListenerContainerFactory(final AmqpProperties amqpProperties,
+    public ConfigurableRabbitListenerContainerFactory(final boolean missingQueuesFatal, final int declarationRetries,
-            final ConnectionFactory rabbitConnectionFactory, final ErrorHandler errorHandler) {
+            final ErrorHandler errorHandler) {
-        this.amqpProperties = amqpProperties;
+        this.declarationRetries = declarationRetries;
-        setErrorHandler(errorHandler);
-        setDefaultRequeueRejected(true);
-        setConnectionFactory(rabbitConnectionFactory);
-        setMissingQueuesFatal(amqpProperties.isMissingQueuesFatal());
-        setConcurrentConsumers(amqpProperties.getInitialConcurrentConsumers());
-        setMaxConcurrentConsumers(amqpProperties.getMaxConcurrentConsumers());
-        setPrefetchCount(amqpProperties.getPrefetchCount());
 
+        setErrorHandler(errorHandler);
+        setMissingQueuesFatal(missingQueuesFatal);
     }
 
     @Override
@@ -51,6 +46,6 @@ public class ConfigurableRabbitListenerContainerFactory extends SimpleRabbitList
     @SuppressWarnings("squid:UnusedProtectedMethod")
     protected void initializeContainer(final SimpleMessageListenerContainer instance) {
         super.initializeContainer(instance);
-        instance.setDeclarationRetries(amqpProperties.getDeclarationRetries());
+        instance.setDeclarationRetries(declarationRetries);
     }
 }

hawkbit-dmf/hawkbit-dmf-amqp/src/main/resources/hawkbit-dmf-defaults.properties

@@ -0,0 +1,19 @@
+#
+# Copyright (c) 2015 Bosch Software Innovations GmbH and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Eclipse Public License v1.0
+# which accompanies this distribution, and is available at
+# http://www.eclipse.org/legal/epl-v10.html
+#
+
+## DMF RabbitMQ configuration - START
+spring.rabbitmq.listener.prefetch=10
+spring.rabbitmq.listener.concurrency=3
+spring.rabbitmq.listener.max-concurrency=10
+spring.rabbitmq.requested-heartbeat=60
+
+hawkbit.dmf.rabbitmq.declaration-retries=10000
+##DMF RabbitMQ configuration - END
+
+