[#19] Enhance device communication API

* added support for device states * added support for request/response commands * 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 * formatted device communication api code Signed-off-by: Matthias Kaemmer <[email protected]>
sotec-iot · Aug 24, 2023 · 50dd661 · 50dd661
1 parent c97a44d
commit 50dd661
Show file tree

Hide file tree

Showing 81 changed files with 3,508 additions and 710 deletions.
diff --git a/device-communication/README.md b/device-communication/README.md
@@ -100,7 +100,9 @@ Attributes:
 
 - deviceId
 - tenantId
-- subject (always set to "command")
+- subject (if not specified set to "command")
+- response-required (optional)
+- correlation-id (optional)
 
 Body:
 
@@ -152,7 +154,7 @@ quarkus:
   builder: docker
   build: true
   push: true
-  image: "gcr.io/sotec-iot-core-dev/hono-device-communication"
+  image: "eclipse/hono-device-communication"
 
 ````
 

diff --git a/device-communication/src/main/java/org/eclipse/hono/communication/api/Application.java b/device-communication/src/main/java/org/eclipse/hono/communication/api/Application.java
@@ -29,19 +29,19 @@
  */
 public class Application extends AbstractServiceApplication {
 
-    private final Logger log = LoggerFactory.getLogger(AbstractServiceApplication.class);
+    private final Logger log = LoggerFactory.getLogger(Application.class);
     private final HttpServer server;
 
     /**
      * Creates new Application with all dependencies.
      *
-     * @param vertx      The quarkus Vertx instance
+     * @param vertx The quarkus Vertx instance
      * @param appConfigs The application configs
-     * @param server     The http server
+     * @param server The http server
      */
     public Application(final Vertx vertx,
-                       final ApplicationConfig appConfigs,
-                       final HttpServer server) {
+            final ApplicationConfig appConfigs,
+            final HttpServer server) {
         super(vertx, appConfigs);
         this.server = server;
     }
@@ -58,5 +58,4 @@ public void doStop() {
         server.stop();
     }
 
-
 }
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,50 +46,49 @@
 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.
      *
-     * @param appConfigs            THe application configurations
-     * @param vertx                 The quarkus Vertx instance
-     * @param httpHandlerManager    The http handler manager
-     * @param databaseService       The database connection
+     * @param appConfigs THe application configurations
+     * @param vertx The quarkus Vertx instance
+     * @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 Vertx vertx,
+            final VertxHttpHandlerManagerService httpHandlerManager,
+            final DatabaseService databaseService,
+            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 +97,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,19 +106,19 @@ public void start() {
 
                 });
 
-        // Wait until application is stopped
         Quarkus.waitForExit();
 
     }
 
     /**
      * Creates the Router object and adds endpoints and handlers.
      *
-     * @param routerBuilder        Vertx RouterBuilder object
+     * @param routerBuilder Vertx RouterBuilder object
      * @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);
         }
@@ -150,22 +150,19 @@ private void addReadinessHandlers(final Router router, final String readinessPat
         final HealthCheckHandler healthCheckHandler = HealthCheckHandler.create(vertx);
 
         healthCheckHandler.register("database-communication-is-ready",
-                promise ->
-                        db.getDbClient().getConnection(connection -> {
-                            if (connection.failed()) {
-                                log.error(connection.cause().getMessage());
-                                promise.tryComplete(Status.KO());
-                            } else {
-                                connection.result().close();
-                                promise.tryComplete(Status.OK());
-                            }
-                        })
-        );
+                promise -> db.getDbClient().getConnection(connection -> {
+                    if (connection.failed()) {
+                        log.error(connection.cause().getMessage());
+                        promise.tryComplete(Status.KO());
+                    } else {
+                        connection.result().close();
+                        promise.tryComplete(Status.OK());
+                    }
+                }));
 
         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,9 +187,9 @@ void startVertxServer(final Router router) {
                 .listen();
 
         serverCreationFuture
-                .onSuccess(server -> log.info(this.serverStartedMsg, serverConfigs.getServerUrl()
-                        , serverConfigs.getServerPort()))
-                .onFailure(error -> log.info(this.serverFailedMsg, error.getMessage()));
+                .onSuccess(server -> log.info("HTTP Server is listening at http://{}:{}", serverConfigs.getServerUrl(),
+                        serverConfigs.getServerPort()))
+                .onFailure(error -> log.info("HTTP Server failed to start: {}", error.getMessage()));
     }
 
     /**
@@ -229,10 +226,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,44 @@
+/*
+ * ***********************************************************
+ *  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";
+    /**
+     * Caption for tenantId.
+     */
+    public static final String TENANT_ID_CAPTION = "tenantId";
+    /**
+     * Caption for deviceId.
+     */
+    public static final String DEVICE_ID_CAPTION = "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
@@ -25,29 +25,23 @@ public final class DeviceConfigsConstants {
      * 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";
+    /**
+     * Subject of config commands.
+     */
+    public static final String CONFIG_SUBJECT = "config";
 
     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,41 @@
+/*
+ * ***********************************************************
+ *  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 {
+
+    /**
+     * 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
+    }
+}