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

Fix DMF context (as system) (#2097)

Browse Source 
Signed-off-by: Avgustin Marinov <Avgustin.Marinov@bosch.com>
This commit is contained in:
Avgustin Marinov
2024-11-22 15:32:00 +02:00
committed by GitHub
parent 49d27b1cd1
commit 9df68e2d97
2 changed files with 35 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -153,12 +153,17 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return; return;
} }
final List<Target> filteredTargetList = getTargetsWithoutPendingCancellations(assignedEvent.getActions().keySet()); systemSecurityContext.runAsSystemAsTenant(() -> {
final List<Target> filteredTargetList = getTargetsWithoutPendingCancellations(
assignedEvent.getTenant(), assignedEvent.getActions().keySet());
if (!filteredTargetList.isEmpty()) { if (!filteredTargetList.isEmpty()) {
log.debug("targetAssignDistributionSet retrieved. I will forward it to DMF broker."); log.debug("targetAssignDistributionSet retrieved. I will forward it to DMF broker.");
sendUpdateMessageToTargets(assignedEvent.getDistributionSetId(), assignedEvent.getActions(), filteredTargetList); sendUpdateMessageToTargets(assignedEvent.getDistributionSetId(), assignedEvent.getActions(), filteredTargetList);
} }
return null;
}, assignedEvent.getTenant());
} }
/** /**
@@ -171,8 +176,12 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
if (!shouldBeProcessed(multiActionEvent)) { if (!shouldBeProcessed(multiActionEvent)) {
return; return;
} }
log.debug("MultiActionEvent received for {}", multiActionEvent.getControllerIds());
sendMultiActionRequestMessages(multiActionEvent.getTenant(), multiActionEvent.getControllerIds()); systemSecurityContext.runAsSystemAsTenant(() -> {
log.debug("MultiActionEvent received for {}", multiActionEvent.getControllerIds());
sendMultiActionRequestMessages(multiActionEvent.getTenant(), multiActionEvent.getControllerIds());
return null;
}, multiActionEvent.getTenant());
} }
protected void sendUpdateMessageToTarget( protected void sendUpdateMessageToTarget(
@@ -229,16 +238,20 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
return; return;
} }
final List<Target> eventTargets = partitionedParallelExecution( systemSecurityContext.runAsSystemAsTenant(() -> {
cancelEvent.getActions().keySet(), targetManagement::getByControllerID); final List<Target> eventTargets = partitionedParallelExecution(
cancelEvent.getActions().keySet(), targetManagement::getByControllerID);
eventTargets.forEach(target -> eventTargets.forEach(target ->
cancelEvent.getActionPropertiesForController(target.getControllerId()) cancelEvent.getActionPropertiesForController(target.getControllerId())
.map(ActionProperties::getId) .map(ActionProperties::getId)
.ifPresent(actionId -> .ifPresent(actionId ->
sendCancelMessageToTarget(cancelEvent.getTenant(), target.getControllerId(), actionId, target.getAddress()) sendCancelMessageToTarget(cancelEvent.getTenant(), target.getControllerId(), actionId, target.getAddress())
) )
); );
return null;
}, cancelEvent.getTenant());
} }
/** /**
@@ -251,6 +264,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
if (!shouldBeProcessed(deleteEvent)) { if (!shouldBeProcessed(deleteEvent)) {
return; return;
} }
sendDeleteMessage(deleteEvent.getTenant(), deleteEvent.getControllerId(), deleteEvent.getTargetAddress()); sendDeleteMessage(deleteEvent.getTenant(), deleteEvent.getControllerId(), deleteEvent.getTargetAddress());
} }
@@ -427,12 +441,13 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
: EventTopic.BATCH_DOWNLOAD_AND_INSTALL; : EventTopic.BATCH_DOWNLOAD_AND_INSTALL;
} }
private List<Target> getTargetsWithoutPendingCancellations(final Set<String> controllerIds) { private List<Target> getTargetsWithoutPendingCancellations(final String tenant, final Set<String> controllerIds) {
return partitionedParallelExecution(controllerIds, partition -> return partitionedParallelExecution(controllerIds, partition ->
targetManagement.getByControllerID(partition).stream() targetManagement.getByControllerID(partition).stream()
.filter(target -> { .filter(target -> {
if (hasPendingCancellations(target.getId())) { if (hasPendingCancellations(target.getId())) {
log.debug("Target {} has pending cancellations. Will not send update message to it.", target.getControllerId()); log.debug("Target {} has pending cancellations. Will not send update message to it.",
target.getControllerId());
return false; return false;
} }
return true; return true;

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

@@ -213,11 +213,9 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
return StringUtils.hasLength(message.getMessageProperties().getCorrelationId()); return StringUtils.hasLength(message.getMessageProperties().getCorrelationId());
} }
// Exception squid:MethodCyclomaticComplexity - false positive, is a simple // Exception squid:MethodCyclomaticComplexity - false positive, is a simple mapping
// mapping
@SuppressWarnings("squid:MethodCyclomaticComplexity") @SuppressWarnings("squid:MethodCyclomaticComplexity")
private static Status mapStatus(final Message message, final DmfActionUpdateStatus actionUpdateStatus, private static Status mapStatus(final Message message, final DmfActionUpdateStatus actionUpdateStatus, final Action action) {
final Action action) {
Status status = null; Status status = null;
switch (actionUpdateStatus.getActionStatus()) { switch (actionUpdateStatus.getActionStatus()) {
case DOWNLOAD: { case DOWNLOAD: {