diff --git a/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java b/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java index 11f892654..bee9b2364 100644 --- a/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java +++ b/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java @@ -26,6 +26,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler; import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; @@ -37,7 +38,8 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.retry.backoff.ExponentialBackOffPolicy; -import org.springframework.retry.support.RetryTemplate;; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.ErrorHandler;; /** * The spring AMQP configuration which is enabled by using the profile @@ -58,6 +60,9 @@ public class AmqpConfiguration { @Autowired private ConnectionFactory rabbitConnectionFactory; + @Autowired + private ErrorHandler errorHandler; + @Configuration @ConditionalOnMissingBean(ConnectionFactory.class) protected static class RabbitConnectionFactoryCreator { @@ -269,7 +274,7 @@ public class AmqpConfiguration { */ @Bean(name = { "listenerContainerFactory" }) public RabbitListenerContainerFactory listenerContainerFactory() { - return new ConfigurableRabbitListenerContainerFactory(amqpProperties, rabbitConnectionFactory); + return new ConfigurableRabbitListenerContainerFactory(amqpProperties, rabbitConnectionFactory, errorHandler); } private static Map getTTLMaxArgsAuthenticationQueue() { @@ -279,4 +284,15 @@ public class AmqpConfiguration { return args; } + /** + * Create default error handler bean. + * + * @return the default error handler bean + */ + @Bean + @ConditionalOnMissingBean + public ErrorHandler errorHandler() { + return new ConditionalRejectingErrorHandler(); + } + } diff --git a/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/ConfigurableRabbitListenerContainerFactory.java b/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/ConfigurableRabbitListenerContainerFactory.java index 14e6f8fcb..e792db696 100644 --- a/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/ConfigurableRabbitListenerContainerFactory.java +++ b/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/ConfigurableRabbitListenerContainerFactory.java @@ -12,6 +12,7 @@ import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFacto 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; /** * {@link RabbitListenerContainerFactory} that can be configured through @@ -30,8 +31,9 @@ public class ConfigurableRabbitListenerContainerFactory extends SimpleRabbitList * to configure the container factory */ public ConfigurableRabbitListenerContainerFactory(final AmqpProperties amqpProperties, - final ConnectionFactory rabbitConnectionFactory) { + final ConnectionFactory rabbitConnectionFactory, final ErrorHandler errorHandler) { this.amqpProperties = amqpProperties; + setErrorHandler(errorHandler); setDefaultRequeueRejected(true); setConnectionFactory(rabbitConnectionFactory); setMissingQueuesFatal(amqpProperties.isMissingQueuesFatal());