SDK DMF Support Initial draft (#1706)

Signed-off-by: Marinov Avgustin <Avgustin.Marinov@bosch.com>

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/ArtifactHandler.java

@@ -0,0 +1,73 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf;
+
+import java.nio.file.Path;
+import java.util.Optional;
+
+/**
+ * Artifact handler provide plug-in endpoint allowing for customization of the artifact processing.
+ * For instance, it could save downloaded files in some location and on successful finish could
+ * apply them.
+ */
+public interface ArtifactHandler {
+
+    ArtifactHandler SKIP = url -> DownloadHandler.SKIP;
+
+    DownloadHandler getDownloadHandler(final String url);
+
+    interface DownloadHandler {
+
+        enum Status {
+            SUCCESS, ERROR
+        }
+
+        DownloadHandler SKIP = new DownloadHandler() {
+            @Override
+            public void read(byte[] buff, int off, int len) {
+                // skip
+            }
+
+            @Override
+            public void finished(Status status) {
+                // skip
+            }
+
+            @Override
+            public Optional<Path> download() {
+                return Optional.empty();
+            }
+        };
+
+        /**
+         * Called on every read chunk of data
+         *
+         * @param buff buffer
+         * @param off offset
+         * @param len read bytes
+         */
+        void read(byte[] buff, int off, int len);
+
+        /**
+         * Called when the download has finished. It could have finished with error in case of network problems,
+         * invalid hashes or size. In case of success the hashes and size are already checked and valid.
+         *
+         * @param status finish status. On error the download shall be discarded and related resources shall be released
+         */
+        void finished(Status status);
+
+        /**
+         * Return download path if existing
+         *
+         * @return the path to the download
+         */
+        Optional<Path> download();
+    }
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DeviceManagement.java

@@ -0,0 +1,67 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf;
+
+import org.springframework.stereotype.Service;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * An in-memory simulated device management to hold the multi-tenants and controller twins in
+ * memory and be able to retrieve them again.
+ */
+@Service
+public class DeviceManagement {
+
+    private final Set<String> tenants = new HashSet<>();
+
+    private final Map<DeviceKey, DmfController> controllers = new ConcurrentHashMap<>();
+
+    public DmfController add(final DmfController controller) {
+        controllers.put(new DeviceKey(controller.getTenantId().toLowerCase(), controller.getControllerId()), controller);
+        tenants.add(controller.getTenantId().toLowerCase());
+        return controller;
+    }
+
+    public Set<String> getTenants() {
+        return tenants;
+    }
+
+    public Collection<DmfController> getControllers() {
+        return controllers.values();
+    }
+
+    public Optional<DmfController> getController(final String tenantId, final String controllerId) {
+        return Optional.ofNullable(controllers.get(new DeviceKey(tenantId.toLowerCase(), controllerId)));
+    }
+
+    public void remove(final String tenant, final String id) {
+        final DmfController controller = controllers.remove(new DeviceKey(tenant.toLowerCase(), id));
+        if (controller != null) {
+            controller.stop();
+        }
+    }
+
+    /**
+     * Clears all stored devices.
+     */
+    public void destroy() {
+        controllers.values().forEach(DmfController::stop);
+        controllers.clear();
+        tenants.clear();
+    }
+
+    private record DeviceKey(String tenant, String id) {}
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfController.java

@@ -0,0 +1,92 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf;
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.hawkbit.dmf.amqp.api.EventTopic;
+import org.eclipse.hawkbit.dmf.json.model.DmfDownloadAndUpdateRequest;
+import org.eclipse.hawkbit.sdk.Controller;
+import org.eclipse.hawkbit.sdk.Tenant;
+import org.eclipse.hawkbit.sdk.dmf.amqp.DmfSenderService;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * Class representing DMF device twin connecting to hawkBit via DMF.
+ */
+@Slf4j
+@Getter
+public class DmfController {
+
+    private static final String LOG_PREFIX = "[{}:{}] ";
+
+    private static final String DEPLOYMENT_BASE_LINK = "deploymentBase";
+    private static final String CONFIRMATION_BASE_LINK = "confirmationBase";
+
+    private final String tenantId;
+    private final String controllerId;
+    private final UpdateHandler updateHandler;
+    private final DmfSenderService dmfSenderService;
+
+    // configuration
+    private final boolean downloadAuthenticationEnabled;
+
+    @Getter @Setter
+    private volatile Map<String, String> attributes = Collections.emptyMap();
+
+    private volatile Long currentActionId;
+
+    private volatile Long lastActionId;
+
+    /**
+     * Creates a new device instance.
+     *
+     * @param tenant the tenant of the device belongs to
+     * @param controller the controller
+     */
+    public DmfController(
+            final Tenant tenant, final Controller controller,
+            final UpdateHandler updateHandler,
+            final DmfSenderService dmfSenderService) {
+        this.tenantId = tenant.getTenantId();
+        downloadAuthenticationEnabled = tenant.isDownloadAuthenticationEnabled();
+        this.controllerId = controller.getControllerId();
+        this.updateHandler = updateHandler == null ? UpdateHandler.SKIP : updateHandler;
+        this.dmfSenderService = dmfSenderService;
+    }
+
+    public void connect() {
+        log.debug(LOG_PREFIX + "Connecting/Polling ...", getTenantId(), getControllerId());
+        dmfSenderService.createOrUpdateThing(getTenantId(), getControllerId());
+        log.debug(LOG_PREFIX + "Done. Create thing sent.", getTenantId(), getControllerId());
+    }
+
+    public void poll() {
+        log.debug(LOG_PREFIX + "Polling ..", getTenantId(), getControllerId());
+        dmfSenderService.createOrUpdateThing(getTenantId(), getControllerId());
+        log.debug(LOG_PREFIX + "Done. Create thing sent.", getTenantId(), getControllerId());
+    }
+
+    public void stop() {
+        lastActionId = null;
+        currentActionId = null;
+    }
+
+    public void processUpdate(final EventTopic actionType, final DmfDownloadAndUpdateRequest updateRequest) {
+        updateHandler.getUpdateProcessor(this, actionType, updateRequest);
+    }
+
+    public void sendFeedback(final UpdateStatus updateStatus) {
+        dmfSenderService.sendFeedback(tenantId, currentActionId, updateStatus);
+    }
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfProperties.java

@@ -0,0 +1,62 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf;
+
+import lombok.Data;
+import lombok.ToString;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+/**
+ * Bean which holds the necessary properties for configuring the AMQP connection.
+ */
+@Data
+@ToString
+@Component
+@ConfigurationProperties(DmfProperties.CONFIGURATION_PREFIX)
+public class DmfProperties {
+
+    /**
+     * The prefix for this configuration.
+     */
+    public static final String CONFIGURATION_PREFIX = "hawkbit.sdk.dmf.amqp";
+
+    /**
+     * The property string of ~.amqp.enabled
+     */
+    public static final String CONFIGURATION_ENABLED_PROPERTY = CONFIGURATION_PREFIX + ".enabled";
+
+    /**
+     * Indicates if the AMQP interface is enabled for the device simulator.
+     */
+    private boolean enabled;
+
+    /**
+     * Set to true for the simulator run DMF health check.
+     */
+    private boolean healthCheckEnabled;
+
+    /**
+     * Queue for receiving DMF messages from update server.
+     */
+    private String receiverConnectorQueueFromSp = "sdk_receiver";
+
+    /**
+     * Exchange for sending DMF messages to update server.
+     */
+    private String senderForSpExchange = "sdk.replyTo";
+
+    /**
+     * Message time to live (ttl) for the deadletter queue. Default ttl is 1 hour.
+     */
+    private int deadLetterTtl = 60_000;
+
+    private String customVhost;
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfSDKConfiguration.java

@@ -0,0 +1,120 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf;
+
+import java.time.Duration;
+import java.util.Map;
+
+import org.eclipse.hawkbit.sdk.dmf.amqp.DmfReceiverService;
+import org.eclipse.hawkbit.sdk.dmf.amqp.DmfSenderService;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueBuilder;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * The spring AMQP configuration to use a AMQP for communication with SP update server.
+ */
+@Configuration
+@EnableConfigurationProperties(DmfProperties.class)
+@ConditionalOnProperty(prefix = DmfProperties.CONFIGURATION_PREFIX, name = "enabled", matchIfMissing = true)
+public class DmfSDKConfiguration {
+
+    @Bean
+    DeviceManagement deviceManagement() {
+        return new DeviceManagement();
+    }
+
+    @Bean
+    DmfSenderService dmfSenderService(
+            final RabbitTemplate rabbitTemplate,
+            final DmfProperties dmfProperties) {
+        return new DmfSenderService(rabbitTemplate, dmfProperties);
+    }
+
+    @Bean
+    DmfReceiverService dmfReceiverService(
+            final RabbitTemplate rabbitTemplate,
+            final DmfSenderService dmfSenderService,
+            final DeviceManagement deviceManagement,
+            final DmfProperties dmfProperties) {
+        return new DmfReceiverService(rabbitTemplate, dmfSenderService, deviceManagement, dmfProperties);
+    }
+
+    @Bean
+    public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
+        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+
+        // It is necessary to define rabbitTemplate as a Bean and set
+        // Jackson2JsonMessageConverter explicitly here in order to convert only
+        // OUTCOMING messages to json. In case of INCOMING messages,
+        // Jackson2JsonMessageConverter can not handle messages with NULL
+        // payload (e.g. REQUEST_ATTRIBUTES_UPDATE), so the
+        // SimpleMessageConverter is used instead per default.
+        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
+
+        return rabbitTemplate;
+    }
+
+    /**
+     * Creates the receiver queue from update server for receiving message from update server.
+     */
+    @Bean
+    Queue receiverConnectorQueueFromHawkBit(final DmfProperties dmfProperties) {
+        return QueueBuilder.nonDurable(dmfProperties.getReceiverConnectorQueueFromSp()).autoDelete()
+                .withArguments(Map.of(
+                        "x-message-ttl", Duration.ofDays(1).toMillis(),
+                        "x-max-length", 100_000))
+                .build();
+    }
+
+    /**
+     * Creates the receiver exchange for sending messages to update server.
+     */
+    @Bean
+    FanoutExchange exchangeQueueToConnector(final DmfProperties dmfProperties) {
+        return new FanoutExchange(dmfProperties.getSenderForSpExchange(), false, true);
+    }
+
+    /**
+     * Create the Binding
+     *
+     * @return the binding and create the queue and exchange
+     */
+    @Bean
+    Binding bindReceiverQueueToSpExchange(final DmfProperties dmfProperties) {
+        return BindingBuilder.bind(receiverConnectorQueueFromHawkBit(dmfProperties))
+                .to(exchangeQueueToConnector(dmfProperties));
+    }
+
+    @Configuration
+    @ConditionalOnProperty(prefix = DmfProperties.CONFIGURATION_PREFIX, name = "customVhost")
+    protected static class CachingConnectionFactoryInitializer {
+
+        CachingConnectionFactoryInitializer(
+                final CachingConnectionFactory connectionFactory, final DmfProperties dmfProperties) {
+            connectionFactory.setVirtualHost(dmfProperties.getCustomVhost());
+        }
+    }
+
+    @ConditionalOnProperty(prefix = DmfProperties.CONFIGURATION_PREFIX, name = "healthCheckEnabled")
+    HealthService healthService(final DeviceManagement deviceManagement, final DmfSenderService dmfSenderService) {
+        return new HealthService(deviceManagement, dmfSenderService);
+    }
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/HealthService.java

@@ -0,0 +1,61 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf;
+
+import org.eclipse.hawkbit.sdk.dmf.amqp.DmfSenderService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.Message;
+import org.springframework.scheduling.annotation.Scheduled;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+
+/**
+ * Handle all incoming Messages from hawkBit update server.
+ */
+class HealthService {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(HealthService.class);
+
+    private final DeviceManagement deviceManagement;
+    private final DmfSenderService dmfSenderService;
+
+    private final Set<String> openPings = Collections.synchronizedSet(new HashSet<>());
+
+    HealthService(final DeviceManagement deviceManagement, final DmfSenderService dmfSenderService) {
+        this.deviceManagement = deviceManagement;
+        this.dmfSenderService = dmfSenderService;
+    }
+
+    @Scheduled(fixedDelay = 5_000, initialDelay = 5_000)
+    void checkDmfHealth() {
+        if (openPings.size() > 5) {
+            LOGGER.error("Currently {} open pings! DMF does not seem to be reachable.", openPings.size());
+        } else {
+            LOGGER.debug("Currently {} open pings", openPings.size());
+        }
+
+        deviceManagement.getTenants().forEach(tenant -> {
+            final String correlationId = UUID.randomUUID().toString();
+            openPings.add(correlationId);
+            LOGGER.debug("Ping tenant {} with correlationId {}", tenant, correlationId);
+            dmfSenderService.ping(tenant, correlationId, this::pingReceived);
+        });
+    }
+
+    void pingReceived(final String correlationId, final Message message) {
+        if (!openPings.remove(correlationId)) {
+            LOGGER.error("Unknown PING_RESPONSE received for correlationId: {}.", correlationId);
+        }
+    }
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/UpdateHandler.java

@@ -0,0 +1,391 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf;
+
+import com.google.common.io.BaseEncoding;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
+import org.apache.hc.client5.http.ssl.SSLConnectionSocketFactory;
+import org.apache.hc.core5.ssl.SSLContextBuilder;
+import org.eclipse.hawkbit.dmf.amqp.api.EventTopic;
+import org.eclipse.hawkbit.dmf.json.model.DmfActionStatus;
+import org.eclipse.hawkbit.dmf.json.model.DmfArtifact;
+import org.eclipse.hawkbit.dmf.json.model.DmfArtifactHash;
+import org.eclipse.hawkbit.dmf.json.model.DmfDownloadAndUpdateRequest;
+import org.eclipse.hawkbit.dmf.json.model.DmfSoftwareModule;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyManagementException;
+import java.security.KeyStoreException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Update handler provide plug-in endpoint allowing for customization of the update processing.
+ */
+public interface UpdateHandler {
+
+    UpdateHandler SKIP = (controller, updateType, modules) ->
+            new UpdateProcessor(controller, updateType, modules, ArtifactHandler.SKIP);
+
+    /**
+     * Creates an update processor for a single software update
+     *
+     * @param controller controller instance
+     * @param eventTopic the event topic for update
+     * @param updateRequest the info for th requested update
+     * @return the update processor
+     */
+    UpdateProcessor getUpdateProcessor(
+            final DmfController controller,
+            final EventTopic eventTopic, final DmfDownloadAndUpdateRequest updateRequest);
+
+    @Slf4j
+    class UpdateProcessor implements Runnable {
+
+        private static final String LOG_PREFIX = "[{}:{}] ";
+
+        private static final String DOWNLOAD_LOG_MESSAGE = "Download ";
+        private static final String EXPECTED = "(Expected: ";
+        private static final String BUT_GOT_LOG_MESSAGE = " but got: ";
+        private static final int MINIMUM_TOKEN_LENGTH_FOR_HINT = 6;
+
+        private final DmfController dmfController;
+
+        private final DmfDownloadAndUpdateRequest updateRequest;
+        private final EventTopic eventTopic;
+
+        private final ArtifactHandler artifactHandler;
+        protected final Map<String, Path> downloads = new HashMap<>();
+
+        public UpdateProcessor(
+                final DmfController dmfController,
+                final EventTopic eventTopic, final DmfDownloadAndUpdateRequest updateRequest,
+                final ArtifactHandler artifactHandler) {
+            this.dmfController = dmfController;
+
+            this.eventTopic = eventTopic;
+            this.updateRequest = updateRequest;
+
+            this.artifactHandler = artifactHandler;
+        }
+
+        @Override
+        public void run() {
+            dmfController.sendFeedback(new UpdateStatus(DmfActionStatus.RUNNING, List.of("Update begin ...")));
+
+            final List<DmfSoftwareModule> modules = updateRequest.getSoftwareModules();
+            if (!CollectionUtils.isEmpty(modules)) {
+                try {
+                    final UpdateStatus updateStatus = download();
+                    dmfController.sendFeedback(updateStatus);
+                    if (updateStatus.status() == DmfActionStatus.ERROR) {
+                        return;
+                    } else {
+                        if (eventTopic != EventTopic.DOWNLOAD) {
+                            dmfController.sendFeedback(update());
+                        }
+                    }
+                } finally {
+                    cleanup();
+                }
+            }
+
+            if (eventTopic == EventTopic.DOWNLOAD) {
+                dmfController.sendFeedback(
+                        new UpdateStatus(DmfActionStatus.FINISHED, List.of("Update (download-only) completed.")));
+            }
+        }
+
+        /**
+         * Extension point. An overriding implementation could completely skip the default download and provide its own.
+         * By contract, it shall fill up {@link #downloads};
+         *
+         * @return the status of the download
+         */
+        protected UpdateStatus download() {
+            final List<DmfSoftwareModule> modules = updateRequest.getSoftwareModules();
+            dmfController.sendFeedback(
+                    new UpdateStatus(
+                            DmfActionStatus.DOWNLOAD,
+                            modules.stream().flatMap(mod -> mod.getArtifacts().stream())
+                                    .map(art -> "Download start for: " + art.getFilename() +
+                                            " with size " + art.getSize() +
+                                            " and hashes " + art.getHashes() + " ...")
+                                    .collect(Collectors.toList())));
+
+            log.info(LOG_PREFIX + "Start download", dmfController.getTenantId(), dmfController.getControllerId());
+
+            final List<UpdateStatus> updateStatusList = new ArrayList<>();
+            modules.forEach(module -> module.getArtifacts().forEach(artifact -> {
+                if (dmfController.isDownloadAuthenticationEnabled()) {
+                    handleArtifact(
+                            updateRequest.getTargetSecurityToken(),
+                            updateStatusList, artifact);
+                } else {
+                    handleArtifact(null, updateStatusList, artifact);
+                }
+            }));
+
+            log.info(LOG_PREFIX + "Download complete.", dmfController.getTenantId(), dmfController.getControllerId());
+
+            final List<String> messages = new LinkedList<>();
+            messages.add("Download complete.");
+            updateStatusList.forEach(download -> messages.addAll(download.messages()));
+            return new UpdateStatus(
+                    updateStatusList.stream().anyMatch(status -> status.status() == DmfActionStatus.ERROR) ?
+                            DmfActionStatus.ERROR : DmfActionStatus.DOWNLOADED,
+                    messages);
+        }
+
+        /**
+         * Extension point. Called after all artifacts has been successfully downloaded. An overriding implementation
+         * may get the {@link #downloads} map and apply them
+         */
+        protected UpdateStatus update() {
+            log.info(LOG_PREFIX + "Updated", dmfController.getTenantId(), dmfController.getControllerId());
+            return new UpdateStatus(DmfActionStatus.FINISHED, List.of("Update complete."));
+        }
+
+        /**
+         * Extension point. Called after download and update has been finished. By default, it deletes all downloaded
+         * files (if any).
+         */
+        protected void cleanup() {
+            downloads.values().forEach(path -> {
+                try {
+                    Files.delete(path);
+                } catch (final IOException e) {
+                    log.warn(LOG_PREFIX + "Failed to cleanup {}",
+                            dmfController.getTenantId(), dmfController.getControllerId(),
+                            path.toFile().getAbsolutePath(), e);
+                }
+            });
+            log.debug(LOG_PREFIX + "Cleaned up", dmfController.getTenantId(), dmfController.getControllerId());
+        }
+
+        private void handleArtifact(
+                final String targetToken,
+                final List<UpdateStatus> status, final DmfArtifact artifact) {
+            if (artifact.getUrls().containsKey("HTTPS")) {
+                status.add(downloadUrl(artifact.getUrls().get("HTTPS"), targetToken,
+                        artifact.getHashes(), artifact.getSize()));
+            } else if (artifact.getUrls().containsKey("HTTP")) {
+                status.add(downloadUrl(artifact.getUrls().get("HTTP"), targetToken,
+                        artifact.getHashes(), artifact.getSize()));
+            }
+        }
+
+        private UpdateStatus downloadUrl(
+                final String url, final String targetToken,
+                final DmfArtifactHash hash, final long size) {
+            if (log.isDebugEnabled()) {
+                log.debug(LOG_PREFIX + "Downloading {} with token {}, expected hash {} and size {}",
+                        dmfController.getTenantId(), dmfController.getControllerId(), url,
+                        hideTokenDetails(targetToken), hash, size);
+            }
+
+            try {
+                return readAndCheckDownloadUrl(url, targetToken, hash, size);
+            } catch (final IOException | KeyManagementException | NoSuchAlgorithmException | KeyStoreException e) {
+                log.error(LOG_PREFIX + "Failed to download {}",
+                        dmfController.getTenantId(), dmfController.getControllerId(), url, e);
+                return new UpdateStatus(
+                        DmfActionStatus.ERROR,
+                        List.of("Failed to download " + url + ": " + e.getMessage()));
+            }
+        }
+
+        private UpdateStatus readAndCheckDownloadUrl(final String url,
+                final String targetToken, final DmfArtifactHash hash, final long size)
+                throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException, IOException {
+            final Validator sizeValidator = sizeValidator(size);
+            final Validator hashValidator = hashValidator(hash);
+            final ArtifactHandler.DownloadHandler downloadHandler = artifactHandler.getDownloadHandler(url);
+
+            try (final CloseableHttpClient httpclient = createHttpClientThatAcceptsAllServerCerts()) {
+                final HttpGet request = new HttpGet(url);
+                if (StringUtils.hasLength(targetToken)) {
+                    request.addHeader(HttpHeaders.AUTHORIZATION, "TargetToken " + targetToken);
+                }
+
+                return httpclient.execute(request, response -> {
+                    try {
+                        if (response.getCode() != HttpStatus.OK.value()) {
+                            throw new IllegalStateException("Unexpected status code: " + response.getCode());
+                        }
+
+                        if (response.getEntity().getContentLength() != size) {
+                            throw new IllegalArgumentException("Wrong content length " + EXPECTED + size + BUT_GOT_LOG_MESSAGE + response.getEntity()
+                                    .getContentLength() + ")!");
+                        }
+
+                        final byte[] buff = new byte[32 * 1024];
+                        try (final InputStream is = response.getEntity().getContent()) {
+                            for (int read; (read = is.read(buff)) != -1; ) {
+                                sizeValidator.read(buff, read);
+                                hashValidator.read(buff, read);
+                                downloadHandler.read(buff, 0, read);
+                            }
+                        }
+                        sizeValidator.validate();
+                        hashValidator.validate();
+
+                        final String message = "Downloaded " + url + " (" + size + " bytes)";
+                        log.debug(LOG_PREFIX + message, dmfController.getTenantId(), dmfController.getControllerId());
+                        downloadHandler.finished(ArtifactHandler.DownloadHandler.Status.SUCCESS);
+                        downloadHandler.download().ifPresent(path -> downloads.put(url, path));
+                        return new UpdateStatus(DmfActionStatus.FINISHED, List.of(message));
+                    } catch (final Exception e) {
+                        final String message = e.getMessage();
+                        if (log.isTraceEnabled()) {
+                            log.error(LOG_PREFIX + DOWNLOAD_LOG_MESSAGE + url + " failed: " + message,
+                                    dmfController.getTenantId(), dmfController.getControllerId(), e);
+                        } else {
+                            log.error(LOG_PREFIX + DOWNLOAD_LOG_MESSAGE + url + " failed: " + message,
+                                    dmfController.getTenantId(), dmfController.getControllerId());
+                        }
+                        downloadHandler.finished(ArtifactHandler.DownloadHandler.Status.ERROR);
+                        return new UpdateStatus(DmfActionStatus.ERROR, List.of(message));
+                    }
+                });
+            }
+        }
+
+        private static String hideTokenDetails(final String targetToken) {
+            if (targetToken == null) {
+                return "<NULL!>";
+            }
+
+            if (targetToken.isEmpty()) {
+                return "<EMPTY!>";
+            }
+
+            if (targetToken.length() <= MINIMUM_TOKEN_LENGTH_FOR_HINT) {
+                return "***";
+            }
+
+            return targetToken.substring(0, 2) + "***" + targetToken.substring(targetToken.length() - 2);
+        }
+
+        private static CloseableHttpClient createHttpClientThatAcceptsAllServerCerts()
+                throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
+            return HttpClients
+                    .custom()
+                    .setConnectionManager(
+                            PoolingHttpClientConnectionManagerBuilder.create()
+                                    .setSSLSocketFactory(
+                                            new SSLConnectionSocketFactory(
+                                                    SSLContextBuilder
+                                                            .create()
+                                                            .loadTrustMaterial(null, (chain, authType) -> true)
+                                                            .build()))
+                                    .build()
+                    )
+                    .build();
+        }
+
+
+        private interface Validator {
+
+            void read(final byte[] buff, final int len);
+
+            void validate();
+        }
+
+        private static Validator sizeValidator(final long size) {
+            return new Validator() {
+
+                private int read;
+
+                @Override
+                public void read(final byte[] buff, final int len) {
+                    read += len;
+                    if (read > size) {
+                        throw new SecurityException("Size mismatch: read more " + EXPECTED + size + BUT_GOT_LOG_MESSAGE + read + ")!");
+                    }
+                }
+
+                @Override
+                public void validate() {
+                    if (read != size) {
+                        throw new SecurityException("Size mismatch " + EXPECTED + size + BUT_GOT_LOG_MESSAGE + read + ")!");
+                    }
+                }
+            };
+        }
+
+        private static Validator hashValidator(final DmfArtifactHash hash) throws NoSuchAlgorithmException {
+
+            class HashValidator {
+
+                private final String expected;
+                private final MessageDigest messageDigest;
+
+                private HashValidator(final String expected, final MessageDigest messageDigest) {
+                    this.expected = expected;
+                    this.messageDigest = messageDigest;
+                }
+
+                private void update(final byte[] buff, final int len) {
+                    messageDigest.update(buff, 0, len);
+                }
+
+                private void check() {
+                    final String actual = BaseEncoding.base16().lowerCase().encode(messageDigest.digest());
+                    if (!actual.equals(expected)) {
+                        throw new SecurityException(
+                                messageDigest.getAlgorithm() + " hash mismatch " + EXPECTED + expected + BUT_GOT_LOG_MESSAGE + actual + ")!");
+                    }
+                }
+            }
+
+            final List<HashValidator> hashValidators = new ArrayList<>(2);
+            if (!ObjectUtils.isEmpty(hash.getSha1())) {
+                hashValidators.add(new HashValidator(hash.getSha1(), MessageDigest.getInstance("SHA-1")));
+            }
+            if (!ObjectUtils.isEmpty(hash.getMd5())) {
+                hashValidators.add(new HashValidator(hash.getMd5(), MessageDigest.getInstance("MD5")));
+            }
+            if (hashValidators.isEmpty()) {
+                throw new SecurityException("No hashes in " + hash + "!");
+            }
+
+            return new Validator() {
+                @Override
+                public void read(final byte[] buff, final int len) {
+                    hashValidators.forEach(hashValidator -> hashValidator.update(buff, len));
+                }
+
+                @Override
+                public void validate() {
+                    hashValidators.forEach(HashValidator::check);
+                }
+            };
+        }
+    }
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/UpdateInfo.java

@@ -0,0 +1,46 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf;
+
+import lombok.Data;
+import lombok.ToString;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * Object for holding attributes for a simulated update for the device simulator.
+ */
+@Data
+@ToString
+public class UpdateInfo implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    private final String tenant;
+    private final String thingId;
+    private final Long actionId;
+
+    private final transient LocalDateTime startCacheTime;
+
+    /**
+     * @param tenant the tenant for this thing and for this simulated update
+     * @param thingId the thing id that this simulated update correlates to
+     * @param actionId the id of the action related to this simulated update
+     */
+    public UpdateInfo(final String tenant, final String thingId, final Long actionId) {
+        this.tenant = tenant;
+        this.thingId = thingId;
+        this.actionId = actionId;
+        this.startCacheTime = LocalDateTime.now();
+    }
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/UpdateStatus.java

@@ -0,0 +1,16 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf;
+
+import org.eclipse.hawkbit.dmf.json.model.DmfActionStatus;
+
+import java.util.List;
+
+public record UpdateStatus(DmfActionStatus status, List<String> messages) {}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/DmfReceiverService.java

@@ -0,0 +1,226 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf.amqp;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.eclipse.hawkbit.dmf.amqp.api.EventTopic;
+import org.eclipse.hawkbit.dmf.amqp.api.MessageHeaderKey;
+import org.eclipse.hawkbit.dmf.amqp.api.MessageType;
+import org.eclipse.hawkbit.dmf.json.model.DmfActionRequest;
+import org.eclipse.hawkbit.dmf.json.model.DmfDownloadAndUpdateRequest;
+import org.eclipse.hawkbit.dmf.json.model.DmfMultiActionRequest;
+import org.eclipse.hawkbit.dmf.json.model.DmfUpdateMode;
+import org.eclipse.hawkbit.sdk.dmf.DeviceManagement;
+import org.eclipse.hawkbit.sdk.dmf.DmfProperties;
+import org.eclipse.hawkbit.sdk.dmf.UpdateInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.AmqpRejectAndDontRequeueException;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.messaging.handler.annotation.Header;
+
+/**
+ * Handle all incoming Messages from hawkBit update server.
+ *
+ */
+public class DmfReceiverService extends MessageService {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DmfReceiverService.class);
+
+    private static final String REGEX_EXTRACT_ACTION_ID = "[^0-9]";
+
+    private final DmfSenderService dmfSenderService;
+    private final DeviceManagement deviceManagement;
+
+    private final Set<Long> openActions = Collections.synchronizedSet(new HashSet<>());
+
+    DmfReceiverService(final RabbitTemplate rabbitTemplate, final DmfSenderService dmfSenderService,
+            final DeviceManagement deviceManagement, final DmfProperties dmfProperties) {
+        super(rabbitTemplate, dmfProperties);
+        this.dmfSenderService = dmfSenderService;
+        this.deviceManagement = deviceManagement;
+    }
+
+    @RabbitListener(queues = "${" + DmfProperties.CONFIGURATION_PREFIX + ".receiverConnectorQueueFromSp}")
+    public void receiveMessageSp(
+            @Header(MessageHeaderKey.TENANT) final String tenant,
+            @Header(name = MessageHeaderKey.THING_ID, required = false) final String controllerId,
+            @Header(MessageHeaderKey.TYPE) final String type,
+            final Message message) {
+        switch (MessageType.valueOf(type)) {
+            case EVENT: {
+                checkContentTypeJson(message);
+                handleEventMessage(message, controllerId);
+                break;
+            }
+            case THING_DELETED: {
+                checkContentTypeJson(message);
+                deviceManagement.remove(tenant, controllerId);
+                break;
+            }
+            case PING_RESPONSE: {
+                final String correlationId = message.getMessageProperties().getCorrelationId();
+                if (LOGGER.isDebugEnabled()) {
+                    LOGGER.debug("Got ping response from tenant {} with correlationId {} with timestamp {}", tenant,
+                            correlationId, new String(message.getBody(), StandardCharsets.UTF_8));
+                }
+                dmfSenderService.pingResponse(controllerId, message);
+                break;
+            }
+            default: {
+                LOGGER.info("No valid message type property.");
+            }
+        }
+    }
+
+    private void handleEventMessage(final Message message, final String thingId) {
+        final Object eventHeader = message.getMessageProperties().getHeaders().get(MessageHeaderKey.TOPIC);
+        if (eventHeader == null) {
+            LOGGER.error("Error \"Event Topic is not set\" reported by message {}", message.getMessageProperties().getMessageId());
+            throw new IllegalArgumentException("Event Topic is not set");
+        }
+
+        // Exception squid:S2259 - Checked before
+        @SuppressWarnings({ "squid:S2259" })
+        final EventTopic eventTopic = EventTopic.valueOf(eventHeader.toString());
+        switch (eventTopic) {
+            case CONFIRM:
+                handleConfirmation(message, thingId);
+                break;
+            case DOWNLOAD_AND_INSTALL:
+            case DOWNLOAD:
+                handleUpdateProcess(message, thingId, eventTopic);
+                break;
+            case CANCEL_DOWNLOAD:
+                handleCancelDownloadAction(message, thingId);
+                break;
+            case REQUEST_ATTRIBUTES_UPDATE:
+                handleAttributeUpdateRequest(message, thingId);
+                break;
+            case MULTI_ACTION:
+                handleMultiActionRequest(message, thingId);
+                break;
+            default:
+                LOGGER.info("No valid event property: {}", eventTopic);
+                break;
+        }
+    }
+
+    private void handleConfirmation(final Message message, final String controllerId) {
+        LOGGER.warn("Handle confirmed received for {}! Skip it!", controllerId);
+    }
+
+    private long extractActionIdFrom(final Message message) {
+        final String messageAsString = message.toString();
+        final String requiredMessageContent = messageAsString
+                .substring(messageAsString.indexOf('{') + 1, messageAsString.indexOf('}'));
+        final String[] splitMessageContent = requiredMessageContent.split(",");
+        return Long.parseLong(splitMessageContent[0].replaceAll(REGEX_EXTRACT_ACTION_ID, ""));
+    }
+
+    private void handleMultiActionRequest(final Message message, final String controllerId) {
+        final DmfMultiActionRequest multiActionRequest = convertMessage(message, DmfMultiActionRequest.class);
+        final String tenant = getTenant(message);
+        final DmfMultiActionRequest.DmfMultiActionElement actionElement = multiActionRequest.getElements().get(0);
+
+        final EventTopic eventTopic = actionElement.getTopic();
+        final DmfActionRequest action = actionElement.getAction();
+        final long actionId = action.getActionId();
+
+        if (openActions.contains(actionId)) {
+            return;
+        }
+
+        openActions.add(actionId);
+
+        switch (eventTopic) {
+        case DOWNLOAD:
+        case DOWNLOAD_AND_INSTALL:
+            if (action instanceof DmfDownloadAndUpdateRequest) {
+                processUpdate(tenant, controllerId, eventTopic, (DmfDownloadAndUpdateRequest) action);
+            }
+            break;
+        case CANCEL_DOWNLOAD:
+            processCancelDownloadAction(controllerId, tenant, action.getActionId());
+            break;
+        default:
+            openActions.remove(actionId);
+            LOGGER.info("No valid event property in MULTI_ACTION.");
+            break;
+        }
+    }
+
+    private void handleAttributeUpdateRequest(final Message message, final String controllerId) {
+        final String tenantId = getTenant(message);
+        deviceManagement.getController(tenantId, controllerId).ifPresent(controller ->
+            dmfSenderService.updateAttributes(tenantId, controllerId, DmfUpdateMode.MERGE, controller.getAttributes()));
+    }
+
+    private static String getTenant(final Message message) {
+        final MessageProperties messageProperties = message.getMessageProperties();
+        final Map<String, Object> headers = messageProperties.getHeaders();
+        return (String) headers.get(MessageHeaderKey.TENANT);
+    }
+
+    private void handleCancelDownloadAction(final Message message, final String thingId) {
+        final String tenant = getTenant(message);
+        final Long actionId = extractActionIdFrom(message);
+
+        processCancelDownloadAction(thingId, tenant, actionId);
+    }
+
+    private void processCancelDownloadAction(final String thingId, final String tenant, final Long actionId) {
+        final UpdateInfo update = new UpdateInfo(tenant, thingId, actionId);
+        dmfSenderService.finishUpdateProcess(update, Collections.singletonList("Simulation canceled"));
+        openActions.remove(actionId);
+    }
+
+    private void handleUpdateProcess(final Message message, final String controllerId, final EventTopic actionType) {
+        final String tenant = getTenant(message);
+        final DmfDownloadAndUpdateRequest downloadAndUpdateRequest = convertMessage(message,
+                DmfDownloadAndUpdateRequest.class);
+        processUpdate(tenant, controllerId, actionType, downloadAndUpdateRequest);
+    }
+
+    private void processUpdate(final String tenant, final String thingId, final EventTopic actionType, final DmfDownloadAndUpdateRequest updateRequest) {
+        deviceManagement.getController(tenant, thingId).ifPresent(controller ->
+            controller.processUpdate(actionType, updateRequest));
+    }
+
+    /**
+     * Method to validate if content type is set in the message properties.
+     *
+     * @param message
+     *            the message to get validated
+     */
+    private static void checkContentTypeJson(final Message message) {
+        if (message.getBody().length == 0) {
+            return;
+        }
+        final MessageProperties messageProperties = message.getMessageProperties();
+        final String headerContentType = (String) messageProperties.getHeaders().get("content-type");
+        if (null != headerContentType) {
+            messageProperties.setContentType(headerContentType);
+        }
+        final String contentType = messageProperties.getContentType();
+        if (contentType != null && contentType.contains("json")) {
+            return;
+        }
+        throw new AmqpRejectAndDontRequeueException("Content-Type is not JSON compatible");
+    }
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/DmfSenderService.java

@@ -0,0 +1,181 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf.amqp;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiConsumer;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
+import org.eclipse.hawkbit.dmf.amqp.api.EventTopic;
+import org.eclipse.hawkbit.dmf.amqp.api.MessageHeaderKey;
+import org.eclipse.hawkbit.dmf.amqp.api.MessageType;
+import org.eclipse.hawkbit.dmf.json.model.DmfActionStatus;
+import org.eclipse.hawkbit.dmf.json.model.DmfActionUpdateStatus;
+import org.eclipse.hawkbit.dmf.json.model.DmfAttributeUpdate;
+import org.eclipse.hawkbit.dmf.json.model.DmfUpdateMode;
+import org.eclipse.hawkbit.sdk.dmf.DmfProperties;
+import org.eclipse.hawkbit.sdk.dmf.UpdateInfo;
+import org.eclipse.hawkbit.sdk.dmf.UpdateStatus;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Sender service to send messages to update server.
+ */
+@Slf4j
+public class DmfSenderService extends MessageService {
+
+    private static final byte[] EMPTY_BODY = new byte[0];
+
+    private final String spExchange;
+    private final ConcurrentHashMap<String, BiConsumer<String, Message>> pingListeners = new ConcurrentHashMap<>();
+
+    DmfSenderService(final RabbitTemplate rabbitTemplate, final DmfProperties dmfProperties) {
+        super(rabbitTemplate, dmfProperties);
+        spExchange = AmqpSettings.DMF_EXCHANGE;
+    }
+
+    public void createOrUpdateThing(final String tenant, final String controllerId) {
+        final MessageProperties messagePropertiesForSP = new MessageProperties();
+        messagePropertiesForSP.setHeader(MessageHeaderKey.TYPE, MessageType.THING_CREATED.name());
+        messagePropertiesForSP.setHeader(MessageHeaderKey.TENANT, tenant);
+        messagePropertiesForSP.setHeader(MessageHeaderKey.THING_ID, controllerId);
+        messagePropertiesForSP.setHeader(MessageHeaderKey.SENDER, "hawkBit-sdk");
+        messagePropertiesForSP.setContentType(MessageProperties.CONTENT_TYPE_JSON);
+        messagePropertiesForSP.setReplyTo(dmfProperties.getSenderForSpExchange());
+
+        sendMessage(spExchange, new Message(EMPTY_BODY, messagePropertiesForSP));
+    }
+
+    /**
+     * Finish the update process. This will send a action status to SP.
+     *
+     * @param update the simulated update object
+     * @param updateResultMessages a description according the update process
+     */
+    public void finishUpdateProcess(final UpdateInfo update, final List<String> updateResultMessages) {
+        sendMessage(spExchange, createActionStatusMessage(update, updateResultMessages, DmfActionStatus.FINISHED));
+    }
+
+    /**
+     * Send a message if the message is not null.
+     *
+     * @param address the exchange name
+     * @param message the amqp message which will be send if its not null
+     */
+    // Exception squid:S4449 - Cannot modify RabbitTemplate method definitions.
+    @SuppressWarnings({ "squid:S4449" })
+    public void sendMessage(final String address, final Message message) {
+        if (message == null) {
+            return;
+        }
+        message.getMessageProperties().getHeaders().remove(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME);
+
+        final String correlationId = UUID.randomUUID().toString();
+
+        if (ObjectUtils.isEmpty(message.getMessageProperties().getCorrelationId())) {
+            message.getMessageProperties().setCorrelationId(correlationId);
+        }
+
+        if (log.isTraceEnabled()) {
+            log.trace("Sending message {} to exchange {} with correlationId {}", message, address, correlationId);
+        } else {
+            log.debug("Sending message to exchange {} with correlationId {}", address, correlationId);
+        }
+
+        rabbitTemplate.send(address, null, message, new CorrelationData(correlationId));
+    }
+
+    public Message convertMessage(final Object object, final MessageProperties messageProperties) {
+        return rabbitTemplate.getMessageConverter().toMessage(object, messageProperties);
+    }
+
+    public void sendFeedback(
+            final String tenant, final Long actionId,
+            final UpdateStatus updateStatus) {
+        final Message message = createActionStatusMessage(tenant, actionId, updateStatus.status(), updateStatus.messages());
+        sendMessage(spExchange, message);
+    }
+
+    public void updateAttributes(
+            final String tenant, final String controllerId,
+            final DmfUpdateMode mode,
+            final String key, final String value) {
+        updateAttributes(tenant, controllerId, mode, Collections.singletonMap(key, value));
+    }
+    public void updateAttributes(
+            final String tenant, final String controllerId,
+            final DmfUpdateMode mode,
+            final Map<String, String> attributes) {
+        final MessageProperties messagePropertiesForSP = new MessageProperties();
+        messagePropertiesForSP.setHeader(MessageHeaderKey.TYPE, MessageType.EVENT.name());
+        messagePropertiesForSP.setHeader(MessageHeaderKey.TOPIC, EventTopic.UPDATE_ATTRIBUTES);
+        messagePropertiesForSP.setHeader(MessageHeaderKey.TENANT, tenant);
+        messagePropertiesForSP.setHeader(MessageHeaderKey.THING_ID, controllerId);
+        messagePropertiesForSP.setContentType(MessageProperties.CONTENT_TYPE_JSON);
+        messagePropertiesForSP.setReplyTo(dmfProperties.getSenderForSpExchange());
+
+        final DmfAttributeUpdate attributeUpdate = new DmfAttributeUpdate();
+        attributeUpdate.setMode(mode);
+        attributeUpdate.getAttributes().putAll(attributes);
+
+        sendMessage(spExchange, convertMessage(attributeUpdate, messagePropertiesForSP));
+    }
+
+    public void ping(final String tenant, final String correlationId, final BiConsumer<String, Message> listener) {
+        final MessageProperties messageProperties = new MessageProperties();
+        messageProperties.getHeaders().put(MessageHeaderKey.TENANT, tenant);
+        messageProperties.getHeaders().put(MessageHeaderKey.TYPE, MessageType.PING.toString());
+        messageProperties.setCorrelationId(correlationId);
+        messageProperties.setReplyTo(dmfProperties.getSenderForSpExchange());
+        messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
+
+        if (listener != null) {
+            pingListeners.put(correlationId, listener);
+        }
+
+        sendMessage(spExchange, new Message(EMPTY_BODY, messageProperties));
+    }
+
+    void pingResponse(final String controllerId, final Message message) {
+        final BiConsumer<String, Message> pingListener = pingListeners.remove(controllerId);
+        if (pingListener != null) {
+            pingListener.accept(controllerId, message);
+        }
+    }
+
+    private Message createActionStatusMessage(final UpdateInfo update,
+            final List<String> updateResultMessages, final DmfActionStatus status) {
+        return createActionStatusMessage(update.getTenant(), update.getActionId(), status, updateResultMessages);
+    }
+
+    private Message createActionStatusMessage(final String tenant, final Long actionId,
+            final DmfActionStatus actionStatus, final List<String> updateResultMessages) {
+        final MessageProperties messageProperties = new MessageProperties();
+        final Map<String, Object> headers = messageProperties.getHeaders();
+        final DmfActionUpdateStatus actionUpdateStatus = new DmfActionUpdateStatus(actionId, actionStatus);
+        headers.put(MessageHeaderKey.TYPE, MessageType.EVENT.name());
+        headers.put(MessageHeaderKey.TENANT, tenant);
+        headers.put(MessageHeaderKey.TOPIC, EventTopic.UPDATE_ACTION_STATUS.name());
+        headers.put(MessageHeaderKey.CONTENT_TYPE, MessageProperties.CONTENT_TYPE_JSON);
+        actionUpdateStatus.addMessage(updateResultMessages);
+
+        return convertMessage(actionUpdateStatus, messageProperties);
+    }
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/MessageService.java

@@ -0,0 +1,42 @@
+/**
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.hawkbit.sdk.dmf.amqp;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.hawkbit.sdk.dmf.DmfProperties;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
+
+/**
+ * Abstract class for sender and receiver service.
+ */
+@Slf4j
+class MessageService {
+
+    protected final RabbitTemplate rabbitTemplate;
+    protected final DmfProperties dmfProperties;
+
+    MessageService(final RabbitTemplate rabbitTemplate, final DmfProperties dmfProperties) {
+        this.rabbitTemplate = rabbitTemplate;
+        this.dmfProperties = dmfProperties;
+    }
+
+    /**
+     * Convert a message body to a given class and set the message header
+     * AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME for Jackson converter.
+     */
+    @SuppressWarnings("unchecked")
+    <T> T convertMessage(final Message message, final Class<T> clazz) {
+        message.getMessageProperties().getHeaders().put(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME,
+                clazz.getTypeName());
+        return (T) rabbitTemplate.getMessageConverter().fromMessage(message);
+    }
+}

hawkbit-sdk/hawkbit-sdk-dmf/src/main/resources/application.properties

@@ -0,0 +1,24 @@
+#
+# Copyright (c) 2023 Bosch.IO GmbH and others
+#
+# This program and the accompanying materials are made
+# available under the terms of the Eclipse Public License 2.0
+# which is available at https://www.eclipse.org/legal/epl-2.0/
+#
+# SPDX-License-Identifier: EPL-2.0
+#
+
+## Configuration for local RabbitMQ integration
+spring.rabbitmq.username=guest
+spring.rabbitmq.password=guest
+spring.rabbitmq.virtualHost=/
+spring.rabbitmq.host=localhost
+spring.rabbitmq.port=5672
+spring.rabbitmq.dynamic=true
+
+## Configuration for sdk dmf
+hawkbit.sdk.dmf.amqp.enabled=true
+hawkbit.sdk.dmf.amqp.receiverConnectorQueueFromSp=sdk_receiver
+hawkbit.sdk.dmf.amqp.senderForSpExchange=sdk.replyTo
+
+