Renamed new DMF parent folder to be consistent with other parents. (#499)
* Fix DMF folder name. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Alligned name. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Delete old gitgnore Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com>
This commit is contained in:
Kai Zimmermann
@@ -0,0 +1,236 @@
|
||||
/**
|
||||
* 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 java.net.URISyntaxException;
|
||||
import java.util.Optional;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.eclipse.hawkbit.api.HostnameResolver;
|
||||
import org.eclipse.hawkbit.artifact.repository.model.DbArtifact;
|
||||
import org.eclipse.hawkbit.artifact.repository.model.DbArtifactHash;
|
||||
import org.eclipse.hawkbit.cache.DownloadArtifactCache;
|
||||
import org.eclipse.hawkbit.cache.DownloadIdCache;
|
||||
import org.eclipse.hawkbit.cache.DownloadType;
|
||||
import org.eclipse.hawkbit.dmf.json.model.Artifact;
|
||||
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.repository.ArtifactManagement;
|
||||
import org.eclipse.hawkbit.repository.ControllerManagement;
|
||||
import org.eclipse.hawkbit.repository.exception.EntityNotFoundException;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.amqp.AmqpRejectAndDontRequeueException;
|
||||
import org.springframework.amqp.core.Message;
|
||||
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.security.authentication.AuthenticationServiceException;
|
||||
import org.springframework.security.authentication.BadCredentialsException;
|
||||
import org.springframework.security.authentication.CredentialsExpiredException;
|
||||
import org.springframework.security.core.context.SecurityContext;
|
||||
import org.springframework.security.core.context.SecurityContextHolder;
|
||||
import org.springframework.web.util.UriComponentsBuilder;
|
||||
|
||||
/**
|
||||
*
|
||||
* {@link AmqpMessageHandlerService} handles all incoming target authentication
|
||||
* AMQP messages that can be used by 3rd party CDN services to check if a target
|
||||
* is permitted to download certain artifact. This is handled by the queue that
|
||||
* is configured for the property
|
||||
* hawkbit.dmf.rabbitmq.authenticationReceiverQueue.
|
||||
*
|
||||
*/
|
||||
public class AmqpAuthenticationMessageHandler extends BaseAmqpService {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(AmqpAuthenticationMessageHandler.class);
|
||||
|
||||
private final AmqpControllerAuthentication authenticationManager;
|
||||
|
||||
private final ArtifactManagement artifactManagement;
|
||||
|
||||
private final DownloadIdCache cache;
|
||||
|
||||
private final HostnameResolver hostnameResolver;
|
||||
|
||||
private final ControllerManagement controllerManagement;
|
||||
|
||||
/**
|
||||
* @param rabbitTemplate
|
||||
* the configured amqp template.
|
||||
* @param artifactManagement
|
||||
* for artifact URI generation
|
||||
* @param cache
|
||||
* for download Ids
|
||||
* @param hostnameResolver
|
||||
* for resolving the host for downloads
|
||||
* @param authenticationManager
|
||||
* for target authentication
|
||||
* @param controllerManagement
|
||||
* for target repo access
|
||||
*/
|
||||
public AmqpAuthenticationMessageHandler(final RabbitTemplate rabbitTemplate,
|
||||
final AmqpControllerAuthentication authenticationManager, final ArtifactManagement artifactManagement,
|
||||
final DownloadIdCache cache, final HostnameResolver hostnameResolver,
|
||||
final ControllerManagement controllerManagement) {
|
||||
super(rabbitTemplate);
|
||||
this.authenticationManager = authenticationManager;
|
||||
this.artifactManagement = artifactManagement;
|
||||
this.cache = cache;
|
||||
this.hostnameResolver = hostnameResolver;
|
||||
this.controllerManagement = controllerManagement;
|
||||
}
|
||||
|
||||
/**
|
||||
* Executed on an authentication request.
|
||||
*
|
||||
* @param message
|
||||
* the amqp message
|
||||
* @return the rpc message back to supplier.
|
||||
*/
|
||||
@RabbitListener(queues = "${hawkbit.dmf.rabbitmq.authenticationReceiverQueue:authentication_receiver}", containerFactory = "listenerContainerFactory")
|
||||
public Message onAuthenticationRequest(final Message message) {
|
||||
checkContentTypeJson(message);
|
||||
final SecurityContext oldContext = SecurityContextHolder.getContext();
|
||||
try {
|
||||
return handleAuthenticationMessage(message);
|
||||
} catch (final RuntimeException ex) {
|
||||
throw new AmqpRejectAndDontRequeueException(ex);
|
||||
} finally {
|
||||
SecurityContextHolder.setContext(oldContext);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* check action for this download purposes, the method will throw an
|
||||
* EntityNotFoundException in case the controller is not allowed to download
|
||||
* this file because it's not assigned to an action and not assigned to this
|
||||
* controller. Otherwise no controllerId is set = anonymous download
|
||||
*
|
||||
* @param secruityToken
|
||||
* the security token which holds the target ID to check on
|
||||
* @param sha1Hash
|
||||
* of the artifact to verify if the given target is allowed to
|
||||
* download it
|
||||
*/
|
||||
private void checkIfArtifactIsAssignedToTarget(final TenantSecurityToken secruityToken, final String sha1Hash) {
|
||||
|
||||
if (secruityToken.getControllerId() != null) {
|
||||
checkByControllerId(sha1Hash, secruityToken.getControllerId());
|
||||
} else if (secruityToken.getTargetId() != null) {
|
||||
checkByTargetId(sha1Hash, secruityToken.getTargetId());
|
||||
} else {
|
||||
LOG.info("anonymous download no authentication check for artifact {}", sha1Hash);
|
||||
return;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private void checkByTargetId(final String sha1Hash, final Long targetId) {
|
||||
LOG.debug("no anonymous download request, doing authentication check for target {} and artifact {}", targetId,
|
||||
sha1Hash);
|
||||
if (!controllerManagement.hasTargetArtifactAssigned(targetId, sha1Hash)) {
|
||||
LOG.info("target {} tried to download artifact {} which is not assigned to the target", targetId, sha1Hash);
|
||||
throw new EntityNotFoundException();
|
||||
}
|
||||
LOG.info("download security check for target {} and artifact {} granted", targetId, sha1Hash);
|
||||
}
|
||||
|
||||
private void checkByControllerId(final String sha1Hash, final String controllerId) {
|
||||
LOG.debug("no anonymous download request, doing authentication check for target {} and artifact {}",
|
||||
controllerId, sha1Hash);
|
||||
if (!controllerManagement.hasTargetArtifactAssigned(controllerId, sha1Hash)) {
|
||||
LOG.info("target {} tried to download artifact {} which is not assigned to the target", controllerId,
|
||||
sha1Hash);
|
||||
throw new EntityNotFoundException();
|
||||
}
|
||||
LOG.info("download security check for target {} and artifact {} granted", controllerId, sha1Hash);
|
||||
}
|
||||
|
||||
private Optional<org.eclipse.hawkbit.repository.model.Artifact> findArtifactByFileResource(
|
||||
final FileResource fileResource) {
|
||||
|
||||
if (fileResource == null) {
|
||||
return Optional.empty();
|
||||
}
|
||||
|
||||
if (fileResource.getSha1() != null) {
|
||||
return artifactManagement.findFirstArtifactBySHA1(fileResource.getSha1());
|
||||
}
|
||||
|
||||
if (fileResource.getFilename() != null) {
|
||||
return artifactManagement.findArtifactByFilename(fileResource.getFilename());
|
||||
}
|
||||
|
||||
if (fileResource.getArtifactId() != null) {
|
||||
return artifactManagement.findArtifact(fileResource.getArtifactId());
|
||||
}
|
||||
|
||||
if (fileResource.getSoftwareModuleFilenameResource() != null) {
|
||||
return artifactManagement.findByFilenameAndSoftwareModule(
|
||||
fileResource.getSoftwareModuleFilenameResource().getFilename(),
|
||||
fileResource.getSoftwareModuleFilenameResource().getSoftwareModuleId());
|
||||
}
|
||||
|
||||
return Optional.empty();
|
||||
}
|
||||
|
||||
private static Artifact convertDbArtifact(final DbArtifact dbArtifact) {
|
||||
final Artifact artifact = new Artifact();
|
||||
artifact.setSize(dbArtifact.getSize());
|
||||
final DbArtifactHash dbArtifactHash = dbArtifact.getHashes();
|
||||
artifact.setHashes(new ArtifactHash(dbArtifactHash.getSha1(), dbArtifactHash.getMd5()));
|
||||
return artifact;
|
||||
}
|
||||
|
||||
private Message handleAuthenticationMessage(final Message message) {
|
||||
final DownloadResponse authentificationResponse = new DownloadResponse();
|
||||
final TenantSecurityToken secruityToken = convertMessage(message, TenantSecurityToken.class);
|
||||
final FileResource fileResource = secruityToken.getFileResource();
|
||||
try {
|
||||
SecurityContextHolder.getContext().setAuthentication(authenticationManager.doAuthenticate(secruityToken));
|
||||
|
||||
final String sha1Hash = findArtifactByFileResource(fileResource)
|
||||
.map(org.eclipse.hawkbit.repository.model.Artifact::getSha1Hash)
|
||||
.orElseThrow(() -> new EntityNotFoundException());
|
||||
|
||||
checkIfArtifactIsAssignedToTarget(secruityToken, sha1Hash);
|
||||
|
||||
final Artifact artifact = convertDbArtifact(artifactManagement.loadArtifactBinary(sha1Hash).orElseThrow(
|
||||
() -> new EntityNotFoundException(org.eclipse.hawkbit.repository.model.Artifact.class, sha1Hash)));
|
||||
|
||||
authentificationResponse.setArtifact(artifact);
|
||||
final String downloadId = UUID.randomUUID().toString();
|
||||
// SHA1 key is set, download by SHA1
|
||||
final DownloadArtifactCache downloadCache = new DownloadArtifactCache(DownloadType.BY_SHA1, sha1Hash);
|
||||
cache.put(downloadId, downloadCache);
|
||||
authentificationResponse
|
||||
.setDownloadUrl(UriComponentsBuilder.fromUri(hostnameResolver.resolveHostname().toURI())
|
||||
.path("/api/v1/downloadserver/downloadId/").path(downloadId).build().toUriString());
|
||||
authentificationResponse.setResponseCode(HttpStatus.OK.value());
|
||||
} catch (final BadCredentialsException | AuthenticationServiceException | CredentialsExpiredException e) {
|
||||
LOG.error("Login failed", e);
|
||||
authentificationResponse.setResponseCode(HttpStatus.FORBIDDEN.value());
|
||||
authentificationResponse.setMessage("Login failed");
|
||||
} catch (final URISyntaxException e) {
|
||||
LOG.error("URI build exception", e);
|
||||
authentificationResponse.setResponseCode(HttpStatus.INTERNAL_SERVER_ERROR.value());
|
||||
authentificationResponse.setMessage("Building download URI failed");
|
||||
} catch (final EntityNotFoundException e) {
|
||||
final String errorMessage = "Artifact for resource " + fileResource + "not found ";
|
||||
LOG.warn(errorMessage, e);
|
||||
authentificationResponse.setResponseCode(HttpStatus.NOT_FOUND.value());
|
||||
authentificationResponse.setMessage(errorMessage);
|
||||
}
|
||||
|
||||
return getMessageConverter().toMessage(authentificationResponse, message.getMessageProperties());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,419 @@
|
||||
/**
|
||||
* 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 java.security.GeneralSecurityException;
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
|
||||
import org.eclipse.hawkbit.api.ArtifactUrlHandler;
|
||||
import org.eclipse.hawkbit.api.HostnameResolver;
|
||||
import org.eclipse.hawkbit.cache.DownloadIdCache;
|
||||
import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
|
||||
import org.eclipse.hawkbit.repository.ArtifactManagement;
|
||||
import org.eclipse.hawkbit.repository.ControllerManagement;
|
||||
import org.eclipse.hawkbit.repository.EntityFactory;
|
||||
import org.eclipse.hawkbit.repository.SystemManagement;
|
||||
import org.eclipse.hawkbit.repository.TargetManagement;
|
||||
import org.eclipse.hawkbit.repository.TenantConfigurationManagement;
|
||||
import org.eclipse.hawkbit.security.DdiSecurityProperties;
|
||||
import org.eclipse.hawkbit.security.SystemSecurityContext;
|
||||
import org.eclipse.hawkbit.tenancy.TenantAware;
|
||||
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.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.ConditionalRejectingErrorHandler;
|
||||
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.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.cloud.bus.ServiceMatcher;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
|
||||
import org.springframework.retry.support.RetryTemplate;
|
||||
import org.springframework.util.ErrorHandler;
|
||||
|
||||
import com.google.common.collect.Maps;
|
||||
|
||||
/**
|
||||
* Spring configuration for AMQP based DMF communication for indirect device
|
||||
* integration.
|
||||
*
|
||||
*/
|
||||
@EnableConfigurationProperties({ AmqpProperties.class, AmqpDeadletterProperties.class })
|
||||
@ConditionalOnProperty(prefix = "hawkbit.dmf.rabbitmq", name = "enabled", matchIfMissing = true)
|
||||
public class AmqpConfiguration {
|
||||
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(AmqpConfiguration.class);
|
||||
|
||||
@Autowired
|
||||
private AmqpProperties amqpProperties;
|
||||
|
||||
@Autowired
|
||||
private AmqpDeadletterProperties amqpDeadletterProperties;
|
||||
|
||||
@Autowired
|
||||
private ConnectionFactory rabbitConnectionFactory;
|
||||
|
||||
@Autowired(required = false)
|
||||
private ServiceMatcher serviceMatcher;
|
||||
|
||||
/**
|
||||
* Register the bean for the custom error handler.
|
||||
*
|
||||
* @return custom error handler
|
||||
*/
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(ErrorHandler.class)
|
||||
public ErrorHandler errorHandler() {
|
||||
return new ConditionalRejectingErrorHandler(
|
||||
new DelayedRequeueExceptionStrategy(amqpProperties.getRequeueDelay()));
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@ConditionalOnMissingBean(ConnectionFactory.class)
|
||||
@ConditionalOnProperty(prefix = "hawkbit.dmf.rabbitmq", name = "enabled", matchIfMissing = true)
|
||||
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}
|
||||
* @throws GeneralSecurityException
|
||||
* in case of problems with enabled SSL connections
|
||||
*/
|
||||
@Bean
|
||||
public ConnectionFactory rabbitConnectionFactory(final RabbitProperties config)
|
||||
throws GeneralSecurityException {
|
||||
final CachingConnectionFactory factory = new CachingConnectionFactory();
|
||||
factory.setRequestedHeartBeat(amqpProperties.getRequestedHeartBeat());
|
||||
factory.setExecutor(threadPoolExecutor);
|
||||
factory.getRabbitConnectionFactory().setHeartbeatExecutor(scheduledExecutorService);
|
||||
factory.setPublisherConfirms(true);
|
||||
|
||||
if (config.getSsl().isEnabled()) {
|
||||
factory.getRabbitConnectionFactory().useSslProtocol();
|
||||
}
|
||||
|
||||
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.
|
||||
* {@link RabbitAdmin#setIgnoreDeclarationExceptions(boolean)}
|
||||
*
|
||||
* @return the bean
|
||||
*/
|
||||
@Bean
|
||||
public RabbitAdmin rabbitAdmin() {
|
||||
final RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitConnectionFactory);
|
||||
rabbitAdmin.setIgnoreDeclarationExceptions(true);
|
||||
return rabbitAdmin;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return {@link RabbitTemplate} with automatic retry, published confirms
|
||||
* and {@link Jackson2JsonMessageConverter}.
|
||||
*/
|
||||
@Bean
|
||||
public RabbitTemplate rabbitTemplate() {
|
||||
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 DMF API receiver queue for retrieving DMF messages.
|
||||
*
|
||||
* @return the receiver queue
|
||||
*/
|
||||
@Bean
|
||||
public Queue dmfReceiverQueue() {
|
||||
return new Queue(amqpProperties.getReceiverQueue(), true, false, false,
|
||||
amqpDeadletterProperties.getDeadLetterExchangeArgs(amqpProperties.getDeadLetterExchange()));
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 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.
|
||||
*
|
||||
* @return the queue
|
||||
*/
|
||||
@Bean
|
||||
public Queue deadLetterQueue() {
|
||||
return amqpDeadletterProperties.createDeadletterQueue(amqpProperties.getDeadLetterQueue());
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the dead letter fanout exchange.
|
||||
*
|
||||
* @return the fanout exchange
|
||||
*/
|
||||
@Bean
|
||||
public FanoutExchange deadLetterExchange() {
|
||||
return new FanoutExchange(amqpProperties.getDeadLetterExchange());
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the Binding deadLetterQueue to deadLetterExchange.
|
||||
*
|
||||
* @return the binding
|
||||
*/
|
||||
@Bean
|
||||
public Binding bindDeadLetterQueueToDeadLetterExchange() {
|
||||
return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange());
|
||||
}
|
||||
|
||||
/**
|
||||
* Create AMQP handler service bean.
|
||||
*
|
||||
* @param rabbitTemplate
|
||||
* for converting messages
|
||||
* @param amqpMessageDispatcherService
|
||||
* to sending events to DMF client
|
||||
* @param controllerManagement
|
||||
* for target repo access
|
||||
* @param entityFactory
|
||||
* to create entities
|
||||
*
|
||||
* @return handler service bean
|
||||
*/
|
||||
@Bean
|
||||
public AmqpMessageHandlerService amqpMessageHandlerService(final RabbitTemplate rabbitTemplate,
|
||||
final AmqpMessageDispatcherService amqpMessageDispatcherService,
|
||||
final ControllerManagement controllerManagement, final EntityFactory entityFactory) {
|
||||
return new AmqpMessageHandlerService(rabbitTemplate, amqpMessageDispatcherService, controllerManagement,
|
||||
entityFactory);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create AMQP handler service bean for authentication messages.
|
||||
*
|
||||
* @param rabbitTemplate
|
||||
* for converting messages
|
||||
* @param authenticationManager
|
||||
* for target authentication
|
||||
* @param artifactManagement
|
||||
* for artifact URI generation
|
||||
* @param downloadIdCache
|
||||
* for download IDs
|
||||
* @param hostnameResolver
|
||||
* for resolving the host for downloads
|
||||
* @param controllerManagement
|
||||
* for target repo access
|
||||
* @return handler service bean
|
||||
*/
|
||||
@Bean
|
||||
public AmqpAuthenticationMessageHandler amqpAuthenticationMessageHandler(final RabbitTemplate rabbitTemplate,
|
||||
final AmqpControllerAuthentication authenticationManager, final ArtifactManagement artifactManagement,
|
||||
final DownloadIdCache downloadIdCache, final HostnameResolver hostnameResolver,
|
||||
final ControllerManagement controllerManagement) {
|
||||
return new AmqpAuthenticationMessageHandler(rabbitTemplate, authenticationManager, artifactManagement,
|
||||
downloadIdCache, hostnameResolver, controllerManagement);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create default amqp sender service bean.
|
||||
*
|
||||
* @return the default amqp sender service bean
|
||||
*/
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public AmqpSenderService amqpSenderServiceBean() {
|
||||
return new DefaultAmqpSenderService(rabbitTemplate());
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the Listener factory.
|
||||
*
|
||||
* @param errorHandler
|
||||
* the error hander
|
||||
* @return the {@link SimpleMessageListenerContainer} that gets used receive
|
||||
* AMQP messages
|
||||
*/
|
||||
@Bean(name = { "listenerContainerFactory" })
|
||||
public RabbitListenerContainerFactory<SimpleMessageListenerContainer> listenerContainerFactory(
|
||||
final ErrorHandler errorHandler) {
|
||||
return new ConfigurableRabbitListenerContainerFactory(amqpProperties, rabbitConnectionFactory, errorHandler);
|
||||
}
|
||||
|
||||
/**
|
||||
* create the authentication bean for controller over amqp.
|
||||
*
|
||||
* @param systemManagement
|
||||
* the systemManagement
|
||||
* @param controllerManagement
|
||||
* the controllerManagement
|
||||
* @param tenantConfigurationManagement
|
||||
* the tenantConfigurationManagement
|
||||
* @param tenantAware
|
||||
* the tenantAware
|
||||
* @param ddiSecruityProperties
|
||||
* the ddiSecruityProperties
|
||||
* @param systemSecurityContext
|
||||
* the systemSecurityContext
|
||||
* @return the bean
|
||||
*/
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(AmqpControllerAuthentication.class)
|
||||
public AmqpControllerAuthentication amqpControllerAuthentication(final SystemManagement systemManagement,
|
||||
final ControllerManagement controllerManagement,
|
||||
final TenantConfigurationManagement tenantConfigurationManagement, final TenantAware tenantAware,
|
||||
final DdiSecurityProperties ddiSecruityProperties, final SystemSecurityContext systemSecurityContext) {
|
||||
return new AmqpControllerAuthentication(systemManagement, controllerManagement, tenantConfigurationManagement,
|
||||
tenantAware, ddiSecruityProperties, systemSecurityContext);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the dispatcher bean.
|
||||
*
|
||||
* @param rabbitTemplate
|
||||
* the rabbitTemplate
|
||||
* @param amqpSenderService
|
||||
* to send AMQP message
|
||||
* @param artifactUrlHandler
|
||||
* for generating download URLs
|
||||
* @param systemSecurityContext
|
||||
* for execution with system permissions
|
||||
* @param systemManagement
|
||||
* the systemManagement
|
||||
* @param targetManagement
|
||||
* to access target information
|
||||
* @return the bean
|
||||
*/
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(AmqpMessageDispatcherService.class)
|
||||
public AmqpMessageDispatcherService amqpMessageDispatcherService(final RabbitTemplate rabbitTemplate,
|
||||
final AmqpSenderService amqpSenderService, final ArtifactUrlHandler artifactUrlHandler,
|
||||
final SystemSecurityContext systemSecurityContext, final SystemManagement systemManagement,
|
||||
final TargetManagement targetManagement) {
|
||||
return new AmqpMessageDispatcherService(rabbitTemplate, amqpSenderService, artifactUrlHandler,
|
||||
systemSecurityContext, systemManagement, targetManagement, serviceMatcher);
|
||||
}
|
||||
|
||||
private static Map<String, Object> getTTLMaxArgsAuthenticationQueue() {
|
||||
final Map<String, Object> args = Maps.newHashMapWithExpectedSize(2);
|
||||
args.put("x-message-ttl", Duration.ofSeconds(30).toMillis());
|
||||
args.put("x-max-length", 1_000);
|
||||
return args;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,167 @@
|
||||
/**
|
||||
* 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 java.util.List;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
|
||||
import org.eclipse.hawkbit.dmf.json.model.TenantSecurityToken;
|
||||
import org.eclipse.hawkbit.im.authentication.TenantAwareAuthenticationDetails;
|
||||
import org.eclipse.hawkbit.repository.ControllerManagement;
|
||||
import org.eclipse.hawkbit.repository.SystemManagement;
|
||||
import org.eclipse.hawkbit.repository.TenantConfigurationManagement;
|
||||
import org.eclipse.hawkbit.security.ControllerPreAuthenticateSecurityTokenFilter;
|
||||
import org.eclipse.hawkbit.security.ControllerPreAuthenticatedAnonymousDownload;
|
||||
import org.eclipse.hawkbit.security.ControllerPreAuthenticatedAnonymousFilter;
|
||||
import org.eclipse.hawkbit.security.ControllerPreAuthenticatedGatewaySecurityTokenFilter;
|
||||
import org.eclipse.hawkbit.security.ControllerPreAuthenticatedSecurityHeaderFilter;
|
||||
import org.eclipse.hawkbit.security.DdiSecurityProperties;
|
||||
import org.eclipse.hawkbit.security.PreAuthTokenSourceTrustAuthenticationProvider;
|
||||
import org.eclipse.hawkbit.security.PreAuthentificationFilter;
|
||||
import org.eclipse.hawkbit.security.SystemSecurityContext;
|
||||
import org.eclipse.hawkbit.tenancy.TenantAware;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.security.core.Authentication;
|
||||
import org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationToken;
|
||||
|
||||
import com.google.common.collect.Lists;
|
||||
|
||||
/**
|
||||
*
|
||||
* A controller which handles the DMF AMQP authentication.
|
||||
*/
|
||||
public class AmqpControllerAuthentication {
|
||||
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(AmqpControllerAuthentication.class);
|
||||
|
||||
private final PreAuthTokenSourceTrustAuthenticationProvider preAuthenticatedAuthenticationProvider = new PreAuthTokenSourceTrustAuthenticationProvider();
|
||||
|
||||
private List<PreAuthentificationFilter> filterChain;
|
||||
|
||||
private final ControllerManagement controllerManagement;
|
||||
|
||||
private final SystemManagement systemManagement;
|
||||
|
||||
private final TenantConfigurationManagement tenantConfigurationManagement;
|
||||
|
||||
private final TenantAware tenantAware;
|
||||
|
||||
private final DdiSecurityProperties ddiSecruityProperties;
|
||||
|
||||
private final SystemSecurityContext systemSecurityContext;
|
||||
|
||||
/**
|
||||
* Constructor.
|
||||
*
|
||||
* @param systemManagement
|
||||
* @param controllerManagement
|
||||
* @param tenantConfigurationManagement
|
||||
* @param tenantAware
|
||||
* current tenant
|
||||
* @param ddiSecruityProperties
|
||||
* security configurations
|
||||
* @param systemSecurityContext
|
||||
* security context
|
||||
*/
|
||||
public AmqpControllerAuthentication(final SystemManagement systemManagement,
|
||||
final ControllerManagement controllerManagement,
|
||||
final TenantConfigurationManagement tenantConfigurationManagement, final TenantAware tenantAware,
|
||||
final DdiSecurityProperties ddiSecruityProperties, final SystemSecurityContext systemSecurityContext) {
|
||||
this.controllerManagement = controllerManagement;
|
||||
this.systemManagement = systemManagement;
|
||||
this.tenantConfigurationManagement = tenantConfigurationManagement;
|
||||
this.tenantAware = tenantAware;
|
||||
this.ddiSecruityProperties = ddiSecruityProperties;
|
||||
this.systemSecurityContext = systemSecurityContext;
|
||||
}
|
||||
|
||||
/**
|
||||
* Called by spring when bean instantiated and autowired.
|
||||
*/
|
||||
@PostConstruct
|
||||
public void postConstruct() {
|
||||
addFilter();
|
||||
}
|
||||
|
||||
private void addFilter() {
|
||||
filterChain = Lists.newArrayListWithExpectedSize(5);
|
||||
|
||||
final ControllerPreAuthenticatedGatewaySecurityTokenFilter gatewaySecurityTokenFilter = new ControllerPreAuthenticatedGatewaySecurityTokenFilter(
|
||||
tenantConfigurationManagement, tenantAware, systemSecurityContext);
|
||||
filterChain.add(gatewaySecurityTokenFilter);
|
||||
|
||||
final ControllerPreAuthenticatedSecurityHeaderFilter securityHeaderFilter = new ControllerPreAuthenticatedSecurityHeaderFilter(
|
||||
ddiSecruityProperties.getRp().getCnHeader(), ddiSecruityProperties.getRp().getSslIssuerHashHeader(),
|
||||
tenantConfigurationManagement, tenantAware, systemSecurityContext);
|
||||
filterChain.add(securityHeaderFilter);
|
||||
|
||||
final ControllerPreAuthenticateSecurityTokenFilter securityTokenFilter = new ControllerPreAuthenticateSecurityTokenFilter(
|
||||
tenantConfigurationManagement, controllerManagement, tenantAware, systemSecurityContext);
|
||||
filterChain.add(securityTokenFilter);
|
||||
|
||||
final ControllerPreAuthenticatedAnonymousDownload anonymousDownloadFilter = new ControllerPreAuthenticatedAnonymousDownload(
|
||||
tenantConfigurationManagement, tenantAware, systemSecurityContext);
|
||||
filterChain.add(anonymousDownloadFilter);
|
||||
|
||||
filterChain.add(new ControllerPreAuthenticatedAnonymousFilter(ddiSecruityProperties));
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs authentication with the security token.
|
||||
*
|
||||
* @param securityToken
|
||||
* the authentication request object
|
||||
* @return the authentication object
|
||||
*/
|
||||
public Authentication doAuthenticate(final TenantSecurityToken securityToken) {
|
||||
resolveTenant(securityToken);
|
||||
PreAuthenticatedAuthenticationToken authentication = new PreAuthenticatedAuthenticationToken(null, null);
|
||||
for (final PreAuthentificationFilter filter : filterChain) {
|
||||
final PreAuthenticatedAuthenticationToken authenticationRest = createAuthentication(filter, securityToken);
|
||||
if (authenticationRest != null) {
|
||||
authentication = authenticationRest;
|
||||
authentication.setDetails(new TenantAwareAuthenticationDetails(securityToken.getTenant(), true));
|
||||
break;
|
||||
}
|
||||
}
|
||||
return preAuthenticatedAuthenticationProvider.authenticate(authentication);
|
||||
|
||||
}
|
||||
|
||||
private void resolveTenant(final TenantSecurityToken securityToken) {
|
||||
if (securityToken.getTenant() == null) {
|
||||
securityToken.setTenant(systemSecurityContext
|
||||
.runAsSystem(() -> systemManagement.getTenantMetadata(securityToken.getTenantId()).getTenant()));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static PreAuthenticatedAuthenticationToken createAuthentication(final PreAuthentificationFilter filter,
|
||||
final TenantSecurityToken secruityToken) {
|
||||
|
||||
if (!filter.isEnable(secruityToken)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final Object principal = filter.getPreAuthenticatedPrincipal(secruityToken);
|
||||
final Object credentials = filter.getPreAuthenticatedCredentials(secruityToken);
|
||||
|
||||
if (principal == null) {
|
||||
LOGGER.debug("No pre-authenticated principal found in message");
|
||||
return null;
|
||||
}
|
||||
|
||||
LOGGER.debug("preAuthenticatedPrincipal = {} trying to authenticate", principal);
|
||||
|
||||
return new PreAuthenticatedAuthenticationToken(principal, credentials);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
/**
|
||||
* 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 java.time.Duration;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.amqp.core.Queue;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
import com.google.common.collect.Maps;
|
||||
|
||||
/**
|
||||
* Bean which holds the necessary properties for configuring the AMQP deadletter
|
||||
* queue.
|
||||
*/
|
||||
@ConfigurationProperties("hawkbit.dmf.rabbitmq.deadLetter")
|
||||
public class AmqpDeadletterProperties {
|
||||
private static final int THREE_WEEKS = 21;
|
||||
|
||||
/**
|
||||
* Message time to live (ttl) for the deadletter queue. Default ttl is 3
|
||||
* weeks.
|
||||
*/
|
||||
private long ttl = Duration.ofDays(THREE_WEEKS).toMillis();
|
||||
|
||||
/**
|
||||
* Return the deadletter arguments.
|
||||
*
|
||||
* @param exchange
|
||||
* the deadletter exchange
|
||||
* @return map which holds the properties
|
||||
*/
|
||||
public Map<String, Object> getDeadLetterExchangeArgs(final String exchange) {
|
||||
final Map<String, Object> args = Maps.newHashMapWithExpectedSize(1);
|
||||
args.put("x-dead-letter-exchange", exchange);
|
||||
return args;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a deadletter queue with ttl for messages
|
||||
*
|
||||
* @param queueName
|
||||
* the deadlette queue name
|
||||
* @return the deadletter queue
|
||||
*/
|
||||
public Queue createDeadletterQueue(final String queueName) {
|
||||
return new Queue(queueName, true, false, false, getTTLArgs());
|
||||
}
|
||||
|
||||
private Map<String, Object> getTTLArgs() {
|
||||
final Map<String, Object> args = Maps.newHashMapWithExpectedSize(1);
|
||||
args.put("x-message-ttl", getTtl());
|
||||
return args;
|
||||
}
|
||||
|
||||
public long getTtl() {
|
||||
return ttl;
|
||||
}
|
||||
|
||||
public void setTtl(final long ttl) {
|
||||
this.ttl = ttl;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,271 @@
|
||||
/**
|
||||
* 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 java.net.URI;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.eclipse.hawkbit.api.ApiType;
|
||||
import org.eclipse.hawkbit.api.ArtifactUrl;
|
||||
import org.eclipse.hawkbit.api.ArtifactUrlHandler;
|
||||
import org.eclipse.hawkbit.api.URLPlaceholder;
|
||||
import org.eclipse.hawkbit.api.URLPlaceholder.SoftwareData;
|
||||
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.Artifact;
|
||||
import org.eclipse.hawkbit.dmf.json.model.ArtifactHash;
|
||||
import org.eclipse.hawkbit.dmf.json.model.DownloadAndUpdateRequest;
|
||||
import org.eclipse.hawkbit.dmf.json.model.SoftwareModule;
|
||||
import org.eclipse.hawkbit.repository.SystemManagement;
|
||||
import org.eclipse.hawkbit.repository.TargetManagement;
|
||||
import org.eclipse.hawkbit.repository.event.remote.TargetAssignDistributionSetEvent;
|
||||
import org.eclipse.hawkbit.repository.event.remote.TargetDeletedEvent;
|
||||
import org.eclipse.hawkbit.repository.event.remote.entity.CancelTargetAssignmentEvent;
|
||||
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.core.Message;
|
||||
import org.springframework.amqp.core.MessageProperties;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.cloud.bus.ServiceMatcher;
|
||||
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
|
||||
import org.springframework.context.event.EventListener;
|
||||
|
||||
/**
|
||||
* {@link AmqpMessageDispatcherService} create all outgoing AMQP messages and
|
||||
* delegate the messages to a {@link AmqpSenderService}.
|
||||
*
|
||||
* Additionally the dispatcher listener/subscribe for some target events e.g.
|
||||
* assignment.
|
||||
*
|
||||
*/
|
||||
public class AmqpMessageDispatcherService extends BaseAmqpService {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(AmqpMessageDispatcherService.class);
|
||||
|
||||
private final ArtifactUrlHandler artifactUrlHandler;
|
||||
private final AmqpSenderService amqpSenderService;
|
||||
private final SystemSecurityContext systemSecurityContext;
|
||||
private final SystemManagement systemManagement;
|
||||
private final TargetManagement targetManagement;
|
||||
private final ServiceMatcher serviceMatcher;
|
||||
|
||||
/**
|
||||
* Constructor.
|
||||
*
|
||||
* @param rabbitTemplate
|
||||
* the rabbitTemplate
|
||||
* @param amqpSenderService
|
||||
* to send AMQP message
|
||||
* @param artifactUrlHandler
|
||||
* for generating download URLs
|
||||
* @param systemSecurityContext
|
||||
* for execution with system permissions
|
||||
* @param systemManagement
|
||||
* the systemManagement
|
||||
* @param targetManagement
|
||||
* to access target information
|
||||
* @param serviceMatcher
|
||||
* to check in cluster case if the message is from the same
|
||||
* cluster node
|
||||
*/
|
||||
public AmqpMessageDispatcherService(final RabbitTemplate rabbitTemplate, final AmqpSenderService amqpSenderService,
|
||||
final ArtifactUrlHandler artifactUrlHandler, final SystemSecurityContext systemSecurityContext,
|
||||
final SystemManagement systemManagement, final TargetManagement targetManagement,
|
||||
final ServiceMatcher serviceMatcher) {
|
||||
super(rabbitTemplate);
|
||||
this.artifactUrlHandler = artifactUrlHandler;
|
||||
this.amqpSenderService = amqpSenderService;
|
||||
this.systemSecurityContext = systemSecurityContext;
|
||||
this.systemManagement = systemManagement;
|
||||
this.targetManagement = targetManagement;
|
||||
this.serviceMatcher = serviceMatcher;
|
||||
}
|
||||
|
||||
/**
|
||||
* Method to send a message to a RabbitMQ Exchange after the Distribution
|
||||
* set has been assign to a Target.
|
||||
*
|
||||
* @param assignedEvent
|
||||
* the object to be send.
|
||||
*/
|
||||
@EventListener(classes = TargetAssignDistributionSetEvent.class)
|
||||
public void targetAssignDistributionSet(final TargetAssignDistributionSetEvent assignedEvent) {
|
||||
if (isNotFromSelf(assignedEvent)) {
|
||||
return;
|
||||
}
|
||||
|
||||
LOG.debug("targetAssignDistributionSet retrieved for controller {}. I will forward it to DMF broker.",
|
||||
assignedEvent.getControllerId());
|
||||
|
||||
targetManagement.findTargetByControllerID(assignedEvent.getControllerId())
|
||||
.ifPresent(target -> sendUpdateMessageToTarget(assignedEvent.getTenant(), target,
|
||||
assignedEvent.getActionId(), assignedEvent.getModules()));
|
||||
|
||||
}
|
||||
|
||||
void sendUpdateMessageToTarget(final String tenant, final Target target, final Long actionId,
|
||||
final Collection<org.eclipse.hawkbit.repository.model.SoftwareModule> modules) {
|
||||
|
||||
final URI targetAdress = target.getAddress();
|
||||
if (!IpUtil.isAmqpUri(targetAdress)) {
|
||||
return;
|
||||
}
|
||||
|
||||
final DownloadAndUpdateRequest downloadAndUpdateRequest = new DownloadAndUpdateRequest();
|
||||
downloadAndUpdateRequest.setActionId(actionId);
|
||||
|
||||
final String targetSecurityToken = systemSecurityContext.runAsSystem(target::getSecurityToken);
|
||||
downloadAndUpdateRequest.setTargetSecurityToken(targetSecurityToken);
|
||||
|
||||
for (final org.eclipse.hawkbit.repository.model.SoftwareModule softwareModule : modules) {
|
||||
final SoftwareModule amqpSoftwareModule = convertToAmqpSoftwareModule(target, softwareModule);
|
||||
downloadAndUpdateRequest.addSoftwareModule(amqpSoftwareModule);
|
||||
}
|
||||
|
||||
final Message message = getMessageConverter().toMessage(downloadAndUpdateRequest,
|
||||
createConnectorMessagePropertiesEvent(tenant, target.getControllerId(),
|
||||
EventTopic.DOWNLOAD_AND_INSTALL));
|
||||
amqpSenderService.sendMessage(message, targetAdress);
|
||||
}
|
||||
|
||||
/**
|
||||
* Method to send a message to a RabbitMQ Exchange after the assignment of
|
||||
* the Distribution set to a Target has been canceled.
|
||||
*
|
||||
* @param cancelEvent
|
||||
* the object to be send.
|
||||
*/
|
||||
@EventListener(classes = CancelTargetAssignmentEvent.class)
|
||||
public void targetCancelAssignmentToDistributionSet(final CancelTargetAssignmentEvent cancelEvent) {
|
||||
if (isNotFromSelf(cancelEvent)) {
|
||||
return;
|
||||
}
|
||||
|
||||
sendCancelMessageToTarget(cancelEvent.getTenant(), cancelEvent.getEntity().getControllerId(),
|
||||
cancelEvent.getActionId(), cancelEvent.getEntity().getAddress());
|
||||
}
|
||||
|
||||
/**
|
||||
* Method to send a message to a RabbitMQ Exchange after a Target was
|
||||
* deleted.
|
||||
*
|
||||
* @param deleteEvent
|
||||
* the TargetDeletedEvent which holds the necessary data for
|
||||
* sending a target delete message.
|
||||
*/
|
||||
@EventListener(classes = TargetDeletedEvent.class)
|
||||
public void targetDelete(final TargetDeletedEvent deleteEvent) {
|
||||
if (isNotFromSelf(deleteEvent)) {
|
||||
return;
|
||||
}
|
||||
sendDeleteMessage(deleteEvent.getTenant(), deleteEvent.getControllerId(), deleteEvent.getTargetAddress());
|
||||
}
|
||||
|
||||
void sendDeleteMessage(final String tenant, final String controllerId, final String targetAddress) {
|
||||
|
||||
if (!hasValidAddress(targetAddress)) {
|
||||
return;
|
||||
}
|
||||
|
||||
final Message message = new Message(null, createConnectorMessagePropertiesDeleteThing(tenant, controllerId));
|
||||
amqpSenderService.sendMessage(message, URI.create(targetAddress));
|
||||
}
|
||||
|
||||
private boolean hasValidAddress(final String targetAddress) {
|
||||
return targetAddress != null && IpUtil.isAmqpUri(URI.create(targetAddress));
|
||||
}
|
||||
|
||||
private boolean isNotFromSelf(final RemoteApplicationEvent event) {
|
||||
return serviceMatcher != null && !serviceMatcher.isFromSelf(event);
|
||||
}
|
||||
|
||||
void sendCancelMessageToTarget(final String tenant, final String controllerId, final Long actionId,
|
||||
final URI address) {
|
||||
if (!IpUtil.isAmqpUri(address)) {
|
||||
return;
|
||||
}
|
||||
final Message message = getMessageConverter().toMessage(actionId,
|
||||
createConnectorMessagePropertiesEvent(tenant, controllerId, EventTopic.CANCEL_DOWNLOAD));
|
||||
|
||||
amqpSenderService.sendMessage(message, address);
|
||||
|
||||
}
|
||||
|
||||
private static MessageProperties createConnectorMessagePropertiesEvent(final String tenant,
|
||||
final String controllerId, final EventTopic topic) {
|
||||
final MessageProperties messageProperties = createConnectorMessageProperties(tenant, controllerId);
|
||||
messageProperties.setHeader(MessageHeaderKey.TOPIC, topic);
|
||||
messageProperties.setHeader(MessageHeaderKey.TYPE, MessageType.EVENT);
|
||||
return messageProperties;
|
||||
}
|
||||
|
||||
private static MessageProperties createConnectorMessagePropertiesDeleteThing(final String tenant,
|
||||
final String controllerId) {
|
||||
final MessageProperties messageProperties = createConnectorMessageProperties(tenant, controllerId);
|
||||
messageProperties.setHeader(MessageHeaderKey.TYPE, MessageType.THING_DELETED);
|
||||
return messageProperties;
|
||||
}
|
||||
|
||||
private static MessageProperties createConnectorMessageProperties(final String tenant, final String controllerId) {
|
||||
final MessageProperties messageProperties = new MessageProperties();
|
||||
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
|
||||
messageProperties.setHeader(MessageHeaderKey.CONTENT_TYPE, MessageProperties.CONTENT_TYPE_JSON);
|
||||
messageProperties.setHeader(MessageHeaderKey.THING_ID, controllerId);
|
||||
messageProperties.setHeader(MessageHeaderKey.TENANT, tenant);
|
||||
return messageProperties;
|
||||
}
|
||||
|
||||
private SoftwareModule convertToAmqpSoftwareModule(final Target target,
|
||||
final org.eclipse.hawkbit.repository.model.SoftwareModule softwareModule) {
|
||||
final SoftwareModule amqpSoftwareModule = new SoftwareModule();
|
||||
amqpSoftwareModule.setModuleId(softwareModule.getId());
|
||||
amqpSoftwareModule.setModuleType(softwareModule.getType().getKey());
|
||||
amqpSoftwareModule.setModuleVersion(softwareModule.getVersion());
|
||||
|
||||
final List<Artifact> artifacts = convertArtifacts(target, softwareModule.getArtifacts());
|
||||
amqpSoftwareModule.setArtifacts(artifacts);
|
||||
return amqpSoftwareModule;
|
||||
}
|
||||
|
||||
private List<Artifact> convertArtifacts(final Target target,
|
||||
final List<org.eclipse.hawkbit.repository.model.Artifact> localArtifacts) {
|
||||
if (localArtifacts.isEmpty()) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
||||
return localArtifacts.stream().map(localArtifact -> convertArtifact(target, localArtifact))
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
|
||||
private Artifact convertArtifact(final Target target,
|
||||
final org.eclipse.hawkbit.repository.model.Artifact localArtifact) {
|
||||
final Artifact artifact = new Artifact();
|
||||
|
||||
artifact.setUrls(artifactUrlHandler
|
||||
.getUrls(new URLPlaceholder(systemManagement.getTenantMetadata().getTenant(),
|
||||
systemManagement.getTenantMetadata().getId(), target.getControllerId(), target.getId(),
|
||||
new SoftwareData(localArtifact.getSoftwareModule().getId(), localArtifact.getFilename(),
|
||||
localArtifact.getId(), localArtifact.getSha1Hash())),
|
||||
ApiType.DMF)
|
||||
.stream().collect(Collectors.toMap(ArtifactUrl::getProtocol, ArtifactUrl::getRef)));
|
||||
|
||||
artifact.setFilename(localArtifact.getFilename());
|
||||
artifact.setHashes(new ArtifactHash(localArtifact.getSha1Hash(), localArtifact.getMd5Hash()));
|
||||
artifact.setSize(localArtifact.getSize());
|
||||
return artifact;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,331 @@
|
||||
/**
|
||||
* 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 java.net.URI;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.apache.commons.lang3.ArrayUtils;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
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.ActionUpdateStatus;
|
||||
import org.eclipse.hawkbit.dmf.json.model.AttributeUpdate;
|
||||
import org.eclipse.hawkbit.im.authentication.SpPermission.SpringEvalExpressions;
|
||||
import org.eclipse.hawkbit.im.authentication.TenantAwareAuthenticationDetails;
|
||||
import org.eclipse.hawkbit.repository.ControllerManagement;
|
||||
import org.eclipse.hawkbit.repository.EntityFactory;
|
||||
import org.eclipse.hawkbit.repository.RepositoryConstants;
|
||||
import org.eclipse.hawkbit.repository.builder.ActionStatusCreate;
|
||||
import org.eclipse.hawkbit.repository.model.Action;
|
||||
import org.eclipse.hawkbit.repository.model.Action.Status;
|
||||
import org.eclipse.hawkbit.repository.model.Target;
|
||||
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.rabbit.annotation.RabbitListener;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.messaging.handler.annotation.Header;
|
||||
import org.springframework.security.authentication.AnonymousAuthenticationToken;
|
||||
import org.springframework.security.core.Authentication;
|
||||
import org.springframework.security.core.authority.SimpleGrantedAuthority;
|
||||
import org.springframework.security.core.context.SecurityContext;
|
||||
import org.springframework.security.core.context.SecurityContextHolder;
|
||||
import org.springframework.security.core.context.SecurityContextImpl;
|
||||
|
||||
/**
|
||||
*
|
||||
* {@link AmqpMessageHandlerService} handles all incoming target interaction
|
||||
* AMQP messages (e.g. create target, check for updates etc.) for the queue
|
||||
* which is configured for the property hawkbit.dmf.rabbitmq.receiverQueue.
|
||||
*
|
||||
*/
|
||||
public class AmqpMessageHandlerService extends BaseAmqpService {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(AmqpMessageHandlerService.class);
|
||||
|
||||
private final AmqpMessageDispatcherService amqpMessageDispatcherService;
|
||||
|
||||
private final ControllerManagement controllerManagement;
|
||||
|
||||
private final EntityFactory entityFactory;
|
||||
|
||||
/**
|
||||
* Constructor.
|
||||
*
|
||||
* @param rabbitTemplate
|
||||
* for converting messages
|
||||
* @param amqpMessageDispatcherService
|
||||
* to sending events to DMF client
|
||||
* @param controllerManagement
|
||||
* for target repo access
|
||||
* @param entityFactory
|
||||
* to create entities
|
||||
*/
|
||||
public AmqpMessageHandlerService(final RabbitTemplate rabbitTemplate,
|
||||
final AmqpMessageDispatcherService amqpMessageDispatcherService,
|
||||
final ControllerManagement controllerManagement, final EntityFactory entityFactory) {
|
||||
super(rabbitTemplate);
|
||||
this.amqpMessageDispatcherService = amqpMessageDispatcherService;
|
||||
this.controllerManagement = controllerManagement;
|
||||
this.entityFactory = entityFactory;
|
||||
}
|
||||
|
||||
/**
|
||||
* Method to handle all incoming DMF amqp messages.
|
||||
*
|
||||
* @param message
|
||||
* incoming message
|
||||
* @param type
|
||||
* the message type
|
||||
* @param tenant
|
||||
* the contentType of the message
|
||||
*
|
||||
* @return a message if <null> no message is send back to sender
|
||||
*/
|
||||
@RabbitListener(queues = "${hawkbit.dmf.rabbitmq.receiverQueue:dmf_receiver}", 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());
|
||||
}
|
||||
|
||||
/**
|
||||
* * Executed if a amqp message arrives.
|
||||
*
|
||||
* @param message
|
||||
* the message
|
||||
* @param type
|
||||
* the type
|
||||
* @param tenant
|
||||
* the tenant
|
||||
* @param virtualHost
|
||||
* the virtual host
|
||||
* @return the rpc message back to supplier.
|
||||
*/
|
||||
public Message onMessage(final Message message, final String type, final String tenant, final String virtualHost) {
|
||||
checkContentTypeJson(message);
|
||||
final SecurityContext oldContext = SecurityContextHolder.getContext();
|
||||
try {
|
||||
final MessageType messageType = MessageType.valueOf(type);
|
||||
switch (messageType) {
|
||||
case THING_CREATED:
|
||||
setTenantSecurityContext(tenant);
|
||||
registerTarget(message, virtualHost);
|
||||
break;
|
||||
case EVENT:
|
||||
setTenantSecurityContext(tenant);
|
||||
final String topicValue = getStringHeaderKey(message, MessageHeaderKey.TOPIC, "EventTopic is null");
|
||||
final EventTopic eventTopic = EventTopic.valueOf(topicValue);
|
||||
handleIncomingEvent(message, eventTopic);
|
||||
break;
|
||||
default:
|
||||
logAndThrowMessageError(message, "No handle method was found for the given message type.");
|
||||
}
|
||||
} catch (final IllegalArgumentException ex) {
|
||||
throw new AmqpRejectAndDontRequeueException("Invalid message!", ex);
|
||||
} finally {
|
||||
SecurityContextHolder.setContext(oldContext);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private static void setSecurityContext(final Authentication authentication) {
|
||||
final SecurityContextImpl securityContextImpl = new SecurityContextImpl();
|
||||
securityContextImpl.setAuthentication(authentication);
|
||||
SecurityContextHolder.setContext(securityContextImpl);
|
||||
}
|
||||
|
||||
private static void setTenantSecurityContext(final String tenantId) {
|
||||
final AnonymousAuthenticationToken authenticationToken = new AnonymousAuthenticationToken(
|
||||
UUID.randomUUID().toString(), "AMQP-Controller",
|
||||
Collections.singletonList(new SimpleGrantedAuthority(SpringEvalExpressions.CONTROLLER_ROLE_ANONYMOUS)));
|
||||
authenticationToken.setDetails(new TenantAwareAuthenticationDetails(tenantId, true));
|
||||
setSecurityContext(authenticationToken);
|
||||
}
|
||||
|
||||
/**
|
||||
* Method to create a new target or to find the target if it already exists.
|
||||
*
|
||||
* @param targetID
|
||||
* the ID of the target/thing
|
||||
* @param ip
|
||||
* the ip of the target/thing
|
||||
*/
|
||||
private void registerTarget(final Message message, final String virtualHost) {
|
||||
final String thingId = getStringHeaderKey(message, MessageHeaderKey.THING_ID, "ThingId is null");
|
||||
final String replyTo = message.getMessageProperties().getReplyTo();
|
||||
|
||||
if (StringUtils.isEmpty(replyTo)) {
|
||||
logAndThrowMessageError(message, "No ReplyTo was set for the createThing message.");
|
||||
}
|
||||
|
||||
final URI amqpUri = IpUtil.createAmqpUri(virtualHost, replyTo);
|
||||
final Target target = controllerManagement.findOrRegisterTargetIfItDoesNotexist(thingId, amqpUri);
|
||||
LOG.debug("Target {} reported online state.", thingId);
|
||||
|
||||
lookIfUpdateAvailable(target);
|
||||
}
|
||||
|
||||
private void lookIfUpdateAvailable(final Target target) {
|
||||
|
||||
final Optional<Action> actionOptional = controllerManagement
|
||||
.findOldestActiveActionByTarget(target.getControllerId());
|
||||
|
||||
if (!actionOptional.isPresent()) {
|
||||
return;
|
||||
}
|
||||
|
||||
final Action action = actionOptional.get();
|
||||
if (action.isCancelingOrCanceled()) {
|
||||
amqpMessageDispatcherService.sendCancelMessageToTarget(target.getTenant(), target.getControllerId(),
|
||||
action.getId(), target.getAddress());
|
||||
return;
|
||||
}
|
||||
|
||||
amqpMessageDispatcherService.sendUpdateMessageToTarget(action.getTenant(), action.getTarget(), action.getId(),
|
||||
action.getDistributionSet().getModules());
|
||||
}
|
||||
|
||||
/**
|
||||
* Method to handle the different topics to an event.
|
||||
*
|
||||
* @param message
|
||||
* the incoming event message.
|
||||
* @param topic
|
||||
* the topic of the event.
|
||||
*/
|
||||
private void handleIncomingEvent(final Message message, final EventTopic topic) {
|
||||
switch (topic) {
|
||||
case UPDATE_ACTION_STATUS:
|
||||
updateActionStatus(message);
|
||||
break;
|
||||
case UPDATE_ATTRIBUTES:
|
||||
updateAttributes(message);
|
||||
break;
|
||||
default:
|
||||
logAndThrowMessageError(message, "Got event without appropriate topic.");
|
||||
break;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private void updateAttributes(final Message message) {
|
||||
final AttributeUpdate attributeUpdate = convertMessage(message, AttributeUpdate.class);
|
||||
final String thingId = getStringHeaderKey(message, MessageHeaderKey.THING_ID, "ThingId is null");
|
||||
|
||||
controllerManagement.updateControllerAttributes(thingId, attributeUpdate.getAttributes());
|
||||
}
|
||||
|
||||
/**
|
||||
* Method to update the action status of an action through the event.
|
||||
*
|
||||
* @param actionUpdateStatus
|
||||
* the object form the ampq message
|
||||
*/
|
||||
private void updateActionStatus(final Message message) {
|
||||
final ActionUpdateStatus actionUpdateStatus = convertMessage(message, ActionUpdateStatus.class);
|
||||
final Action action = checkActionExist(message, actionUpdateStatus);
|
||||
|
||||
final List<String> messages = actionUpdateStatus.getMessage();
|
||||
if (ArrayUtils.isNotEmpty(message.getMessageProperties().getCorrelationId())) {
|
||||
messages.add(RepositoryConstants.SERVER_MESSAGE_PREFIX + "DMF message correlation-id "
|
||||
+ convertCorrelationId(message));
|
||||
}
|
||||
|
||||
final Status status = mapStatus(message, actionUpdateStatus, action);
|
||||
final ActionStatusCreate actionStatus = entityFactory.actionStatus().create(action.getId()).status(status)
|
||||
.messages(messages);
|
||||
|
||||
final Action addUpdateActionStatus = getUpdateActionStatus(status, actionStatus);
|
||||
|
||||
if (!addUpdateActionStatus.isActive()) {
|
||||
lookIfUpdateAvailable(action.getTarget());
|
||||
}
|
||||
}
|
||||
|
||||
private Status mapStatus(final Message message, final ActionUpdateStatus actionUpdateStatus, final Action action) {
|
||||
Status status = null;
|
||||
switch (actionUpdateStatus.getActionStatus()) {
|
||||
case DOWNLOAD:
|
||||
status = Status.DOWNLOAD;
|
||||
break;
|
||||
case RETRIEVED:
|
||||
status = Status.RETRIEVED;
|
||||
break;
|
||||
case RUNNING:
|
||||
status = Status.RUNNING;
|
||||
break;
|
||||
case CANCELED:
|
||||
status = Status.CANCELED;
|
||||
break;
|
||||
case FINISHED:
|
||||
status = Status.FINISHED;
|
||||
break;
|
||||
case ERROR:
|
||||
status = Status.ERROR;
|
||||
break;
|
||||
case WARNING:
|
||||
status = Status.WARNING;
|
||||
break;
|
||||
case CANCEL_REJECTED:
|
||||
status = hanldeCancelRejectedState(message, action);
|
||||
break;
|
||||
default:
|
||||
logAndThrowMessageError(message, "Status for action does not exisit.");
|
||||
}
|
||||
|
||||
return status;
|
||||
}
|
||||
|
||||
private Status hanldeCancelRejectedState(final Message message, final Action action) {
|
||||
if (action.isCancelingOrCanceled()) {
|
||||
return Status.CANCEL_REJECTED;
|
||||
}
|
||||
logAndThrowMessageError(message,
|
||||
"Cancel rejected message is not allowed, if action is on state: " + action.getStatus());
|
||||
return null;
|
||||
|
||||
}
|
||||
|
||||
private static String convertCorrelationId(final Message message) {
|
||||
return new String(message.getMessageProperties().getCorrelationId(), StandardCharsets.UTF_8);
|
||||
}
|
||||
|
||||
private Action getUpdateActionStatus(final Status status, final ActionStatusCreate actionStatus) {
|
||||
if (Status.CANCELED.equals(status)) {
|
||||
return controllerManagement.addCancelActionStatus(actionStatus);
|
||||
}
|
||||
return controllerManagement.addUpdateActionStatus(actionStatus);
|
||||
}
|
||||
|
||||
// Exception squid:S3655 - logAndThrowMessageError throws exception, i.e.
|
||||
// get will not be called
|
||||
@SuppressWarnings("squid:S3655")
|
||||
private Action checkActionExist(final Message message, final ActionUpdateStatus actionUpdateStatus) {
|
||||
final Long actionId = actionUpdateStatus.getActionId();
|
||||
|
||||
LOG.debug("Target notifies intermediate about action {} with status {}.", actionId,
|
||||
actionUpdateStatus.getActionStatus());
|
||||
|
||||
final Optional<Action> findActionWithDetails = controllerManagement.findActionWithDetails(actionId);
|
||||
if (!findActionWithDetails.isPresent()) {
|
||||
logAndThrowMessageError(message,
|
||||
"Got intermediate notification about action " + actionId + " but action does not exist");
|
||||
}
|
||||
|
||||
return findActionWithDetails.get();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,196 @@
|
||||
/**
|
||||
* 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 java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
/**
|
||||
* Bean which holds the necessary properties for configuring the AMQP
|
||||
* connection.
|
||||
*
|
||||
*/
|
||||
@ConfigurationProperties("hawkbit.dmf.rabbitmq")
|
||||
public class AmqpProperties {
|
||||
|
||||
private static final int ONE_MINUTE = 60;
|
||||
|
||||
private static final int DEFAULT_QUEUE_DECLARATION_RETRIES = 50;
|
||||
|
||||
private static final int DEFAULT_INITIAL_CONSUMERS = 3;
|
||||
|
||||
private static final int DEFAULT_PREFETCH_COUNT = 10;
|
||||
|
||||
private static final int DEFAULT_MAX_CONSUMERS = 10;
|
||||
|
||||
private static final long DEFAULT_REQUEUE_DELAY = 0;
|
||||
|
||||
/**
|
||||
* Enable DMF API based on AMQP 0.9
|
||||
*/
|
||||
private boolean enabled = true;
|
||||
|
||||
/**
|
||||
* DMF API dead letter queue.
|
||||
*/
|
||||
private String deadLetterQueue = "dmf_connector_deadletter_ttl";
|
||||
|
||||
/**
|
||||
* DMF API dead letter exchange.
|
||||
*/
|
||||
private String deadLetterExchange = "dmf.connector.deadletter";
|
||||
|
||||
/**
|
||||
* 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;
|
||||
|
||||
/**
|
||||
* Requested heartbeat interval from broker in {@link TimeUnit#SECONDS}.
|
||||
*/
|
||||
private int requestedHeartBeat = (int) TimeUnit.SECONDS.toSeconds(ONE_MINUTE);
|
||||
|
||||
/**
|
||||
* Sets an upper limit to the number of consumers.
|
||||
*/
|
||||
private int maxConcurrentConsumers = DEFAULT_MAX_CONSUMERS;
|
||||
|
||||
/**
|
||||
* 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 = DEFAULT_PREFETCH_COUNT;
|
||||
|
||||
/**
|
||||
* Initial number of consumers. Is scaled up if necessary up to
|
||||
* {@link #maxConcurrentConsumers}.
|
||||
*/
|
||||
private int initialConcurrentConsumers = DEFAULT_INITIAL_CONSUMERS;
|
||||
|
||||
/**
|
||||
* 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 = DEFAULT_QUEUE_DECLARATION_RETRIES;
|
||||
|
||||
/**
|
||||
* Delay for messages that are requeued in milliseconds.
|
||||
*/
|
||||
private long requeueDelay = DEFAULT_REQUEUE_DELAY;
|
||||
|
||||
public long getRequeueDelay() {
|
||||
return requeueDelay;
|
||||
}
|
||||
|
||||
public void setRequeueDelay(final long requeueDelay) {
|
||||
this.requeueDelay = requeueDelay;
|
||||
}
|
||||
|
||||
public int getDeclarationRetries() {
|
||||
return declarationRetries;
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
public boolean isMissingQueuesFatal() {
|
||||
return missingQueuesFatal;
|
||||
}
|
||||
|
||||
public void setMissingQueuesFatal(final boolean missingQueuesFatal) {
|
||||
this.missingQueuesFatal = missingQueuesFatal;
|
||||
}
|
||||
|
||||
public String getDeadLetterExchange() {
|
||||
return deadLetterExchange;
|
||||
}
|
||||
|
||||
public void setDeadLetterExchange(final String deadLetterExchange) {
|
||||
this.deadLetterExchange = deadLetterExchange;
|
||||
}
|
||||
|
||||
public String getDeadLetterQueue() {
|
||||
return deadLetterQueue;
|
||||
}
|
||||
|
||||
public void setDeadLetterQueue(final String deadLetterQueue) {
|
||||
this.deadLetterQueue = deadLetterQueue;
|
||||
}
|
||||
|
||||
public String getReceiverQueue() {
|
||||
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;
|
||||
}
|
||||
|
||||
public boolean isEnabled() {
|
||||
return enabled;
|
||||
}
|
||||
|
||||
public void setEnabled(final boolean enabled) {
|
||||
this.enabled = enabled;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
/**
|
||||
* 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 java.net.URI;
|
||||
|
||||
import javax.validation.constraints.NotNull;
|
||||
|
||||
import org.springframework.amqp.core.Message;
|
||||
|
||||
/**
|
||||
* Interface to send a amqp message.
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface AmqpSenderService {
|
||||
|
||||
/**
|
||||
* Send the given message to the given uri. The uri contains the (virtual)
|
||||
* host and exchange e.g amqp://host/exchange.
|
||||
*
|
||||
* @param message
|
||||
* the amqp message
|
||||
* @param replyTo
|
||||
* the reply to uri
|
||||
*/
|
||||
void sendMessage(@NotNull Message message, @NotNull URI replyTo);
|
||||
|
||||
/**
|
||||
* Extract the exchange from the uri. Default implementation removes the
|
||||
* first /.
|
||||
*
|
||||
* @param amqpUri
|
||||
* the amqp uri
|
||||
* @return the exchange.
|
||||
*/
|
||||
default String extractExchange(final URI amqpUri) {
|
||||
return amqpUri.getPath().substring(1);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
/**
|
||||
* 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 java.util.Map;
|
||||
|
||||
import javax.validation.constraints.NotNull;
|
||||
|
||||
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.core.RabbitTemplate;
|
||||
import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
|
||||
import org.springframework.amqp.support.converter.MessageConversionException;
|
||||
import org.springframework.amqp.support.converter.MessageConverter;
|
||||
|
||||
/**
|
||||
* A base class which provide basis amqp staff.
|
||||
*/
|
||||
public class BaseAmqpService {
|
||||
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(BaseAmqpService.class);
|
||||
private final RabbitTemplate rabbitTemplate;
|
||||
|
||||
/**
|
||||
* Constructor.
|
||||
*
|
||||
* @param rabbitTemplate
|
||||
* the rabbit template.
|
||||
*/
|
||||
public BaseAmqpService(final RabbitTemplate rabbitTemplate) {
|
||||
this.rabbitTemplate = rabbitTemplate;
|
||||
}
|
||||
|
||||
protected static void checkContentTypeJson(final Message message) {
|
||||
final MessageProperties messageProperties = message.getMessageProperties();
|
||||
if (messageProperties.getContentType() != null && messageProperties.getContentType().contains("json")) {
|
||||
return;
|
||||
}
|
||||
throw new AmqpRejectAndDontRequeueException("Content-Type is not JSON compatible");
|
||||
}
|
||||
|
||||
/**
|
||||
* Is needed to convert a incoming message to is originally object type.
|
||||
*
|
||||
* @param message
|
||||
* the message to convert.
|
||||
* @param clazz
|
||||
* the class of the originally object.
|
||||
* @return the converted object
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public <T> T convertMessage(@NotNull final Message message, final Class<T> clazz) {
|
||||
checkMessageBody(message);
|
||||
message.getMessageProperties().getHeaders().put(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME,
|
||||
clazz.getName());
|
||||
return (T) rabbitTemplate.getMessageConverter().fromMessage(message);
|
||||
}
|
||||
|
||||
protected MessageConverter getMessageConverter() {
|
||||
return rabbitTemplate.getMessageConverter();
|
||||
}
|
||||
|
||||
private static boolean isMessageBodyEmpty(final Message message) {
|
||||
return message.getBody() == null || message.getBody().length == 0;
|
||||
}
|
||||
|
||||
protected void checkMessageBody(@NotNull final Message message) {
|
||||
if (isMessageBodyEmpty(message)) {
|
||||
throw new MessageConversionException("Message body cannot be null");
|
||||
}
|
||||
}
|
||||
|
||||
protected String getStringHeaderKey(final Message message, final String key, final String errorMessageIfNull) {
|
||||
final Map<String, Object> header = message.getMessageProperties().getHeaders();
|
||||
final Object value = header.get(key);
|
||||
if (value == null) {
|
||||
logAndThrowMessageError(message, errorMessageIfNull);
|
||||
return null;
|
||||
}
|
||||
return value.toString();
|
||||
}
|
||||
|
||||
protected static final void logAndThrowMessageError(final Message message, final String error) {
|
||||
LOGGER.warn("Warning! \"{}\" reported by message: {}", error, message);
|
||||
throw new AmqpRejectAndDontRequeueException(error);
|
||||
}
|
||||
|
||||
protected RabbitTemplate getRabbitTemplate() {
|
||||
return rabbitTemplate;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean message properties before sending a message.
|
||||
*
|
||||
* @param message
|
||||
* the message to cleaned up
|
||||
*/
|
||||
protected void cleanMessageHeaderProperties(final Message message) {
|
||||
message.getMessageProperties().getHeaders().remove(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
/**
|
||||
* 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;
|
||||
import org.springframework.util.ErrorHandler;
|
||||
|
||||
/**
|
||||
* {@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
|
||||
* @param errorHandler
|
||||
* the error handler which should be use
|
||||
*/
|
||||
public ConfigurableRabbitListenerContainerFactory(final AmqpProperties amqpProperties,
|
||||
final ConnectionFactory rabbitConnectionFactory, final ErrorHandler errorHandler) {
|
||||
this.amqpProperties = amqpProperties;
|
||||
setErrorHandler(errorHandler);
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
/**
|
||||
* 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 java.net.URI;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
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
|
||||
* message to the configured spring rabbitmq connections. The exchange is
|
||||
* extracted from the uri.
|
||||
*/
|
||||
public class DefaultAmqpSenderService implements AmqpSenderService {
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(DefaultAmqpSenderService.class);
|
||||
|
||||
private final RabbitTemplate internalAmqpTemplate;
|
||||
|
||||
/**
|
||||
* Constructor.
|
||||
*
|
||||
* @param internalAmqpTemplate
|
||||
* the amqp template
|
||||
*/
|
||||
public DefaultAmqpSenderService(final RabbitTemplate internalAmqpTemplate) {
|
||||
this.internalAmqpTemplate = internalAmqpTemplate;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sendMessage(final Message message, final URI replyTo) {
|
||||
if (!IpUtil.isAmqpUri(replyTo)) {
|
||||
return;
|
||||
}
|
||||
|
||||
final String correlationId = UUID.randomUUID().toString();
|
||||
final String exchange = extractExchange(replyTo);
|
||||
message.getMessageProperties().setCorrelationId(correlationId.getBytes(StandardCharsets.UTF_8));
|
||||
|
||||
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));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
/**
|
||||
* 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 java.util.concurrent.TimeUnit;
|
||||
|
||||
import javax.validation.ConstraintViolationException;
|
||||
|
||||
import org.eclipse.hawkbit.repository.exception.CancelActionNotAllowedException;
|
||||
import org.eclipse.hawkbit.repository.exception.EntityNotFoundException;
|
||||
import org.eclipse.hawkbit.repository.exception.InvalidTargetAddressException;
|
||||
import org.eclipse.hawkbit.repository.exception.TenantNotExistException;
|
||||
import org.eclipse.hawkbit.repository.exception.ToManyAttributeEntriesException;
|
||||
import org.eclipse.hawkbit.repository.exception.TooManyStatusEntriesException;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
|
||||
import org.springframework.amqp.rabbit.listener.FatalExceptionStrategy;
|
||||
import org.springframework.amqp.support.converter.MessageConversionException;
|
||||
import org.springframework.messaging.MessageHandlingException;
|
||||
|
||||
/**
|
||||
* Custom {@link FatalExceptionStrategy} that markes defined hawkBit internal
|
||||
* exceptions not to be requeued. In addition it throttles in case of a requeue
|
||||
* by means of blocking the processing thread for a certain amount of time. That
|
||||
* avoids a back and forth between broker and hawkBit at maximum speed.
|
||||
*
|
||||
*/
|
||||
public class DelayedRequeueExceptionStrategy extends ConditionalRejectingErrorHandler.DefaultExceptionStrategy {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(DelayedRequeueExceptionStrategy.class);
|
||||
|
||||
private final long delay;
|
||||
|
||||
/**
|
||||
* @param delay
|
||||
* in {@link TimeUnit#MILLISECONDS} before requeue.
|
||||
*/
|
||||
public DelayedRequeueExceptionStrategy(final long delay) {
|
||||
this.delay = delay;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean isUserCauseFatal(final Throwable cause) {
|
||||
if (invalidMessage(cause)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
LOG.error("Found a message that has to be requeued. Processing with delay of {}ms: ", delay, cause);
|
||||
|
||||
try {
|
||||
TimeUnit.MILLISECONDS.sleep(delay);
|
||||
} catch (final InterruptedException e) {
|
||||
LOG.error("Delay interrupted!", e);
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
private boolean invalidMessage(final Throwable cause) {
|
||||
return doesNotExist(cause) || quotaHit(cause) || invalidContent(cause) || invalidState(cause);
|
||||
}
|
||||
|
||||
private boolean invalidState(final Throwable cause) {
|
||||
return cause instanceof CancelActionNotAllowedException;
|
||||
}
|
||||
|
||||
private boolean quotaHit(final Throwable cause) {
|
||||
return cause instanceof TooManyStatusEntriesException || cause instanceof ToManyAttributeEntriesException;
|
||||
}
|
||||
|
||||
private boolean doesNotExist(final Throwable cause) {
|
||||
return cause instanceof TenantNotExistException || cause instanceof EntityNotFoundException;
|
||||
}
|
||||
|
||||
private boolean invalidContent(final Throwable cause) {
|
||||
return cause instanceof ConstraintViolationException || cause instanceof InvalidTargetAddressException
|
||||
|| cause instanceof MessageConversionException || cause instanceof MessageHandlingException;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
/**
|
||||
* 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.context.annotation.ComponentScan;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
|
||||
/**
|
||||
* Enable Device Management Federation API.
|
||||
*/
|
||||
@Configuration
|
||||
@ComponentScan
|
||||
@Import(AmqpConfiguration.class)
|
||||
public class DmfApiConfiguration {
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user