diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/pom.xml b/hawkbit-sdk/hawkbit-sdk-dmf/pom.xml new file mode 100644 index 000000000..441309f50 --- /dev/null +++ b/hawkbit-sdk/hawkbit-sdk-dmf/pom.xml @@ -0,0 +1,44 @@ + + + 4.0.0 + + org.eclipse.hawkbit + hawkbit-sdk + ${revision} + + + hawkbit-sdk-dmf + hawkBit :: SDK :: DMF SDK + DMF SDK that could be used for development of DMF integrations on JVM based languages + + + + org.eclipse.hawkbit + hawkbit-sdk-commons + ${project.version} + + + + org.eclipse.hawkbit + hawkbit-dmf-api + ${project.version} + + + + org.springframework.amqp + spring-rabbit + + + \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/ArtifactHandler.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/ArtifactHandler.java new file mode 100644 index 000000000..548628ff8 --- /dev/null +++ b/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 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 download(); + } +} diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DeviceManagement.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DeviceManagement.java new file mode 100644 index 000000000..3adf778a1 --- /dev/null +++ b/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 tenants = new HashSet<>(); + + private final Map 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 getTenants() { + return tenants; + } + + public Collection getControllers() { + return controllers.values(); + } + + public Optional 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) {} +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfController.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfController.java new file mode 100644 index 000000000..f6aadc663 --- /dev/null +++ b/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 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); + } +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfProperties.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfProperties.java new file mode 100644 index 000000000..18950cdc9 --- /dev/null +++ b/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; +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfSDKConfiguration.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfSDKConfiguration.java new file mode 100644 index 000000000..e0373bb25 --- /dev/null +++ b/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); + } +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/HealthService.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/HealthService.java new file mode 100644 index 000000000..9fca84b0a --- /dev/null +++ b/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 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); + } + } +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/UpdateHandler.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/UpdateHandler.java new file mode 100644 index 000000000..5766ede57 --- /dev/null +++ b/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 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 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 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 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 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 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 ""; + } + + if (targetToken.isEmpty()) { + return ""; + } + + 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 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); + } + }; + } + } +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/UpdateInfo.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/UpdateInfo.java new file mode 100644 index 000000000..bf81d43a1 --- /dev/null +++ b/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(); + } +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/UpdateStatus.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/UpdateStatus.java new file mode 100644 index 000000000..0bf847ae8 --- /dev/null +++ b/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 messages) {} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/DmfReceiverService.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/DmfReceiverService.java new file mode 100644 index 000000000..bf810b52e --- /dev/null +++ b/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 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 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"); + } +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/DmfSenderService.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/DmfSenderService.java new file mode 100644 index 000000000..d4b9a89ec --- /dev/null +++ b/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> 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 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 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 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 pingListener = pingListeners.remove(controllerId); + if (pingListener != null) { + pingListener.accept(controllerId, message); + } + } + + private Message createActionStatusMessage(final UpdateInfo update, + final List 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 updateResultMessages) { + final MessageProperties messageProperties = new MessageProperties(); + final Map 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); + } +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/MessageService.java b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/MessageService.java new file mode 100644 index 000000000..640a79f0b --- /dev/null +++ b/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 convertMessage(final Message message, final Class clazz) { + message.getMessageProperties().getHeaders().put(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME, + clazz.getTypeName()); + return (T) rabbitTemplate.getMessageConverter().fromMessage(message); + } +} \ No newline at end of file diff --git a/hawkbit-sdk/hawkbit-sdk-dmf/src/main/resources/application.properties b/hawkbit-sdk/hawkbit-sdk-dmf/src/main/resources/application.properties new file mode 100644 index 000000000..967eb7664 --- /dev/null +++ b/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 + + diff --git a/hawkbit-sdk/pom.xml b/hawkbit-sdk/pom.xml index 1d9328dae..bd7be7b2e 100644 --- a/hawkbit-sdk/pom.xml +++ b/hawkbit-sdk/pom.xml @@ -25,6 +25,7 @@ hawkbit-sdk-commons hawkbit-sdk-device + hawkbit-sdk-dmf hawkbit-sdk-demo \ No newline at end of file