Integrating GCP IoT Core with HawkBit

hawkbit-device-simulator/.gitignore

@@ -1,2 +1,5 @@
 /bin/
 /target/
+/pom2.xml
+/pom3.xml
+/pom4.xml

hawkbit-device-simulator/README.md

@@ -12,19 +12,6 @@ Or:
 run org.eclipse.hawkbit.simulator.DeviceSimulator
 ```
 
-## Deploy to cloud foundry environment
-
-- Go to ```target``` subfolder.
-- Run ```cf push```
-
-## Notes
-
-The simulator has user authentication enabled in **cloud profile**. Default credentials:
-*  username : admin
-*  passwd : admin
-
-This can be configured/disabled by spring boot properties
-
 ## hawkBit APIs
 
 The simulator supports `DDI` as well as the `DMF` integration APIs.

hawkbit-device-simulator/pom.xml

@@ -104,11 +104,15 @@
       </dependency>
       <dependency>
          <groupId>org.springframework.cloud</groupId>
-         <artifactId>spring-cloud-commons</artifactId>
+         <artifactId>spring-cloud-context</artifactId>
       </dependency>
       <dependency>
          <groupId>org.springframework.cloud</groupId>
-         <artifactId>spring-cloud-context</artifactId>
+         <artifactId>spring-cloud-starter-openfeign</artifactId>
+      </dependency>
+      <dependency>
+         <groupId>io.github.openfeign</groupId>
+         <artifactId>feign-jackson</artifactId>
       </dependency>
       <dependency>
          <groupId>com.google.guava</groupId>

hawkbit-device-simulator/src/main/java/org/eclipse/hawkbit/simulator/AllowAllWebSecurityConfig.java

@@ -0,0 +1,25 @@
+/**
+ * Copyright (c) 2019 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.simulator;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.config.annotation.web.configuration.WebSecurityConfigurerAdapter;
+
+/**
+ * Spring security configuration to grant access for all incomming requests.
+ */
+@Configuration
+public class AllowAllWebSecurityConfig extends WebSecurityConfigurerAdapter {
+
+    @Override
+    protected void configure(final HttpSecurity httpSec) throws Exception {
+        httpSec.csrf().disable().authorizeRequests().antMatchers("/**").permitAll().anyRequest().authenticated();
+    }
+}

hawkbit-device-simulator/src/main/java/org/eclipse/hawkbit/simulator/SimulatedDeviceFactory.java

@@ -18,7 +18,7 @@
 import org.eclipse.hawkbit.simulator.amqp.DmfSenderService;
 import org.eclipse.hawkbit.simulator.http.GatewayTokenInterceptor;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.netflix.feign.support.ResponseEntityDecoder;
+import org.springframework.cloud.openfeign.support.ResponseEntityDecoder;
 import org.springframework.hateoas.hal.Jackson2HalModule;
 import org.springframework.stereotype.Service;
 

hawkbit-device-simulator/src/main/java/org/eclipse/hawkbit/simulator/SimulationController.java

@@ -8,17 +8,21 @@
  */
 package org.eclipse.hawkbit.simulator;
 
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Collections;
+import java.util.Optional;
+
 import org.eclipse.hawkbit.simulator.AbstractSimulatedDevice.Protocol;
 import org.eclipse.hawkbit.simulator.amqp.AmqpProperties;
+import org.eclipse.hawkbit.simulator.amqp.DmfSenderService;
+import org.eclipse.hawkbit.simulator.amqp.SimulatedUpdate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
-import java.net.MalformedURLException;
-import java.net.URL;
-
 /**
  * REST endpoint for controlling the device simulator.
  */
@@ -33,8 +37,11 @@ public class SimulationController {
 
     private final SimulationProperties simulationProperties;
 
+    private Optional<DmfSenderService> spSenderService = Optional.empty();
+
     @Autowired
-    public SimulationController(final DeviceSimulatorRepository repository, final SimulatedDeviceFactory deviceFactory, final AmqpProperties amqpProperties, final SimulationProperties simulationProperties) {
+    public SimulationController(final DeviceSimulatorRepository repository, final SimulatedDeviceFactory deviceFactory,
+            final AmqpProperties amqpProperties, final SimulationProperties simulationProperties) {
         this.repository = repository;
         this.deviceFactory = deviceFactory;
         this.amqpProperties = amqpProperties;
@@ -56,8 +63,7 @@ public SimulationController(final DeviceSimulatorRepository repository, final Si
      *            the URL endpoint to be used of the hawkbit-update-server for
      *            DDI devices
      * @param pollDelay
-     *            number of delay in seconds to delay polling of DDI
-     *            devices
+     *            number of delay in seconds to delay polling of DDI devices
      * @param gatewayToken
      *            the hawkbit-update-server gatewaytoken in case authentication
      *            is enforced in hawkbit
@@ -87,9 +93,7 @@ ResponseEntity<String> start(@RequestParam(value = "name", defaultValue = "simul
         }
 
         if (protocol == Protocol.DMF_AMQP && isDmfDisabled()) {
-            return ResponseEntity.badRequest()
-                    .body("The AMQP interface has been disabled, to use DMF protocol you need to enable the AMQP interface via '"
-                            + AmqpProperties.CONFIGURATION_PREFIX + ".enabled=true'");
+            return createAmqpDisabledResponse();
         }
 
         for (int i = 0; i < amount; i++) {
@@ -102,11 +106,17 @@ ResponseEntity<String> start(@RequestParam(value = "name", defaultValue = "simul
         return ResponseEntity.ok("Updated " + amount + " " + protocol + " connected targets!");
     }
 
+    private ResponseEntity<String> createAmqpDisabledResponse() {
+        return ResponseEntity.badRequest().body(
+                "The AMQP interface has been disabled, to use DMF protocol you need to enable the AMQP interface via '"
+                        + AmqpProperties.CONFIGURATION_PREFIX + ".enabled=true'");
+    }
+
     /**
      * Update an attribute of a device.
      *
      * NOTE: This represents not the expected client behaviour for DDI, since a
-     *       DDI client shall only update its attributes if requested by hawkBit.
+     * DDI client shall only update its attributes if requested by hawkBit.
      *
      * @param tenant
      *            The tenant the device belongs to
@@ -150,7 +160,7 @@ ResponseEntity<String> update(@RequestParam(value = "tenant", required = false)
      *         if not found.
      */
     @GetMapping("/remove")
-    ResponseEntity remove(@RequestParam(value = "tenant", required = false) final String tenant,
+    ResponseEntity<String> remove(@RequestParam(value = "tenant", required = false) final String tenant,
             @RequestParam(value = "controllerid") final String controllerId) {
 
         final AbstractSimulatedDevice controller = repository
@@ -176,6 +186,39 @@ ResponseEntity<String> reset() {
         return ResponseEntity.ok("All simulated devices have been removed.");
     }
 
+    /**
+     * Report action as FINISHED Sends an UpdateActionStatus event with value
+     * FINISHED
+     *
+     * @return A response string that the action_finished event was sent
+     */
+    @GetMapping("/finishAction")
+    ResponseEntity<String> finishAction(@RequestParam(value = "actionId") final long actionId,
+            @RequestParam(value = "tenant", required = false) final String tenant,
+            @RequestParam(value = "controllerId") final String controllerId) {
+
+        if (!spSenderService.isPresent()) {
+            return createAmqpDisabledResponse();
+        }
+
+        final String theTenant = tenant != null && !tenant.isEmpty() ? tenant : simulationProperties.getDefaultTenant();
+        final AbstractSimulatedDevice device = repository.get(theTenant, controllerId);
+
+        if (device == null) {
+            return ResponseEntity.notFound().build();
+        }
+
+        spSenderService.get().finishUpdateProcess(new SimulatedUpdate(device.getTenant(), device.getId(), actionId),
+                Collections.singletonList("Simulation Finished."));
+
+        return ResponseEntity.ok(String.format("Action with id: [%d] reported as FINISHED", actionId));
+    }
+
+    @Autowired(required = false)
+    public void setSpSenderService(final DmfSenderService spSenderService) {
+        this.spSenderService = Optional.of(spSenderService);
+    }
+
     private boolean isDmfDisabled() {
         return !amqpProperties.isEnabled();
     }

hawkbit-device-simulator/src/main/java/org/eclipse/hawkbit/simulator/UpdateStatus.java

@@ -44,12 +44,12 @@ public UpdateStatus(final ResponseStatus responseStatus, final String message) {
     }
 
     /**
-     * Constructor including status message.
+     * Constructor including status messages.
      * 
      * @param responseStatus
      *            of the update
-     * @param messages
-     *            of the update status
+     * @param statusMessages
+     *            list of status messages
      */
     public UpdateStatus(final ResponseStatus responseStatus, final List<String> statusMessages) {
         this(responseStatus);

hawkbit-device-simulator/src/main/java/org/eclipse/hawkbit/simulator/amqp/AmqpConfiguration.java

@@ -14,23 +14,19 @@
 import org.eclipse.hawkbit.simulator.DeviceSimulatorRepository;
 import org.eclipse.hawkbit.simulator.DeviceSimulatorUpdater;
 import org.eclipse.hawkbit.simulator.SimulationProperties;
-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.RabbitTemplate;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.retry.backoff.ExponentialBackOffPolicy;
-import org.springframework.retry.support.RetryTemplate;
 
 import com.google.common.collect.Maps;
 
@@ -43,33 +39,6 @@
 @ConditionalOnProperty(prefix = AmqpProperties.CONFIGURATION_PREFIX, name = "enabled")
 public class AmqpConfiguration {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(AmqpConfiguration.class);
-
-    @Autowired
-    private AmqpProperties amqpProperties;
-
-    @Bean
-    RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
-        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
-        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 correlation ID {} confirmed by broker.", correlationData.getId());
-            } else {
-                LOGGER.error("Broker is unable to handle message with correlation ID {} : {}", correlationData.getId(),
-                        cause);
-            }
-
-        });
-
-        return rabbitTemplate;
-    }
-
     @Bean
     DmfReceiverService dmfReceiverService(final RabbitTemplate rabbitTemplate, final AmqpProperties amqpProperties,
             final DmfSenderService spSenderService, final DeviceSimulatorUpdater deviceUpdater,
@@ -83,18 +52,38 @@ DmfSenderService dmfSenderService(final RabbitTemplate rabbitTemplate, final Amq
         return new DmfSenderService(rabbitTemplate, amqpProperties, simulationProperties);
     }
 
+    @Bean
+    public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
+        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+
+        // It is necessary to define rabbitTemplate as a Bean and set
+        // Jackson2JsonMessageConverter explicitly here in order to convert only
+        // OUTCOMING messages to json. In case of INCOMING messages,
+        // Jackson2JsonMessageConverter can not handle messages with NULL
+        // payload (e.g. REQUEST_ATTRIBUTES_UPDATE), so the
+        // SimpleMessageConverter is used instead per default.
+        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
+
+        return rabbitTemplate;
+    }
+
     /**
      * Creates the receiver queue from update server for receiving message from
      * update server.
      *
      * @return the queue
      */
     @Bean
-    Queue receiverConnectorQueueFromHawkBit() {
-        final Map<String, Object> arguments = getTTLMaxArgs();
-
+    Queue receiverConnectorQueueFromHawkBit(final AmqpProperties amqpProperties) {
         return QueueBuilder.nonDurable(amqpProperties.getReceiverConnectorQueueFromSp()).autoDelete()
-                .withArguments(arguments).build();
+                .withArguments(getTTLMaxArgs()).build();
+    }
+
+    private static Map<String, Object> getTTLMaxArgs() {
+        final Map<String, Object> args = Maps.newHashMapWithExpectedSize(2);
+        args.put("x-message-ttl", Duration.ofDays(1).toMillis());
+        args.put("x-max-length", 100_000);
+        return args;
     }
 
     /**
@@ -103,7 +92,7 @@ Queue receiverConnectorQueueFromHawkBit() {
      * @return the exchange
      */
     @Bean
-    FanoutExchange exchangeQueueToConnector() {
+    FanoutExchange exchangeQueueToConnector(final AmqpProperties amqpProperties) {
         return new FanoutExchange(amqpProperties.getSenderForSpExchange(), false, true);
     }
 
@@ -115,15 +104,18 @@ FanoutExchange exchangeQueueToConnector() {
      * @return the binding and create the queue and exchange
      */
     @Bean
-    Binding bindReceiverQueueToSpExchange() {
-        return BindingBuilder.bind(receiverConnectorQueueFromHawkBit()).to(exchangeQueueToConnector());
+    Binding bindReceiverQueueToSpExchange(final AmqpProperties amqpProperties) {
+        return BindingBuilder.bind(receiverConnectorQueueFromHawkBit(amqpProperties))
+                .to(exchangeQueueToConnector(amqpProperties));
     }
 
-    private static Map<String, Object> getTTLMaxArgs() {
-        final Map<String, Object> args = Maps.newHashMapWithExpectedSize(2);
-        args.put("x-message-ttl", Duration.ofDays(1).toMillis());
-        args.put("x-max-length", 100_000);
-        return args;
-    }
+    @Configuration
+    @ConditionalOnProperty(prefix = AmqpProperties.CONFIGURATION_PREFIX, name = "customVhost")
+    protected static class CachingConnectionFactoryInitializer {
 
+        CachingConnectionFactoryInitializer(final CachingConnectionFactory connectionFactory,
+                final AmqpProperties amqpProperties) {
+            connectionFactory.setVirtualHost(amqpProperties.getCustomVhost());
+        }
+    }
 }

hawkbit-device-simulator/src/main/java/org/eclipse/hawkbit/simulator/amqp/AmqpProperties.java

@@ -51,6 +51,8 @@ public class AmqpProperties {
      */
     private int deadLetterTtl = 60_000;
 
+    private String customVhost;
+
     public boolean isCheckDmfHealth() {
         return checkDmfHealth;
     }
@@ -90,4 +92,12 @@ public boolean isEnabled() {
     public void setEnabled(final boolean enabled) {
         this.enabled = enabled;
     }
+
+    public String getCustomVhost() {
+        return customVhost;
+    }
+
+    public void setCustomVhost(final String customVhost) {
+        this.customVhost = customVhost;
+    }
 }