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

Fix scheduled executor, auth exchange and simulator poll.

Browse Source 
Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com>
This commit is contained in:
kaizimmerm
2016-06-24 13:59:19 +02:00
parent 7857107b46
commit 23cb62b9d9
17 changed files with 185 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -8,8 +8,11 @@
*/
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 java.util.concurrent.ThreadPoolExecutor;
import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
import org.slf4j.Logger;
@@ -18,6 +21,7 @@ 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.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -51,10 +55,6 @@ public class AmqpConfiguration {
@Autowired
private AmqpDeadletterProperties amqpDeadletterProperties;
@Autowired
@Qualifier("threadPoolExecutor")
private ThreadPoolExecutor threadPoolExecutor;
@Autowired
private ConnectionFactory rabbitConnectionFactory;
@@ -66,8 +66,8 @@ public class AmqpConfiguration {
private AmqpProperties amqpProperties;
@Autowired
@Qualifier("threadPoolExecutor")
private ThreadPoolExecutor threadPoolExecutor;
@Qualifier("asyncExecutor")
private Executor threadPoolExecutor;
@Autowired
private ScheduledExecutorService scheduledExecutorService;
@@ -145,26 +145,71 @@ public class AmqpConfiguration {
}
/**
* Create the sp receiver queue.
* Create the DMF API receiver queue for
*
* @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.
*
@@ -181,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#receiverQueue()} to
* {@link AmqpConfiguration#senderExchange()}.
*
* @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());
}
/**
@@ -245,4 +279,11 @@ public class AmqpConfiguration {
return containerFactory;
}
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;
}
}

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

@@ -116,7 +116,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
}
/**
* Method to handle all incoming amqp messages.
* Method to handle all incoming DMF amqp messages.
*
* @param message
* incoming message
@@ -133,6 +133,12 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
return onMessage(message, type, tenant, getRabbitTemplate().getConnectionFactory().getVirtualHost());
}
@RabbitListener(queues = "${hawkbit.dmf.rabbitmq.authenticationReceiverQueue}", containerFactory = "listenerContainerFactory")
public Message onAuthenticationRequest(final Message message,
@Header(MessageHeaderKey.TENANT) final String tenant) {
return onAuthenticationRequest(message);
}
public Message onMessage(final Message message, final String type, final String tenant, final String virtualHost) {
checkContentTypeJson(message);
final SecurityContext oldContext = SecurityContextHolder.getContext();
@@ -149,8 +155,7 @@ 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.");
}
@@ -164,6 +169,20 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
return null;
}
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);
}
}
private Message handleAuthentifiactionMessage(final Message message) {
final DownloadResponse authentificationResponse = new DownloadResponse();
final MessageProperties messageProperties = message.getMessageProperties();

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

@@ -31,10 +31,16 @@ 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.
*/
@@ -62,6 +68,14 @@ public class AmqpProperties {
*/
private int initialConcurrentConsumers = 3;
public String getAuthenticationReceiverQueue() {
return authenticationReceiverQueue;
}
public void setAuthenticationReceiverQueue(final String authenticationReceiverQueue) {
this.authenticationReceiverQueue = authenticationReceiverQueue;
}
public int getPrefetchCount() {
return prefetchCount;
}
@@ -147,10 +161,6 @@ public class AmqpProperties {
return receiverQueue;
}
public void setReceiverQueue(final String receiverQueue) {
this.receiverQueue = receiverQueue;
}
public int getRequestedHeartBeat() {
return requestedHeartBeat;
}
@@ -159,4 +169,8 @@ public class AmqpProperties {
this.requestedHeartBeat = requestedHeartBeat;
}
public void setReceiverQueue(final String receiverQueue) {
this.receiverQueue = receiverQueue;
}
}

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

@@ -31,7 +31,7 @@ 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.DelegatingSecurityContextExecutor;
import org.springframework.security.concurrent.DelegatingSecurityContextExecutorService;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -78,18 +78,17 @@ public class AmqpTestConfiguration {
* @return ExecutorService with security context availability in thread
* execution..
*/
@Bean
@Bean(destroyMethod = "shutdown")
@ConditionalOnMissingBean
public Executor asyncExecutor() {
return new DelegatingSecurityContextExecutor(threadPoolExecutor());
return new DelegatingSecurityContextExecutorService(threadPoolExecutor());
}
/**
* @return central ThreadPoolExecutor for general purpose multi threaded
* operations. Tries an orderly shutdown when destroyed.
*/
@Bean(destroyMethod = "shutdown")
public ThreadPoolExecutor threadPoolExecutor() {
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());

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.sha1("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.sha1("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.sha1("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/AmqpMessageHandlerServiceTest.java
View File

@@ -273,14 +273,13 @@ 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 MessageProperties messageProperties = createMessageProperties(null);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, "123", FileResource.sha1("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);
@@ -292,7 +291,7 @@ 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 MessageProperties messageProperties = createMessageProperties(null);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, "123", FileResource.sha1("12345"));
final Message message = amqpMessageHandlerService.getMessageConverter().toMessage(securityToken,
messageProperties);
@@ -303,8 +302,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);
@@ -316,7 +314,7 @@ 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 MessageProperties messageProperties = createMessageProperties(null);
final TenantSecurityToken securityToken = new TenantSecurityToken(TENANT, "123", FileResource.sha1("12345"));
final Message message = amqpMessageHandlerService.getMessageConverter().toMessage(securityToken,
messageProperties);
@@ -334,8 +332,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);
@@ -411,7 +408,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);