Skip to content

feat(dmf): add externalRef to outgoing DMF messages #3363

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
herdt-michael wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat(dmf): add externalRef to outgoing DMF messages #3363

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 21 additions & 12 deletions ...hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,9 +141,10 @@ protected void sendUpdateMessageToTarget(
}

protected DmfDownloadAndUpdateRequest createDownloadAndUpdateRequest(
final Target target, final Long actionId, final Map<SoftwareModule, Map<String, String>> softwareModules) {
final Target target, final Long actionId, final String externalRef,
final Map<SoftwareModule, Map<String, String>> softwareModules) {
return new DmfDownloadAndUpdateRequest(
actionId, asSystem(target::getSecurityToken), convertToAmqpSoftwareModules(target, softwareModules));
actionId, asSystem(target::getSecurityToken), convertToAmqpSoftwareModules(target, softwareModules), externalRef);
}

/**
Expand All @@ -158,9 +159,9 @@ protected void targetCancelAssignmentToDistributionSet(final CancelTargetAssignm
final List<Target> eventTargets = partitionedParallelExecution(cancelEvent.getActions().keySet(), targetManagement::findByControllerId);
eventTargets.forEach(target ->
cancelEvent.getActionPropertiesForController(target.getControllerId())
.map(ActionProperties::getId)
.ifPresent(actionId -> sendCancelMessageToTarget
(cancelEvent.getTenant(), target.getControllerId(), actionId, IpUtil.addressToUri(target.getAddress()))));
.ifPresent(action -> sendCancelMessageToTarget(
cancelEvent.getTenant(), target.getControllerId(), action.getId(), action.getExternalRef(),
IpUtil.addressToUri(target.getAddress()))));
}

/**
Expand Down Expand Up @@ -196,20 +197,24 @@ protected void sendPingResponseToDmfReceiver(final Message ping, final String te
IpUtil.createAmqpUri(virtualHost, ping.getMessageProperties().getReplyTo()));
}

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 String externalRef, final URI address) {
if (!IpUtil.isAmqpUri(address)) {
return;
}

final Message message = getMessageConverter().toMessage(
new DmfActionRequest(actionId), createConnectorMessagePropertiesEvent(tenant, controllerId, EventTopic.CANCEL_DOWNLOAD));
new DmfActionRequest(actionId, externalRef),
createConnectorMessagePropertiesEvent(tenant, controllerId, EventTopic.CANCEL_DOWNLOAD));

amqpSenderService.sendMessage(message, address);
}

protected DmfConfirmRequest createConfirmRequest(
final Target target, final Long actionId, final Map<SoftwareModule, Map<String, String>> softwareModules) {
return new DmfConfirmRequest(actionId, asSystem(target::getSecurityToken), convertToAmqpSoftwareModules(target, softwareModules));
final Target target, final Long actionId, final String externalRef,
final Map<SoftwareModule, Map<String, String>> softwareModules) {
return new DmfConfirmRequest(
actionId, asSystem(target::getSecurityToken), convertToAmqpSoftwareModules(target, softwareModules), externalRef);
}

/**
Expand Down Expand Up @@ -344,9 +349,9 @@ private void sendSingleUpdateMessage(
if (action.isWaitingConfirmation()) {
// For the moment the confirmation request is the same as download and update request.
// It can be modified not to expose all the software modules in the future.
request = createConfirmRequest(target, action.getId(), modules);
request = createConfirmRequest(target, action.getId(), action.getExternalRef(), modules);
} else {
request = createDownloadAndUpdateRequest(target, action.getId(), modules);
request = createDownloadAndUpdateRequest(target, action.getId(), action.getExternalRef(), modules);
}

final Message message = getMessageConverter().toMessage(
Expand Down Expand Up @@ -450,7 +455,11 @@ private void sendBatchUpdateMessage(
final List<DmfTarget> dmfTargets = targets.stream()
.filter(target -> IpUtil.isAmqpUri(IpUtil.addressToUri(target.getAddress())))
// as system - the security token is sent to DMF receiver
.map(t -> new DmfTarget(actions.get(t.getControllerId()).getId(), t.getControllerId(), asSystem(t::getSecurityToken)))
.map(t -> {
final ActionProperties action = actions.get(t.getControllerId());
return new DmfTarget(
action.getId(), t.getControllerId(), asSystem(t::getSecurityToken), action.getExternalRef());
})
.toList();

// due to the fact that all targets in a batch use the same set of software modules we don't generate target-specific urls
Expand Down
7 changes: 3 additions & 4 deletions ...mf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
*/
package org.eclipse.hawkbit.amqp;

import jakarta.validation.constraints.NotNull;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
Expand All @@ -18,9 +19,6 @@
import java.util.UUID;
import java.util.function.Function;
import java.util.stream.Collectors;

import jakarta.validation.constraints.NotNull;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.hawkbit.audit.AuditLog;
import org.eclipse.hawkbit.auth.SpRole;
Expand Down Expand Up @@ -317,7 +315,8 @@ private void sendUpdateCommandToTarget(final Target target) {
final Action action = actionOptional.get();
if (action.isCancelingOrCanceled()) {
amqpMessageDispatcherService.sendCancelMessageToTarget(
target.getTenant(), target.getControllerId(), action.getId(), IpUtil.addressToUri(target.getAddress()));
target.getTenant(), target.getControllerId(), action.getId(), action.getExternalRef(),
IpUtil.addressToUri(target.getAddress()));
} else {
amqpMessageDispatcherService.sendUpdateMessageToTarget(
new ActionProperties(action), action.getTarget(), getSoftwareModulesWithMetadata(action.getDistributionSet()));
Expand Down
3 changes: 3 additions & 0 deletions ...bit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest {
private static final URI AMQP_URI = IpUtil.createAmqpUri("vHost", "mytest");
private static final String TEST_TOKEN = "testToken";
private static final String CONTROLLER_ID = "1";
private static final String EXTERNAL_REF = "external-system-reference";

private AmqpMessageDispatcherService amqpMessageDispatcherService;
private RabbitTemplate rabbitTemplate;
Expand Down Expand Up @@ -243,6 +244,7 @@ void testSendCancelRequest() {
when(action.getId()).thenReturn(1L);
when(action.getTenant()).thenReturn(TENANT);
when(action.getTarget()).thenReturn(testTarget);
when(action.getExternalRef()).thenReturn(EXTERNAL_REF);
final CancelTargetAssignmentEvent cancelTargetAssignmentDistributionSetEvent = new CancelTargetAssignmentEvent(action);
final CancelTargetAssignmentServiceEvent serviceCancelTargetAssignmentDistributionSetEvent =
new CancelTargetAssignmentServiceEvent(cancelTargetAssignmentDistributionSetEvent);
Expand Down Expand Up @@ -322,6 +324,7 @@ private void assertCancelMessage(final Message sendMessage) {
assertThat(actionId.getActionId())
.as("Action ID should be 1")
.isOne();
assertThat(actionId.getExternalRef()).isEqualTo(EXTERNAL_REF);
assertThat(sendMessage.getMessageProperties().getHeaders())
.as("The topc in the message should be a CANCEL_DOWNLOAD value")
.containsEntry(MessageHeaderKey.TOPIC, EventTopic.CANCEL_DOWNLOAD);
Expand Down
45 changes: 37 additions & 8 deletions ...mqp/src/test/java/org/eclipse/hawkbit/integration/AbstractAmqpServiceIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

import static org.assertj.core.api.Assertions.assertThat;

import com.cronutils.utils.StringUtils;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
Expand All @@ -20,8 +21,6 @@
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;

import com.cronutils.utils.StringUtils;
import org.assertj.core.api.HamcrestCondition;
import org.eclipse.hawkbit.amqp.DmfApiConfiguration;
import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
Expand Down Expand Up @@ -143,11 +142,16 @@ protected DistributionSetAssignmentResult registerTargetAndAssignDistributionSet
}

protected void assertCancelActionMessage(final Long actionId, final String controllerId) {
assertCancelActionMessage(actionId, controllerId, null);
}

protected void assertCancelActionMessage(final Long actionId, final String controllerId, final String externalRef) {
final Message replyMessage = assertReplyMessageHeader(EventTopic.CANCEL_DOWNLOAD, controllerId);

final DmfActionRequest actionUpdateStatus = (DmfActionRequest) getDmfClient().getMessageConverter()
.fromMessage(replyMessage);
assertThat(actionUpdateStatus.getActionId()).isEqualTo(actionId);
assertThat(actionUpdateStatus.getExternalRef()).isEqualTo(externalRef);
}

protected void assertDeleteMessage(final String target) {
Expand Down Expand Up @@ -186,19 +190,36 @@ protected void assertPingReplyMessage(final String correlationId) {

protected void assertDmfDownloadAndUpdateRequest(
final DmfDownloadAndUpdateRequest request, final Set<SoftwareModule> softwareModules, final String controllerId) {
assertDmfDownloadAndUpdateRequest(request, softwareModules, controllerId, null);
}

protected void assertDmfDownloadAndUpdateRequest(
final DmfDownloadAndUpdateRequest request, final Set<SoftwareModule> softwareModules,
final String controllerId, final String externalRef) {
assertSoftwareModules(softwareModules, request.getSoftwareModules());
final Target updatedTarget = waitUntilIsPresent(() -> targetManagement.findByControllerId(controllerId));
assertThat(updatedTarget).isNotNull();
assertThat(updatedTarget.getSecurityToken()).isEqualTo(request.getTargetSecurityToken());
assertThat(request.getExternalRef()).isEqualTo(externalRef);
}

protected void assertDownloadAndInstallMessage(final Set<SoftwareModule> softwareModules,
final String controllerId) {
assertAssignmentMessage(softwareModules, controllerId, EventTopic.DOWNLOAD_AND_INSTALL);
assertDownloadAndInstallMessage(softwareModules, controllerId, null);
}

protected void assertDownloadAndInstallMessage(final Set<SoftwareModule> softwareModules,
final String controllerId, final String externalRef) {
assertAssignmentMessage(softwareModules, controllerId, EventTopic.DOWNLOAD_AND_INSTALL, externalRef);
}

protected void assertDownloadMessage(final Set<SoftwareModule> dsModules, final String controllerId) {
assertAssignmentMessage(dsModules, controllerId, EventTopic.DOWNLOAD);
assertDownloadMessage(dsModules, controllerId, null);
}

protected void assertDownloadMessage(final Set<SoftwareModule> dsModules,
final String controllerId, final String externalRef) {
assertAssignmentMessage(dsModules, controllerId, EventTopic.DOWNLOAD, externalRef);
}

protected void createAndSendThingCreated(final String controllerId) {
Expand Down Expand Up @@ -401,31 +422,39 @@ protected void assertSoftwareModules(final Set<SoftwareModule> expectedSoftwareM
}

protected void assertConfirmMessage(final Set<SoftwareModule> dsModules, final String controllerId) {
assertConfirmMessage(dsModules, controllerId, null);
}

protected void assertConfirmMessage(final Set<SoftwareModule> dsModules,
final String controllerId, final String externalRef) {

final Message replyMessage = assertReplyMessageHeader(EventTopic.CONFIRM, controllerId);
assertAllTargetsCount(1);

final DmfConfirmRequest confirmRequest = (DmfConfirmRequest) getDmfClient()
.getMessageConverter().fromMessage(replyMessage);

assertConfirmRequest(confirmRequest, dsModules, controllerId);
assertConfirmRequest(confirmRequest, dsModules, controllerId, externalRef);
}

protected void assertConfirmRequest(final DmfConfirmRequest request, final Set<SoftwareModule> softwareModules, final String controllerId) {
protected void assertConfirmRequest(final DmfConfirmRequest request, final Set<SoftwareModule> softwareModules,
final String controllerId, final String externalRef) {
assertSoftwareModules(softwareModules, request.getSoftwareModules());
final Target updatedTarget = waitUntilIsPresent(() -> targetManagement.findByControllerId(controllerId));
assertThat(updatedTarget).isNotNull();
assertThat(updatedTarget.getSecurityToken()).isEqualTo(request.getTargetSecurityToken());
assertThat(request.getExternalRef()).isEqualTo(externalRef);
}

private void assertAssignmentMessage(final Set<SoftwareModule> dsModules, final String controllerId, final EventTopic topic) {
private void assertAssignmentMessage(final Set<SoftwareModule> dsModules, final String controllerId,
final EventTopic topic, final String externalRef) {
final Message replyMessage = assertReplyMessageHeader(topic, controllerId);
assertAllTargetsCount(1);

final DmfDownloadAndUpdateRequest downloadAndUpdateRequest = (DmfDownloadAndUpdateRequest) getDmfClient()
.getMessageConverter().fromMessage(replyMessage);

assertDmfDownloadAndUpdateRequest(downloadAndUpdateRequest, dsModules, controllerId);
assertDmfDownloadAndUpdateRequest(downloadAndUpdateRequest, dsModules, controllerId, externalRef);
}

private void registerAndAssertTargetWithExistingTenant(
Expand Down
Loading
Loading