@@ -39,6 +39,12 @@ public class AmqpProperties {
|
|||||||
*/
|
*/
|
||||||
private String deadLetterExchange = "simulator.deadletter";
|
private String deadLetterExchange = "simulator.deadletter";
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Message time to live (ttl) for the deadletter queue. Default ttl is 1
|
||||||
|
* hour.
|
||||||
|
*/
|
||||||
|
private int deadLetterTtl = 60_000;
|
||||||
|
|
||||||
public String getReceiverConnectorQueueFromSp() {
|
public String getReceiverConnectorQueueFromSp() {
|
||||||
return receiverConnectorQueueFromSp;
|
return receiverConnectorQueueFromSp;
|
||||||
}
|
}
|
||||||
@@ -70,4 +76,12 @@ public class AmqpProperties {
|
|||||||
public void setSenderForSpExchange(final String senderForSpExchange) {
|
public void setSenderForSpExchange(final String senderForSpExchange) {
|
||||||
this.senderForSpExchange = senderForSpExchange;
|
this.senderForSpExchange = senderForSpExchange;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public int getDeadLetterTtl() {
|
||||||
|
return deadLetterTtl;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void setDeadLetterTtl(final int deadLetterTtl) {
|
||||||
|
this.deadLetterTtl = deadLetterTtl;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,9 +8,6 @@
|
|||||||
*/
|
*/
|
||||||
package org.eclipse.hawkbit.amqp;
|
package org.eclipse.hawkbit.amqp;
|
||||||
|
|
||||||
import java.util.HashMap;
|
|
||||||
import java.util.Map;
|
|
||||||
|
|
||||||
import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
|
import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
|
||||||
import org.springframework.amqp.core.Binding;
|
import org.springframework.amqp.core.Binding;
|
||||||
import org.springframework.amqp.core.BindingBuilder;
|
import org.springframework.amqp.core.BindingBuilder;
|
||||||
@@ -31,15 +28,29 @@ import org.springframework.context.annotation.Bean;
|
|||||||
* {@code amqp} to use a AMQP for communication with SP enabled devices.
|
* {@code amqp} to use a AMQP for communication with SP enabled devices.
|
||||||
*
|
*
|
||||||
*/
|
*/
|
||||||
@EnableConfigurationProperties(AmqpProperties.class)
|
@EnableConfigurationProperties({ AmqpProperties.class, AmqpDeadletterProperties.class })
|
||||||
public class AmqpConfiguration {
|
public class AmqpConfiguration {
|
||||||
|
|
||||||
@Autowired
|
@Autowired
|
||||||
protected AmqpProperties amqpProperties;
|
protected AmqpProperties amqpProperties;
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
protected AmqpDeadletterProperties amqpDeadletterProperties;
|
||||||
|
|
||||||
@Autowired
|
@Autowired
|
||||||
private ConnectionFactory connectionFactory;
|
private ConnectionFactory connectionFactory;
|
||||||
|
|
||||||
|
// /**
|
||||||
|
// * Method to set the Jackson2JsonMessageConverter.
|
||||||
|
// *
|
||||||
|
// * @return the Jackson2JsonMessageConverter
|
||||||
|
// */
|
||||||
|
// @Bean
|
||||||
|
// public RabbitAdmin rabbitAdmin(final RabbitAdmin rabbitAdmin) {
|
||||||
|
// rabbitAdmin.setIgnoreDeclarationExceptions(true);
|
||||||
|
// return rabbitAdmin;
|
||||||
|
// }
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Method to set the Jackson2JsonMessageConverter.
|
* Method to set the Jackson2JsonMessageConverter.
|
||||||
*
|
*
|
||||||
@@ -59,7 +70,8 @@ public class AmqpConfiguration {
|
|||||||
*/
|
*/
|
||||||
@Bean
|
@Bean
|
||||||
public Queue receiverQueue() {
|
public Queue receiverQueue() {
|
||||||
return new Queue(amqpProperties.getReceiverQueue(), true, false, false, getDeadLetterExchangeArgs());
|
return new Queue(amqpProperties.getReceiverQueue(), true, false, false,
|
||||||
|
amqpDeadletterProperties.getDeadLetterExchangeArgs(amqpProperties.getDeadLetterExchange()));
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -79,7 +91,7 @@ public class AmqpConfiguration {
|
|||||||
*/
|
*/
|
||||||
@Bean
|
@Bean
|
||||||
public Queue deadLetterQueue() {
|
public Queue deadLetterQueue() {
|
||||||
return new Queue(amqpProperties.getDeadLetterQueue());
|
return amqpDeadletterProperties.createDeadletterQueue(amqpProperties.getDeadLetterQueue());
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -149,10 +161,4 @@ public class AmqpConfiguration {
|
|||||||
return containerFactory;
|
return containerFactory;
|
||||||
}
|
}
|
||||||
|
|
||||||
private Map<String, Object> getDeadLetterExchangeArgs() {
|
|
||||||
final Map<String, Object> args = new HashMap<>();
|
|
||||||
args.put("x-dead-letter-exchange", amqpProperties.getDeadLetterExchange());
|
|
||||||
return args;
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,69 @@
|
|||||||
|
/**
|
||||||
|
* 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
|
||||||
|
*/
|
||||||
|
package org.eclipse.hawkbit.amqp;
|
||||||
|
|
||||||
|
import java.util.HashMap;
|
||||||
|
import java.util.Map;
|
||||||
|
|
||||||
|
import org.springframework.amqp.core.Queue;
|
||||||
|
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Bean which holds the necessary properties for configuring the AMQP deadletter
|
||||||
|
* queue.
|
||||||
|
*/
|
||||||
|
@ConfigurationProperties("hawkbit.dmf.rabbitmq.deadLetter")
|
||||||
|
public class AmqpDeadletterProperties {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Message time to live (ttl) for the deadletter queue. Default ttl is 3
|
||||||
|
* weeks.
|
||||||
|
*/
|
||||||
|
private int ttl = 1_814_400_000;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Return the deadletter arguments.
|
||||||
|
*
|
||||||
|
* @param exchange
|
||||||
|
* the deadletter exchange
|
||||||
|
* @return map which holds the properties
|
||||||
|
*/
|
||||||
|
public Map<String, Object> getDeadLetterExchangeArgs(final String exchange) {
|
||||||
|
final Map<String, Object> args = new HashMap<>();
|
||||||
|
args.put("x-dead-letter-exchange", exchange);
|
||||||
|
return args;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Create a deadletter queue with ttl for messages
|
||||||
|
*
|
||||||
|
* @param queueName
|
||||||
|
* the deadlette queue name
|
||||||
|
* @return the deadletter queue
|
||||||
|
*/
|
||||||
|
public Queue createDeadletterQueue(final String queueName) {
|
||||||
|
// getTTLArgs()
|
||||||
|
return new Queue(queueName, true, false, false, null);
|
||||||
|
}
|
||||||
|
|
||||||
|
private Map<String, Object> getTTLArgs() {
|
||||||
|
final Map<String, Object> args = new HashMap<>();
|
||||||
|
args.put("x-message-ttl", getTtl());
|
||||||
|
return args;
|
||||||
|
}
|
||||||
|
|
||||||
|
public int getTtl() {
|
||||||
|
return ttl;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void setTtl(final int ttl) {
|
||||||
|
this.ttl = ttl;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user