Query Executor integrated in the Digital Adapter

Author: piconem

src/main/java/it/wldt/adapter/digital/DigitalAdapter.java

@@ -12,6 +12,7 @@
 import it.wldt.exception.EventBusException;
 import it.wldt.exception.WldtDigitalTwinStateEventException;
 import it.wldt.exception.WldtRuntimeException;
+import it.wldt.storage.query.QueryExecutor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -22,20 +23,17 @@
  * Authors:
  *          Marco Picone, Ph.D. (picone.m@gmail.com)
  *          Marta Spadoni (marta.spadoni2@studio.unibo.it)
- *
- * Date: 01/02/2023
+ * Date: 05/08/2024
  * Project: White Label Digital Twin Java Framework - (whitelabel-digitaltwin)
- *
  * This class defines the core capabilities and responsibilities of a Digital Adapter in the Digital Twin's Instance
  * A Digital Adapter is the architectural component in charge of handling the interaction of the Digital Twin
  * with the external digital world exposing information coming from the Digital Twin State and receiving information and
  * action from other digital applications or twins.
- *
  * This class can be extended in order to create custom Digital Adapter shaping specific behaviours of the twin and
  * allowing a simplified interaction with the external digital world.
- *
  * Multiple Digital Adapters can be active at the same time on the Digital Twin with the aim to handle different
  * interaction with the digital layer according to the nature of the twin and its operation context.
+ * The class integrates a Query Executor to send query to the storage layer in both synchronous and asynchronous way.
  */
 public abstract class DigitalAdapter<C> extends DigitalTwinWorker implements WldtEventListener, LifeCycleListener {
 
@@ -67,17 +65,21 @@ public abstract class DigitalAdapter<C> extends DigitalTwinWorker implements Wld
 
     private DigitalAdapterLifeCycleListener digitalAdapterLifeCycleListener;
 
+    // Query Executor to send query to the storage layer in both synchronous and asynchronous way
+    protected QueryExecutor queryExecutor = null;
+
     private DigitalAdapter() {
+        super();
     }
 
     public DigitalAdapter(String id, C configuration) {
-        super();
+        this(id);
         this.id = id;
         this.configuration = configuration;
     }
 
     public DigitalAdapter(String id) {
-        super();
+        this();
         this.id = id;
     }
 
@@ -314,6 +316,11 @@ abstract protected void onStateUpdate(DigitalTwinState newDigitalTwinState,
     @Override
     public void onWorkerStart() throws WldtRuntimeException {
         try{
+
+            // Init the Query Executor
+            if(this.queryExecutor == null)
+                this.queryExecutor = new QueryExecutor(this.digitalTwinId, String.format("query-executor-%s", this.id));
+
             onAdapterStart();
         }catch (Exception e){
             throw new WldtRuntimeException(e.getLocalizedMessage());

src/test/java/it/wldt/process/digital/DemoDigitalAdapter.java

@@ -7,6 +7,9 @@
 import it.wldt.core.state.DigitalTwinStateEventNotification;
 import it.wldt.exception.EventBusException;
 import it.wldt.process.metrics.SharedTestMetrics;
+import it.wldt.storage.query.IQueryResultListener;
+import it.wldt.storage.query.QueryRequest;
+import it.wldt.storage.query.QueryResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.util.ArrayList;
@@ -95,4 +98,28 @@ public <T> void invokeAction(String actionKey, T body){
             e.printStackTrace();
         }
     }
+
+    /**
+     * This method is used to test the sync query execution
+     * @param queryRequest - The query request
+     * @return - The query result
+     */
+    public QueryResult<?> testSyncQuery(QueryRequest queryRequest){
+        if(this.queryExecutor != null){
+            return this.queryExecutor.syncQueryExecute(queryRequest);
+        }
+        else
+            return null;
+    }
+
+    /**
+     * This method is used to test the async query execution
+     * @param queryRequest - The query request
+     * @param queryResultListener - The query result listener
+     */
+    public void testAsyncQuery(QueryRequest queryRequest, IQueryResultListener queryResultListener){
+        if(this.queryExecutor != null){
+            this.queryExecutor.asyncQueryExecute(queryRequest, queryResultListener);
+        }
+    }
 }

src/test/java/it/wldt/storage/StorageQueryTester.java

@@ -1174,4 +1174,66 @@ public void testSyncLifeCycleQuery() throws InterruptedException {
         // Check the last Life Cycle State is SYNCHRONIZED
         assertEquals(LifeCycleState.SYNCHRONIZED, ((LifeCycleVariationRecord)resultList.get(0)).getLifeCycleState());
     }
+
+    @Test
+    @Order(14)
+    public void testDigitalAdapterQueryExecutorSync() throws WldtConfigurationException, EventBusException, ModelException, InterruptedException, WldtRuntimeException, StorageException {
+
+        //Set EventBus Logger
+        WldtEventBus.getInstance().setEventLogger(new DefaultWldtEventLogger());
+
+        //Wait until all the messages have been received
+        Thread.sleep((DemoPhysicalAdapter.DEFAULT_MESSAGE_SLEEP_PERIOD_MS + ((DemoPhysicalAdapter.DEFAULT_TARGET_PHYSICAL_ASSET_PROPERTY_UPDATE_MESSAGES + DemoPhysicalAdapter.DEFAULT_TARGET_PHYSICAL_ASSET_EVENT_UPDATES) * DemoPhysicalAdapter.DEFAULT_MESSAGE_SLEEP_PERIOD_MS)));
+
+        Thread.sleep(5000);
+
+        // Create Query Request to the Storage Manager for the Last Digital Twin State
+        QueryRequest queryRequest = new QueryRequest();
+        queryRequest.setResourceType(QueryResourceType.DIGITAL_TWIN_STATE);
+        queryRequest.setRequestType(QueryRequestType.LAST_VALUE);
+
+        // Send the Query Request to the Storage Manager for the target DT
+        QueryResult<?> queryResult = digitalAdapter.testSyncQuery(queryRequest);
+
+        assertNotNull(queryResult);
+        assertEquals(queryResult.getOriginalRequest().getRequestType(), QueryRequestType.LAST_VALUE);
+        assertEquals(queryResult.getOriginalRequest().getResourceType(), QueryResourceType.DIGITAL_TWIN_STATE);
+
+    }
+
+    @Test
+    @Order(15)
+    public void testDigitalAdapterQueryExecutorAsync() throws InterruptedException {
+
+        //Set EventBus Logger
+        WldtEventBus.getInstance().setEventLogger(new DefaultWldtEventLogger());
+
+        //Wait until all the messages have been received
+        Thread.sleep((DemoPhysicalAdapter.DEFAULT_MESSAGE_SLEEP_PERIOD_MS + ((DemoPhysicalAdapter.DEFAULT_TARGET_PHYSICAL_ASSET_PROPERTY_UPDATE_MESSAGES + DemoPhysicalAdapter.DEFAULT_TARGET_PHYSICAL_ASSET_EVENT_UPDATES) * DemoPhysicalAdapter.DEFAULT_MESSAGE_SLEEP_PERIOD_MS)));
+
+        Thread.sleep(5000);
+
+        // Create Query Request to the Storage Manager for the Last Digital Twin State
+        QueryRequest queryRequest = new QueryRequest();
+        queryRequest.setResourceType(QueryResourceType.DIGITAL_TWIN_STATE);
+        queryRequest.setRequestType(QueryRequestType.LAST_VALUE);
+
+        final QueryResult<?>[] receivedQueryResult = {null};
+
+        // Send the Query Request to the Storage Manager for the target DT
+        digitalAdapter.testAsyncQuery(queryRequest, new IQueryResultListener() {
+            @Override
+            public void onQueryResult(QueryResult<?> queryResult) {
+                receivedQueryResult[0] = queryResult;
+            }
+        });
+
+        Thread.sleep(5000);
+
+        assertNotNull(receivedQueryResult[0]);
+        assertEquals(receivedQueryResult[0].getOriginalRequest().getRequestType(), QueryRequestType.LAST_VALUE);
+        assertEquals(receivedQueryResult[0].getOriginalRequest().getResourceType(), QueryResourceType.DIGITAL_TWIN_STATE);
+
+        Thread.sleep(10000);
+    }
 }