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

Code format hawkbit-dmf-amqp (#1936)

Browse Source 
Signed-off-by: Marinov Avgustin <Avgustin.Marinov@bosch.com>
This commit is contained in:
Avgustin Marinov
2024-11-05 11:24:04 +02:00
committed by GitHub
parent ea56402aaa
commit ec5b797d41
25 changed files with 1172 additions and 1256 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,19 +31,17 @@ public abstract class AbstractAmqpErrorHandler<T> implements AmqpErrorHandler {
/**
* Returns the class of the exception.
*
* @return
* the exception class
* @return the exception class
*/
public abstract Class<T> getExceptionClass();
/**
* Returns the customized error message.
*
* @return
* the customized error message
* @return the customized error message
*/
public String getErrorMessage(Throwable throwable){
return AmqpErrorMessageComposer.constructErrorMessage(throwable);
public String getErrorMessage(Throwable throwable) {
return AmqpErrorMessageComposer.constructErrorMessage(throwable);
}
}

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

@@ -9,6 +9,11 @@
*/
package org.eclipse.hawkbit.amqp;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.hawkbit.api.ArtifactUrlHandler;
import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
@@ -46,11 +51,6 @@ import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.ErrorHandler;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* Spring configuration for AMQP based DMF communication for indirect device
* integration.
@@ -76,9 +76,7 @@ public class AmqpConfiguration {
/**
* Creates a custom error handler bean.
*
* @param handlers
* list of {@link AmqpErrorHandler} handlers
* @param handlers list of {@link AmqpErrorHandler} handlers
* @return the delegating error handler bean
*/
@Bean
@@ -253,15 +251,10 @@ public class AmqpConfiguration {
/**
* Create AMQP handler service bean.
*
* @param rabbitTemplate
* for converting messages
* @param amqpMessageDispatcherService
* to sending events to DMF client
* @param controllerManagement
* for target repo access
* @param entityFactory
* to create entities
*
* @param rabbitTemplate for converting messages
* @param amqpMessageDispatcherService to sending events to DMF client
* @param controllerManagement for target repo access
* @param entityFactory to create entities
* @return handler service bean
*/
@Bean

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

@@ -35,9 +35,8 @@ public class AmqpDeadletterProperties {
/**
* Return the deadletter arguments.
*
* @param exchange
* the deadletter exchange
*
* @param exchange the deadletter exchange
* @return map which holds the properties
*/
public Map<String, Object> getDeadLetterExchangeArgs(final String exchange) {
@@ -48,7 +47,7 @@ public class AmqpDeadletterProperties {
/**
* Create a deadletter queue with ttl for messages
*
*
* @param queueName the deadletter queue name
* @return the deadletter queue
*/

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

@@ -19,10 +19,8 @@ public interface AmqpErrorHandler {
/**
* Handles the error based on the type of exception
*
* @param throwable
* the throwable
* @param chain
* an {@link AmqpErrorHandlerChain}
* @param throwable the throwable
* @param chain an {@link AmqpErrorHandlerChain}
*/
void doHandle(final Throwable throwable, final AmqpErrorHandlerChain chain);
}

22
hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpErrorHandlerChain.java
View File

@@ -9,27 +9,26 @@
*/
package org.eclipse.hawkbit.amqp;
import org.springframework.util.ErrorHandler;
import java.util.Iterator;
import java.util.List;
import org.springframework.util.ErrorHandler;
/**
* An error handler chain that delegates the error to the matching error handler based on the type of exception
*/
public final class AmqpErrorHandlerChain {
private final Iterator<AmqpErrorHandler> iterator;
private final ErrorHandler defaultHandler;
/**
* Constructor.
*
* @param iterator
* the {@link AmqpErrorHandler} iterator
* @param defaultHandler
* the default handler
* @param iterator the {@link AmqpErrorHandler} iterator
* @param defaultHandler the default handler
*/
private AmqpErrorHandlerChain(Iterator<AmqpErrorHandler> iterator, ErrorHandler defaultHandler) {
private AmqpErrorHandlerChain(Iterator<AmqpErrorHandler> iterator, ErrorHandler defaultHandler) {
this.iterator = iterator;
this.defaultHandler = defaultHandler;
}
@@ -37,10 +36,8 @@ public final class AmqpErrorHandlerChain {
/**
* Returns an {@link AmqpErrorHandlerChain}
*
* @param errorHandlers
* {@link List} of error handlers
* @param defaultHandler
* the default error handler
* @param errorHandlers {@link List} of error handlers
* @param defaultHandler the default error handler
* @return an {@link AmqpErrorHandlerChain}
*/
public static AmqpErrorHandlerChain getHandlerChain(final List<AmqpErrorHandler> errorHandlers, final ErrorHandler defaultHandler) {
@@ -50,8 +47,7 @@ public final class AmqpErrorHandlerChain {
/**
* Handles the error based on the type of exception
*
* @param error
* the throwable containing the cause of exception
* @param error the throwable containing the cause of exception
*/
public void handle(final Throwable error) {
if (iterator.hasNext()) {

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

@@ -27,10 +27,8 @@ public final class AmqpErrorMessageComposer {
/**
* Constructs an error message based on failed message content
*
* @param throwable
* the throwable containing failed message content
* @return
* meaningful error message
* @param throwable the throwable containing failed message content
* @return meaningful error message
*/
public static String constructErrorMessage(final Throwable throwable) {
StringBuilder completeErrorMessage = new StringBuilder();

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

@@ -106,26 +106,16 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
/**
* Constructor.
*
* @param rabbitTemplate
* the rabbitTemplate
* @param amqpSenderService
* to send AMQP message
* @param artifactUrlHandler
* for generating download URLs
* @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 distributionSetManagement
* to retrieve modules
* @param tenantConfigurationManagement
* to access tenant configuration
*
* @param rabbitTemplate the rabbitTemplate
* @param amqpSenderService to send AMQP message
* @param artifactUrlHandler for generating download URLs
* @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 distributionSetManagement to retrieve modules
* @param tenantConfigurationManagement to access tenant configuration
*/
protected AmqpMessageDispatcherService(final RabbitTemplate rabbitTemplate,
final AmqpMessageSenderService amqpSenderService, final ArtifactUrlHandler artifactUrlHandler,
@@ -147,12 +137,16 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
this.tenantConfigurationManagement = tenantConfigurationManagement;
}
public boolean isBatchAssignmentsEnabled() {
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.
*
* @param assignedEvent
* the object to be send.
* @param assignedEvent the object to be send.
*/
@EventListener(classes = TargetAssignDistributionSetEvent.class)
protected void targetAssignDistributionSet(final TargetAssignDistributionSetEvent assignedEvent) {
@@ -173,8 +167,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
/**
* Listener for Multi-Action events.
*
* @param multiActionEvent
* the Multi-Action event to be processed
* @param multiActionEvent the Multi-Action event to be processed
*/
@EventListener(classes = MultiActionEvent.class)
protected void onMultiAction(final MultiActionEvent multiActionEvent) {
@@ -185,6 +178,261 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
sendMultiActionRequestMessages(multiActionEvent.getTenant(), multiActionEvent.getControllerIds());
}
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,
final Function<Action, Map<SoftwareModule, List<SoftwareModuleMetadata>>> getSoftwareModuleMetaData) {
final URI targetAddress = target.getAddress();
if (!IpUtil.isAmqpUri(targetAddress) || CollectionUtils.isEmpty(actions)) {
return;
}
final DmfMultiActionRequest multiActionRequest = new DmfMultiActionRequest();
actions.forEach(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,
createConnectorMessagePropertiesEvent(tenant, target.getControllerId(), EventTopic.MULTI_ACTION));
amqpSenderService.sendMessage(message, targetAddress);
}
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)));
}
return request;
}
/**
* Method to send a message to a RabbitMQ Exchange after the assignment of
* the Distribution set to a Target has been canceled.
*
* @param cancelEvent that is to be converted to a DMF message
*/
@EventListener(classes = CancelTargetAssignmentEvent.class)
protected void targetCancelAssignmentToDistributionSet(final CancelTargetAssignmentEvent cancelEvent) {
if (!shouldBeProcessed(cancelEvent)) {
return;
}
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());
});
});
}
/**
* 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.
*/
@EventListener(classes = TargetDeletedEvent.class)
protected void targetDelete(final TargetDeletedEvent deleteEvent) {
if (!shouldBeProcessed(deleteEvent)) {
return;
}
sendDeleteMessage(deleteEvent.getTenant(), deleteEvent.getControllerId(), deleteEvent.getTargetAddress());
}
@EventListener(classes = TargetAttributesRequestedEvent.class)
protected void targetTriggerUpdateAttributes(final TargetAttributesRequestedEvent updateAttributesEvent) {
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())
.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
.setCorrelationId(ping.getMessageProperties().getCorrelationId())
.setHeader(MessageHeaderKey.TYPE, MessageType.PING_RESPONSE).setHeader(MessageHeaderKey.TENANT, tenant)
.build();
amqpSenderService.sendMessage(message,
IpUtil.createAmqpUri(virtualHost, ping.getMessageProperties().getReplyTo()));
}
protected boolean shouldBeProcessed(final RemoteApplicationEvent event) {
return isFromSelf(event);
}
protected void sendCancelMessageToTarget(final String tenant, final String controllerId, final Long actionId,
final URI address) {
if (!IpUtil.isAmqpUri(address)) {
return;
}
final DmfActionRequest actionRequest = new DmfActionRequest();
actionRequest.setActionId(actionId);
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) {
final DmfTarget dmfTarget = new DmfTarget();
dmfTarget.setActionId(actionId);
dmfTarget.setControllerId(target.getControllerId());
dmfTarget.setTargetSecurityToken(systemSecurityContext.runAsSystem(target::getSecurityToken));
return dmfTarget;
}
/**
* Creates a Confirmation request.
*
* @param target the target
* @param actionId the actionId
* @param softwareModules the software modules
* @return
*/
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)));
}
return request;
}
private static DmfActionRequest createPlainActionRequest(final Action action) {
final DmfActionRequest actionRequest = new DmfActionRequest();
actionRequest.setActionId(action.getId());
return actionRequest;
}
/**
* 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.
*/
private static EventTopic getEventTypeForTarget(final ActionProperties action) {
if (action.isWaitingConfirmation()) {
return EventTopic.CONFIRM;
}
return (Action.ActionType.DOWNLOAD_ONLY == action.getActionType() || !action.isMaintenanceWindowAvailable())
? EventTopic.DOWNLOAD
: EventTopic.DOWNLOAD_AND_INSTALL;
}
/**
* 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
*/
private static EventTopic getEventTypeForAction(final Action action) {
if (action.isCancelingOrCanceled()) {
return EventTopic.CANCEL_DOWNLOAD;
}
return getEventTypeForTarget(new ActionProperties(action));
}
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
final Iterable<List<T>> partitions = ListUtils.partition(IterableUtils.toList(controllerIds), MAX_PROCESSING_SIZE);
// Preserve the security context because it gets lost when executing
// loading calls in new threads
final SecurityContext context = SecurityContextHolder.getContext();
// Handling remote request in parallel streams
return StreamSupport.stream(partitions.spliterator(), true) //
.flatMap(partition -> withSecurityContext(() -> loadingFunction.apply(partition), context).stream())
.collect(Collectors.toList());
}
return loadingFunction.apply(controllerIds);
}
private static <T> T withSecurityContext(final Supplier<T> callable, final SecurityContext securityContext) {
final SecurityContext oldContext = SecurityContextHolder.getContext();
try {
SecurityContextHolder.setContext(securityContext);
return callable.get();
} finally {
SecurityContextHolder.setContext(oldContext);
}
}
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) {
final MessageProperties messageProperties = createConnectorMessageProperties(tenant, controllerId);
messageProperties.setHeader(MessageHeaderKey.TYPE, MessageType.THING_DELETED);
return messageProperties;
}
private static MessageProperties createConnectorMessageProperties(final String tenant, final String controllerId) {
final MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setHeader(MessageHeaderKey.CONTENT_TYPE, MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setHeader(MessageHeaderKey.THING_ID, controllerId);
messageProperties.setHeader(MessageHeaderKey.TENANT, tenant);
return messageProperties;
}
private static MessageProperties createMessagePropertiesBatch(final String tenant, final EventTopic topic) {
final MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setHeader(MessageHeaderKey.CONTENT_TYPE, MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setHeader(MessageHeaderKey.TENANT, tenant);
messageProperties.setHeader(MessageHeaderKey.TOPIC, topic);
messageProperties.setHeader(MessageHeaderKey.TYPE, MessageType.EVENT);
return messageProperties;
}
private static EventTopic getBatchEventTopicForAction(final ActionProperties action) {
return (Action.ActionType.DOWNLOAD_ONLY == action.getActionType() || !action.isMaintenanceWindowAvailable())
? EventTopic.BATCH_DOWNLOAD
: EventTopic.BATCH_DOWNLOAD_AND_INSTALL;
}
private List<Target> getTargetsWithoutPendingCancellations(final Set<String> controllerIds) {
return partitionedParallelExecution(controllerIds, partition -> {
return targetManagement.getByControllerID(partition).stream().filter(target -> {
@@ -207,13 +455,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
});
}
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);
}
private void sendUpdateMessageToTargets(final Map<String, ActionProperties> actionsPropsByTargetId,
final List<Target> targets, final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
@@ -248,27 +489,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
}
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;
}
final DmfMultiActionRequest multiActionRequest = new DmfMultiActionRequest();
actions.forEach(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,
createConnectorMessagePropertiesEvent(tenant, target.getControllerId(), EventTopic.MULTI_ACTION));
amqpSenderService.sendMessage(message, targetAddress);
}
private DmfActionRequest createDmfActionRequest(final Target target, final Action action,
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
if (action.isCancelingOrCanceled()) {
@@ -279,137 +499,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return createDownloadAndUpdateRequest(target, action.getId(), softwareModules);
}
private static DmfActionRequest createPlainActionRequest(final Action action) {
final DmfActionRequest actionRequest = new DmfActionRequest();
actionRequest.setActionId(action.getId());
return actionRequest;
}
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)));
}
return request;
}
/**
* 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.
*/
private static EventTopic getEventTypeForTarget(final ActionProperties action) {
if (action.isWaitingConfirmation()) {
return EventTopic.CONFIRM;
}
return (Action.ActionType.DOWNLOAD_ONLY == action.getActionType() || !action.isMaintenanceWindowAvailable())
? EventTopic.DOWNLOAD
: EventTopic.DOWNLOAD_AND_INSTALL;
}
/**
* 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
*/
private static EventTopic getEventTypeForAction(final Action action) {
if (action.isCancelingOrCanceled()) {
return EventTopic.CANCEL_DOWNLOAD;
}
return getEventTypeForTarget(new ActionProperties(action));
}
/**
* Method to send a message to a RabbitMQ Exchange after the assignment of
* the Distribution set to a Target has been canceled.
*
* @param cancelEvent
* that is to be converted to a DMF message
*/
@EventListener(classes = CancelTargetAssignmentEvent.class)
protected void targetCancelAssignmentToDistributionSet(final CancelTargetAssignmentEvent cancelEvent) {
if (!shouldBeProcessed(cancelEvent)) {
return;
}
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());
});
});
}
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
final Iterable<List<T>> partitions = ListUtils.partition(IterableUtils.toList(controllerIds), MAX_PROCESSING_SIZE);
// Preserve the security context because it gets lost when executing
// loading calls in new threads
final SecurityContext context = SecurityContextHolder.getContext();
// Handling remote request in parallel streams
return StreamSupport.stream(partitions.spliterator(), true) //
.flatMap(partition -> withSecurityContext(() -> loadingFunction.apply(partition), context).stream())
.collect(Collectors.toList());
}
return loadingFunction.apply(controllerIds);
}
private static <T> T withSecurityContext(final Supplier<T> callable, final SecurityContext securityContext) {
final SecurityContext oldContext = SecurityContextHolder.getContext();
try {
SecurityContextHolder.setContext(securityContext);
return callable.get();
} finally {
SecurityContextHolder.setContext(oldContext);
}
}
/**
* 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.
*/
@EventListener(classes = TargetDeletedEvent.class)
protected void targetDelete(final TargetDeletedEvent deleteEvent) {
if (!shouldBeProcessed(deleteEvent)) {
return;
}
sendDeleteMessage(deleteEvent.getTenant(), deleteEvent.getControllerId(), deleteEvent.getTargetAddress());
}
@EventListener(classes = TargetAttributesRequestedEvent.class)
protected void targetTriggerUpdateAttributes(final TargetAttributesRequestedEvent updateAttributesEvent) {
sendUpdateAttributesMessageToTarget(updateAttributesEvent.getTenant(), updateAttributesEvent.getControllerId(),
updateAttributesEvent.getTargetAddress());
}
private void sendSingleUpdateMessage(final ActionProperties action, final Target target,
final Map<SoftwareModule, List<SoftwareModuleMetadata>> modules) {
@@ -434,17 +523,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
amqpSenderService.sendMessage(message, targetAddress);
}
protected void sendPingReponseToDmfReceiver(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)
.build();
amqpSenderService.sendMessage(message,
IpUtil.createAmqpUri(virtualHost, ping.getMessageProperties().getReplyTo()));
}
private void sendDeleteMessage(final String tenant, final String controllerId, final String targetAddress) {
if (!hasValidAddress(targetAddress)) {
return;
@@ -459,10 +537,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return targetAddress != null && IpUtil.isAmqpUri(URI.create(targetAddress));
}
protected boolean shouldBeProcessed(final RemoteApplicationEvent event) {
return isFromSelf(event);
}
private boolean isFromSelf(final RemoteApplicationEvent event) {
return serviceMatcher == null || serviceMatcher.isFromSelf(event);
}
@@ -471,22 +545,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return deploymentManagement.hasPendingCancellations(targetId);
}
protected void sendCancelMessageToTarget(final String tenant, final String controllerId, final Long actionId,
final URI address) {
if (!IpUtil.isAmqpUri(address)) {
return;
}
final DmfActionRequest actionRequest = new DmfActionRequest();
actionRequest.setActionId(actionId);
final Message message = getMessageConverter().toMessage(actionRequest,
createConnectorMessagePropertiesEvent(tenant, controllerId, EventTopic.CANCEL_DOWNLOAD));
amqpSenderService.sendMessage(message, address);
}
private void sendUpdateAttributesMessageToTarget(final String tenant, final String controllerId,
final String targetAddress) {
if (!hasValidAddress(targetAddress)) {
@@ -499,30 +557,6 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
amqpSenderService.sendMessage(message, URI.create(targetAddress));
}
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) {
final MessageProperties messageProperties = createConnectorMessageProperties(tenant, controllerId);
messageProperties.setHeader(MessageHeaderKey.TYPE, MessageType.THING_DELETED);
return messageProperties;
}
private static MessageProperties createConnectorMessageProperties(final String tenant, final String controllerId) {
final MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setHeader(MessageHeaderKey.CONTENT_TYPE, MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setHeader(MessageHeaderKey.THING_ID, controllerId);
messageProperties.setHeader(MessageHeaderKey.TENANT, tenant);
return messageProperties;
}
private DmfSoftwareModule convertToAmqpSoftwareModule(final Target target,
final Entry<SoftwareModule, List<SoftwareModuleMetadata>> entry) {
final DmfSoftwareModule amqpSoftwareModule = new DmfSoftwareModule();
@@ -558,9 +592,9 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
final TenantMetaData metaData = systemManagement.getTenantMetadata();
artifact.setUrls(artifactUrlHandler
.getUrls(new URLPlaceholder(metaData.getTenant(),
metaData.getId(), target.getControllerId(), target.getId(),
new SoftwareData(localArtifact.getSoftwareModule().getId(), localArtifact.getFilename(),
localArtifact.getId(), localArtifact.getSha1Hash())),
metaData.getId(), target.getControllerId(), target.getId(),
new SoftwareData(localArtifact.getSoftwareModule().getId(), localArtifact.getFilename(),
localArtifact.getId(), localArtifact.getSha1Hash())),
ApiType.DMF)
.stream().collect(Collectors.toMap(ArtifactUrl::getProtocol, ArtifactUrl::getRef)));
@@ -608,55 +642,4 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
createMessagePropertiesBatch(firstAction.getTenant(), getBatchEventTopicForAction(firstAction)));
amqpSenderService.sendMessage(message, firstTarget.getAddress());
}
protected DmfTarget convertToDmfTarget(final Target target, final Long actionId) {
final DmfTarget dmfTarget = new DmfTarget();
dmfTarget.setActionId(actionId);
dmfTarget.setControllerId(target.getControllerId());
dmfTarget.setTargetSecurityToken(systemSecurityContext.runAsSystem(target::getSecurityToken));
return dmfTarget;
}
private static MessageProperties createMessagePropertiesBatch(final String tenant, final EventTopic topic) {
final MessageProperties messageProperties = new MessageProperties();
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setHeader(MessageHeaderKey.CONTENT_TYPE, MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setHeader(MessageHeaderKey.TENANT, tenant);
messageProperties.setHeader(MessageHeaderKey.TOPIC, topic);
messageProperties.setHeader(MessageHeaderKey.TYPE, MessageType.EVENT);
return messageProperties;
}
public boolean isBatchAssignmentsEnabled() {
return systemSecurityContext.runAsSystem(() -> tenantConfigurationManagement
.getConfigurationValue(BATCH_ASSIGNMENTS_ENABLED, Boolean.class).getValue());
}
private static EventTopic getBatchEventTopicForAction(final ActionProperties action) {
return (Action.ActionType.DOWNLOAD_ONLY == action.getActionType() || !action.isMaintenanceWindowAvailable())
? EventTopic.BATCH_DOWNLOAD
: EventTopic.BATCH_DOWNLOAD_AND_INSTALL;
}
/**
* Creates a Confirmation request.
* @param target the target
* @param actionId the actionId
* @param softwareModules the software modules
* @return
*/
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)));
}
return request;
}
}

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

@@ -73,38 +73,25 @@ import org.springframework.util.StringUtils;
@Slf4j
public class AmqpMessageHandlerService extends BaseAmqpService {
private final AmqpMessageDispatcherService amqpMessageDispatcherService;
private ControllerManagement controllerManagement;
private final ConfirmationManagement confirmationManagement;
private final EntityFactory entityFactory;
private final TenantConfigurationManagement tenantConfigurationManagement;
private final SystemSecurityContext systemSecurityContext;
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;
private final TenantConfigurationManagement tenantConfigurationManagement;
private final SystemSecurityContext systemSecurityContext;
private ControllerManagement controllerManagement;
/**
* Constructor.
*
* @param rabbitTemplate
* for converting messages
* @param amqpMessageDispatcherService
* to sending events to DMF client
* @param controllerManagement
* for target repo access
* @param entityFactory
* to create entities
* @param systemSecurityContext
* the system Security Context
* @param tenantConfigurationManagement
* the tenant configuration Management
* @param confirmationManagement
* the confirmation management
*
* @param rabbitTemplate for converting messages
* @param amqpMessageDispatcherService to sending events to DMF client
* @param controllerManagement for target repo access
* @param entityFactory to create entities
* @param systemSecurityContext the system Security Context
* @param tenantConfigurationManagement the tenant configuration Management
* @param confirmationManagement the confirmation management
*/
public AmqpMessageHandlerService(final RabbitTemplate rabbitTemplate,
final AmqpMessageDispatcherService amqpMessageDispatcherService,
@@ -123,12 +110,9 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
/**
* Method to handle all incoming DMF amqp messages.
*
* @param message
* incoming message
* @param type
* the message type
* @param tenant
* the contentType of the message
* @param message incoming message
* @param type the message type
* @param tenant the contentType of the message
* @return a message if <null> no message is send back to sender
*/
@RabbitListener(queues = "${hawkbit.dmf.rabbitmq.receiverQueue:dmf_receiver}", containerFactory = "listenerContainerFactory")
@@ -140,15 +124,11 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
/**
* * Executed if a amqp message arrives.
*
* @param message
* the message
* @param type
* the type
* @param tenant
* the tenant
* @param virtualHost
* the virtual host
*
* @param message the message
* @param type the type
* @param tenant the tenant
* @param virtualHost the virtual host
* @return the rpc message back to supplier.
*/
public Message onMessage(final Message message, final String type, final String tenant, final String virtualHost) {
@@ -160,28 +140,28 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
try {
final MessageType messageType = MessageType.valueOf(type);
switch (messageType) {
case THING_CREATED:
setTenantSecurityContext(tenant);
registerTarget(message, virtualHost);
break;
case THING_REMOVED:
setTenantSecurityContext(tenant);
deleteTarget(message);
break;
case EVENT:
checkContentTypeJson(message);
setTenantSecurityContext(tenant);
handleIncomingEvent(message);
break;
case PING:
if (isCorrelationIdNotEmpty(message)) {
amqpMessageDispatcherService.sendPingReponseToDmfReceiver(message, tenant, virtualHost);
}
break;
default:
logAndThrowMessageError(message, "No handle method was found for the given message type.");
case THING_CREATED:
setTenantSecurityContext(tenant);
registerTarget(message, virtualHost);
break;
case THING_REMOVED:
setTenantSecurityContext(tenant);
deleteTarget(message);
break;
case EVENT:
checkContentTypeJson(message);
setTenantSecurityContext(tenant);
handleIncomingEvent(message);
break;
case PING:
if (isCorrelationIdNotEmpty(message)) {
amqpMessageDispatcherService.sendPingReponseToDmfReceiver(message, tenant, virtualHost);
}
break;
default:
logAndThrowMessageError(message, "No handle method was found for the given message type.");
}
} catch(AssignmentQuotaExceededException ex) {
} catch (AssignmentQuotaExceededException ex) {
throw new AmqpRejectAndDontRequeueException("Could not handle message due to quota violation!", ex);
} catch (final IllegalArgumentException ex) {
throw new AmqpRejectAndDontRequeueException("Invalid message!", ex);
@@ -191,6 +171,11 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
return null;
}
// for testing
public void setControllerManagement(final ControllerManagement controllerManagement) {
this.controllerManagement = controllerManagement;
}
private static void setSecurityContext(final Authentication authentication) {
final SecurityContextImpl securityContextImpl = new SecurityContextImpl();
securityContextImpl.setAuthentication(authentication);
@@ -205,15 +190,92 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
setSecurityContext(authenticationToken);
}
private static boolean isOptionalMessageBodyEmpty(final Message message) {
// empty byte array message body is serialized to double-quoted string
// by message converter and should also be considered as empty
return isMessageBodyEmpty(message) || EMPTY_MESSAGE_BODY.equals(new String(message.getBody()));
}
private static boolean shouldTargetProceed(final Action action) {
return !action.isActive() || (action.hasMaintenanceSchedule() && action.isMaintenanceWindowAvailable());
}
private static boolean isCorrelationIdNotEmpty(final Message message) {
return StringUtils.hasLength(message.getMessageProperties().getCorrelationId());
}
// Exception squid:MethodCyclomaticComplexity - false positive, is a simple
// mapping
@SuppressWarnings("squid:MethodCyclomaticComplexity")
private static Status mapStatus(final Message message, final DmfActionUpdateStatus actionUpdateStatus,
final Action action) {
Status status = null;
switch (actionUpdateStatus.getActionStatus()) {
case DOWNLOAD:
status = Status.DOWNLOAD;
break;
case RETRIEVED:
status = Status.RETRIEVED;
break;
case RUNNING:
case CONFIRMED:
status = Status.RUNNING;
break;
case CANCELED:
status = Status.CANCELED;
break;
case FINISHED:
status = Status.FINISHED;
break;
case ERROR:
status = Status.ERROR;
break;
case WARNING:
status = Status.WARNING;
break;
case DOWNLOADED:
status = Status.DOWNLOADED;
break;
case CANCEL_REJECTED:
status = handleCancelRejectedState(message, action);
break;
case DENIED:
status = Status.WAIT_FOR_CONFIRMATION;
break;
default:
logAndThrowMessageError(message, "Status for action does not exisit.");
}
return status;
}
private static Status handleCancelRejectedState(final Message message, final Action action) {
if (action.isCancelingOrCanceled()) {
return Status.CANCEL_REJECTED;
}
logAndThrowMessageError(message,
"Cancel rejected message is not allowed, if action is on state: " + action.getStatus());
return null;
}
/**
* Retrieve the update mode from the given update message.
*/
private static UpdateMode getUpdateMode(final DmfAttributeUpdate update) {
final DmfUpdateMode mode = update.getMode();
if (mode != null) {
return UpdateMode.valueOf(mode.name());
}
return null;
}
/**
* 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 virtualHost
* the virtual host
* @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);
@@ -235,7 +297,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
final DmfAttributeUpdate thingAttributeUpdateBody = thingCreateBody.getAttributeUpdate();
log.debug("Received \"THING_CREATED\" AMQP message for thing \"{}\" with target name \"{}\" and type " +
"\"{}\".", thingId, thingCreateBody.getName(), thingCreateBody.getType());
"\"{}\".", thingId, thingCreateBody.getName(), thingCreateBody.getType());
target = controllerManagement.findOrRegisterTargetIfItDoesNotExist(thingId, amqpUri,
thingCreateBody.getName(), thingCreateBody.getType());
@@ -253,12 +315,6 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
}
}
private static boolean isOptionalMessageBodyEmpty(final Message message) {
// empty byte array message body is serialized to double-quoted string
// by message converter and should also be considered as empty
return isMessageBodyEmpty(message) || EMPTY_MESSAGE_BODY.equals(new String(message.getBody()));
}
private void sendUpdateCommandToTarget(final Target target) {
if (isMultiAssignmentsEnabled()) {
sendCurrentActionsAsMultiActionToTarget(target);
@@ -314,23 +370,22 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
/**
* Method to handle the different topics to an event.
*
* @param message
* the incoming event message.
* @param message the incoming event message.
*/
private void handleIncomingEvent(final Message message) {
switch (EventTopic.valueOf(getStringHeaderKey(message, MessageHeaderKey.TOPIC, "EventTopic is null"))) {
case UPDATE_ACTION_STATUS:
updateActionStatus(message);
break;
case UPDATE_ATTRIBUTES:
updateAttributes(message);
break;
case UPDATE_AUTO_CONFIRM:
setAutoConfirmationState(message);
break;
default:
logAndThrowMessageError(message, "Got event without appropriate topic.");
break;
case UPDATE_ACTION_STATUS:
updateActionStatus(message);
break;
case UPDATE_ATTRIBUTES:
updateAttributes(message);
break;
case UPDATE_AUTO_CONFIRM:
setAutoConfirmationState(message);
break;
default:
logAndThrowMessageError(message, "Got event without appropriate topic.");
break;
}
}
@@ -345,7 +400,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
final String thingId = getStringHeaderKey(message, MessageHeaderKey.THING_ID, THING_ID_NULL);
controllerManagement.updateControllerAttributes(thingId, attributeUpdate.getAttributes(),
getUpdateMode(attributeUpdate));
getUpdateMode(attributeUpdate));
}
private void setAutoConfirmationState(final Message message) {
@@ -367,8 +422,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
/**
* Method to update the action status of an action through the event.
*
* @param message
* the object form the ampq message
* @param message the object form the ampq message
*/
private void updateActionStatus(final Message message) {
final DmfActionUpdateStatus actionUpdateStatus = convertMessage(message, DmfActionUpdateStatus.class);
@@ -408,68 +462,6 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
}
}
private static boolean shouldTargetProceed(final Action action) {
return !action.isActive() || (action.hasMaintenanceSchedule() && action.isMaintenanceWindowAvailable());
}
private static boolean isCorrelationIdNotEmpty(final Message message) {
return StringUtils.hasLength(message.getMessageProperties().getCorrelationId());
}
// Exception squid:MethodCyclomaticComplexity - false positive, is a simple
// mapping
@SuppressWarnings("squid:MethodCyclomaticComplexity")
private static Status mapStatus(final Message message, final DmfActionUpdateStatus actionUpdateStatus,
final Action action) {
Status status = null;
switch (actionUpdateStatus.getActionStatus()) {
case DOWNLOAD:
status = Status.DOWNLOAD;
break;
case RETRIEVED:
status = Status.RETRIEVED;
break;
case RUNNING:
case CONFIRMED:
status = Status.RUNNING;
break;
case CANCELED:
status = Status.CANCELED;
break;
case FINISHED:
status = Status.FINISHED;
break;
case ERROR:
status = Status.ERROR;
break;
case WARNING:
status = Status.WARNING;
break;
case DOWNLOADED:
status = Status.DOWNLOADED;
break;
case CANCEL_REJECTED:
status = handleCancelRejectedState(message, action);
break;
case DENIED:
status = Status.WAIT_FOR_CONFIRMATION;
break;
default:
logAndThrowMessageError(message, "Status for action does not exisit.");
}
return status;
}
private static Status handleCancelRejectedState(final Message message, final Action action) {
if (action.isCancelingOrCanceled()) {
return Status.CANCEL_REJECTED;
}
logAndThrowMessageError(message,
"Cancel rejected message is not allowed, if action is on state: " + action.getStatus());
return null;
}
// Exception squid:S3655 - logAndThrowMessageError throws exception, i.e.
// get will not be called
@SuppressWarnings("squid:S3655")
@@ -488,17 +480,6 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
return findActionWithDetails.get();
}
/**
* Retrieve the update mode from the given update message.
*/
private static UpdateMode getUpdateMode(final DmfAttributeUpdate update) {
final DmfUpdateMode mode = update.getMode();
if (mode != null) {
return UpdateMode.valueOf(mode.name());
}
return null;
}
private boolean isMultiAssignmentsEnabled() {
return getConfigValue(MULTI_ASSIGNMENTS_ENABLED, Boolean.class);
}
@@ -507,9 +488,4 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
return systemSecurityContext
.runAsSystem(() -> tenantConfigurationManagement.getConfigurationValue(key, valueType).getValue());
}
// for testing
public void setControllerManagement(final ControllerManagement controllerManagement) {
this.controllerManagement = controllerManagement;
}
}

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

@@ -24,11 +24,9 @@ public interface AmqpMessageSenderService {
/**
* Send the given message to the given uri. The uri contains the (virtual)
* host and exchange e.g amqp://host/exchange.
*
* @param message
* the amqp message
* @param replyTo
* the reply to uri
*
* @param message the amqp message
* @param replyTo the reply to uri
*/
void sendMessage(@NotNull final Message message, @NotNull final URI replyTo);
}

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

@@ -15,7 +15,6 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* Bean which holds the necessary properties for configuring the AMQP
* connection.
*
*/
@Data
@ConfigurationProperties("hawkbit.dmf.rabbitmq")

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

@@ -32,29 +32,18 @@ public class BaseAmqpService {
/**
* Constructor.
*
* @param rabbitTemplate
* the rabbit template.
*
* @param rabbitTemplate the rabbit template.
*/
public BaseAmqpService(final RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
protected static void checkContentTypeJson(final Message message) {
final MessageProperties messageProperties = message.getMessageProperties();
if (messageProperties.getContentType() != null && messageProperties.getContentType().contains("json")) {
return;
}
throw new AmqpRejectAndDontRequeueException("Content-Type is not JSON compatible");
}
/**
* Is needed to convert a incoming message to is originally object type.
*
* @param message
* the message to convert.
* @param clazz
* the class of the originally object.
* @param message the message to convert.
* @param clazz the class of the originally object.
* @return the converted object
*/
@SuppressWarnings("unchecked")
@@ -65,14 +54,27 @@ public class BaseAmqpService {
return (T) rabbitTemplate.getMessageConverter().fromMessage(message);
}
protected MessageConverter getMessageConverter() {
return rabbitTemplate.getMessageConverter();
protected static void checkContentTypeJson(final Message message) {
final MessageProperties messageProperties = message.getMessageProperties();
if (messageProperties.getContentType() != null && messageProperties.getContentType().contains("json")) {
return;
}
throw new AmqpRejectAndDontRequeueException("Content-Type is not JSON compatible");
}
protected static boolean isMessageBodyEmpty(final Message message) {
return message.getBody() == null || message.getBody().length == 0;
}
protected static final void logAndThrowMessageError(final Message message, final String error) {
log.debug("Warning! \"{}\" reported by message: {}", error, message);
throw new AmqpRejectAndDontRequeueException(error);
}
protected MessageConverter getMessageConverter() {
return rabbitTemplate.getMessageConverter();
}
protected void checkMessageBody(@NotNull final Message message) {
if (isMessageBodyEmpty(message)) {
throw new MessageConversionException("Message body cannot be null");
@@ -89,20 +91,14 @@ public class BaseAmqpService {
return value.toString();
}
protected static final void logAndThrowMessageError(final Message message, final String error) {
log.debug("Warning! \"{}\" reported by message: {}", error, message);
throw new AmqpRejectAndDontRequeueException(error);
}
protected RabbitTemplate getRabbitTemplate() {
return rabbitTemplate;
}
/**
* Clean message properties before sending a message.
*
* @param message
* the message to cleaned up
*
* @param message the message to cleaned up
*/
protected void cleanMessageHeaderProperties(final Message message) {
message.getMessageProperties().getHeaders().remove(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME);

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

@@ -18,7 +18,6 @@ import org.springframework.util.ErrorHandler;
/**
* {@link RabbitListenerContainerFactory} that can be configured through
* hawkBit's {@link AmqpProperties}.
*
*/
public class ConfigurableRabbitListenerContainerFactory extends SimpleRabbitListenerContainerFactory {
@@ -26,14 +25,11 @@ public class ConfigurableRabbitListenerContainerFactory extends SimpleRabbitList
/**
* Constructor.
*
* @param missingQueuesFatal
* the missingQueuesFatal to set.
*
* @param missingQueuesFatal the missingQueuesFatal to set.
* @param declarationRetries The number of retries
* @param errorHandler the error handler which should be use
* @see SimpleMessageListenerContainer#setMissingQueuesFatal
* @param declarationRetries
* The number of retries
* @param errorHandler
* the error handler which should be use
*/
public ConfigurableRabbitListenerContainerFactory(final boolean missingQueuesFatal, final int declarationRetries,
final ErrorHandler errorHandler) {

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

@@ -29,7 +29,7 @@ public class DefaultAmqpMessageSenderService extends BaseAmqpService implements
/**
* Constructor.
*
*
* @param rabbitTemplate the AMQP template
*/
public DefaultAmqpMessageSenderService(final RabbitTemplate rabbitTemplate) {

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

@@ -14,11 +14,11 @@ import java.util.concurrent.TimeUnit;
import jakarta.validation.ConstraintViolationException;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.hawkbit.repository.exception.AssignmentQuotaExceededException;
import org.eclipse.hawkbit.repository.exception.CancelActionNotAllowedException;
import org.eclipse.hawkbit.repository.exception.EntityNotFoundException;
import org.eclipse.hawkbit.repository.exception.InvalidTargetAddressException;
import org.eclipse.hawkbit.repository.exception.InvalidTargetAttributeException;
import org.eclipse.hawkbit.repository.exception.AssignmentQuotaExceededException;
import org.eclipse.hawkbit.repository.exception.TenantNotExistException;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.rabbit.listener.FatalExceptionStrategy;

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

@@ -10,6 +10,7 @@
package org.eclipse.hawkbit.amqp;
import java.util.List;
import jakarta.validation.constraints.NotNull;
import lombok.extern.slf4j.Slf4j;
@@ -21,16 +22,15 @@ import org.springframework.util.ErrorHandler;
*/
@Slf4j
public class DelegatingConditionalErrorHandler implements ErrorHandler {
private final List<AmqpErrorHandler> handlers;
private final ErrorHandler defaultHandler;
/**
* Constructor
*
* @param handlers
* {@link List} of error handlers
* @param defaultHandler
* the default error handler
* @param handlers {@link List} of error handlers
* @param defaultHandler the default error handler
*/
public DelegatingConditionalErrorHandler(final List<AmqpErrorHandler> handlers, @NotNull final ErrorHandler defaultHandler) {
this.handlers = handlers;
@@ -53,7 +53,7 @@ public class DelegatingConditionalErrorHandler implements ErrorHandler {
}
private boolean includesAmqpRejectException(final Throwable t) {
if (t instanceof AmqpRejectAndDontRequeueException){
if (t instanceof AmqpRejectAndDontRequeueException) {
return true;
}
if (t.getCause() != null) {

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

@@ -30,11 +30,11 @@ public class MessageConversionExceptionHandler extends AbstractAmqpErrorHandler<
//since the detailed error message lies in the first parent of current throwable we retrieve it
// and append it to the errorMessage
final Optional<String> detailedErrorMessage = getFirstAncestralErrorMessage(throwable.getCause());
return detailedErrorMessage.isPresent()? (detailedErrorMessage.get() + errorMessage) : errorMessage;
return detailedErrorMessage.isPresent() ? (detailedErrorMessage.get() + errorMessage) : errorMessage;
}
private Optional<String> getFirstAncestralErrorMessage(final Throwable throwable) {
if(throwable.getCause() instanceof InvalidFormatException) {
if (throwable.getCause() instanceof InvalidFormatException) {
return Optional.of(throwable.getCause().getMessage());
}
return Optional.empty();