diff --git a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java index 88122059f..4eb95a26f 100644 --- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java +++ b/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()); } - /** - * Returns the Listener factory. - * - * @param errorHandler - * the error hander - * @return the {@link SimpleMessageListenerContainer} that gets used receive - * AMQP messages - */ - @Bean(name = { "listenerContainerFactory" }) + @Bean + @ConditionalOnMissingBean(name = "listenerContainerFactory") public RabbitListenerContainerFactory listenerContainerFactory( - final ErrorHandler errorHandler) { - return new ConfigurableRabbitListenerContainerFactory(amqpProperties, rabbitConnectionFactory, errorHandler); + final SimpleRabbitListenerContainerFactoryConfigurer configurer, final ErrorHandler errorHandler) { + final ConfigurableRabbitListenerContainerFactory factory = new ConfigurableRabbitListenerContainerFactory( + amqpProperties.isMissingQueuesFatal(), amqpProperties.getDeclarationRetries(), errorHandler); + configurer.configure(factory, rabbitConnectionFactory); + return factory; } /** diff --git a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpProperties.java b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpProperties.java index 1e33b8106..6336f1b41 100644 --- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpProperties.java +++ b/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; } diff --git a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/ConfigurableRabbitListenerContainerFactory.java b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/ConfigurableRabbitListenerContainerFactory.java index b213e744b..e0610e274 100644 --- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/ConfigurableRabbitListenerContainerFactory.java +++ b/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 - * for the container factory - * @param amqpProperties - * to configure the container factory + * @param missingQueuesFatal + * the missingQueuesFatal to set. + * @see SimpleMessageListenerContainer#setMissingQueuesFatal + * @param declarationRetries + * The number of retries * @param errorHandler * the error handler which should be use */ - public ConfigurableRabbitListenerContainerFactory(final AmqpProperties amqpProperties, - final ConnectionFactory rabbitConnectionFactory, final ErrorHandler errorHandler) { - this.amqpProperties = amqpProperties; - setErrorHandler(errorHandler); - setDefaultRequeueRejected(true); - setConnectionFactory(rabbitConnectionFactory); - setMissingQueuesFatal(amqpProperties.isMissingQueuesFatal()); - setConcurrentConsumers(amqpProperties.getInitialConcurrentConsumers()); - setMaxConcurrentConsumers(amqpProperties.getMaxConcurrentConsumers()); - setPrefetchCount(amqpProperties.getPrefetchCount()); + public ConfigurableRabbitListenerContainerFactory(final boolean missingQueuesFatal, final int declarationRetries, + final ErrorHandler errorHandler) { + this.declarationRetries = declarationRetries; + 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); } } diff --git a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/resources/hawkbit-dmf-defaults.properties b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/resources/hawkbit-dmf-defaults.properties new file mode 100644 index 000000000..c7616efd0 --- /dev/null +++ b/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 + +