lingwei.kong/hawkbit
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Code refactoring of hawkbit-dmf-amqp (#2054)

Browse Source 
Signed-off-by: Avgustin Marinov <Avgustin.Marinov@bosch.com>
This commit is contained in:
Avgustin Marinov
2024-11-16 20:17:13 +02:00
committed by GitHub
parent ce846ebe81
commit ca2c50ffa5
21 changed files with 235 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
hawkbit-dmf/hawkbit-dmf-amqp/pom.xml
View File

@@ -9,7 +9,6 @@
SPDX-License-Identifier: EPL-2.0
-->
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
@@ -43,6 +42,7 @@
<artifactId>hawkbit-dmf-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
@@ -67,6 +67,7 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-collections4</artifactId>

6
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AbstractAmqpErrorHandler.java
View File

@@ -18,8 +18,7 @@ public abstract class AbstractAmqpErrorHandler<T> implements AmqpErrorHandler {
@Override
public void doHandle(Throwable throwable, AmqpErrorHandlerChain chain) {
// retrieving the cause of throwable as it contains the actual class of
// exception
// retrieving the cause of throwable as it contains the actual class of exception
final Throwable cause = throwable.getCause();
if (getExceptionClass().isAssignableFrom(cause.getClass())) {
throw new AmqpRejectAndDontRequeueException(getErrorMessage(throwable));
@@ -43,5 +42,4 @@ public abstract class AbstractAmqpErrorHandler<T> implements AmqpErrorHandler {
public String getErrorMessage(Throwable throwable) {
return AmqpErrorMessageComposer.constructErrorMessage(throwable);
}
}
}

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

@@ -127,8 +127,7 @@ public class AmqpConfiguration {
}
/**
* @return {@link RabbitTemplate} with automatic retry, published confirms and
* {@link Jackson2JsonMessageConverter}.
* @return {@link RabbitTemplate} with automatic retry, published confirms and {@link Jackson2JsonMessageConverter}.
*/
@Bean
public RabbitTemplate rabbitTemplate() {
@@ -246,7 +245,8 @@ public class AmqpConfiguration {
final SystemSecurityContext systemSecurityContext,
final TenantConfigurationManagement tenantConfigurationManagement,
final ConfirmationManagement confirmationManagement) {
return new AmqpMessageHandlerService(rabbitTemplate, amqpMessageDispatcherService, controllerManagement,
return new AmqpMessageHandlerService(
rabbitTemplate, amqpMessageDispatcherService, controllerManagement,
entityFactory, systemSecurityContext, tenantConfigurationManagement, confirmationManagement);
}
@@ -279,7 +279,8 @@ public class AmqpConfiguration {
@Bean
@ConditionalOnMissingBean(AmqpMessageDispatcherService.class)
AmqpMessageDispatcherService amqpMessageDispatcherService(final RabbitTemplate rabbitTemplate,
AmqpMessageDispatcherService amqpMessageDispatcherService(
final RabbitTemplate rabbitTemplate,
final AmqpMessageSenderService amqpSenderService, final ArtifactUrlHandler artifactUrlHandler,
final SystemSecurityContext systemSecurityContext, final SystemManagement systemManagement,
final TargetManagement targetManagement, final DistributionSetManagement distributionSetManagement,

15
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpDeadletterProperties.java
View File

@@ -28,15 +28,14 @@ public class AmqpDeadletterProperties {
private static final int THREE_WEEKS = 21;
/**
* Message time to live (ttl) for the deadletter queue. Default ttl is 3
* weeks.
* Message time to live (ttl) for the dead letter queue. Default ttl is 3 weeks.
*/
private long ttl = Duration.ofDays(THREE_WEEKS).toMillis();
/**
* Return the deadletter arguments.
* Return the dead letter arguments.
*
* @param exchange the deadletter exchange
* @param exchange the dead letter exchange
* @return map which holds the properties
*/
public Map<String, Object> getDeadLetterExchangeArgs(final String exchange) {
@@ -46,10 +45,10 @@ public class AmqpDeadletterProperties {
}
/**
* Create a deadletter queue with ttl for messages
* Create a dead letter queue with ttl for messages
*
* @param queueName the deadletter queue name
* @return the deadletter queue
* @param queueName the dead letter queue name
* @return the dead letter queue
*/
public Queue createDeadletterQueue(final String queueName) {
return new Queue(queueName, true, false, false, getTTLArgs());
@@ -60,4 +59,4 @@ public class AmqpDeadletterProperties {
args.put("x-message-ttl", getTtl());
return args;
}
}
}

3
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpErrorHandler.java
View File

@@ -10,8 +10,7 @@
package org.eclipse.hawkbit.amqp;
/**
* Interface declaration of {@link AmqpErrorHandler} that handles errors based on the
* types of exception.
* Interface declaration of {@link AmqpErrorHandler} that handles errors based on the types of exception.
*/
@FunctionalInterface
public interface AmqpErrorHandler {

13
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpErrorMessageComposer.java
View File

@@ -31,9 +31,7 @@ public final class AmqpErrorMessageComposer {
* @return meaningful error message
*/
public static String constructErrorMessage(final Throwable throwable) {
StringBuilder completeErrorMessage = new StringBuilder();
final String mainErrorMsg = throwable.getCause().getMessage();
if (throwable instanceof ListenerExecutionFailedException) {
Collection<Message> failedMessages = ((ListenerExecutionFailedException) throwable).getFailedMessages();
// since the intended message content is always on top of the collection, we only extract the first one
@@ -41,12 +39,11 @@ public final class AmqpErrorMessageComposer {
final byte[] amqpFailedMsgBody = failedMessage.getBody();
final Map<String, Object> amqpFailedMsgHeaders = failedMessage.getMessageProperties().getHeaders();
String amqpFailedMsgConcatenatedHeaders = amqpFailedMsgHeaders.keySet().stream()
.map(key -> key + "=" + amqpFailedMsgHeaders.get(key)).collect(Collectors.joining(", ", "{", "}"));
completeErrorMessage.append(mainErrorMsg).append(new String(amqpFailedMsgBody))
.append(amqpFailedMsgConcatenatedHeaders);
return completeErrorMessage.toString();
final String amqpFailedMsgConcatenatedHeaders = amqpFailedMsgHeaders.keySet().stream()
.map(key -> key + "=" + amqpFailedMsgHeaders.get(key))
.collect(Collectors.joining(", ", "{", "}"));
return mainErrorMsg + new String(amqpFailedMsgBody) + amqpFailedMsgConcatenatedHeaders;
}
return mainErrorMsg;
}
}
}

207
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java
View File

@@ -84,8 +84,7 @@ import org.springframework.util.CollectionUtils;
* {@link AmqpMessageDispatcherService} create all outgoing AMQP messages and
* delegate the messages to a {@link AmqpMessageSenderService}.
*
* Additionally the dispatcher listener/subscribe for some target events e.g.
* assignment.
* Additionally, the dispatcher listener/subscribe for some target events e.g. assignment.
*/
@Slf4j
public class AmqpMessageDispatcherService extends BaseAmqpService {
@@ -112,12 +111,12 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
* @param systemSecurityContext for execution with system permissions
* @param systemManagement the systemManagement
* @param targetManagement to access target information
* @param serviceMatcher to check in cluster case if the message is from the same
* cluster node
* @param serviceMatcher to check in cluster case if the message is from the same cluster node
* @param distributionSetManagement to retrieve modules
* @param tenantConfigurationManagement to access tenant configuration
*/
protected AmqpMessageDispatcherService(final RabbitTemplate rabbitTemplate,
protected AmqpMessageDispatcherService(
final RabbitTemplate rabbitTemplate,
final AmqpMessageSenderService amqpSenderService, final ArtifactUrlHandler artifactUrlHandler,
final SystemSecurityContext systemSecurityContext, final SystemManagement systemManagement,
final TargetManagement targetManagement, final ServiceMatcher serviceMatcher,
@@ -138,15 +137,15 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
}
public boolean isBatchAssignmentsEnabled() {
return systemSecurityContext.runAsSystem(() -> tenantConfigurationManagement
.getConfigurationValue(BATCH_ASSIGNMENTS_ENABLED, Boolean.class).getValue());
return systemSecurityContext.runAsSystem(() ->
tenantConfigurationManagement
.getConfigurationValue(BATCH_ASSIGNMENTS_ENABLED, Boolean.class).getValue());
}
/**
* Method to send a message to a RabbitMQ Exchange after the Distribution
* set has been assign to a Target.
* Method to send a message to a RabbitMQ Exchange after the Distribution set has been assign to a Target.
*
* @param assignedEvent the object to be send.
* @param assignedEvent the object to be sent.
*/
@EventListener(classes = TargetAssignDistributionSetEvent.class)
protected void targetAssignDistributionSet(final TargetAssignDistributionSetEvent assignedEvent) {
@@ -154,13 +153,11 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return;
}
final List<Target> filteredTargetList = getTargetsWithoutPendingCancellations(
assignedEvent.getActions().keySet());
final List<Target> filteredTargetList = getTargetsWithoutPendingCancellations(assignedEvent.getActions().keySet());
if (!filteredTargetList.isEmpty()) {
log.debug("targetAssignDistributionSet retrieved. I will forward it to DMF broker.");
sendUpdateMessageToTargets(assignedEvent.getDistributionSetId(), assignedEvent.getActions(),
filteredTargetList);
sendUpdateMessageToTargets(assignedEvent.getDistributionSetId(), assignedEvent.getActions(), filteredTargetList);
}
}
@@ -178,16 +175,17 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
sendMultiActionRequestMessages(multiActionEvent.getTenant(), multiActionEvent.getControllerIds());
}
protected void sendUpdateMessageToTarget(final ActionProperties actionsProps, final Target target,
protected void sendUpdateMessageToTarget(
final ActionProperties actionsProps, final Target target,
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
final Map<String, ActionProperties> actionProp = new HashMap<>();
actionProp.put(target.getControllerId(), actionsProps);
sendUpdateMessageToTargets(actionProp, Collections.singletonList(target), softwareModules);
}
protected void sendMultiActionRequestToTarget(final String tenant, final Target target, final List<Action> actions,
protected void sendMultiActionRequestToTarget(
final String tenant, final Target target, final List<Action> actions,
final Function<Action, Map<SoftwareModule, List<SoftwareModuleMetadata>>> getSoftwareModuleMetaData) {
final URI targetAddress = target.getAddress();
if (!IpUtil.isAmqpUri(targetAddress) || CollectionUtils.isEmpty(actions)) {
return;
@@ -195,26 +193,26 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
final DmfMultiActionRequest multiActionRequest = new DmfMultiActionRequest();
actions.forEach(action -> {
final DmfActionRequest actionRequest = createDmfActionRequest(target, action,
getSoftwareModuleMetaData.apply(action));
final DmfActionRequest actionRequest = createDmfActionRequest(target, action, getSoftwareModuleMetaData.apply(action));
final int weight = deploymentManagement.getWeightConsideringDefault(action);
multiActionRequest.addElement(getEventTypeForAction(action), actionRequest, weight);
});
final Message message = getMessageConverter().toMessage(multiActionRequest,
final Message message = getMessageConverter().toMessage(
multiActionRequest,
createConnectorMessagePropertiesEvent(tenant, target.getControllerId(), EventTopic.MULTI_ACTION));
amqpSenderService.sendMessage(message, targetAddress);
}
protected DmfDownloadAndUpdateRequest createDownloadAndUpdateRequest(final Target target, final Long actionId,
protected DmfDownloadAndUpdateRequest createDownloadAndUpdateRequest(
final Target target, final Long actionId,
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
final DmfDownloadAndUpdateRequest request = new DmfDownloadAndUpdateRequest();
request.setActionId(actionId);
request.setTargetSecurityToken(systemSecurityContext.runAsSystem(target::getSecurityToken));
if (softwareModules != null) {
softwareModules.entrySet()
.forEach(entry -> request.addSoftwareModule(convertToAmqpSoftwareModule(target, entry)));
softwareModules.entrySet().forEach(entry -> request.addSoftwareModule(convertToAmqpSoftwareModule(target, entry)));
}
return request;
}
@@ -231,24 +229,22 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return;
}
final List<Target> eventTargets = partitionedParallelExecution(cancelEvent.getActions().keySet(),
targetManagement::getByControllerID);
final List<Target> eventTargets = partitionedParallelExecution(
cancelEvent.getActions().keySet(), targetManagement::getByControllerID);
eventTargets.forEach(target -> {
cancelEvent.getActionPropertiesForController(target.getControllerId()).map(ActionProperties::getId)
.ifPresent(actionId -> {
sendCancelMessageToTarget(cancelEvent.getTenant(), target.getControllerId(), actionId,
target.getAddress());
});
});
eventTargets.forEach(target ->
cancelEvent.getActionPropertiesForController(target.getControllerId())
.map(ActionProperties::getId)
.ifPresent(actionId ->
sendCancelMessageToTarget(cancelEvent.getTenant(), target.getControllerId(), actionId, target.getAddress())
)
);
}
/**
* Method to send a message to a RabbitMQ Exchange after a Target was
* deleted.
* Method to send a message to a RabbitMQ Exchange after a Target was deleted.
*
* @param deleteEvent the TargetDeletedEvent which holds the necessary data for
* sending a target delete message.
* @param deleteEvent the TargetDeletedEvent which holds the necessary data for sending a target delete message.
*/
@EventListener(classes = TargetDeletedEvent.class)
protected void targetDelete(final TargetDeletedEvent deleteEvent) {
@@ -260,15 +256,18 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
@EventListener(classes = TargetAttributesRequestedEvent.class)
protected void targetTriggerUpdateAttributes(final TargetAttributesRequestedEvent updateAttributesEvent) {
sendUpdateAttributesMessageToTarget(updateAttributesEvent.getTenant(), updateAttributesEvent.getControllerId(),
sendUpdateAttributesMessageToTarget(
updateAttributesEvent.getTenant(), updateAttributesEvent.getControllerId(),
updateAttributesEvent.getTargetAddress());
}
protected void sendPingReponseToDmfReceiver(final Message ping, final String tenant, final String virtualHost) {
final Message message = MessageBuilder.withBody(String.valueOf(System.currentTimeMillis()).getBytes())
protected void sendPingResponseToDmfReceiver(final Message ping, final String tenant, final String virtualHost) {
final Message message = MessageBuilder
.withBody(String.valueOf(System.currentTimeMillis()).getBytes())
.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
.setCorrelationId(ping.getMessageProperties().getCorrelationId())
.setHeader(MessageHeaderKey.TYPE, MessageType.PING_RESPONSE).setHeader(MessageHeaderKey.TENANT, tenant)
.setHeader(MessageHeaderKey.TYPE, MessageType.PING_RESPONSE)
.setHeader(MessageHeaderKey.TENANT, tenant)
.build();
amqpSenderService.sendMessage(message,
@@ -279,8 +278,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return isFromSelf(event);
}
protected void sendCancelMessageToTarget(final String tenant, final String controllerId, final Long actionId,
final URI address) {
protected void sendCancelMessageToTarget(final String tenant, final String controllerId, final Long actionId, final URI address) {
if (!IpUtil.isAmqpUri(address)) {
return;
}
@@ -288,11 +286,11 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
final DmfActionRequest actionRequest = new DmfActionRequest();
actionRequest.setActionId(actionId);
final Message message = getMessageConverter().toMessage(actionRequest,
final Message message = getMessageConverter().toMessage(
actionRequest,
createConnectorMessagePropertiesEvent(tenant, controllerId, EventTopic.CANCEL_DOWNLOAD));
amqpSenderService.sendMessage(message, address);
}
protected DmfTarget convertToDmfTarget(final Target target, final Long actionId) {
@@ -309,18 +307,18 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
* @param target the target
* @param actionId the actionId
* @param softwareModules the software modules
* @return
* @return confirm request
*/
protected DmfConfirmRequest createConfirmRequest(final Target target, final Long actionId, final Map<SoftwareModule,
List<SoftwareModuleMetadata>> softwareModules) {
protected DmfConfirmRequest createConfirmRequest(
final Target target, final Long actionId, final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
final DmfConfirmRequest request = new DmfConfirmRequest();
request.setActionId(actionId);
request.setTargetSecurityToken(systemSecurityContext.runAsSystem(target::getSecurityToken));
//Software modules can be filtered in the future exposing only the needed.
if (softwareModules != null) {
softwareModules.entrySet()
.forEach(entry -> request.addSoftwareModule(convertToAmqpSoftwareModule(target, entry)));
softwareModules.entrySet().forEach(entry ->
request.addSoftwareModule(convertToAmqpSoftwareModule(target, entry)));
}
return request;
}
@@ -332,12 +330,9 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
}
/**
* Method to get the type of event depending on whether the action is a
* DOWNLOAD_ONLY action or if it has a valid maintenance window available or
* not based on defined maintenance schedule. In case of no maintenance
* schedule or if there is a valid window available, the topic
* {@link EventTopic#DOWNLOAD_AND_INSTALL} is returned else
* {@link EventTopic#DOWNLOAD} is returned.
* Method to get the type of event depending on whether the action is a DOWNLOAD_ONLY action or if it has a valid maintenance window
* available or not based on defined maintenance schedule. In case of no maintenance schedule or if there is a valid window available,
* the topic {@link EventTopic#DOWNLOAD_AND_INSTALL} is returned else {@link EventTopic#DOWNLOAD} is returned.
*
* @param action current action properties.
* @return {@link EventTopic} to use for message.
@@ -352,8 +347,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
}
/**
* Determines the {@link EventTopic} for the given {@link Action}, depending
* on its action type.
* Determines the {@link EventTopic} for the given {@link Action}, depending on its action type.
*
* @param action to obtain the corresponding {@link EventTopic} for
* @return the {@link EventTopic} for this action
@@ -365,8 +359,8 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return getEventTypeForTarget(new ActionProperties(action));
}
private static <T, R> List<R> partitionedParallelExecution(final Collection<T> controllerIds,
final Function<Collection<T>, List<R>> loadingFunction) {
private static <T, R> List<R> partitionedParallelExecution(
final Collection<T> controllerIds, final Function<Collection<T>, List<R>> loadingFunction) {
// Ensure not exceeding the max value of MAX_PROCESSING_SIZE
if (controllerIds.size() > MAX_PROCESSING_SIZE) {
// Split the provided collection
@@ -392,16 +386,16 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
}
}
private static MessageProperties createConnectorMessagePropertiesEvent(final String tenant,
final String controllerId, final EventTopic topic) {
private static MessageProperties createConnectorMessagePropertiesEvent(
final String tenant, final String controllerId, final EventTopic topic) {
final MessageProperties messageProperties = createConnectorMessageProperties(tenant, controllerId);
messageProperties.setHeader(MessageHeaderKey.TOPIC, topic);
messageProperties.setHeader(MessageHeaderKey.TYPE, MessageType.EVENT);
return messageProperties;
}
private static MessageProperties createConnectorMessagePropertiesDeleteThing(final String tenant,
final String controllerId) {
private static MessageProperties createConnectorMessagePropertiesDeleteThing(
final String tenant, final String controllerId) {
final MessageProperties messageProperties = createConnectorMessageProperties(tenant, controllerId);
messageProperties.setHeader(MessageHeaderKey.TYPE, MessageType.THING_DELETED);
return messageProperties;
@@ -434,30 +428,28 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
}
private List<Target> getTargetsWithoutPendingCancellations(final Set<String> controllerIds) {
return partitionedParallelExecution(controllerIds, partition -> {
return targetManagement.getByControllerID(partition).stream().filter(target -> {
if (hasPendingCancellations(target.getId())) {
log.debug("Target {} has pending cancellations. Will not send update message to it.",
target.getControllerId());
return false;
}
return true;
}).collect(Collectors.toList());
});
return partitionedParallelExecution(controllerIds, partition ->
targetManagement.getByControllerID(partition).stream()
.filter(target -> {
if (hasPendingCancellations(target.getId())) {
log.debug("Target {} has pending cancellations. Will not send update message to it.", target.getControllerId());
return false;
}
return true;
}).collect(Collectors.toList()));
}
private void sendUpdateMessageToTargets(final Long dsId, final Map<String, ActionProperties> actionsPropsByTargetId,
final List<Target> targets) {
private void sendUpdateMessageToTargets(
final Long dsId, final Map<String, ActionProperties> actionsPropsByTargetId, final List<Target> targets) {
distributionSetManagement.get(dsId).ifPresent(ds -> {
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules = getSoftwareModulesWithMetadata(
ds);
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules = getSoftwareModulesWithMetadata(ds);
sendUpdateMessageToTargets(actionsPropsByTargetId, targets, softwareModules);
});
}
private void sendUpdateMessageToTargets(final Map<String, ActionProperties> actionsPropsByTargetId,
private void sendUpdateMessageToTargets(
final Map<String, ActionProperties> actionsPropsByTargetId,
final List<Target> targets, final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
if (!targets.isEmpty() && isBatchAssignmentsEnabled()) {
sendBatchUpdateMessage(actionsPropsByTargetId, targets, softwareModules);
} else {
@@ -469,16 +461,15 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
}
private void sendMultiActionRequestMessages(final String tenant, final List<String> controllerIds) {
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModuleMetadata = new HashMap<>();
targetManagement.getByControllerID(controllerIds).stream()
.filter(target -> IpUtil.isAmqpUri(target.getAddress())).forEach(target -> {
final List<Action> activeActions = deploymentManagement
.findActiveActionsWithHighestWeight(target.getControllerId(), MAX_ACTION_COUNT);
activeActions.forEach(action -> action.getDistributionSet().getModules().forEach(
module -> softwareModuleMetadata.computeIfAbsent(module, this::getSoftwareModuleMetadata)));
activeActions.forEach(action ->
action.getDistributionSet().getModules().forEach(module ->
softwareModuleMetadata.computeIfAbsent(module, this::getSoftwareModuleMetadata)));
if (!activeActions.isEmpty()) {
sendMultiActionRequestToTarget(tenant, target, activeActions,
@@ -486,10 +477,10 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
.collect(Collectors.toMap(m -> m, softwareModuleMetadata::get)));
}
});
}
private DmfActionRequest createDmfActionRequest(final Target target, final Action action,
private DmfActionRequest createDmfActionRequest(
final Target target, final Action action,
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
if (action.isCancelingOrCanceled()) {
return createPlainActionRequest(action);
@@ -499,9 +490,9 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return createDownloadAndUpdateRequest(target, action.getId(), softwareModules);
}
private void sendSingleUpdateMessage(final ActionProperties action, final Target target,
private void sendSingleUpdateMessage(
final ActionProperties action, final Target target,
final Map<SoftwareModule, List<SoftwareModuleMetadata>> modules) {
final String tenant = action.getTenant();
final URI targetAddress = target.getAddress();
@@ -518,7 +509,8 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
request = createDownloadAndUpdateRequest(target, action.getId(), modules);
}
final Message message = getMessageConverter().toMessage(request,
final Message message = getMessageConverter().toMessage(
request,
createConnectorMessagePropertiesEvent(tenant, target.getControllerId(), getEventTypeForTarget(action)));
amqpSenderService.sendMessage(message, targetAddress);
}
@@ -528,8 +520,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return;
}
final Message message = new Message("".getBytes(),
createConnectorMessagePropertiesDeleteThing(tenant, controllerId));
final Message message = new Message("".getBytes(), createConnectorMessagePropertiesDeleteThing(tenant, controllerId));
amqpSenderService.sendMessage(message, URI.create(targetAddress));
}
@@ -545,19 +536,20 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return deploymentManagement.hasPendingCancellations(targetId);
}
private void sendUpdateAttributesMessageToTarget(final String tenant, final String controllerId,
final String targetAddress) {
private void sendUpdateAttributesMessageToTarget(final String tenant, final String controllerId, final String targetAddress) {
if (!hasValidAddress(targetAddress)) {
return;
}
final Message message = new Message("".getBytes(),
final Message message = new Message(
"".getBytes(),
createConnectorMessagePropertiesEvent(tenant, controllerId, EventTopic.REQUEST_ATTRIBUTES_UPDATE));
amqpSenderService.sendMessage(message, URI.create(targetAddress));
}
private DmfSoftwareModule convertToAmqpSoftwareModule(final Target target,
private DmfSoftwareModule convertToAmqpSoftwareModule(
final Target target,
final Entry<SoftwareModule, List<SoftwareModuleMetadata>> entry) {
final DmfSoftwareModule amqpSoftwareModule = new DmfSoftwareModule();
amqpSoftwareModule.setModuleId(entry.getKey().getId());
@@ -596,7 +588,8 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
new SoftwareData(localArtifact.getSoftwareModule().getId(), localArtifact.getFilename(),
localArtifact.getId(), localArtifact.getSha1Hash())),
ApiType.DMF)
.stream().collect(Collectors.toMap(ArtifactUrl::getProtocol, ArtifactUrl::getRef)));
.stream()
.collect(Collectors.toMap(ArtifactUrl::getProtocol, ArtifactUrl::getRef)));
artifact.setFilename(localArtifact.getFilename());
artifact.setHashes(new DmfArtifactHash(localArtifact.getSha1Hash(), localArtifact.getMd5Hash()));
@@ -605,8 +598,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return artifact;
}
private Map<SoftwareModule, List<SoftwareModuleMetadata>> getSoftwareModulesWithMetadata(
final DistributionSet distributionSet) {
private Map<SoftwareModule, List<SoftwareModuleMetadata>> getSoftwareModulesWithMetadata(final DistributionSet distributionSet) {
return distributionSet.getModules().stream().collect(Collectors.toMap(m -> m, this::getSoftwareModuleMetadata));
}
@@ -615,11 +607,14 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
PageRequest.of(0, RepositoryConstants.MAX_META_DATA_COUNT), module.getId()).getContent();
}
private void sendBatchUpdateMessage(final Map<String, ActionProperties> actions, final List<Target> targets,
private void sendBatchUpdateMessage(
final Map<String, ActionProperties> actions, final List<Target> targets,
final Map<SoftwareModule, List<SoftwareModuleMetadata>> modules) {
final List<DmfTarget> dmfTargets = targets.stream().filter(target -> IpUtil.isAmqpUri(target.getAddress()))
.map(t -> convertToDmfTarget(t, actions.get(t.getControllerId()).getId())).collect(Collectors.toList());
final List<DmfTarget> dmfTargets = targets.stream()
.filter(target -> IpUtil.isAmqpUri(target.getAddress()))
.map(t -> convertToDmfTarget(t, actions.get(t.getControllerId()).getId()))
.collect(Collectors.toList());
final DmfBatchDownloadAndUpdateRequest batchRequest = new DmfBatchDownloadAndUpdateRequest();
batchRequest.setTimestamp(System.currentTimeMillis());
@@ -630,16 +625,16 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
// target-specific urls
final Target firstTarget = targets.get(0);
if (modules != null) {
modules.entrySet()
.forEach(entry -> batchRequest.addSoftwareModule(convertToAmqpSoftwareModule(firstTarget, entry)));
modules.entrySet().forEach(entry ->
batchRequest.addSoftwareModule(convertToAmqpSoftwareModule(firstTarget, entry)));
}
// we use only the first action when constructing message as Tenant and
// action type are the same
// we use only the first action when constructing message as Tenant and action type are the same
// since all actions have the same trigger
final ActionProperties firstAction = actions.values().iterator().next();
final Message message = getMessageConverter().toMessage(batchRequest,
final Message message = getMessageConverter().toMessage(
batchRequest,
createMessagePropertiesBatch(firstAction.getTenant(), getBatchEventTopicForAction(firstAction)));
amqpSenderService.sendMessage(message, firstTarget.getAddress());
}
}
}

140
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerService.java
View File

@@ -33,7 +33,6 @@ import org.eclipse.hawkbit.dmf.json.model.DmfAutoConfirmation;
import org.eclipse.hawkbit.dmf.json.model.DmfCreateThing;
import org.eclipse.hawkbit.dmf.json.model.DmfUpdateMode;
import org.eclipse.hawkbit.im.authentication.SpPermission.SpringEvalExpressions;
import org.eclipse.hawkbit.tenancy.TenantAwareAuthenticationDetails;
import org.eclipse.hawkbit.repository.ConfirmationManagement;
import org.eclipse.hawkbit.repository.ControllerManagement;
import org.eclipse.hawkbit.repository.EntityFactory;
@@ -51,6 +50,7 @@ import org.eclipse.hawkbit.repository.model.SoftwareModule;
import org.eclipse.hawkbit.repository.model.SoftwareModuleMetadata;
import org.eclipse.hawkbit.repository.model.Target;
import org.eclipse.hawkbit.security.SystemSecurityContext;
import org.eclipse.hawkbit.tenancy.TenantAwareAuthenticationDetails;
import org.eclipse.hawkbit.util.IpUtil;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
@@ -63,6 +63,7 @@ import org.springframework.security.core.authority.SimpleGrantedAuthority;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.core.context.SecurityContextImpl;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/**
@@ -75,6 +76,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
private static final String THING_ID_NULL = "ThingId is null";
private static final String EMPTY_MESSAGE_BODY = "\"\"";
private final AmqpMessageDispatcherService amqpMessageDispatcherService;
private final ConfirmationManagement confirmationManagement;
private final EntityFactory entityFactory;
@@ -117,14 +119,15 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
* @return a message if <null> no message is send back to sender
*/
@RabbitListener(queues = "${hawkbit.dmf.rabbitmq.receiverQueue:dmf_receiver}", containerFactory = "listenerContainerFactory")
public Message onMessage(final Message message,
public Message onMessage(
final Message message,
@Header(name = MessageHeaderKey.TYPE, required = false) final String type,
@Header(name = MessageHeaderKey.TENANT, required = false) final String tenant) {
return onMessage(message, type, tenant, getRabbitTemplate().getConnectionFactory().getVirtualHost());
}
/**
* * Executed if a amqp message arrives.
* Executed if a amqp message arrives.
*
* @param message the message
* @param type the type
@@ -133,7 +136,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
* @return the rpc message back to supplier.
*/
public Message onMessage(final Message message, final String type, final String tenant, final String virtualHost) {
if (StringUtils.isEmpty(type) || StringUtils.isEmpty(tenant)) {
if (ObjectUtils.isEmpty(type) || ObjectUtils.isEmpty(tenant)) {
throw new AmqpRejectAndDontRequeueException("Invalid message! tenant and type header are mandatory!");
}
@@ -141,26 +144,31 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
try {
final MessageType messageType = MessageType.valueOf(type);
switch (messageType) {
case THING_CREATED:
case THING_CREATED: {
setTenantSecurityContext(tenant);
registerTarget(message, virtualHost);
break;
case THING_REMOVED:
}
case THING_REMOVED: {
setTenantSecurityContext(tenant);
deleteTarget(message);
break;
case EVENT:
}
case EVENT: {
checkContentTypeJson(message);
setTenantSecurityContext(tenant);
handleIncomingEvent(message);
break;
case PING:
}
case PING: {
if (isCorrelationIdNotEmpty(message)) {
amqpMessageDispatcherService.sendPingReponseToDmfReceiver(message, tenant, virtualHost);
amqpMessageDispatcherService.sendPingResponseToDmfReceiver(message, tenant, virtualHost);
}
break;
default:
}
default: {
logAndThrowMessageError(message, "No handle method was found for the given message type.");
}
}
} catch (AssignmentQuotaExceededException ex) {
throw new AmqpRejectAndDontRequeueException("Could not handle message due to quota violation!", ex);
@@ -212,39 +220,50 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
final Action action) {
Status status = null;
switch (actionUpdateStatus.getActionStatus()) {
case DOWNLOAD:
case DOWNLOAD: {
status = Status.DOWNLOAD;
break;
case RETRIEVED:
}
case RETRIEVED: {
status = Status.RETRIEVED;
break;
}
case RUNNING:
case CONFIRMED:
case CONFIRMED: {
status = Status.RUNNING;
break;
case CANCELED:
}
case CANCELED: {
status = Status.CANCELED;
break;
case FINISHED:
}
case FINISHED: {
status = Status.FINISHED;
break;
case ERROR:
}
case ERROR: {
status = Status.ERROR;
break;
case WARNING:
}
case WARNING: {
status = Status.WARNING;
break;
case DOWNLOADED:
}
case DOWNLOADED: {
status = Status.DOWNLOADED;
break;
case CANCEL_REJECTED:
}
case CANCEL_REJECTED: {
status = handleCancelRejectedState(message, action);
break;
case DENIED:
}
case DENIED: {
status = Status.WAIT_FOR_CONFIRMATION;
break;
default:
}
default: {
logAndThrowMessageError(message, "Status for action does not exisit.");
}
}
return status;
@@ -254,7 +273,8 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
if (action.isCancelingOrCanceled()) {
return Status.CANCEL_REJECTED;
}
logAndThrowMessageError(message,
logAndThrowMessageError(
message,
"Cancel rejected message is not allowed, if action is on state: " + action.getStatus());
return null;
}
@@ -274,15 +294,14 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
* Method to create a new target or to find the target if it already exists
* and update its poll time, status and optionally its name and attributes.
*
* @param message the message that contains replyTo property and optionally the
* name and attributes in body
* @param message the message that contains replyTo property and optionally the name and attributes in body
* @param virtualHost the virtual host
*/
private void registerTarget(final Message message, final String virtualHost) {
final String thingId = getStringHeaderKey(message, MessageHeaderKey.THING_ID, THING_ID_NULL);
final String replyTo = message.getMessageProperties().getReplyTo();
if (StringUtils.isEmpty(replyTo)) {
if (ObjectUtils.isEmpty(replyTo)) {
logAndThrowMessageError(message, "No ReplyTo was set for the createThing message.");
}
@@ -297,22 +316,22 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
final DmfCreateThing thingCreateBody = convertMessage(message, DmfCreateThing.class);
final DmfAttributeUpdate thingAttributeUpdateBody = thingCreateBody.getAttributeUpdate();
log.debug("Received \"THING_CREATED\" AMQP message for thing \"{}\" with target name \"{}\" and type " +
"\"{}\".", thingId, thingCreateBody.getName(), thingCreateBody.getType());
log.debug(
"Received \"THING_CREATED\" AMQP message for thing \"{}\" with target name \"{}\" and type \"{}\".",
thingId, thingCreateBody.getName(), thingCreateBody.getType());
target = controllerManagement.findOrRegisterTargetIfItDoesNotExist(thingId, amqpUri,
thingCreateBody.getName(), thingCreateBody.getType());
target = controllerManagement.findOrRegisterTargetIfItDoesNotExist(
thingId, amqpUri, thingCreateBody.getName(), thingCreateBody.getType());
if (thingAttributeUpdateBody != null) {
controllerManagement.updateControllerAttributes(thingId, thingAttributeUpdateBody.getAttributes(),
getUpdateMode(thingAttributeUpdateBody));
controllerManagement.updateControllerAttributes(
thingId, thingAttributeUpdateBody.getAttributes(), getUpdateMode(thingAttributeUpdateBody));
}
}
log.debug("Target {} reported online state.", thingId);
sendUpdateCommandToTarget(target);
} catch (final EntityAlreadyExistsException e) {
throw new AmqpRejectAndDontRequeueException(
"Tried to register previously registered target, message will be ignored!", e);
throw new AmqpRejectAndDontRequeueException("Tried to register previously registered target, message will be ignored!", e);
}
}
@@ -325,23 +344,21 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
}
private void sendCurrentActionsAsMultiActionToTarget(final Target target) {
final List<Action> actions = controllerManagement.findActiveActionsWithHighestWeight(target.getControllerId(),
MAX_ACTION_COUNT);
final List<Action> actions = controllerManagement.findActiveActionsWithHighestWeight(target.getControllerId(), MAX_ACTION_COUNT);
final Set<DistributionSet> distributionSets = actions.stream().map(Action::getDistributionSet)
.collect(Collectors.toSet());
final Set<DistributionSet> distributionSets = actions.stream().map(Action::getDistributionSet).collect(Collectors.toSet());
final Map<Long, Map<SoftwareModule, List<SoftwareModuleMetadata>>> softwareModulesPerDistributionSet = distributionSets
.stream().collect(Collectors.toMap(DistributionSet::getId, this::getSoftwareModulesWithMetadata));
amqpMessageDispatcherService.sendMultiActionRequestToTarget(target.getTenant(), target, actions,
amqpMessageDispatcherService.sendMultiActionRequestToTarget(
target.getTenant(), target, actions,
action -> softwareModulesPerDistributionSet.get(action.getDistributionSet().getId()));
}
private void sendOldestActionToTarget(final Target target) {
final Optional<Action> actionOptional = controllerManagement
.findActiveActionWithHighestWeight(target.getControllerId());
final Optional<Action> actionOptional = controllerManagement.findActiveActionWithHighestWeight(target.getControllerId());
if (!actionOptional.isPresent()) {
if (actionOptional.isEmpty()) {
return;
}
@@ -355,8 +372,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
}
}
private Map<SoftwareModule, List<SoftwareModuleMetadata>> getSoftwareModulesWithMetadata(
final DistributionSet distributionSet) {
private Map<SoftwareModule, List<SoftwareModuleMetadata>> getSoftwareModulesWithMetadata(final DistributionSet distributionSet) {
final List<Long> smIds = distributionSet.getModules().stream().map(SoftwareModule::getId)
.collect(Collectors.toList());
@@ -375,20 +391,23 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
*/
private void handleIncomingEvent(final Message message) {
switch (EventTopic.valueOf(getStringHeaderKey(message, MessageHeaderKey.TOPIC, "EventTopic is null"))) {
case UPDATE_ACTION_STATUS:
case UPDATE_ACTION_STATUS: {
updateActionStatus(message);
break;
case UPDATE_ATTRIBUTES:
}
case UPDATE_ATTRIBUTES: {
updateAttributes(message);
break;
case UPDATE_AUTO_CONFIRM:
}
case UPDATE_AUTO_CONFIRM: {
setAutoConfirmationState(message);
break;
default:
}
default: {
logAndThrowMessageError(message, "Got event without appropriate topic.");
break;
}
}
}
private void deleteTarget(final Message message) {
@@ -400,8 +419,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
final DmfAttributeUpdate attributeUpdate = convertMessage(message, DmfAttributeUpdate.class);
final String thingId = getStringHeaderKey(message, MessageHeaderKey.THING_ID, THING_ID_NULL);
controllerManagement.updateControllerAttributes(thingId, attributeUpdate.getAttributes(),
getUpdateMode(attributeUpdate));
controllerManagement.updateControllerAttributes(thingId, attributeUpdate.getAttributes(), getUpdateMode(attributeUpdate));
}
private void setAutoConfirmationState(final Message message) {
@@ -439,22 +457,19 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
final Status status = mapStatus(message, actionUpdateStatus, action);
final Action updatedAction;
if (actionUpdateStatus.getActionStatus() == DmfActionStatus.CONFIRMED) {
updatedAction = confirmationManagement.confirmAction(action.getId(),
actionUpdateStatus.getCode().orElse(null), messages);
} else if (actionUpdateStatus.getActionStatus() == DmfActionStatus.DENIED) {
updatedAction = confirmationManagement.denyAction(action.getId(), actionUpdateStatus.getCode().orElse(null),
messages);
updatedAction = confirmationManagement.denyAction(action.getId(), actionUpdateStatus.getCode().orElse(null), messages);
} else {
final ActionStatusCreate actionStatus = entityFactory.actionStatus().create(action.getId()).status(status)
.messages(messages);
final ActionStatusCreate actionStatus = entityFactory.actionStatus().create(action.getId()).status(status).messages(messages);
actionUpdateStatus.getCode().ifPresent(code -> {
actionStatus.code(code);
actionStatus.message("Device reported status code: " + code);
});
updatedAction = ((Status.CANCELED == status) || (Status.CANCEL_REJECTED == status)) ?
controllerManagement.addCancelActionStatus(actionStatus)
updatedAction = ((Status.CANCELED == status) || (Status.CANCEL_REJECTED == status))
? controllerManagement.addCancelActionStatus(actionStatus)
: controllerManagement.addUpdateActionStatus(actionStatus);
}
@@ -463,8 +478,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
}
}
// Exception squid:S3655 - logAndThrowMessageError throws exception, i.e.
// get will not be called
// Exception squid:S3655 - logAndThrowMessageError throws exception, i.e. get will not be called
@SuppressWarnings("squid:S3655")
private Action checkActionExist(final Message message, final DmfActionUpdateStatus actionUpdateStatus) {
final Long actionId = actionUpdateStatus.getActionId();
@@ -473,9 +487,8 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
actionUpdateStatus.getActionStatus());
final Optional<Action> findActionWithDetails = controllerManagement.findActionWithDetails(actionId);
if (!findActionWithDetails.isPresent()) {
logAndThrowMessageError(message,
"Got intermediate notification about action " + actionId + " but action does not exist");
if (findActionWithDetails.isEmpty()) {
logAndThrowMessageError(message, "Got intermediate notification about action " + actionId + " but action does not exist");
}
return findActionWithDetails.get();
@@ -486,7 +499,6 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
}
private <T extends Serializable> T getConfigValue(final String key, final Class<T> valueType) {
return systemSecurityContext
.runAsSystem(() -> tenantConfigurationManagement.getConfigurationValue(key, valueType).getValue());
return systemSecurityContext.runAsSystem(() -> tenantConfigurationManagement.getConfigurationValue(key, valueType).getValue());
}
}

2
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageSenderService.java
View File

@@ -23,7 +23,7 @@ public interface AmqpMessageSenderService {
/**
* Send the given message to the given uri. The uri contains the (virtual)
* host and exchange e.g amqp://host/exchange.
* host and exchange e.g. amqp://host/exchange.
*
* @param message the amqp message
* @param replyTo the reply to uri

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

@@ -21,7 +21,6 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
public class AmqpProperties {
private static final int DEFAULT_QUEUE_DECLARATION_RETRIES = 50;
private static final long DEFAULT_REQUEUE_DELAY = 0;
/**
@@ -67,4 +66,4 @@ public class AmqpProperties {
* Delay for messages that are requeued in milliseconds.
*/
private long requeueDelay = DEFAULT_REQUEUE_DELAY;
}
}

6
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/BaseAmqpService.java
View File

@@ -50,8 +50,7 @@ public class BaseAmqpService {
@SuppressWarnings("unchecked")
public <T> T convertMessage(@NotNull final Message message, final Class<T> clazz) {
checkMessageBody(message);
message.getMessageProperties().getHeaders().put(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME,
clazz.getName());
message.getMessageProperties().getHeaders().put(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME, clazz.getName());
return (T) rabbitTemplate.getMessageConverter().fromMessage(message);
}
@@ -104,5 +103,4 @@ public class BaseAmqpService {
protected void cleanMessageHeaderProperties(final Message message) {
message.getMessageProperties().getHeaders().remove(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME);
}
}
}

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

@@ -31,20 +31,19 @@ public class ConfigurableRabbitListenerContainerFactory extends SimpleRabbitList
* @param errorHandler the error handler which should be use
* @see SimpleMessageListenerContainer#setMissingQueuesFatal
*/
public ConfigurableRabbitListenerContainerFactory(final boolean missingQueuesFatal, final int declarationRetries,
final ErrorHandler errorHandler) {
public ConfigurableRabbitListenerContainerFactory(
final boolean missingQueuesFatal, final int declarationRetries, final ErrorHandler errorHandler) {
this.declarationRetries = declarationRetries;
setErrorHandler(errorHandler);
setMissingQueuesFatal(missingQueuesFatal);
}
@Override
// Exception squid:UnusedProtectedMethod - called by
// AbstractRabbitListenerContainerFactory
@SuppressWarnings("squid:UnusedProtectedMethod")
protected void initializeContainer(final SimpleMessageListenerContainer instance,
final RabbitListenerEndpoint endpoint) {
@Override
protected void initializeContainer(final SimpleMessageListenerContainer instance, final RabbitListenerEndpoint endpoint) {
super.initializeContainer(instance, endpoint);
instance.setDeclarationRetries(declarationRetries);
}

3
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DefaultAmqpMessageSenderService.java
View File

@@ -61,5 +61,4 @@ public class DefaultAmqpMessageSenderService extends BaseAmqpService implements
protected static boolean isCorrelationIdEmpty(final Message message) {
return !StringUtils.hasLength(message.getMessageProperties().getCorrelationId());
}
}
}

5
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DelayedRequeueExceptionStrategy.java
View File

@@ -86,7 +86,8 @@ public class DelayedRequeueExceptionStrategy extends ConditionalRejectingErrorHa
}
private static boolean isMessageException(final Throwable cause) {
return cause instanceof InvalidTargetAddressException || cause instanceof MessageConversionException
|| cause instanceof MessageHandlingException;
return cause instanceof InvalidTargetAddressException ||
cause instanceof MessageConversionException ||
cause instanceof MessageHandlingException;
}
}

2
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DelegatingConditionalErrorHandler.java
View File

@@ -61,4 +61,4 @@ public class DelegatingConditionalErrorHandler implements ErrorHandler {
}
return false;
}
}
}

4
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DmfApiConfiguration.java
View File

@@ -19,6 +19,4 @@ import org.springframework.context.annotation.Import;
@Configuration
@ComponentScan
@Import(AmqpConfiguration.class)
public class DmfApiConfiguration {
}
public class DmfApiConfiguration {}