[#19] Enhance device communication API

* added support for device states * added automatic Pub/Sub tenant topic & subscription creation and deletion * added automatically sending of a config in case a config is requested or a new one created * extended config table with error field Signed-off-by: Matthias Kaemmer <[email protected]>
sotec-iot · Aug 2, 2023 · c965112 · c965112
1 parent c01bccb
commit c965112
Show file tree

Hide file tree

Showing 44 changed files with 2,828 additions and 341 deletions.
diff --git a/...ation/src/main/java/org/eclipse/hono/communication/api/DeviceCommunicationHttpServer.java b/...ation/src/main/java/org/eclipse/hono/communication/api/DeviceCommunicationHttpServer.java
@@ -22,9 +22,10 @@
 
 import javax.inject.Singleton;
 
-import org.eclipse.hono.communication.api.service.DatabaseSchemaCreator;
-import org.eclipse.hono.communication.api.service.DatabaseService;
 import org.eclipse.hono.communication.api.service.VertxHttpHandlerManagerService;
+import org.eclipse.hono.communication.api.service.communication.InternalTopicManager;
+import org.eclipse.hono.communication.api.service.database.DatabaseSchemaCreator;
+import org.eclipse.hono.communication.api.service.database.DatabaseService;
 import org.eclipse.hono.communication.core.app.ApplicationConfig;
 import org.eclipse.hono.communication.core.app.ServerConfig;
 import org.eclipse.hono.communication.core.http.AbstractVertxHttpServer;
@@ -45,22 +46,22 @@
 import io.vertx.ext.web.openapi.RouterBuilder;
 import io.vertx.ext.web.validation.BadRequestException;
 
-
 /**
  * Vertx HTTP Server for the device communication api.
  */
 @Singleton
 public class DeviceCommunicationHttpServer extends AbstractVertxHttpServer implements HttpServer {
+
     private final Logger log = LoggerFactory.getLogger(DeviceCommunicationHttpServer.class);
     private final String serverStartedMsg = "HTTP Server is listening at http://{}:{}";
     private final String serverFailedMsg = "HTTP Server failed to start: {}";
     private final VertxHttpHandlerManagerService httpHandlerManager;
 
     private final DatabaseService db;
     private final DatabaseSchemaCreator databaseSchemaCreator;
+    private final InternalTopicManager internalTopicManager;
     private List<HttpEndpointHandler> httpEndpointHandlers;
 
-
     /**
      * Creates a new DeviceCommunicationHttpServer with all dependencies.
      *
@@ -69,26 +70,27 @@ public class DeviceCommunicationHttpServer extends AbstractVertxHttpServer imple
      * @param httpHandlerManager    The http handler manager
      * @param databaseService       The database connection
      * @param databaseSchemaCreator The database migrations service
+     * @param internalTopicManager  The internal topic manager
      */
     public DeviceCommunicationHttpServer(final ApplicationConfig appConfigs,
                                          final Vertx vertx,
                                          final VertxHttpHandlerManagerService httpHandlerManager,
                                          final DatabaseService databaseService,
-                                         final DatabaseSchemaCreator databaseSchemaCreator) {
+                                         final DatabaseSchemaCreator databaseSchemaCreator, final InternalTopicManager internalTopicManager) {
         super(appConfigs, vertx);
         this.httpHandlerManager = httpHandlerManager;
         this.databaseSchemaCreator = databaseSchemaCreator;
         this.httpEndpointHandlers = new ArrayList<>();
         this.db = databaseService;
+        this.internalTopicManager = internalTopicManager;
     }
 
-
     @Override
     public void start() {
-        //Create Database Tables
         databaseSchemaCreator.createDBTables();
 
-        // Create Endpoints Router
+        internalTopicManager.initPubSub();
+
         this.httpEndpointHandlers = httpHandlerManager.getAvailableHandlerServices();
         RouterBuilder.create(this.vertx, appConfigs.getServerConfig().getOpenApiFilePath())
                 .onSuccess(routerBuilder -> {
@@ -97,7 +99,7 @@ public void start() {
                 })
                 .onFailure(error -> {
                     if (error != null) {
-                        log.error("Can not create Router: {}", error.getMessage());
+                        log.error("Can not create Router {}", error.getMessage());
                     } else {
                         log.error("Can not create Router");
                     }
@@ -106,7 +108,6 @@ public void start() {
 
                 });
 
-        // Wait until application is stopped
         Quarkus.waitForExit();
 
     }
@@ -118,7 +119,8 @@ public void start() {
      * @param httpEndpointHandlers All available http endpoint handlers
      * @return The created Router object
      */
-    Router createRouterWithEndpoints(final RouterBuilder routerBuilder, final List<HttpEndpointHandler> httpEndpointHandlers) {
+    Router createRouterWithEndpoints(final RouterBuilder routerBuilder,
+                                     final List<HttpEndpointHandler> httpEndpointHandlers) {
         for (HttpEndpointHandler handlerService : httpEndpointHandlers) {
             handlerService.addRoutes(routerBuilder);
         }
@@ -165,7 +167,6 @@ private void addReadinessHandlers(final Router router, final String readinessPat
         router.get(readinessPath).handler(healthCheckHandler);
     }
 
-
     private void addLivenessHandlers(final Router router, final String livenessPath) {
         log.info("Adding liveness path: {}", livenessPath);
         final HealthCheckHandler healthCheckHandler = HealthCheckHandler.create(vertx);
@@ -190,8 +191,8 @@ void startVertxServer(final Router router) {
                 .listen();
 
         serverCreationFuture
-                .onSuccess(server -> log.info(this.serverStartedMsg, serverConfigs.getServerUrl()
-                        , serverConfigs.getServerPort()))
+                .onSuccess(server -> log.info(this.serverStartedMsg, serverConfigs.getServerUrl(),
+                        serverConfigs.getServerPort()))
                 .onFailure(error -> log.info(this.serverFailedMsg, error.getMessage()));
     }
 
@@ -229,10 +230,8 @@ void addDefault404ExceptionHandler(final RoutingContext routingContext) {
         }
     }
 
-
     @Override
     public void stop() {
-        // stop server custom functionality
         db.close();
 
     }

diff --git a/...unication/src/main/java/org/eclipse/hono/communication/api/config/ApiCommonConstants.java b/...unication/src/main/java/org/eclipse/hono/communication/api/config/ApiCommonConstants.java
@@ -0,0 +1,36 @@
+/*
+ * ***********************************************************
+ *  Copyright (c) 2023 Contributors to the Eclipse Foundation
+ *  <p>
+ *  See the NOTICE file(s) distributed with this work for additional
+ *  information regarding copyright ownership.
+ *  <p>
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Eclipse Public License 2.0 which is available at
+ *  http://www.eclipse.org/legal/epl-2.0
+ *  <p>
+ *  SPDX-License-Identifier: EPL-2.0
+ * **********************************************************
+ *
+ */
+
+package org.eclipse.hono.communication.api.config;
+
+/**
+ * Device constant values.
+ */
+public final class ApiCommonConstants {
+
+    /**
+     * Path parameter name for tenantId.
+     */
+    public static final String TENANT_PATH_PARAMS = "tenantid";
+    /**
+     * Path parameter name for deviceId.
+     */
+    public static final String DEVICE_PATH_PARAMS = "deviceid";
+
+    private ApiCommonConstants() {
+        // avoid instantiation
+    }
+}
diff --git a/...ation/src/main/java/org/eclipse/hono/communication/api/config/DeviceConfigsConstants.java b/...ation/src/main/java/org/eclipse/hono/communication/api/config/DeviceConfigsConstants.java
@@ -21,33 +21,24 @@
  */
 public final class DeviceConfigsConstants {
 
+    public static ApiCommonConstants API_COMMON;
     /**
      * OpenApi GET device configs operation id.
      */
     public static final String LIST_CONFIG_VERSIONS_OP_ID = "listConfigVersions";
-    /**
-     * Path parameter name for tenantId.
-     */
-    public static final String TENANT_PATH_PARAMS = "tenantid";
-    /**
-     * Path parameter name for deviceId.
-     */
-    public static final String DEVICE_PATH_PARAMS = "deviceid";
     /**
      * Path parameter name for number of versions.
      */
     public static final String NUM_VERSION_QUERY_PARAMS = "numVersions";
-
     /**
      * Sql migrations script path.
      */
-    public static final String CREATE_SQL_SCRIPT_PATH = "db/create_device_config_table.sql";
+    public static final String CREATE_SQL_SCRIPT_PATH = "db/v1_create_config_table.sql";
     /**
      * OpenApi POST device configs operation id.
      */
     public static final String POST_MODIFY_DEVICE_CONFIG_OP_ID = "modifyCloudToDeviceConfig";
 
     private DeviceConfigsConstants() {
-        // avoid instantiation
     }
 }
diff --git a/...cation/src/main/java/org/eclipse/hono/communication/api/config/DeviceStatesConstants.java b/...cation/src/main/java/org/eclipse/hono/communication/api/config/DeviceStatesConstants.java
@@ -0,0 +1,42 @@
+/*
+ * ***********************************************************
+ *  Copyright (c) 2023 Contributors to the Eclipse Foundation
+ *  <p>
+ *  See the NOTICE file(s) distributed with this work for additional
+ *  information regarding copyright ownership.
+ *  <p>
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Eclipse Public License 2.0 which is available at
+ *  http://www.eclipse.org/legal/epl-2.0
+ *  <p>
+ *  SPDX-License-Identifier: EPL-2.0
+ * **********************************************************
+ *
+ */
+
+package org.eclipse.hono.communication.api.config;
+
+/**
+ * Device states constant values.
+ */
+public final class DeviceStatesConstants {
+
+    public static ApiCommonConstants API_COMMON;
+    /**
+     * OpenApi GET device states operation id.
+     */
+    public static final String LIST_STATES_OP_ID = "listStates";
+    /**
+     * Path parameter name for number of states.
+     */
+
+    public static final String NUM_STATES_QUERY_PARAMS = "numStates";
+    /**
+     * Sql migrations script path.
+     */
+    public static final String CREATE_SQL_SCRIPT_PATH = "db/v1_create_state_table.sql";
+
+    private DeviceStatesConstants() {
+        // avoid instantiation
+    }
+}
diff --git a/device-communication/src/main/java/org/eclipse/hono/communication/api/data/DeviceConfig.java b/device-communication/src/main/java/org/eclipse/hono/communication/api/data/DeviceConfig.java
@@ -21,8 +21,6 @@
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
-
-
 /**
  * The device configuration.
  **/

diff --git a/...mmunication/src/main/java/org/eclipse/hono/communication/api/data/DeviceConfigEntity.java b/...mmunication/src/main/java/org/eclipse/hono/communication/api/data/DeviceConfigEntity.java
@@ -18,6 +18,8 @@
 
 import java.util.Objects;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
+
 /**
  * The device configuration entity object.
  **/
@@ -31,6 +33,7 @@ public class DeviceConfigEntity {
     private String cloudUpdateTime;
     private String deviceAckTime;
     private String binaryData;
+    private String deviceAckError;
 
 
     /**
@@ -48,47 +51,66 @@ public void setVersion(final int version) {
         this.version = version;
     }
 
+    @JsonProperty("tenantId")
     public String getTenantId() {
         return tenantId;
     }
 
+    @JsonProperty("tenant_id")
     public void setTenantId(final String tenantId) {
         this.tenantId = tenantId;
     }
 
+    @JsonProperty("deviceId")
     public String getDeviceId() {
         return deviceId;
     }
 
+    @JsonProperty("device_id")
     public void setDeviceId(final String deviceId) {
         this.deviceId = deviceId;
     }
 
-
+    @JsonProperty("cloudUpdateTime")
     public String getCloudUpdateTime() {
         return cloudUpdateTime;
     }
 
+    @JsonProperty("cloud_update_time")
     public void setCloudUpdateTime(final String cloudUpdateTime) {
         this.cloudUpdateTime = cloudUpdateTime;
     }
 
+    @JsonProperty("deviceAckTime")
     public String getDeviceAckTime() {
         return deviceAckTime;
     }
 
+    @JsonProperty("device_ack_time")
     public void setDeviceAckTime(final String deviceAckTime) {
         this.deviceAckTime = deviceAckTime;
     }
 
+    @JsonProperty("binaryData")
     public String getBinaryData() {
         return binaryData;
     }
 
+    @JsonProperty("binary_data")
     public void setBinaryData(final String binaryData) {
         this.binaryData = binaryData;
     }
 
+    @JsonProperty("device_ack_error")
+    public String getDeviceAckError() {
+        return deviceAckError;
+    }
+
+    @JsonProperty("device_ack_error")
+    public void setDeviceAckError(final String deviceAckError) {
+        this.deviceAckError = deviceAckError;
+    }
+
 
     @Override
     public String toString() {
@@ -99,6 +121,7 @@ public String toString() {
                 ", cloudUpdateTime='" + cloudUpdateTime + '\'' +
                 ", deviceAckTime='" + deviceAckTime + '\'' +
                 ", binaryData='" + binaryData + '\'' +
+                ", device_ack_error='" + deviceAckError + '\'' +
                 '}';
     }
 
@@ -111,12 +134,12 @@ public boolean equals(final Object o) {
             return false;
         }
         final var that = (DeviceConfigEntity) o;
-        return version == that.version && tenantId.equals(that.tenantId) && deviceId.equals(that.deviceId) && cloudUpdateTime.equals(that.cloudUpdateTime) && Objects.equals(deviceAckTime, that.deviceAckTime) && binaryData.equals(that.binaryData);
+        return version == that.version && tenantId.equals(that.tenantId) && deviceId.equals(that.deviceId) && cloudUpdateTime.equals(that.cloudUpdateTime) && Objects.equals(deviceAckTime, that.deviceAckTime) && binaryData.equals(that.binaryData) && deviceAckError.equals(that.deviceAckError);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(version, tenantId, deviceId, cloudUpdateTime, deviceAckTime, binaryData);
+        return Objects.hash(version, tenantId, deviceId, cloudUpdateTime, deviceAckTime, binaryData, deviceAckError);
     }
-Original file line number
+Diff line change
@@ Expand Up / @@ -21,8 +21,6 @@ @@
     import com.fasterxml.jackson.annotation.JsonInclude;
     import com.fasterxml.jackson.annotation.JsonProperty;
     /**
      * The device configuration.
      **/
@@ Expand Down @@