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

Code format hawkbit (#1948)

Browse Source 
Signed-off-by: Marinov Avgustin <Avgustin.Marinov@bosch.com>
This commit is contained in:
Avgustin Marinov
2024-11-05 11:41:56 +02:00
committed by GitHub
parent 3e469fa58c
commit d842bc2aaa
108 changed files with 17957 additions and 12571 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
hawkbit-sdk/hawkbit-sdk-dmf/pom.xml
View File

@@ -10,35 +10,35 @@
-->
<project
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.eclipse.hawkbit</groupId>
<artifactId>hawkbit-sdk</artifactId>
<version>${revision}</version>
</parent>
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.eclipse.hawkbit</groupId>
<artifactId>hawkbit-sdk</artifactId>
<version>${revision}</version>
</parent>
<artifactId>hawkbit-sdk-dmf</artifactId>
<name>hawkBit :: SDK :: DMF SDK</name>
<description>DMF SDK that could be used for development of DMF integrations on JVM based languages</description>
<artifactId>hawkbit-sdk-dmf</artifactId>
<name>hawkBit :: SDK :: DMF SDK</name>
<description>DMF SDK that could be used for development of DMF integrations on JVM based languages</description>
<dependencies>
<dependency>
<groupId>org.eclipse.hawkbit</groupId>
<artifactId>hawkbit-sdk-commons</artifactId>
<version>${project.version}</version>
</dependency>
<dependencies>
<dependency>
<groupId>org.eclipse.hawkbit</groupId>
<artifactId>hawkbit-sdk-commons</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.hawkbit</groupId>
<artifactId>hawkbit-dmf-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.hawkbit</groupId>
<artifactId>hawkbit-dmf-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
</dependency>
</dependencies>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
</dependency>
</dependencies>
</project>

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

@@ -9,6 +9,10 @@
*/
package org.eclipse.hawkbit.sdk.dmf;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
@@ -19,10 +23,6 @@ import org.eclipse.hawkbit.sdk.Controller;
import org.eclipse.hawkbit.sdk.Tenant;
import org.eclipse.hawkbit.sdk.dmf.amqp.DmfSender;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
/**
* Class representing DMF device twin connecting to hawkBit via DMF.
*/

10
hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/DmfTenant.java
View File

@@ -9,6 +9,11 @@
*/
package org.eclipse.hawkbit.sdk.dmf;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
import lombok.Getter;
import org.eclipse.hawkbit.sdk.Controller;
import org.eclipse.hawkbit.sdk.Tenant;
@@ -16,11 +21,6 @@ import org.eclipse.hawkbit.sdk.dmf.amqp.Amqp;
import org.eclipse.hawkbit.sdk.dmf.amqp.VHost;
import org.springframework.amqp.core.Message;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
/**
* An in-memory simulated DMF Tenant to hold the controller twins in
* memory and be able to retrieve them again.

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

@@ -9,6 +9,22 @@
*/
package org.eclipse.hawkbit.sdk.dmf;
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.HexFormat;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.hc.client5.http.classic.methods.HttpGet;
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
@@ -29,22 +45,6 @@ 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.HexFormat;
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.
*/
@@ -68,20 +68,16 @@ public interface UpdateHandler {
@Slf4j
class UpdateProcessor implements Runnable {
protected final Map<String, Path> downloads = new HashMap<>();
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,
@@ -189,94 +185,6 @@ public interface UpdateHandler {
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!>";
@@ -310,14 +218,6 @@ public interface UpdateHandler {
.build();
}
private interface Validator {
void read(final byte[] buff, final int len);
void validate();
}
private static Validator sizeValidator(final long size) {
return new Validator() {
@@ -377,6 +277,7 @@ public interface UpdateHandler {
}
return new Validator() {
@Override
public void read(final byte[] buff, final int len) {
hashValidators.forEach(hashValidator -> hashValidator.update(buff, len));
@@ -388,5 +289,101 @@ public interface UpdateHandler {
}
};
}
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 interface Validator {
void read(final byte[] buff, final int len);
void validate();
}
}
}

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

@@ -9,8 +9,8 @@
*/
package org.eclipse.hawkbit.sdk.dmf;
import org.eclipse.hawkbit.dmf.json.model.DmfActionStatus;
import java.util.List;
import org.eclipse.hawkbit.dmf.json.model.DmfActionStatus;
public record UpdateStatus(DmfActionStatus status, List<String> messages) {}

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

@@ -9,6 +9,10 @@
*/
package org.eclipse.hawkbit.sdk.dmf.amqp;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.ConcurrentHashMap;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.hawkbit.sdk.Tenant.DMF;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
@@ -16,10 +20,6 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.util.ObjectUtils;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.ConcurrentHashMap;
/**
* Abstract class for connecting to AMQP host.
*/
@@ -41,7 +41,7 @@ public class Amqp {
public VHost getVhost(final DMF dmf, final boolean initVHost) {
final String vHost = dmf == null || ObjectUtils.isEmpty(dmf.getVirtualHost()) ?
(rabbitProperties.getVirtualHost() == null ? "/" :rabbitProperties.getVirtualHost()) :
(rabbitProperties.getVirtualHost() == null ? "/" : rabbitProperties.getVirtualHost()) :
dmf.getVirtualHost();
return vHosts.computeIfAbsent(vHost, vh -> new VHost(getConnectionFactory(dmf, vHost), amqpProperties, initVHost));
}

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

@@ -12,7 +12,6 @@ package org.eclipse.hawkbit.sdk.dmf.amqp;
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.

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

@@ -9,6 +9,8 @@
*/
package org.eclipse.hawkbit.sdk.dmf.amqp;
import static org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings.DMF_EXCHANGE;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -31,17 +33,14 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
import org.springframework.util.ObjectUtils;
import static org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings.DMF_EXCHANGE;
/**
* Sender service to send messages to update server.
*/
@Slf4j
public class DmfSender {
private static final byte[] EMPTY_BODY = new byte[0];
protected final RabbitTemplate rabbitTemplate;
private static final byte[] EMPTY_BODY = new byte[0];
private final AmqpProperties amqpProperties;
private final ConcurrentHashMap<String, BiConsumer<String, Message>> pingListeners = new ConcurrentHashMap<>();

207
hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/amqp/VHost.java
View File

@@ -9,6 +9,15 @@
*/
package org.eclipse.hawkbit.sdk.dmf.amqp;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.hawkbit.dmf.amqp.api.EventTopic;
import org.eclipse.hawkbit.dmf.amqp.api.MessageHeaderKey;
@@ -33,24 +42,15 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
* Abstract class for sender and receiver service.
*/
@Slf4j
public class VHost extends DmfSender implements MessageListener {
private static final String REGEX_EXTRACT_ACTION_ID = "[^0-9]";
private final SimpleMessageListenerContainer container;
private final ConcurrentHashMap<String, DmfTenant> dmfTenants = new ConcurrentHashMap<>();
private final Set<Long> openActions = Collections.synchronizedSet(new HashSet<>());
public VHost(final ConnectionFactory connectionFactory, final AmqpProperties amqpProperties) {
@@ -80,7 +80,7 @@ public class VHost extends DmfSender implements MessageListener {
rabbitAdmin.declareExchange(exchange);
rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange));
}
container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueueNames(amqpProperties.getReceiverConnectorQueueFromSp());
@@ -88,11 +88,6 @@ public class VHost extends DmfSender implements MessageListener {
container.start();
}
void stop() {
container.stop();
rabbitTemplate.destroy();
}
public void register(final DmfTenant dmfTenant) {
dmfTenants.put(dmfTenant.getTenant().getTenantId(), dmfTenant);
}
@@ -100,8 +95,8 @@ public class VHost extends DmfSender implements MessageListener {
@Override
public void onMessage(final Message message) {
final String tenantId = getTenant(message);
final String controllerId = (String)message.getMessageProperties().getHeaders().get(MessageHeaderKey.THING_ID);
final String type = (String)message.getMessageProperties().getHeaders().get(MessageHeaderKey.TYPE);
final String controllerId = (String) message.getMessageProperties().getHeaders().get(MessageHeaderKey.THING_ID);
final String type = (String) message.getMessageProperties().getHeaders().get(MessageHeaderKey.TYPE);
log.info("Message received for target {}, value : {}", controllerId, message);
switch (MessageType.valueOf(type)) {
@@ -130,6 +125,62 @@ public class VHost extends DmfSender implements MessageListener {
}
}
protected void handleAttributeUpdateRequest(final Message message, final String controllerId) {
final String tenantId = getTenant(message);
Optional.ofNullable(dmfTenants.get(tenantId))
.flatMap(dmfTenant -> dmfTenant.getController(controllerId))
.ifPresent(controller ->
updateAttributes(tenantId, controllerId, DmfUpdateMode.MERGE, controller.getAttributes()));
}
protected void handleCancelDownloadAction(final Message message, final String thingId) {
final String tenant = getTenant(message);
final Long actionId = extractActionIdFrom(message);
processCancelDownloadAction(thingId, tenant, actionId);
}
protected void handleUpdateProcess(final Message message, final String controllerId, final EventTopic actionType) {
final String tenant = getTenant(message);
final DmfDownloadAndUpdateRequest downloadAndUpdateRequest = convertMessage(message,
DmfDownloadAndUpdateRequest.class);
dmfTenants.get(tenant).getController(controllerId)
.ifPresent(controller -> controller.setCurrentActionId(downloadAndUpdateRequest.getActionId()));
processUpdate(tenant, controllerId, actionType, downloadAndUpdateRequest);
}
void stop() {
container.stop();
rabbitTemplate.destroy();
}
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);
}
/**
* 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");
}
private void handleEventMessage(final Message message, final String thingId) {
final Object eventHeader = message.getMessageProperties().getHeaders().get(MessageHeaderKey.TOPIC);
if (eventHeader == null) {
@@ -138,27 +189,26 @@ public class VHost extends DmfSender implements MessageListener {
}
// Exception squid:S2259 - Checked before
@SuppressWarnings({ "squid:S2259" })
final EventTopic eventTopic = EventTopic.valueOf(eventHeader.toString());
@SuppressWarnings({ "squid:S2259" }) final EventTopic eventTopic = EventTopic.valueOf(eventHeader.toString());
switch (eventTopic) {
case CONFIRM:
handleConfirmation(message, thingId);
break;
case DOWNLOAD_AND_INSTALL, 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:
log.info("No valid event property: {}", eventTopic);
break;
case CONFIRM:
handleConfirmation(message, thingId);
break;
case DOWNLOAD_AND_INSTALL, 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:
log.info("No valid event property: {}", eventTopic);
break;
}
}
@@ -166,7 +216,6 @@ public class VHost extends DmfSender implements MessageListener {
log.warn("Handle confirmed received for {}! Skip it!", controllerId);
}
private static final String REGEX_EXTRACT_ACTION_ID = "[^0-9]";
private long extractActionIdFrom(final Message message) {
final String messageAsString = message.toString();
final String requiredMessageContent = messageAsString
@@ -191,57 +240,29 @@ public class VHost extends DmfSender implements MessageListener {
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);
log.info("No valid event property in MULTI_ACTION.");
break;
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);
log.info("No valid event property in MULTI_ACTION.");
break;
}
}
protected void handleAttributeUpdateRequest(final Message message, final String controllerId) {
final String tenantId = getTenant(message);
Optional.ofNullable(dmfTenants.get(tenantId))
.flatMap(dmfTenant -> dmfTenant.getController(controllerId))
.ifPresent(controller ->
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);
}
protected 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) {
finishUpdateProcess(thingId, actionId, Collections.singletonList("Simulation canceled"));
openActions.remove(actionId);
}
protected void handleUpdateProcess(final Message message, final String controllerId, final EventTopic actionType) {
final String tenant = getTenant(message);
final DmfDownloadAndUpdateRequest downloadAndUpdateRequest = convertMessage(message,
DmfDownloadAndUpdateRequest.class);
dmfTenants.get(tenant).getController(controllerId).ifPresent(controller -> controller.setCurrentActionId(downloadAndUpdateRequest.getActionId()));
processUpdate(tenant, controllerId, actionType, downloadAndUpdateRequest);
}
private void processUpdate(final String tenantId, final String controllerId, final EventTopic actionType, final DmfDownloadAndUpdateRequest updateRequest) {
private void processUpdate(final String tenantId, final String controllerId, final EventTopic actionType,
final DmfDownloadAndUpdateRequest updateRequest) {
Optional.ofNullable(dmfTenants.get(tenantId))
.flatMap(dmfTenant -> dmfTenant.getController(controllerId))
.ifPresent(controller -> controller.processUpdate(actionType, updateRequest));
@@ -257,26 +278,4 @@ public class VHost extends DmfSender implements MessageListener {
clazz.getTypeName());
return (T) rabbitTemplate.getMessageConverter().fromMessage(message);
}
/**
* 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");
}
}

12
hawkbit-sdk/hawkbit-sdk-dmf/src/main/java/org/eclipse/hawkbit/sdk/dmf/health/HealthService.java
View File

@@ -9,18 +9,18 @@
*/
package org.eclipse.hawkbit.sdk.dmf.health;
import org.eclipse.hawkbit.sdk.dmf.DmfTenant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.scheduling.annotation.Scheduled;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import org.eclipse.hawkbit.sdk.dmf.DmfTenant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.scheduling.annotation.Scheduled;
/**
* Handle all incoming Messages from hawkBit update server.
*/