Merge pull request #24 from oracle-devrel/dbextractreplay

Dbextractreplay

Author: martatolosa

IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileDataInputMode.java

@@ -0,0 +1,5 @@
+package com.oracle.demo.timg.iot.iotdbjdbc.dataread.filereader;
+
+public enum FileDataInputMode {
+	REAL_TIME, HIGH_SPEED
+}

IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java

@@ -0,0 +1,31 @@
+package com.oracle.demo.timg.iot.iotdbjdbc.dataread.filereader;
+
+public class FileReaderProperties {
+	public final static String NORMALIZED_DATA_FILE_INPUT = "iotdatacache.filereader.normalizeddata";
+	public final static String NORMALIZED_DATA_FILE_INPUT_ENABLED = NORMALIZED_DATA_FILE_INPUT + ".enabled";
+	public final static String NORMALIZED_DATA_FILE_INPUT_ORDER = NORMALIZED_DATA_FILE_INPUT + ".order";
+	public final static String NORMALIZED_DATA_FILE_INPUT_REPLAY = NORMALIZED_DATA_FILE_INPUT + ".replay";
+	public final static String NORMALIZED_DATA_FILE_INPUT_REPLAY_DURATION = NORMALIZED_DATA_FILE_INPUT_REPLAY
+			+ ".duration";
+	public static final String NORMALIZED_DATA_FILE_INPUT_REPLAY_START_OFFSET = NORMALIZED_DATA_FILE_INPUT_REPLAY
+			+ ".startoffset";
+	public final static String NORMALIZED_DATA_FILE_INPUT_REPLAY_OUTPUT_END_TIME_OBSERVED = NORMALIZED_DATA_FILE_INPUT_REPLAY
+			+ ".endtimeobserved";
+	public final static String NORMALIZED_DATA_FILE_INPUT_REPLAY_HIGH_SPEED_PLAYBACK_DELAY = NORMALIZED_DATA_FILE_INPUT_REPLAY
+			+ ".highspeeddelay";
+	public final static String NORMALIZED_DATA_FILE_INPUT_MODE = NORMALIZED_DATA_FILE_INPUT + ".mode";
+	public final static String NORMALIZED_DATA_FILE_INPUT_SOURCE_FILE = NORMALIZED_DATA_FILE_INPUT + ".sourcefile";
+
+	public final static String RAW_DATA_FILE_INPUT = "iotdatacache.filereader.rawdata";
+	public final static String RAW_DATA_FILE_INPUT_ENABLED = RAW_DATA_FILE_INPUT + ".enabled";
+	public final static String RAW_DATA_FILE_INPUT_ORDER = RAW_DATA_FILE_INPUT + ".order";
+	public final static String RAW_DATA_FILE_INPUT_REPLAY = RAW_DATA_FILE_INPUT + ".replay";
+	public final static String RAW_DATA_FILE_INPUT_REPLAY_DURATION = RAW_DATA_FILE_INPUT_REPLAY + ".duration";
+	public static final String RAW_DATA_FILE_INPUT_REPLAY_START_OFFSET = RAW_DATA_FILE_INPUT_REPLAY + ".startoffset";
+	public final static String RAW_DATA_FILE_INPUT_REPLAY_OUTPUT_END_TIME_OBSERVED = RAW_DATA_FILE_INPUT_REPLAY
+			+ ".endtimeobserved";
+	public final static String RAW_DATA_FILE_INPUT_REPLAY_HIGH_SPEED_PLAYBACK_DELAY = RAW_DATA_FILE_INPUT_REPLAY
+			+ ".highspeeddelay";
+	public final static String RAW_DATA_FILE_INPUT_MODE = RAW_DATA_FILE_INPUT + ".mode";
+	public final static String RAW_DATA_FILE_INPUT_SOURCE_FILE = RAW_DATA_FILE_INPUT + ".sourcefile";
+}

IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java

@@ -36,8 +36,6 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on
  */
 package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.filters.rawdata;
 
-import java.sql.PreparedStatement;
-
 import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.RawData;
 import com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.RawDataMessageHandler;
 import com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.filters.common.DeviceModelMessageFilterCoreOrig;
@@ -58,9 +56,6 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on
 @Requires(property = "iotdatacache.schemaname")
 @Log
 public class RawDataDeviceModelMessageFilter extends DeviceModelMessageFilterCoreOrig implements RawDataMessageHandler {
-
-	private PreparedStatement selectModelIdByInstanceIdPS;
-
 	@Inject
 	public RawDataDeviceModelMessageFilter(DBConnectionSupplier dbConnectionSupplier,
 			@Property(name = "iotdatacache.schemaname") String schemaName,

IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/filters/rawdata/RawDataDeviceModelMessageFilter.java

@@ -51,8 +51,8 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on
 
 import io.micronaut.context.annotation.Property;
 import io.micronaut.context.event.ShutdownEvent;
-import io.micronaut.context.event.StartupEvent;
 import io.micronaut.runtime.event.annotation.EventListener;
+import jakarta.annotation.PostConstruct;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import jakarta.validation.constraints.NotEmpty;
@@ -75,25 +75,29 @@ public class DeviceModelInstancesCache {
 	// we we don't know about this then we will try an individual load
 	public final static String SELECT_MODEL_NAME_BY_MODEL_ID = "SELECT JSON_VALUE(dtm.data, '$.displayName' ) AS modelname FROM digital_twin_models dtm WHERE JSON_VALUE (dtm.data, '$._id' ) = ? ";
 	public final static String SELECT_MODEL_ID_BY_MODEL_NAME = "SELECT  JSON_VALUE (dtm.data, '$._id' )  AS modelid FROM digital_twin_models dtm WHERE JSON_VALUE(dtm.data, '$.displayName' ) = ? ";
-	public final static String SELECT_MODEL_ID_DISPLAY_NAME_AND_EXTERNAL_KEY_BY_INSTANCE_ID = "SELECT JSON_VALUE (dti.data, '$.digitalTwinModelId' ) AS modelid, JSON_VALUE (dti.data, '$.externalKey' ) AS externalkey, JSON_VALUE(dti.data, '$.displayName' ) FROM digital_twin_instances dti WHERE JSON_VALUE(dti.data,  '$._id'  ) = ?";
+	public final static String SELECT_MODEL_ID_DISPLAY_NAME_AND_EXTERNAL_KEY_BY_INSTANCE_ID = "SELECT JSON_VALUE (dti.data, '$.digitalTwinModelId' ) AS modelid, JSON_VALUE (dti.data, '$.externalKey' ) AS externalkey, JSON_VALUE(dti.data, '$.displayName' )AS displayname FROM digital_twin_instances dti WHERE JSON_VALUE(dti.data,  '$._id'  ) = ?";
+	public final static String SELECT_MODEL_ID_INSTANCE_ID_AND_EXTERNAL_KEY_BY_INSTANCE_DISPLAY_NAME = "SELECT JSON_VALUE (dti.data, '$.digitalTwinModelId' ) AS modelid, JSON_VALUE (dti.data, '$.externalKey' ) AS externalkey, JSON_VALUE (dti.data, '$._id' ) AS instanceid FROM digital_twin_instances dti WHERE JSON_VALUE(dti.data, '$.displayName' ) = ?";
 
 	private final String schemaName;
 	private final DBConnectionSupplier dbConnectionSupplier;
 	private Connection connection;
 
 	private final Map<String, String> instanceIdToModelId = Collections.synchronizedMap(new HashMap<>());
 	private final Map<String, String> instanceIdToExternalKey = Collections.synchronizedMap(new HashMap<>());
-	private final Map<String, String> instanceIdToInstanceName = Collections.synchronizedMap(new HashMap<>());
+	private final Map<String, String> instanceIdToInstanceDisplayName = Collections.synchronizedMap(new HashMap<>());
+	private final Map<String, String> instanceDisplayNameToInstanceId = Collections.synchronizedMap(new HashMap<>());
 	private final Map<String, String> instanceIdToModelName = Collections.synchronizedMap(new HashMap<>());
 	private final Map<String, String> externalKeyToInstanceId = Collections.synchronizedMap(new HashMap<>());
 	private final Map<String, String> modelIdToModelName = Collections.synchronizedMap(new HashMap<>());
 	private final Map<String, String> modelNameToModelId = Collections.synchronizedMap(new HashMap<>());
 	private final Set<String> foundMissingModelIds = Collections.synchronizedSet(new HashSet<>());
 	private final Set<String> foundMissingModelNames = Collections.synchronizedSet(new HashSet<>());
 	private final Set<String> foundMissingInstanceIds = Collections.synchronizedSet(new HashSet<>());
+	private final Set<String> foundMissingInstanceDisplayNames = Collections.synchronizedSet(new HashSet<>());
 	private final Set<String> foundMissingExternalKeys = Collections.synchronizedSet(new HashSet<>());
 
-	private PreparedStatement selectModelIdByInstanceIdPS;
+	private PreparedStatement selectInstanceDetailsByInstanceIdPS;
+	private PreparedStatement selectInstanceDetailsByInstanceDisplayNamePS;
 	private PreparedStatement selectModelNameByModelIdPS;
 	private PreparedStatement selectModelIdByModelNamePS;
 	private final boolean preloadExistingModels;
@@ -113,9 +117,9 @@ public DeviceModelInstancesCache(DBConnectionSupplier dbConnectionSupplier,
 		this.preloadExistingInstances = preloadExistingInstances;
 	}
 
-	@EventListener
-	public void onStartup(StartupEvent event) {
-		log.info("Startup event received for DeviceModelInstancesCache");
+	@PostConstruct
+	public void postConstruct() {
+		log.info("Post Construct event received for DeviceModelInstancesCache");
 		try {
 			configure();
 		} catch (Exception e) {
@@ -149,8 +153,10 @@ public void configure() throws Exception {
 			// set this up so we can re-use it later if we need to query for an instance we
 			// didn't know about
 			log.fine("Creating prepared statements");
-			selectModelIdByInstanceIdPS = connection
+			selectInstanceDetailsByInstanceIdPS = connection
 					.prepareStatement(SELECT_MODEL_ID_DISPLAY_NAME_AND_EXTERNAL_KEY_BY_INSTANCE_ID);
+			selectInstanceDetailsByInstanceDisplayNamePS = connection
+					.prepareStatement(SELECT_MODEL_ID_INSTANCE_ID_AND_EXTERNAL_KEY_BY_INSTANCE_DISPLAY_NAME);
 			selectModelNameByModelIdPS = connection.prepareStatement(SELECT_MODEL_NAME_BY_MODEL_ID);
 			selectModelIdByModelNamePS = connection.prepareStatement(SELECT_MODEL_ID_BY_MODEL_NAME);
 			log.fine("Prepared statements created");
@@ -360,7 +366,16 @@ private void preloadExistingInstances() throws SQLException {
 				String instanceDisplayName = rs.getString(INSTANCE_ID_COLUMN_DISPLAY_NAME);
 				String modelName = modelIdToModelName.get(modelIdExistingInstance);
 				instanceIdToModelId.put(instanceIdExistingInstance, modelIdExistingInstance);
-				instanceIdToInstanceName.put(instanceIdExistingInstance, instanceDisplayName);
+				instanceIdToInstanceDisplayName.put(instanceIdExistingInstance, instanceDisplayName);
+				if (instanceDisplayNameToInstanceId.containsKey(instanceDisplayName)) {
+					log.warning("Instance display name cache already contains key " + instanceDisplayName
+							+ " connected to instance id " + instanceDisplayNameToInstanceId.get(instanceDisplayName)
+							+ ", duplicates are not added");
+				} else {
+					instanceDisplayNameToInstanceId.put(instanceDisplayName, instanceIdExistingInstance);
+					log.info("Added instance name " + instanceDisplayName + " to instanceId "
+							+ instanceIdExistingInstance + " mapping");
+				}
 				log.info("Added instance id " + instanceIdExistingInstance + " named " + instanceDisplayName
 						+ " to modelId " + modelIdExistingInstance + " mapping");
 				instanceIdToModelName.put(instanceIdExistingInstance, modelName);
@@ -451,10 +466,10 @@ public String getInstanceDisplayNameByInstanceId(@NotNull @NotEmpty String insta
 		}
 		// do we already have the info ? note that empty string and null are valid
 		// responses here.
-		if (instanceIdToInstanceName.containsKey(instanceId)) {
+		if (instanceIdToInstanceDisplayName.containsKey(instanceId)) {
 			// we have the key, the model could be a string, null blank etc if one hasn't
 			// been set, but that's still valid.
-			return instanceIdToInstanceName.get(instanceId);
+			return instanceIdToInstanceDisplayName.get(instanceId);
 		}
 		// we don't have a cached version
 		// let's try and locate it
@@ -474,6 +489,54 @@ public String getInstanceDisplayNameByInstanceId(@NotNull @NotEmpty String insta
 		}
 	}
 
+	/**
+	 * try to get the modelId from the cache or if there is no cachedata
+	 * 
+	 * @param instanceId          the instance to locate
+	 * @param cacheMissingResults if true and we already have looked but not found
+	 *                            this then don't look again
+	 * @return the instance display name is there is one
+	 * @throws MissingInstanceException throws an exception if we can't locate the
+	 *                                  instance (either in the known missing cache
+	 *                                  if cacheMissingResults is true, or in the
+	 *                                  IoT service otherwise)
+	 * @throws SQLException             if there was a problem querying the iot
+	 *                                  service
+	 */
+	public String getInstanceIdByInstanceDisplayName(@NotNull @NotEmpty String instanceDisplayName,
+			boolean cacheMissingResults) throws MissingInstanceException, SQLException {
+		boolean knownMissing = foundMissingInstanceDisplayNames.contains(instanceDisplayName);
+		// are we looking at the cache ?
+		if (cacheMissingResults && knownMissing) {
+			// we know it's missing, and we are not checking an other time
+			throw new MissingInstanceException(
+					"No instance found in cache and not checking again for instanceDisplayName" + instanceDisplayName);
+		}
+		// do we already have the info ? note that empty string and null are valid
+		// responses here.
+		if (instanceDisplayNameToInstanceId.containsKey(instanceDisplayName)) {
+			// we have the key, the model could be a string, null blank etc if one hasn't
+			// been set, but that's still valid.
+			return instanceDisplayNameToInstanceId.get(instanceDisplayName);
+		}
+		// we don't have a cached version
+		// let's try and locate it
+		try {
+			InstanceKeyInfo ike = loadInstanceByInstanceDisplayName(instanceDisplayName);
+			// OK we have something, was it previously tagged as knownMissing ? if so remove
+			// the id
+			if (knownMissing) {
+				foundMissingInstanceDisplayNames.remove(instanceDisplayName);
+			}
+			return ike.getInstanceId();
+		} catch (MissingInstanceException e) {
+			// cache the missing result for later use
+			foundMissingInstanceDisplayNames.add(instanceDisplayName);
+			// then throw the exception
+			throw e;
+		}
+	}
+
 	/**
 	 * try to get the external key from the cache or if there is no cachedata
 	 * 
@@ -568,10 +631,10 @@ public String getModelNameByInstanceId(@NotNull @NotEmpty String instanceId, boo
 	 */
 	private InstanceKeyInfo loadInstanceByInstanceId(@NotNull @NotEmpty String instanceId)
 			throws SQLException, MissingInstanceException {
-		synchronized (selectModelIdByInstanceIdPS) {
-			selectModelIdByInstanceIdPS.setString(1, instanceId);
+		synchronized (selectInstanceDetailsByInstanceIdPS) {
+			selectInstanceDetailsByInstanceIdPS.setString(1, instanceId);
 			// get all of the results
-			try (ResultSet rs = selectModelIdByInstanceIdPS.executeQuery()) {
+			try (ResultSet rs = selectInstanceDetailsByInstanceIdPS.executeQuery()) {
 				if (rs.next()) {
 					String modelId = rs.getString(MODEL_ID_COLUMN_NAME);
 					String externalKey = rs.getString(EXTERNAL_KEY_COLUMN_NAME);
@@ -580,7 +643,15 @@ private InstanceKeyInfo loadInstanceByInstanceId(@NotNull @NotEmpty String insta
 					instanceIdToModelId.put(instanceId, modelId);
 					instanceIdToModelName.put(instanceId, modelName);
 					instanceIdToExternalKey.put(instanceId, externalKey);
-					instanceIdToInstanceName.put(instanceId, instanceDisplayName);
+					instanceIdToInstanceDisplayName.put(instanceId, instanceDisplayName);
+					if (instanceDisplayNameToInstanceId.containsKey(instanceDisplayName)) {
+						log.warning("Instance display name cache already contains key " + instanceDisplayName
+								+ " connected to instance id "
+								+ instanceDisplayNameToInstanceId.get(instanceDisplayName)
+								+ ", duplicates are not added");
+					} else {
+						instanceDisplayNameToInstanceId.put(instanceDisplayName, instanceId);
+					}
 					externalKeyToInstanceId.put(externalKey, instanceId);
 					return new InstanceKeyInfo(instanceId, modelId, externalKey, instanceDisplayName);
 				} else {
@@ -593,6 +664,51 @@ private InstanceKeyInfo loadInstanceByInstanceId(@NotNull @NotEmpty String insta
 		}
 	}
 
+	/**
+	 * on demand load an entry in the mode details cache
+	 * 
+	 * @param instanceId
+	 * @throws SQLException
+	 * @throws MissingInstanceException
+	 * @return the located modelId, null for instances with no model
+	 */
+	private InstanceKeyInfo loadInstanceByInstanceDisplayName(@NotNull @NotEmpty String instanceDisplayName)
+			throws SQLException, MissingInstanceException {
+		synchronized (selectInstanceDetailsByInstanceDisplayNamePS) {
+			selectInstanceDetailsByInstanceDisplayNamePS.setString(1, instanceDisplayName);
+			// get all of the results
+			try (ResultSet rs = selectInstanceDetailsByInstanceDisplayNamePS.executeQuery()) {
+				if (rs.next()) {
+					String modelId = rs.getString(MODEL_ID_COLUMN_NAME);
+					String externalKey = rs.getString(EXTERNAL_KEY_COLUMN_NAME);
+					String instanceId = rs.getString(INSTANCE_ID_COLUMN_NAME);
+					String modelName = modelIdToModelName.get(modelId);
+					instanceIdToModelId.put(instanceId, modelId);
+					instanceIdToModelName.put(instanceId, modelName);
+					instanceIdToExternalKey.put(instanceId, externalKey);
+					instanceIdToInstanceDisplayName.put(instanceId, instanceDisplayName);
+					if (instanceDisplayNameToInstanceId.containsKey(instanceDisplayName)) {
+						log.warning("Instance display name cache already contains key " + instanceDisplayName
+								+ " connected to instance id "
+								+ instanceDisplayNameToInstanceId.get(instanceDisplayName)
+								+ ", duplicates are not added");
+					} else {
+						instanceDisplayNameToInstanceId.put(instanceDisplayName, instanceId);
+					}
+					externalKeyToInstanceId.put(externalKey, instanceId);
+					return new InstanceKeyInfo(instanceId, modelId, externalKey, instanceDisplayName);
+				} else {
+					throw new MissingInstanceException("No instance found for instance id " + instanceDisplayName);
+				}
+			} catch (SQLException e) {
+				log.severe(
+						"SQLException getting existing model / instance mappings in loadInstanceByInstanceDisplayName, "
+								+ e.getLocalizedMessage());
+				throw e;
+			}
+		}
+	}
+
 	@Data
 	@AllArgsConstructor
 	private class InstanceKeyInfo {
@@ -623,6 +739,7 @@ public void unconfigure() throws Exception {
 			foundMissingModelIds.clear();
 			foundMissingModelNames.clear();
 			foundMissingInstanceIds.clear();
+			foundMissingInstanceDisplayNames.clear();
 			foundMissingExternalKeys.clear();
 			configured = false;
 		}
@@ -639,6 +756,7 @@ public String getConfig() {
 				+ modelIdToModelName + ")" + " model ids, " + modelNameToModelId.size() + " ( " + modelNameToModelId
 				+ ")" + " model names, " + foundMissingModelIds.size() + " found missing model ids"
 				+ foundMissingModelNames.size() + " found missing model names, " + foundMissingInstanceIds.size()
+				+ " found missing instance display names, " + foundMissingInstanceDisplayNames.size()
 				+ " found missing instance ids. preloadExistingModels=" + preloadExistingModels
 				+ ", preloadExistingInstances=" + preloadExistingInstances;
 	}

IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java

@@ -0,0 +1,15 @@
+package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter;
+
+public class FileWriterProperties {
+	public final static String NORMALIZED_DATA_FILE_OUTPUT = "messagehandler.output.normalizeddata.fileoutput";
+	public final static String NORMALIZED_DATA_FILE_OUTPUT_ENABLED = NORMALIZED_DATA_FILE_OUTPUT + ".enabled";
+	public final static String NORMALIZED_DATA_FILE_OUTPUT_ORDER = NORMALIZED_DATA_FILE_OUTPUT + ".order";
+	public final static String NORMALIZED_DATA_FILE_OUTPUT_DURATION = NORMALIZED_DATA_FILE_OUTPUT + ".duration";
+	public final static String NORMALIZED_DATA_FILE_OUTPUT_TARGET_FILE = NORMALIZED_DATA_FILE_OUTPUT + ".targetfile";
+
+	public final static String RAW_DATA_FILE_OUTPUT = "messagehandler.output.rawdata.fileoutput";
+	public final static String RAW_DATA_FILE_OUTPUT_ENABLED = RAW_DATA_FILE_OUTPUT + ".enabled";
+	public final static String RAW_DATA_FILE_OUTPUT_ORDER = RAW_DATA_FILE_OUTPUT + ".order";
+	public final static String RAW_DATA_FILE_OUTPUT_DURATION = RAW_DATA_FILE_OUTPUT + ".duration";
+	public final static String RAW_DATA_FILE_OUTPUT_TARGET_FILE = RAW_DATA_FILE_OUTPUT + ".targetfile";
+}

IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/FileWriterProperties.java

@@ -0,0 +1,16 @@
+package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter;
+
+import io.micronaut.serde.annotation.Serdeable;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+import lombok.extern.java.Log;
+
+@Log
+@Data
+@NoArgsConstructor
+@SuperBuilder
+@Serdeable
+public abstract class IoTDataCoreFileVersion {
+	private String digitalTwinInstanceDisplayName;
+}