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

Merge branch 'master' into Feature_Improve_Code_Quality

Browse Source 
Conflicts:
	hawkbit-core/src/main/java/org/eclipse/hawkbit/api/ArtifactUrlHandlerProperties.java
	hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerService.java
	hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerServiceTest.java
	hawkbit-repository/src/main/java/org/eclipse/hawkbit/eventbus/CacheFieldEntityListener.java
	hawkbit-security-core/src/main/java/org/eclipse/hawkbit/security/SystemSecurityContext.java
	hawkbit-ui/src/main/java/org/eclipse/hawkbit/ui/artifacts/details/ArtifactBeanQuery.java


Signed-off-by: Michael Hirsch <michael.hirsch@bosch-si.com>
This commit is contained in:
Michael Hirsch
2016-07-25 10:08:24 +02:00
parent d2826a21c2 0380fdf8ba
commit f3b472bcd2
805 changed files with 31882 additions and 22829 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java
View File

@@ -8,21 +8,36 @@
*/
package org.eclipse.hawkbit.amqp;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.rabbit.config.SimpleRabbitListenerContainerFactory;
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.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.support.RetryTemplate;;
/**
* The spring AMQP configuration which is enabled by using the profile
@@ -32,14 +47,65 @@ import org.springframework.context.annotation.Bean;
@EnableConfigurationProperties({ AmqpProperties.class, AmqpDeadletterProperties.class })
public class AmqpConfiguration {
@Autowired
protected AmqpProperties amqpProperties;
private static final Logger LOGGER = LoggerFactory.getLogger(AmqpConfiguration.class);
@Autowired
protected AmqpDeadletterProperties amqpDeadletterProperties;
private AmqpProperties amqpProperties;
@Autowired
private ConnectionFactory connectionFactory;
private AmqpDeadletterProperties amqpDeadletterProperties;
@Autowired
private ConnectionFactory rabbitConnectionFactory;
@Configuration
@ConditionalOnMissingBean(ConnectionFactory.class)
protected static class RabbitConnectionFactoryCreator {
@Autowired
private AmqpProperties amqpProperties;
@Autowired
@Qualifier("asyncExecutor")
private Executor threadPoolExecutor;
@Autowired
private ScheduledExecutorService scheduledExecutorService;
/**
* {@link ConnectionFactory} with enabled publisher confirms and
* heartbeat.
*
* @param config
* with standard {@link RabbitProperties}
* @return {@link ConnectionFactory}
*/
@Bean
public ConnectionFactory rabbitConnectionFactory(final RabbitProperties config) {
final CachingConnectionFactory factory = new CachingConnectionFactory();
factory.setRequestedHeartBeat(amqpProperties.getRequestedHeartBeat());
factory.setExecutor(threadPoolExecutor);
factory.getRabbitConnectionFactory().setHeartbeatExecutor(scheduledExecutorService);
factory.setPublisherConfirms(true);
final String addresses = config.getAddresses();
factory.setAddresses(addresses);
if (config.getHost() != null) {
factory.setHost(config.getHost());
factory.setPort(config.getPort());
}
if (config.getUsername() != null) {
factory.setUsername(config.getUsername());
}
if (config.getPassword() != null) {
factory.setPassword(config.getPassword());
}
if (config.getVirtualHost() != null) {
factory.setVirtualHost(config.getVirtualHost());
}
return factory;
}
}
/**
* Create a {@link RabbitAdmin} and ignore declaration exceptions.
@@ -49,44 +115,101 @@ public class AmqpConfiguration {
*/
@Bean
public RabbitAdmin rabbitAdmin() {
final RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
final RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitConnectionFactory);
rabbitAdmin.setIgnoreDeclarationExceptions(true);
return rabbitAdmin;
}
/**
* Method to set the Jackson2JsonMessageConverter.
*
* @return the Jackson2JsonMessageConverter
* @return {@link RabbitTemplate} with automatic retry, published confirms
* and {@link Jackson2JsonMessageConverter}.
*/
@Bean
public RabbitTemplate rabbitTemplate() {
final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
final RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory);
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
final RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setBackOffPolicy(new ExponentialBackOffPolicy());
rabbitTemplate.setRetryTemplate(retryTemplate);
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
LOGGER.debug("Message with {} confirmed by broker.", correlationData);
} else {
LOGGER.error("Broker is unable to handle message with {} : {}", correlationData, cause);
}
});
return rabbitTemplate;
}
/**
* Create the sp receiver queue.
* Create the DMF API receiver queue for retrieving DMF messages.
*
* @return the receiver queue
*/
@Bean
public Queue receiverQueue() {
public Queue dmfReceiverQueue() {
return new Queue(amqpProperties.getReceiverQueue(), true, false, false,
amqpDeadletterProperties.getDeadLetterExchangeArgs(amqpProperties.getDeadLetterExchange()));
}
/**
* Create the dead letter fanout exchange.
* Create the DMF API receiver queue for authentication requests called by
* 3rd party artifact storages for download authorization by devices.
*
* @return the receiver queue
*/
@Bean
public Queue authenticationReceiverQueue() {
return QueueBuilder.nonDurable(amqpProperties.getAuthenticationReceiverQueue()).autoDelete()
.withArguments(getTTLMaxArgsAuthenticationQueue()).build();
}
/**
* Create DMF exchange.
*
* @return the fanout exchange
*/
@Bean
public FanoutExchange senderExchange() {
public FanoutExchange dmfSenderExchange() {
return new FanoutExchange(AmqpSettings.DMF_EXCHANGE);
}
/**
* Create the Binding {@link AmqpConfiguration#dmfReceiverQueue()} to
* {@link AmqpConfiguration#dmfSenderExchange()}.
*
* @return the binding and create the queue and exchange
*/
@Bean
public Binding bindDmfSenderExchangeToDmfQueue() {
return BindingBuilder.bind(dmfReceiverQueue()).to(dmfSenderExchange());
}
/**
* Create authentication exchange.
*
* @return the fanout exchange
*/
@Bean
public FanoutExchange authenticationExchange() {
return new FanoutExchange(AmqpSettings.AUTHENTICATION_EXCHANGE, false, true);
}
/**
* Create the Binding
* {@link AmqpConfiguration#authenticationReceiverQueue()} to
* {@link AmqpConfiguration#authenticationExchange()}.
*
* @return the binding and create the queue and exchange
*/
@Bean
public Binding bindAuthenticationSenderExchangeToAuthenticationQueue() {
return BindingBuilder.bind(authenticationReceiverQueue()).to(authenticationExchange());
}
/**
* Create dead letter queue.
*
@@ -103,29 +226,18 @@ public class AmqpConfiguration {
* @return the fanout exchange
*/
@Bean
public FanoutExchange exchangeDeadLetter() {
public FanoutExchange deadLetterExchange() {
return new FanoutExchange(amqpProperties.getDeadLetterExchange());
}
/**
* Create the Binding deadLetterQueue to exchangeDeadLetter.
* Create the Binding deadLetterQueue to deadLetterExchange.
*
* @return the binding
*/
@Bean
public Binding bindDeadLetterQueueToLwm2mExchange() {
return BindingBuilder.bind(deadLetterQueue()).to(exchangeDeadLetter());
}
/**
* Create the Binding {@link AmqpConfiguration#receiverQueueFromSp()} to
* {@link AmqpConfiguration#senderConnectorToSpExchange()}.
*
* @return the binding and create the queue and exchange
*/
@Bean
public Binding bindSenderExchangeToSpQueue() {
return BindingBuilder.bind(receiverQueue()).to(senderExchange());
public Binding bindDeadLetterQueueToDeadLetterExchange() {
return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange());
}
/**
@@ -156,12 +268,15 @@ public class AmqpConfiguration {
* AMQP messages
*/
@Bean(name = { "listenerContainerFactory" })
public SimpleRabbitListenerContainerFactory listenerContainerFactory() {
final SimpleRabbitListenerContainerFactory containerFactory = new SimpleRabbitListenerContainerFactory();
containerFactory.setDefaultRequeueRejected(false);
containerFactory.setConnectionFactory(connectionFactory);
containerFactory.setMissingQueuesFatal(amqpProperties.isMissingQueuesFatal());
return containerFactory;
public RabbitListenerContainerFactory<SimpleMessageListenerContainer> listenerContainerFactory() {
return new ConfigurableRabbitListenerContainerFactory(amqpProperties, rabbitConnectionFactory);
}
private static Map<String, Object> getTTLMaxArgsAuthenticationQueue() {
final Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", Duration.ofSeconds(30).toMillis());
args.put("x-max-length", 1_000);
return args;
}
}

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

@@ -25,7 +25,7 @@ import org.eclipse.hawkbit.dmf.json.model.DownloadAndUpdateRequest;
import org.eclipse.hawkbit.dmf.json.model.SoftwareModule;
import org.eclipse.hawkbit.eventbus.EventSubscriber;
import org.eclipse.hawkbit.eventbus.event.CancelTargetAssignmentEvent;
import org.eclipse.hawkbit.eventbus.event.TargetAssignDistributionSetEvent;
import org.eclipse.hawkbit.repository.eventbus.event.TargetAssignDistributionSetEvent;
import org.eclipse.hawkbit.repository.model.LocalArtifact;
import org.eclipse.hawkbit.util.IpUtil;
import org.springframework.amqp.core.Message;
@@ -114,7 +114,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
}
private MessageProperties createConnectorMessageProperties(final String tenant, final String controllerId,
private static MessageProperties createConnectorMessageProperties(final String tenant, final String controllerId,
final EventTopic topic) {
final MessageProperties messageProperties = createMessageProperties();
messageProperties.setHeader(MessageHeaderKey.TOPIC, topic);

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

@@ -14,6 +14,7 @@ import java.util.Collections;
import java.util.List;
import java.util.UUID;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.hawkbit.api.HostnameResolver;
import org.eclipse.hawkbit.artifact.repository.model.DbArtifact;
@@ -30,12 +31,15 @@ import org.eclipse.hawkbit.dmf.json.model.ArtifactHash;
import org.eclipse.hawkbit.dmf.json.model.DownloadResponse;
import org.eclipse.hawkbit.dmf.json.model.TenantSecurityToken;
import org.eclipse.hawkbit.dmf.json.model.TenantSecurityToken.FileResource;
import org.eclipse.hawkbit.eventbus.event.TargetAssignDistributionSetEvent;
import org.eclipse.hawkbit.im.authentication.SpPermission.SpringEvalExpressions;
import org.eclipse.hawkbit.im.authentication.TenantAwareAuthenticationDetails;
import org.eclipse.hawkbit.repository.ArtifactManagement;
import org.eclipse.hawkbit.repository.ControllerManagement;
import org.eclipse.hawkbit.repository.EntityFactory;
import org.eclipse.hawkbit.repository.RepositoryConstants;
import org.eclipse.hawkbit.repository.eventbus.event.TargetAssignDistributionSetEvent;
import org.eclipse.hawkbit.repository.exception.EntityNotFoundException;
import org.eclipse.hawkbit.repository.exception.TenantNotExistException;
import org.eclipse.hawkbit.repository.model.Action;
import org.eclipse.hawkbit.repository.model.Action.Status;
import org.eclipse.hawkbit.repository.model.ActionStatus;
@@ -43,9 +47,11 @@ import org.eclipse.hawkbit.repository.model.DistributionSet;
import org.eclipse.hawkbit.repository.model.LocalArtifact;
import org.eclipse.hawkbit.repository.model.SoftwareModule;
import org.eclipse.hawkbit.repository.model.Target;
import org.eclipse.hawkbit.security.SystemSecurityContext;
import org.eclipse.hawkbit.util.IpUtil;
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;
@@ -97,6 +103,12 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
@Autowired
private HostnameResolver hostnameResolver;
@Autowired
private EntityFactory entityFactory;
@Autowired
private SystemSecurityContext systemSecurityContext;
/**
* Constructor.
*
@@ -107,29 +119,39 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
super(defaultTemplate);
}
// Method is not unused. It is called by the spring rabbit listener.
@SuppressWarnings("squid:UnusedPrivateMethod")
@RabbitListener(queues = "${hawkbit.dmf.rabbitmq.receiverQueue}", containerFactory = "listenerContainerFactory")
private Message onMessage(final Message message, @Header(MessageHeaderKey.TYPE) final String type,
@Header(MessageHeaderKey.TENANT) final String tenant) {
return onMessage(message, type, tenant, getRabbitTemplate().getConnectionFactory().getVirtualHost());
}
/**
* Method to handle all incoming amqp messages.
* Method to handle all incoming DMF amqp messages.
*
* @param message
* incoming message
* @param type
* the message type
* @param contentType
* the contentType of the message
* @param tenant
* the contentType of the message
* @param virtualHost
* the virtual host
*
* @return a message if <null> no message is send back to sender
*/
@RabbitListener(queues = "${hawkbit.dmf.rabbitmq.receiverQueue}", containerFactory = "listenerContainerFactory")
public Message onMessage(final Message message, @Header(MessageHeaderKey.TYPE) final String type,
@Header(MessageHeaderKey.TENANT) final String tenant) {
return onMessage(message, type, tenant, getRabbitTemplate().getConnectionFactory().getVirtualHost());
}
@RabbitListener(queues = "${hawkbit.dmf.rabbitmq.authenticationReceiverQueue}", containerFactory = "listenerContainerFactory")
public Message onAuthenticationRequest(final Message message) {
checkContentTypeJson(message);
final SecurityContext oldContext = SecurityContextHolder.getContext();
try {
return handleAuthentifiactionMessage(message);
} catch (final IllegalArgumentException ex) {
throw new AmqpRejectAndDontRequeueException("Invalid message!", ex);
} catch (final TenantNotExistException teex) {
throw new AmqpRejectAndDontRequeueException(teex);
} finally {
SecurityContextHolder.setContext(oldContext);
}
}
public Message onMessage(final Message message, final String type, final String tenant, final String virtualHost) {
checkContentTypeJson(message);
final SecurityContext oldContext = SecurityContextHolder.getContext();
@@ -146,11 +168,13 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
final EventTopic eventTopic = EventTopic.valueOf(topicValue);
handleIncomingEvent(message, eventTopic);
break;
case AUTHENTIFICATION:
return handleAuthentifiactionMessage(message);
default:
logAndThrowMessageError(message, "No handle method was found for the given message type.");
}
} catch (final IllegalArgumentException ex) {
throw new AmqpRejectAndDontRequeueException("Invalid message!", ex);
} catch (final TenantNotExistException teex) {
throw new AmqpRejectAndDontRequeueException(teex);
} finally {
SecurityContextHolder.setContext(oldContext);
}
@@ -306,9 +330,10 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
final DistributionSet distributionSet = action.getDistributionSet();
final List<SoftwareModule> softwareModuleList = controllerManagement
.findSoftwareModulesByDistributionSet(distributionSet);
final String targetSecurityToken = systemSecurityContext.runAsSystem(() -> target.getSecurityToken());
eventBus.post(new TargetAssignDistributionSetEvent(target.getOptLockRevision(), target.getTenant(),
target.getControllerId(), action.getId(), softwareModuleList, target.getTargetInfo().getAddress(),
target.getSecurityToken()));
targetSecurityToken));
}
@@ -338,11 +363,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
final ActionUpdateStatus actionUpdateStatus = convertMessage(message, ActionUpdateStatus.class);
final Action action = checkActionExist(message, actionUpdateStatus);
final ActionStatus actionStatus = new ActionStatus();
actionUpdateStatus.getMessage().forEach(actionStatus::addMessage);
actionStatus.setAction(action);
actionStatus.setOccurredAt(System.currentTimeMillis());
final ActionStatus actionStatus = createActionStatus(message, actionUpdateStatus, action);
switch (actionUpdateStatus.getActionStatus()) {
case DOWNLOAD:
@@ -373,18 +394,37 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
logAndThrowMessageError(message, "Status for action does not exisit.");
}
final Action addUpdateActionStatus = getUpdateActionStatus(action, actionStatus);
final Action addUpdateActionStatus = getUpdateActionStatus(actionStatus);
if (!addUpdateActionStatus.isActive()) {
lookIfUpdateAvailable(action.getTarget());
}
}
private Action getUpdateActionStatus(final Action action, final ActionStatus actionStatus) {
if (actionStatus.getStatus().equals(Status.CANCELED)) {
return controllerManagement.addCancelActionStatus(actionStatus, action);
private ActionStatus createActionStatus(final Message message, final ActionUpdateStatus actionUpdateStatus,
final Action action) {
final ActionStatus actionStatus = entityFactory.generateActionStatus();
actionUpdateStatus.getMessage().forEach(actionStatus::addMessage);
if (ArrayUtils.isNotEmpty(message.getMessageProperties().getCorrelationId())) {
actionStatus.addMessage(RepositoryConstants.SERVER_MESSAGE_PREFIX + "DMF message correlation-id "
+ convertCorrelationId(message));
}
return controllerManagement.addUpdateActionStatus(actionStatus, action);
actionStatus.setAction(action);
actionStatus.setOccurredAt(System.currentTimeMillis());
return actionStatus;
}
private static String convertCorrelationId(final Message message) {
return new String(message.getMessageProperties().getCorrelationId());
}
private Action getUpdateActionStatus(final ActionStatus actionStatus) {
if (actionStatus.getStatus().equals(Status.CANCELED)) {
return controllerManagement.addCancelActionStatus(actionStatus);
}
return controllerManagement.addUpdateActionStatus(actionStatus);
}
private Action checkActionExist(final Message message, final ActionUpdateStatus actionUpdateStatus) {
@@ -424,7 +464,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
if (messageProperties.getContentType() != null && messageProperties.getContentType().contains("json")) {
return;
}
throw new IllegalArgumentException("Content-Type is not JSON compatible");
throw new AmqpRejectAndDontRequeueException("Content-Type is not JSON compatible");
}
void setControllerManagement(final ControllerManagement controllerManagement) {
@@ -451,4 +491,11 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
this.eventBus = eventBus;
}
void setEntityFactory(final EntityFactory entityFactory) {
this.entityFactory = entityFactory;
}
void setSystemSecurityContext(final SystemSecurityContext systemSecurityContext) {
this.systemSecurityContext = systemSecurityContext;
}
}

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

@@ -8,6 +8,8 @@
*/
package org.eclipse.hawkbit.amqp;
import java.util.concurrent.TimeUnit;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -29,14 +31,97 @@ public class AmqpProperties {
private String deadLetterExchange = "dmf.connector.deadletter";
/**
* DMF API receiving queue.
* DMF API receiving queue for EVENT or THING_CREATED message.
*/
private String receiverQueue = "dmf_receiver";
/**
* Authentication request called by 3rd party artifact storages for download
* authorizations.
*/
private String authenticationReceiverQueue = "authentication_receiver";
/**
* Missing queue fatal.
*/
private boolean missingQueuesFatal = false;
private boolean missingQueuesFatal;
/**
* Requested heartbeat interval from broker in {@link TimeUnit#SECONDS}.
*/
private int requestedHeartBeat = (int) TimeUnit.SECONDS.toSeconds(60);
/**
* Sets an upper limit to the number of consumers.
*/
private int maxConcurrentConsumers = 10;
/**
* Tells the broker how many messages to send to each consumer in a single
* request. Often this can be set quite high to improve throughput.
*/
private int prefetchCount = 10;
/**
* Initial number of consumers. Is scaled up if necessary up to
* {@link #maxConcurrentConsumers}.
*/
private int initialConcurrentConsumers = 3;
/**
* The number of retry attempts when passive queue declaration fails.
* Passive queue declaration occurs when the consumer starts or, when
* consuming from multiple queues, when not all queues were available during
* initialization.
*/
private int declarationRetries = 50;
/**
* @return the declarationRetries
*/
public int getDeclarationRetries() {
return declarationRetries;
}
/**
* @param declarationRetries
* the declarationRetries to set
*/
public void setDeclarationRetries(final int declarationRetries) {
this.declarationRetries = declarationRetries;
}
public String getAuthenticationReceiverQueue() {
return authenticationReceiverQueue;
}
public void setAuthenticationReceiverQueue(final String authenticationReceiverQueue) {
this.authenticationReceiverQueue = authenticationReceiverQueue;
}
public int getPrefetchCount() {
return prefetchCount;
}
public void setPrefetchCount(final int prefetchCount) {
this.prefetchCount = prefetchCount;
}
public int getInitialConcurrentConsumers() {
return initialConcurrentConsumers;
}
public void setInitialConcurrentConsumers(final int initialConcurrentConsumers) {
this.initialConcurrentConsumers = initialConcurrentConsumers;
}
public int getMaxConcurrentConsumers() {
return maxConcurrentConsumers;
}
public void setMaxConcurrentConsumers(final int maxConcurrentConsumers) {
this.maxConcurrentConsumers = maxConcurrentConsumers;
}
/**
* Is missingQueuesFatal enabled
@@ -99,7 +184,16 @@ public class AmqpProperties {
return receiverQueue;
}
public int getRequestedHeartBeat() {
return requestedHeartBeat;
}
public void setRequestedHeartBeat(final int requestedHeartBeat) {
this.requestedHeartBeat = requestedHeartBeat;
}
public void setReceiverQueue(final String receiverQueue) {
this.receiverQueue = receiverQueue;
}
}

5
hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/BaseAmqpService.java
View File

@@ -15,6 +15,7 @@ import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
@@ -99,8 +100,8 @@ public class BaseAmqpService {
}
protected final void logAndThrowMessageError(final Message message, final String error) {
LOGGER.warn("Error \"{}\" reported by message: {}", error, message);
throw new IllegalArgumentException(error);
LOGGER.warn("Warning! \"{}\" reported by message: {}", error, message);
throw new AmqpRejectAndDontRequeueException(error);
}
protected RabbitTemplate getRabbitTemplate() {

52
hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/ConfigurableRabbitListenerContainerFactory.java Normal file
View File

@@ -0,0 +1,52 @@
/**
* Copyright (c) 2015 Bosch Software Innovations GmbH and others.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* which accompanies this distribution, and is available at
* http://www.eclipse.org/legal/epl-v10.html
*/
package org.eclipse.hawkbit.amqp;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
/**
* {@link RabbitListenerContainerFactory} that can be configured through
* hawkBit's {@link AmqpProperties}.
*
*/
public class ConfigurableRabbitListenerContainerFactory extends SimpleRabbitListenerContainerFactory {
private final AmqpProperties amqpProperties;
/**
* Constructor.
*
* @param rabbitConnectionFactory
* for the container factory
* @param amqpProperties
* to configure the container factory
*/
public ConfigurableRabbitListenerContainerFactory(final AmqpProperties amqpProperties,
final ConnectionFactory rabbitConnectionFactory) {
this.amqpProperties = amqpProperties;
setDefaultRequeueRejected(true);
setConnectionFactory(rabbitConnectionFactory);
setMissingQueuesFatal(amqpProperties.isMissingQueuesFatal());
setConcurrentConsumers(amqpProperties.getInitialConcurrentConsumers());
setMaxConcurrentConsumers(amqpProperties.getMaxConcurrentConsumers());
setPrefetchCount(amqpProperties.getPrefetchCount());
}
@Override
// Exception squid:UnusedProtectedMethod - called by
// AbstractRabbitListenerContainerFactory
@SuppressWarnings("squid:UnusedProtectedMethod")
protected void initializeContainer(final SimpleMessageListenerContainer instance) {
super.initializeContainer(instance);
instance.setDeclarationRetries(amqpProperties.getDeclarationRetries());
}
}

17
hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DefaultAmqpSenderService.java
View File

@@ -9,10 +9,14 @@
package org.eclipse.hawkbit.amqp;
import java.net.URI;
import java.util.UUID;
import org.eclipse.hawkbit.util.IpUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
/**
* A default implementation for the sender service. The service sends all amqp
@@ -20,6 +24,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
* extracted from the uri.
*/
public class DefaultAmqpSenderService implements AmqpSenderService {
private static final Logger LOGGER = LoggerFactory.getLogger(DefaultAmqpSenderService.class);
private final RabbitTemplate internalAmqpTemplate;
@@ -39,7 +44,17 @@ public class DefaultAmqpSenderService implements AmqpSenderService {
return;
}
internalAmqpTemplate.send(extractExchange(replyTo), null, message);
final String correlationId = UUID.randomUUID().toString();
final String exchange = extractExchange(replyTo);
message.getMessageProperties().setCorrelationId(correlationId.getBytes());
if (LOGGER.isTraceEnabled()) {
LOGGER.trace("Sending message {} to exchange {} with correlationId {}", message, exchange, correlationId);
} else {
LOGGER.debug("Sending message to exchange {} with correlationId {}", exchange, correlationId);
}
internalAmqpTemplate.send(exchange, null, message, new CorrelationData(correlationId));
}
}

78
hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/AmqpTestConfiguration.java
View File

@@ -8,20 +8,48 @@
*/
package org.eclipse.hawkbit;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.eclipse.hawkbit.amqp.AmqpProperties;
import org.eclipse.hawkbit.amqp.AmqpSenderService;
import org.eclipse.hawkbit.amqp.DefaultAmqpSenderService;
import org.eclipse.hawkbit.repository.jpa.model.helper.SystemSecurityContextHolder;
import org.eclipse.hawkbit.security.SystemSecurityContext;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.security.concurrent.DelegatingSecurityContextExecutorService;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
/**
*
*/
@Configuration
@EnableConfigurationProperties({ AmqpProperties.class })
public class AmqpTestConfiguration {
/**
* @return the {@link SystemSecurityContext} singleton bean which make it
* accessible in beans which cannot access the service directly,
* e.g. JPA entities.
*/
@Bean
public SystemSecurityContextHolder systemSecurityContextHolder() {
return SystemSecurityContextHolder.getInstance();
}
/**
* Method to set the Jackson2JsonMessageConverter.
@@ -45,4 +73,54 @@ public class AmqpTestConfiguration {
public AmqpSenderService amqpSenderServiceBean(final RabbitTemplate rabbitTemplate) {
return new DefaultAmqpSenderService(rabbitTemplate);
}
/**
* @return ExecutorService with security context availability in thread
* execution..
*/
@Bean(destroyMethod = "shutdown")
@ConditionalOnMissingBean
public Executor asyncExecutor() {
return new DelegatingSecurityContextExecutorService(threadPoolExecutor());
}
/**
* @return central ThreadPoolExecutor for general purpose multi threaded
* operations. Tries an orderly shutdown when destroyed.
*/
private ThreadPoolExecutor threadPoolExecutor() {
final BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(10);
final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 10, 1000, TimeUnit.MILLISECONDS,
blockingQueue, new ThreadFactoryBuilder().setNameFormat("central-executor-pool-%d").build());
return threadPoolExecutor;
}
/**
* @return {@link TaskExecutor} for task execution
*/
@Bean
@ConditionalOnMissingBean
public TaskExecutor taskExecutor() {
return new ConcurrentTaskExecutor(asyncExecutor());
}
/**
* @return {@link ScheduledExecutorService} based on
* {@link #threadPoolTaskScheduler()}.
*/
@Bean
@ConditionalOnMissingBean
public ScheduledExecutorService scheduledExecutorService() {
return threadPoolTaskScheduler().getScheduledExecutor();
}
/**
* @return {@link ThreadPoolTaskScheduler} for scheduled operations.
*/
@Bean
@ConditionalOnMissingBean
public ThreadPoolTaskScheduler threadPoolTaskScheduler() {
return new ThreadPoolTaskScheduler();
}
}

19
hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpControllerAuthenticationTest.java
View File

@@ -158,7 +158,7 @@ public class AmqpControllerAuthenticationTest {
@Test
@Description("Tests authentication message without principal")
public void testAuthenticationMessageBadCredantialsWithoutPricipal() {
final MessageProperties messageProperties = createMessageProperties(MessageType.AUTHENTIFICATION);
final MessageProperties messageProperties = createMessageProperties(null);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, CONTROLLLER_ID,
FileResource.createFileResourceBySha1("12345"));
@@ -166,8 +166,7 @@ public class AmqpControllerAuthenticationTest {
messageProperties);
// test
final Message onMessage = amqpMessageHandlerService.onMessage(message, MessageType.AUTHENTIFICATION.name(),
TENANT, "vHost");
final Message onMessage = amqpMessageHandlerService.onAuthenticationRequest(message);
// verify
final DownloadResponse downloadResponse = (DownloadResponse) messageConverter.fromMessage(onMessage);
@@ -178,7 +177,7 @@ public class AmqpControllerAuthenticationTest {
@Test
@Description("Tests authentication message without wrong credential")
public void testAuthenticationMessageBadCredantialsWithWrongCredential() {
final MessageProperties messageProperties = createMessageProperties(MessageType.AUTHENTIFICATION);
final MessageProperties messageProperties = createMessageProperties(null);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, CONTROLLLER_ID,
FileResource.createFileResourceBySha1("12345"));
when(tenantConfigurationManagement.getConfigurationValue(
@@ -189,8 +188,7 @@ public class AmqpControllerAuthenticationTest {
messageProperties);
// test
final Message onMessage = amqpMessageHandlerService.onMessage(message, MessageType.AUTHENTIFICATION.name(),
TENANT, "vHost");
final Message onMessage = amqpMessageHandlerService.onAuthenticationRequest(message);
// verify
final DownloadResponse downloadResponse = (DownloadResponse) messageConverter.fromMessage(onMessage);
@@ -201,7 +199,7 @@ public class AmqpControllerAuthenticationTest {
@Test
@Description("Tests authentication message successfull")
public void testSuccessfullMessageAuthentication() {
final MessageProperties messageProperties = createMessageProperties(MessageType.AUTHENTIFICATION);
final MessageProperties messageProperties = createMessageProperties(null);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, CONTROLLLER_ID,
FileResource.createFileResourceBySha1("12345"));
when(tenantConfigurationManagement.getConfigurationValue(
@@ -212,8 +210,7 @@ public class AmqpControllerAuthenticationTest {
messageProperties);
// test
final Message onMessage = amqpMessageHandlerService.onMessage(message, MessageType.AUTHENTIFICATION.name(),
TENANT, "vHost");
final Message onMessage = amqpMessageHandlerService.onAuthenticationRequest(message);
// verify
final DownloadResponse downloadResponse = (DownloadResponse) messageConverter.fromMessage(onMessage);
@@ -232,7 +229,9 @@ public class AmqpControllerAuthenticationTest {
private MessageProperties createMessageProperties(final MessageType type, final String replyTo) {
final MessageProperties messageProperties = new MessageProperties();
messageProperties.setHeader(MessageHeaderKey.TYPE, type.name());
if (type != null) {
messageProperties.setHeader(MessageHeaderKey.TYPE, type.name());
}
messageProperties.setHeader(MessageHeaderKey.TENANT, TENANT);
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setReplyTo(replyTo);

19
hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherServiceTest.java
View File

@@ -25,8 +25,6 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import org.eclipse.hawkbit.AbstractIntegrationTestWithMongoDB;
import org.eclipse.hawkbit.TestDataUtil;
import org.eclipse.hawkbit.api.ArtifactUrlHandler;
import org.eclipse.hawkbit.artifact.repository.model.DbArtifact;
import org.eclipse.hawkbit.dmf.amqp.api.EventTopic;
@@ -34,11 +32,12 @@ import org.eclipse.hawkbit.dmf.amqp.api.MessageHeaderKey;
import org.eclipse.hawkbit.dmf.amqp.api.MessageType;
import org.eclipse.hawkbit.dmf.json.model.DownloadAndUpdateRequest;
import org.eclipse.hawkbit.eventbus.event.CancelTargetAssignmentEvent;
import org.eclipse.hawkbit.eventbus.event.TargetAssignDistributionSetEvent;
import org.eclipse.hawkbit.repository.eventbus.event.TargetAssignDistributionSetEvent;
import org.eclipse.hawkbit.repository.model.Artifact;
import org.eclipse.hawkbit.repository.model.DistributionSet;
import org.eclipse.hawkbit.repository.model.LocalArtifact;
import org.eclipse.hawkbit.repository.model.SoftwareModule;
import org.eclipse.hawkbit.repository.test.util.AbstractIntegrationTestWithMongoDB;
import org.eclipse.hawkbit.util.IpUtil;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
@@ -48,6 +47,7 @@ import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.test.context.ActiveProfiles;
import ru.yandex.qatools.allure.annotations.Description;
@@ -57,6 +57,7 @@ import ru.yandex.qatools.allure.annotations.Stories;
@ActiveProfiles({ "test" })
@Features("Component Tests - Device Management Federation API")
@Stories("AmqpMessage Dispatcher Service Test")
@SpringApplicationConfiguration(classes = { org.eclipse.hawkbit.RepositoryApplicationConfiguration.class })
public class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTestWithMongoDB {
private static final String TENANT = "default";
@@ -100,6 +101,7 @@ public class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTestWit
amqpMessageDispatcherService.targetAssignDistributionSet(targetAssignDistributionSetEvent);
final Message sendMessage = createArgumentCapture(targetAssignDistributionSetEvent.getTargetAdress());
final DownloadAndUpdateRequest downloadAndUpdateRequest = assertDownloadAndInstallMessage(sendMessage);
assertThat(downloadAndUpdateRequest.getTargetSecurityToken()).isEqualTo(TEST_TOKEN);
assertTrue("No softwaremmodule should be contained in the request",
downloadAndUpdateRequest.getSoftwareModules().isEmpty());
}
@@ -107,8 +109,7 @@ public class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTestWit
@Test
@Description("Verfies that download and install event with 3 software moduls and no artifacts works")
public void testSendDownloadRequesWithSoftwareModulesAndNoArtifacts() {
final DistributionSet dsA = TestDataUtil.generateDistributionSet("", softwareManagement,
distributionSetManagement);
final DistributionSet dsA = testdataFactory.createDistributionSet("");
final TargetAssignDistributionSetEvent targetAssignDistributionSetEvent = new TargetAssignDistributionSetEvent(
1L, TENANT, CONTROLLER_ID, 1L, dsA.getModules(), AMQP_URI, TEST_TOKEN);
amqpMessageDispatcherService.targetAssignDistributionSet(targetAssignDistributionSetEvent);
@@ -116,6 +117,7 @@ public class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTestWit
final DownloadAndUpdateRequest downloadAndUpdateRequest = assertDownloadAndInstallMessage(sendMessage);
assertEquals("Expecting a size of 3 software modules in the reuqest", 3,
downloadAndUpdateRequest.getSoftwareModules().size());
assertThat(downloadAndUpdateRequest.getTargetSecurityToken()).isEqualTo(TEST_TOKEN);
for (final org.eclipse.hawkbit.dmf.json.model.SoftwareModule softwareModule : downloadAndUpdateRequest
.getSoftwareModules()) {
assertTrue("Artifact list for softwaremodule should be empty", softwareModule.getArtifacts().isEmpty());
@@ -137,11 +139,10 @@ public class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTestWit
@Test
@Description("Verfies that download and install event with software moduls and artifacts works")
public void testSendDownloadRequest() {
final DistributionSet dsA = TestDataUtil.generateDistributionSet("", softwareManagement,
distributionSetManagement);
final DistributionSet dsA = testdataFactory.createDistributionSet("");
final SoftwareModule module = dsA.getModules().iterator().next();
final List<DbArtifact> receivedList = new ArrayList<>();
for (final Artifact artifact : TestDataUtil.generateArtifacts(artifactManagement, module.getId())) {
for (final Artifact artifact : testdataFactory.createLocalArtifacts(module.getId())) {
module.addArtifact((LocalArtifact) artifact);
receivedList.add(new DbArtifact());
}
@@ -155,6 +156,8 @@ public class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTestWit
final DownloadAndUpdateRequest downloadAndUpdateRequest = assertDownloadAndInstallMessage(sendMessage);
assertEquals("DownloadAndUpdateRequest event should contains 3 software modules", 3,
downloadAndUpdateRequest.getSoftwareModules().size());
assertThat(downloadAndUpdateRequest.getTargetSecurityToken()).isEqualTo(TEST_TOKEN);
for (final org.eclipse.hawkbit.dmf.json.model.SoftwareModule softwareModule : downloadAndUpdateRequest
.getSoftwareModules()) {
if (!softwareModule.getModuleId().equals(module.getId())) {

112
hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerServiceTest.java
View File

@@ -36,17 +36,23 @@ import org.eclipse.hawkbit.dmf.json.model.ActionUpdateStatus;
import org.eclipse.hawkbit.dmf.json.model.DownloadResponse;
import org.eclipse.hawkbit.dmf.json.model.TenantSecurityToken;
import org.eclipse.hawkbit.dmf.json.model.TenantSecurityToken.FileResource;
import org.eclipse.hawkbit.eventbus.event.TargetAssignDistributionSetEvent;
import org.eclipse.hawkbit.repository.ArtifactManagement;
import org.eclipse.hawkbit.repository.ControllerManagement;
import org.eclipse.hawkbit.repository.EntityFactory;
import org.eclipse.hawkbit.repository.eventbus.event.TargetAssignDistributionSetEvent;
import org.eclipse.hawkbit.repository.exception.EntityNotFoundException;
import org.eclipse.hawkbit.repository.jpa.model.JpaAction;
import org.eclipse.hawkbit.repository.jpa.model.JpaActionStatus;
import org.eclipse.hawkbit.repository.jpa.model.JpaSoftwareModule;
import org.eclipse.hawkbit.repository.jpa.model.JpaTarget;
import org.eclipse.hawkbit.repository.jpa.model.helper.SecurityTokenGeneratorHolder;
import org.eclipse.hawkbit.repository.model.Action;
import org.eclipse.hawkbit.repository.model.Action.Status;
import org.eclipse.hawkbit.repository.model.LocalArtifact;
import org.eclipse.hawkbit.repository.model.SoftwareModule;
import org.eclipse.hawkbit.repository.model.Target;
import org.eclipse.hawkbit.repository.model.helper.SecurityTokenGeneratorHolder;
import org.eclipse.hawkbit.repository.model.TargetInfo;
import org.eclipse.hawkbit.security.SecurityTokenGenerator;
import org.eclipse.hawkbit.security.SystemSecurityContext;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -54,6 +60,7 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Matchers;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -82,6 +89,9 @@ public class AmqpMessageHandlerServiceTest {
@Mock
private ControllerManagement controllerManagementMock;
@Mock
private EntityFactory entityFactoryMock;
@Mock
private ArtifactManagement artifactManagementMock;
@@ -103,6 +113,9 @@ public class AmqpMessageHandlerServiceTest {
@Mock
private RabbitTemplate rabbitTemplate;
@Mock
private SystemSecurityContext systemSecurityContextMock;
@Before
public void before() throws Exception {
messageConverter = new Jackson2JsonMessageConverter();
@@ -114,6 +127,8 @@ public class AmqpMessageHandlerServiceTest {
amqpMessageHandlerService.setCache(cacheMock);
amqpMessageHandlerService.setHostnameResolver(hostnameResolverMock);
amqpMessageHandlerService.setEventBus(eventBus);
amqpMessageHandlerService.setEntityFactory(entityFactoryMock);
amqpMessageHandlerService.setSystemSecurityContext(systemSecurityContextMock);
}
@@ -125,8 +140,8 @@ public class AmqpMessageHandlerServiceTest {
final Message message = new Message(new byte[0], messageProperties);
try {
amqpMessageHandlerService.onMessage(message, MessageType.THING_CREATED.name(), TENANT, "vHost");
fail("IllegalArgumentException was excepeted due to worng content type");
} catch (final IllegalArgumentException e) {
fail("AmqpRejectAndDontRequeueException was excepeted due to worng content type");
} catch (final AmqpRejectAndDontRequeueException e) {
}
}
@@ -160,8 +175,8 @@ public class AmqpMessageHandlerServiceTest {
try {
amqpMessageHandlerService.onMessage(message, MessageType.THING_CREATED.name(), TENANT, "vHost");
fail("IllegalArgumentException was excepeted since no replyTo header was set");
} catch (final IllegalArgumentException exception) {
fail("AmqpRejectAndDontRequeueException was excepeted since no replyTo header was set");
} catch (final AmqpRejectAndDontRequeueException exception) {
// test ok - exception was excepted
}
@@ -174,8 +189,8 @@ public class AmqpMessageHandlerServiceTest {
final Message message = messageConverter.toMessage(new byte[0], messageProperties);
try {
amqpMessageHandlerService.onMessage(message, MessageType.THING_CREATED.name(), TENANT, "vHost");
fail("IllegalArgumentException was excepeted since no thingID was set");
} catch (final IllegalArgumentException exception) {
fail("AmqpRejectAndDontRequeueException was excepeted since no thingID was set");
} catch (final AmqpRejectAndDontRequeueException exception) {
// test ok - exception was excepted
}
}
@@ -190,8 +205,8 @@ public class AmqpMessageHandlerServiceTest {
try {
amqpMessageHandlerService.onMessage(message, type, TENANT, "vHost");
fail("IllegalArgumentException was excepeted due to unknown message type");
} catch (final IllegalArgumentException exception) {
fail("AmqpRejectAndDontRequeueException was excepeted due to unknown message type");
} catch (final AmqpRejectAndDontRequeueException exception) {
// test ok - exception was excepted
}
}
@@ -203,22 +218,22 @@ public class AmqpMessageHandlerServiceTest {
final Message message = new Message(new byte[0], messageProperties);
try {
amqpMessageHandlerService.onMessage(message, MessageType.EVENT.name(), TENANT, "vHost");
fail("IllegalArgumentException was excepeted due to unknown message type");
} catch (final IllegalArgumentException e) {
fail("AmqpRejectAndDontRequeueException was excepeted due to unknown message type");
} catch (final AmqpRejectAndDontRequeueException e) {
}
try {
messageProperties.setHeader(MessageHeaderKey.TOPIC, "wrongTopic");
amqpMessageHandlerService.onMessage(message, MessageType.EVENT.name(), TENANT, "vHost");
fail("IllegalArgumentException was excepeted due to unknown topic");
} catch (final IllegalArgumentException e) {
fail("AmqpRejectAndDontRequeueException was excepeted due to unknown topic");
} catch (final AmqpRejectAndDontRequeueException e) {
}
messageProperties.setHeader(MessageHeaderKey.TOPIC, EventTopic.CANCEL_DOWNLOAD.name());
try {
amqpMessageHandlerService.onMessage(message, MessageType.EVENT.name(), TENANT, "vHost");
fail("IllegalArgumentException was excepeted because there was no event topic");
} catch (final IllegalArgumentException exception) {
fail("AmqpRejectAndDontRequeueException was excepeted because there was no event topic");
} catch (final AmqpRejectAndDontRequeueException exception) {
// test ok - exception was excepted
}
@@ -236,8 +251,8 @@ public class AmqpMessageHandlerServiceTest {
try {
amqpMessageHandlerService.onMessage(message, MessageType.EVENT.name(), TENANT, "vHost");
fail("IllegalArgumentException was excepeted since no action id was set");
} catch (final IllegalArgumentException exception) {
fail("AmqpRejectAndDontRequeueException was excepeted since no action id was set");
} catch (final AmqpRejectAndDontRequeueException exception) {
// test ok - exception was excepted
}
}
@@ -253,8 +268,8 @@ public class AmqpMessageHandlerServiceTest {
try {
amqpMessageHandlerService.onMessage(message, MessageType.EVENT.name(), TENANT, "vHost");
fail("IllegalArgumentException was excepeted since no action id was set");
} catch (final IllegalArgumentException exception) {
fail("AmqpRejectAndDontRequeueException was excepeted since no action id was set");
} catch (final AmqpRejectAndDontRequeueException exception) {
// test ok - exception was excepted
}
@@ -263,14 +278,14 @@ public class AmqpMessageHandlerServiceTest {
@Test
@Description("Tests that an download request is denied for an artifact which does not exists")
public void authenticationRequestDeniedForArtifactWhichDoesNotExists() {
final MessageProperties messageProperties = createMessageProperties(MessageType.AUTHENTIFICATION);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, "123", FileResource.createFileResourceBySha1("12345"));
final MessageProperties messageProperties = createMessageProperties(null);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, "123",
FileResource.createFileResourceBySha1("12345"));
final Message message = amqpMessageHandlerService.getMessageConverter().toMessage(securityToken,
messageProperties);
// test
final Message onMessage = amqpMessageHandlerService.onMessage(message, MessageType.AUTHENTIFICATION.name(),
TENANT, "vHost");
final Message onMessage = amqpMessageHandlerService.onAuthenticationRequest(message);
// verify
final DownloadResponse downloadResponse = (DownloadResponse) messageConverter.fromMessage(onMessage);
@@ -282,8 +297,9 @@ public class AmqpMessageHandlerServiceTest {
@Test
@Description("Tests that an download request is denied for an artifact which is not assigned to the requested target")
public void authenticationRequestDeniedForArtifactWhichIsNotAssignedToTarget() {
final MessageProperties messageProperties = createMessageProperties(MessageType.AUTHENTIFICATION);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, "123", FileResource.createFileResourceBySha1("12345"));
final MessageProperties messageProperties = createMessageProperties(null);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, "123",
FileResource.createFileResourceBySha1("12345"));
final Message message = amqpMessageHandlerService.getMessageConverter().toMessage(securityToken,
messageProperties);
@@ -293,8 +309,7 @@ public class AmqpMessageHandlerServiceTest {
.thenThrow(EntityNotFoundException.class);
// test
final Message onMessage = amqpMessageHandlerService.onMessage(message, MessageType.AUTHENTIFICATION.name(),
TENANT, "vHost");
final Message onMessage = amqpMessageHandlerService.onAuthenticationRequest(message);
// verify
final DownloadResponse downloadResponse = (DownloadResponse) messageConverter.fromMessage(onMessage);
@@ -306,8 +321,9 @@ public class AmqpMessageHandlerServiceTest {
@Test
@Description("Tests that an download request is allowed for an artifact which exists and assigned to the requested target")
public void authenticationRequestAllowedForArtifactWhichExistsAndAssignedToTarget() throws MalformedURLException {
final MessageProperties messageProperties = createMessageProperties(MessageType.AUTHENTIFICATION);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, "123", FileResource.createFileResourceBySha1("12345"));
final MessageProperties messageProperties = createMessageProperties(null);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, "123",
FileResource.createFileResourceBySha1("12345"));
final Message message = amqpMessageHandlerService.getMessageConverter().toMessage(securityToken,
messageProperties);
@@ -324,8 +340,7 @@ public class AmqpMessageHandlerServiceTest {
when(hostnameResolverMock.resolveHostname()).thenReturn(new URL("http://localhost"));
// test
final Message onMessage = amqpMessageHandlerService.onMessage(message, MessageType.AUTHENTIFICATION.name(),
TENANT, "vHost");
final Message onMessage = amqpMessageHandlerService.onAuthenticationRequest(message);
// verify
final DownloadResponse downloadResponse = (DownloadResponse) messageConverter.fromMessage(onMessage);
@@ -346,7 +361,8 @@ public class AmqpMessageHandlerServiceTest {
// Mock
final Action action = createActionWithTarget(22L, Status.FINISHED);
when(controllerManagementMock.findActionWithDetails(Matchers.any())).thenReturn(action);
when(controllerManagementMock.addUpdateActionStatus(Matchers.any(), Matchers.any())).thenReturn(action);
when(controllerManagementMock.addUpdateActionStatus(Matchers.any())).thenReturn(action);
when(entityFactoryMock.generateActionStatus()).thenReturn(new JpaActionStatus());
// for the test the same action can be used
final List<Action> actionList = new ArrayList<>();
actionList.add(action);
@@ -356,6 +372,8 @@ public class AmqpMessageHandlerServiceTest {
when(controllerManagementMock.findSoftwareModulesByDistributionSet(Matchers.any()))
.thenReturn(softwareModuleList);
when(systemSecurityContextMock.runAsSystem(anyObject())).thenReturn("securityToken");
final MessageProperties messageProperties = createMessageProperties(MessageType.EVENT);
messageProperties.setHeader(MessageHeaderKey.TOPIC, EventTopic.UPDATE_ACTION_STATUS.name());
final ActionUpdateStatus actionUpdateStatus = createActionUpdateStatus(ActionStatus.FINISHED, 23L);
@@ -400,7 +418,9 @@ public class AmqpMessageHandlerServiceTest {
private MessageProperties createMessageProperties(final MessageType type, final String replyTo) {
final MessageProperties messageProperties = new MessageProperties();
messageProperties.setHeader(MessageHeaderKey.TYPE, type.name());
if (type != null) {
messageProperties.setHeader(MessageHeaderKey.TYPE, type.name());
}
messageProperties.setHeader(MessageHeaderKey.TENANT, TENANT);
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setReplyTo(replyTo);
@@ -409,7 +429,7 @@ public class AmqpMessageHandlerServiceTest {
private List<SoftwareModule> createSoftwareModuleList() {
final List<SoftwareModule> softwareModuleList = new ArrayList<>();
final SoftwareModule softwareModule = new SoftwareModule();
final JpaSoftwareModule softwareModule = new JpaSoftwareModule();
softwareModule.setId(777L);
softwareModuleList.add(softwareModule);
return softwareModuleList;
@@ -420,14 +440,18 @@ public class AmqpMessageHandlerServiceTest {
initalizeSecurityTokenGenerator();
// Mock
final Action action = new Action();
action.setId(targetId);
action.setStatus(status);
action.setTenant("DEFAULT");
final Target target = new Target("target1");
action.setTarget(target);
return action;
final JpaAction actionMock = mock(JpaAction.class);
final JpaTarget targetMock = mock(JpaTarget.class);
final TargetInfo targetInfoMock = mock(TargetInfo.class);
when(actionMock.getId()).thenReturn(targetId);
when(actionMock.getStatus()).thenReturn(status);
when(actionMock.getTenant()).thenReturn("DEFAULT");
when(actionMock.getTarget()).thenReturn(targetMock);
when(targetMock.getControllerId()).thenReturn("target1");
when(targetMock.getSecurityToken()).thenReturn("securityToken");
when(targetMock.getTargetInfo()).thenReturn(targetInfoMock);
when(targetInfoMock.getAddress()).thenReturn(null);
return actionMock;
}
private void initalizeSecurityTokenGenerator() throws IllegalAccessException {

36
hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/util/PropertyBasedArtifactUrlHandlerTest.java
View File

@@ -10,17 +10,13 @@ package org.eclipse.hawkbit.util;
import static org.junit.Assert.assertEquals;
import org.eclipse.hawkbit.AbstractIntegrationTestWithMongoDB;
import org.eclipse.hawkbit.AmqpTestConfiguration;
import org.eclipse.hawkbit.RepositoryApplicationConfiguration;
import org.eclipse.hawkbit.TestConfiguration;
import org.eclipse.hawkbit.TestDataUtil;
import org.eclipse.hawkbit.api.ArtifactUrlHandler;
import org.eclipse.hawkbit.api.UrlProtocol;
import org.eclipse.hawkbit.repository.model.DistributionSet;
import org.eclipse.hawkbit.repository.model.LocalArtifact;
import org.eclipse.hawkbit.repository.model.SoftwareModule;
import org.eclipse.hawkbit.tenancy.TenantAware;
import org.eclipse.hawkbit.repository.test.util.AbstractIntegrationTestWithMongoDB;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -35,27 +31,27 @@ import ru.yandex.qatools.allure.annotations.Stories;
*/
@Features("Component Tests - Artifact URL Handler")
@Stories("Test to generate the artifact download URL")
@SpringApplicationConfiguration(classes = { RepositoryApplicationConfiguration.class, TestConfiguration.class,
AmqpTestConfiguration.class })
@SpringApplicationConfiguration(classes = { AmqpTestConfiguration.class,
org.eclipse.hawkbit.RepositoryApplicationConfiguration.class })
public class PropertyBasedArtifactUrlHandlerTest extends AbstractIntegrationTestWithMongoDB {
private static final String HTTPS_LOCALHOST = "https://localhost:8080/";
private static final String HTTP_LOCALHOST = "http://localhost:8080/";
@Autowired
private ArtifactUrlHandler urlHandlerProperties;
@Autowired
private TenantAware tenantAware;
private LocalArtifact localArtifact;
private final String controllerId = "Test";
private static final String CONTROLLER_ID = "Test";
private String fileName;
private Long softwareModuleId;
private String sha1Hash;
@Before
public void setup() {
final DistributionSet dsA = TestDataUtil.generateDistributionSet("", softwareManagement,
distributionSetManagement);
final DistributionSet dsA = testdataFactory.createDistributionSet("");
final SoftwareModule module = dsA.getModules().iterator().next();
localArtifact = (LocalArtifact) TestDataUtil.generateArtifacts(artifactManagement, module.getId()).stream()
.findAny().get();
localArtifact = testdataFactory.createLocalArtifacts(module.getId()).stream().findAny().get();
softwareModuleId = localArtifact.getSoftwareModule().getId();
fileName = localArtifact.getFilename();
sha1Hash = localArtifact.getSha1Hash();
@@ -66,10 +62,10 @@ public class PropertyBasedArtifactUrlHandlerTest extends AbstractIntegrationTest
@Description("Tests the generation of http download url.")
public void testHttpUrl() {
final String url = urlHandlerProperties.getUrl(controllerId, softwareModuleId, fileName, sha1Hash,
final String url = urlHandlerProperties.getUrl(CONTROLLER_ID, softwareModuleId, fileName, sha1Hash,
UrlProtocol.HTTP);
assertEquals("http is build incorrect",
"http://localhost/" + tenantAware.getCurrentTenant() + "/controller/v1/" + controllerId
HTTP_LOCALHOST + tenantAware.getCurrentTenant() + "/controller/v1/" + CONTROLLER_ID
+ "/softwaremodules/" + localArtifact.getSoftwareModule().getId() + "/artifacts/"
+ localArtifact.getFilename(),
url);
@@ -78,10 +74,10 @@ public class PropertyBasedArtifactUrlHandlerTest extends AbstractIntegrationTest
@Test
@Description("Tests the generation of https download url.")
public void testHttpsUrl() {
final String url = urlHandlerProperties.getUrl(controllerId, softwareModuleId, fileName, sha1Hash,
final String url = urlHandlerProperties.getUrl(CONTROLLER_ID, softwareModuleId, fileName, sha1Hash,
UrlProtocol.HTTPS);
assertEquals("https is build incorrect",
"https://localhost/" + tenantAware.getCurrentTenant() + "/controller/v1/" + controllerId
HTTPS_LOCALHOST + tenantAware.getCurrentTenant() + "/controller/v1/" + CONTROLLER_ID
+ "/softwaremodules/" + localArtifact.getSoftwareModule().getId() + "/artifacts/"
+ localArtifact.getFilename(),
url);
@@ -90,10 +86,10 @@ public class PropertyBasedArtifactUrlHandlerTest extends AbstractIntegrationTest
@Test
@Description("Tests the generation of coap download url.")
public void testCoapUrl() {
final String url = urlHandlerProperties.getUrl(controllerId, softwareModuleId, fileName, sha1Hash,
final String url = urlHandlerProperties.getUrl(CONTROLLER_ID, softwareModuleId, fileName, sha1Hash,
UrlProtocol.COAP);
assertEquals("coap is build incorrect", "coap://127.0.0.1:5683/fw/" + tenantAware.getCurrentTenant() + "/"
+ controllerId + "/sha1/" + localArtifact.getSha1Hash(), url);
+ CONTROLLER_ID + "/sha1/" + localArtifact.getSha1Hash(), url);
}
}

38
hawkbit-dmf-amqp/src/test/resources/application-test.properties Normal file
View File

@@ -0,0 +1,38 @@
#
# Copyright (c) 2015 Bosch Software Innovations GmbH and others.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Eclipse Public License v1.0
# which accompanies this distribution, and is available at
# http://www.eclipse.org/legal/epl-v10.html
#
# supported: H2, MYSQL
hawkbit.server.database=H2
spring.jpa.database=${hawkbit.server.database}
flyway.sqlMigrationSuffix=${spring.jpa.database}.sql
# effective DB setting
spring.datasource.url=${${hawkbit.server.database}.spring.datasource.url}
spring.datasource.driverClassName=${${hawkbit.server.database}.spring.datasource.driverClassName}
spring.datasource.username=${${hawkbit.server.database}.spring.datasource.username}
spring.datasource.password=${${hawkbit.server.database}.spring.datasource.password}
# H2
##;AUTOCOMMIT=ON
H2.spring.datasource.url=jdbc:h2:mem:sp-db;DB_CLOSE_ON_EXIT=FALSE
#H2.spring.datasource.url=jdbc:h2:./db/sp-db;AUTO_SERVER=TRUE;DB_CLOSE_ON_EXIT=FALSE;MVCC=TRUE
H2.spring.datasource.driverClassName=org.h2.Driver
H2.spring.datasource.username=sa
H2.spring.datasource.password=sa
# MYSQL
MYSQL.spring.datasource.url=jdbc:mysql://localhost:3306/sp_test
MYSQL.spring.datasource.driverClassName=org.mariadb.jdbc.Driver
MYSQL.spring.datasource.username=root
MYSQL.spring.datasource.password=

35
hawkbit-dmf-amqp/src/test/resources/logback.xml Normal file
View File

@@ -0,0 +1,35 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
Copyright (c) 2015 Bosch Software Innovations GmbH and others.
All rights reserved. This program and the accompanying materials
are made available under the terms of the Eclipse Public License v1.0
which accompanies this distribution, and is available at
http://www.eclipse.org/legal/epl-v10.html
-->
<configuration>
<include resource="org/springframework/boot/logging/logback/base.xml" />
<logger name="org.eclipse.hawkbit.eventbus.DeadEventListener" level="WARN" />
<Logger name="org.springframework.boot.actuate.audit.listener.AuditListener" level="WARN" />
<Logger name="org.hibernate.validator.internal.util.Version" level="WARN" />
<Logger name="org.apache.coyote.http11.Http11NioProtocol" level="WARN" />
<Logger name="org.apache.tomcat.util.net.NioSelectorPool" level="WARN" />
<Logger name="org.apache.tomcat.jdbc.pool.ConnectionPool" level="DEBUG" />
<Logger name="org.apache.catalina.startup.DigesterFactory" level="ERROR" />
<!-- <Logger name="org.eclipse.hawkbit.rest.util.MockMvcResultPrinter" level="DEBUG" /> -->
<!-- Security Log with hints on potential attacks -->
<logger name="server-security" level="INFO" />
<Root level="INFO">
<AppenderRef ref="Console" />
</Root>
</configuration>