From 24e3b2be1f211702b8e4f4641c2eeab907dfe822 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 15:15:32 +0100 Subject: [PATCH 01/24] added support for mapping instance names to instance id's --- .../iotdbutils/DeviceModelInstancesCache.java | 126 +++++++++++++++++- 1 file changed, 120 insertions(+), 6 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java index a357f4f..ad23b1d 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java @@ -75,7 +75,8 @@ 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; @@ -84,6 +85,7 @@ public class DeviceModelInstancesCache { private final Map instanceIdToModelId = Collections.synchronizedMap(new HashMap<>()); private final Map instanceIdToExternalKey = Collections.synchronizedMap(new HashMap<>()); private final Map instanceIdToInstanceName = Collections.synchronizedMap(new HashMap<>()); + private final Map instanceDisplayNameToInstanceId = Collections.synchronizedMap(new HashMap<>()); private final Map instanceIdToModelName = Collections.synchronizedMap(new HashMap<>()); private final Map externalKeyToInstanceId = Collections.synchronizedMap(new HashMap<>()); private final Map modelIdToModelName = Collections.synchronizedMap(new HashMap<>()); @@ -91,9 +93,11 @@ public class DeviceModelInstancesCache { private final Set foundMissingModelIds = Collections.synchronizedSet(new HashSet<>()); private final Set foundMissingModelNames = Collections.synchronizedSet(new HashSet<>()); private final Set foundMissingInstanceIds = Collections.synchronizedSet(new HashSet<>()); + private final Set foundMissingInstanceDisplayNames = Collections.synchronizedSet(new HashSet<>()); private final Set foundMissingExternalKeys = Collections.synchronizedSet(new HashSet<>()); - private PreparedStatement selectModelIdByInstanceIdPS; + private PreparedStatement selectInstanceDetailsByInstanceIdPS; + private PreparedStatement selectInstanceDetailsByInstanceDisplayNamePS; private PreparedStatement selectModelNameByModelIdPS; private PreparedStatement selectModelIdByModelNamePS; private final boolean preloadExistingModels; @@ -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"); @@ -361,6 +367,13 @@ private void preloadExistingInstances() throws SQLException { String modelName = modelIdToModelName.get(modelIdExistingInstance); instanceIdToModelId.put(instanceIdExistingInstance, modelIdExistingInstance); instanceIdToInstanceName.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 id " + instanceIdExistingInstance + " named " + instanceDisplayName + " to modelId " + modelIdExistingInstance + " mapping"); instanceIdToModelName.put(instanceIdExistingInstance, modelName); @@ -474,6 +487,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 +629,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); @@ -581,6 +642,14 @@ private InstanceKeyInfo loadInstanceByInstanceId(@NotNull @NotEmpty String insta instanceIdToModelName.put(instanceId, modelName); instanceIdToExternalKey.put(instanceId, externalKey); instanceIdToInstanceName.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 +662,49 @@ 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 = selectInstanceDetailsByInstanceIdPS.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); + instanceIdToInstanceName.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, " + e.getLocalizedMessage()); + throw e; + } + } + } + @Data @AllArgsConstructor private class InstanceKeyInfo { @@ -623,6 +735,7 @@ public void unconfigure() throws Exception { foundMissingModelIds.clear(); foundMissingModelNames.clear(); foundMissingInstanceIds.clear(); + foundMissingInstanceDisplayNames.clear(); foundMissingExternalKeys.clear(); configured = false; } @@ -639,6 +752,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; } From b09cada719f0f4649885711d85c548274eb41b11 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 15:18:32 +0100 Subject: [PATCH 02/24] made the timestamps use SSSSSS so micro seconds --- .../timg/iot/iotproxygateway/iotdata/IoTGatewayConfigData.java | 2 +- .../timg/iot/iotproxygateway/iotdata/IoTGatewayStatsData.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/iotdata/IoTGatewayConfigData.java b/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/iotdata/IoTGatewayConfigData.java index 3884e42..690d7dd 100644 --- a/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/iotdata/IoTGatewayConfigData.java +++ b/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/iotdata/IoTGatewayConfigData.java @@ -64,7 +64,7 @@ public class IoTGatewayConfigData { // so we can if we want we can have the timestamp at the outer (envelope) level // or within the payload, @Builder.Default - @JsonFormat(pattern = "uuuu-MM-dd'T'HH:mm:ss.SSSSSXXX") + @JsonFormat(pattern = "uuuu-MM-dd'T'HH:mm:ss.SSSSSSXXX") private ZonedDateTime timestamp = ZonedDateTime.now(utcTz); private final String devicekey = null; private GatewayConfigData payload; diff --git a/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/iotdata/IoTGatewayStatsData.java b/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/iotdata/IoTGatewayStatsData.java index f647d09..26db352 100644 --- a/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/iotdata/IoTGatewayStatsData.java +++ b/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/iotdata/IoTGatewayStatsData.java @@ -64,7 +64,7 @@ public class IoTGatewayStatsData { // so we can if we want we can have the timestamp at the outer (envelope) level // or within the payload, @Builder.Default - @JsonFormat(pattern = "uuuu-MM-dd'T'HH:mm:ss.SSSSSXXX") + @JsonFormat(pattern = "uuuu-MM-dd'T'HH:mm:ss.SSSSSSXXX") private ZonedDateTime timestamp = ZonedDateTime.now(utcTz); private final String devicekey = null; private GatewayStatsData payload; From 7b119d661a16d756d2265c5a9c9bc791e38393b0 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 15:19:00 +0100 Subject: [PATCH 03/24] added support for writing the event streams out to a file --- .../filewriter/FileWriterProperties.java | 15 ++ .../filewriter/IoTDataCoreFileVersion.java | 16 +++ .../filewriter/NormalizedDataFileOutput.java | 128 ++++++++++++++++++ .../filewriter/NormalizedDataFileVersion.java | 87 ++++++++++++ .../outputs/filewriter/RawDataFileOutput.java | 128 ++++++++++++++++++ .../filewriter/RawDataFileVersion.java | 102 ++++++++++++++ 6 files changed, 476 insertions(+) create mode 100644 IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/FileWriterProperties.java create mode 100644 IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/IoTDataCoreFileVersion.java create mode 100644 IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileOutput.java create mode 100644 IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java create mode 100644 IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileOutput.java create mode 100644 IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/FileWriterProperties.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/FileWriterProperties.java new file mode 100644 index 0000000..5ef66b8 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/FileWriterProperties.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 + ".target_file"; + + 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 + ".target_file"; +} diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/IoTDataCoreFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/IoTDataCoreFileVersion.java new file mode 100644 index 0000000..20eccdb --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/IoTDataCoreFileVersion.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(toBuilder = true) +@Serdeable +public abstract class IoTDataCoreFileVersion { + private String digitalTwinInstanceDisplayName; +} diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileOutput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileOutput.java new file mode 100644 index 0000000..8b896b6 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileOutput.java @@ -0,0 +1,128 @@ +/*Copyright (c) 2025 Oracle and/or its affiliates. + +The Universal Permissive License (UPL), Version 1.0 + +Subject to the condition set forth below, permission is hereby granted to any +person obtaining a copy of this software, associated documentation and/or data +(collectively the "Software"), free of charge and under any and all copyright +rights in the Software, and any and all patent rights owned or freely +licensable by each licensor hereunder covering either (i) the unmodified +Software as contributed to or provided by such licensor, or (ii) the Larger +Works (as defined below), to deal in both + +(a) the Software, and +(b) any piece of software and/or hardware listed in the lrgrwrks.txt file if +one is included with the Software (each a "Larger Work" to which the Software +is contributed by such licensors), + +without restriction, including without limitation the rights to copy, create +derivative works of, display, perform, and distribute the Software and make, +use, sell, offer for sale, import, export, have made, and have sold the +Software and the Larger Work(s), and to sublicense the foregoing rights on +either these or other terms. + +This license is subject to the following condition: +The above copyright notice and either this complete permission notice or at +a minimum a reference to the UPL must be included in all copies or +substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + */ +package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter; + +import java.io.BufferedWriter; +import java.io.FileWriter; +import java.io.IOException; +import java.time.Duration; +import java.time.Instant; + +import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.NormalizedData; +import com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.NormalizedDataMessageHandler; +import com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.iotdbutils.DeviceModelInstancesCache; + +import io.micronaut.context.annotation.Property; +import io.micronaut.context.annotation.Requires; +import io.micronaut.serde.ObjectMapper; +import jakarta.inject.Inject; +import jakarta.inject.Singleton; +import lombok.extern.java.Log; + +@Singleton +@Requires(property = FileWriterProperties.NORMALIZED_DATA_FILE_OUTPUT_ENABLED, value = "true", defaultValue = "false") +@Log +public class NormalizedDataFileOutput implements NormalizedDataMessageHandler { + private final ObjectMapper mapper; + private final DeviceModelInstancesCache deviceModelInstancesCache; + private final int order; + private final Duration recordDuration; + private final String outputFile; + private BufferedWriter output; + private Instant endTime; + + @Inject + public NormalizedDataFileOutput(ObjectMapper mapper, DeviceModelInstancesCache deviceModelInstancesCache, + @Property(name = FileWriterProperties.NORMALIZED_DATA_FILE_OUTPUT_ORDER) int order, + @Property(name = FileWriterProperties.NORMALIZED_DATA_FILE_OUTPUT_DURATION, defaultValue = "1h") Duration recordDuration, + @Property(name = FileWriterProperties.NORMALIZED_DATA_FILE_OUTPUT_TARGET_FILE) String outputFile) + throws IOException { + this.mapper = mapper; + this.deviceModelInstancesCache = deviceModelInstancesCache; + this.order = order; + this.recordDuration = recordDuration; + this.outputFile = outputFile; + } + + @Override + public void configure() throws Exception { + this.output = new BufferedWriter(new FileWriter(outputFile)); + this.endTime = Instant.now().plus(recordDuration); + } + + @Override + public void unconfigure() throws Exception { + this.output.close(); + this.output = null; + } + + @Override + public int getOrder() { + return order; + } + + @Override + public String getName() { + return "Normalized Data file output"; + } + + @Override + public String getConfig() { + return getName() + ", order " + order + " writing to " + outputFile + " for " + recordDuration; + } + + @Override + public NormalizedData[] processNormalizedData(NormalizedData input) throws Exception { + // whatever happens we just pass this on + NormalizedData[] returnData = new NormalizedData[1]; + returnData[0] = input; + // if the data + if (Instant.now().isAfter(endTime)) { + return returnData; + } + String deviceDisplayName = deviceModelInstancesCache + .getInstanceDisplayNameByInstanceId(input.getDigitalTwinInstanceId(), true); + NormalizedDataFileVersion dataFileVersion = NormalizedDataFileVersion.buildFrom(input, deviceDisplayName); + String outputString = mapper.writeValueAsString(dataFileVersion); + log.info("Saving " + outputString); + output.write(outputString); + output.newLine(); + + return returnData; + } + +} diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java new file mode 100644 index 0000000..13ad136 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -0,0 +1,87 @@ +/*Copyright (c) 2026 Oracle and/or its affiliates. + +The Universal Permissive License (UPL), Version 1.0 + +Subject to the condition set forth below, permission is hereby granted to any +person obtaining a copy of this software, associated documentation and/or data +(collectively the "Software"), free of charge and under any and all copyright +rights in the Software, and any and all patent rights owned or freely +licensable by each licensor hereunder covering either (i) the unmodified +Software as contributed to or provided by such licensor, or (ii) the Larger +Works (as defined below), to deal in both + +(a) the Software, and +(b) any piece of software and/or hardware listed in the lrgrwrks.txt file if +one is included with the Software (each a "Larger Work" to which the Software +is contributed by such licensors), + +without restriction, including without limitation the rights to copy, create +derivative works of, display, perform, and distribute the Software and make, +use, sell, offer for sale, import, export, have made, and have sold the +Software and the Larger Work(s), and to sublicense the foregoing rights on +either these or other terms. + +This license is subject to the following condition: +The above copyright notice and either this complete permission notice or at +a minimum a reference to the UPL must be included in all copies or +substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + */ + +package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter; + +import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.NormalizedData; + +import io.micronaut.serde.annotation.Serdeable; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.NoArgsConstructor; +import lombok.experimental.SuperBuilder; +import lombok.extern.java.Log; +import oracle.sql.json.OracleJsonValue; +import oracle.sql.json.OracleJsonValue.OracleJsonType; + +@Log +@Data +@EqualsAndHashCode(callSuper = true) +@SuperBuilder(toBuilder = true) +@NoArgsConstructor +@AllArgsConstructor +@Serdeable +public class NormalizedDataFileVersion extends IoTDataCoreFileVersion { + private String contentPath; + private String timeObserved; + private String contentType; + private String content; + private OracleJsonValue contentJsonValue; + private OracleJsonType contentJsonType; + + // build our version, but replace the instance id with the instance display + // name, that is (hopefully) portable (if it's been set) + public static NormalizedDataFileVersion buildFrom(NormalizedData input, String instanceDisplayName) { + return NormalizedDataFileVersion.builder().digitalTwinInstanceDisplayName(instanceDisplayName) + .contentPath(input.getContentPath()).timeObserved(input.getTimeObserved()) + .contentType(input.getContentType()).content(input.getContent()) + .contentJsonValue(input.getContentJsonValue()).contentJsonType(input.getContentJsonType()).build(); + } + + // build the NormalizedData version from our input + + public NormalizedData buildTo(String instanceId) { + return NormalizedDataFileVersion.buildTo(this, instanceId); + } + + public static NormalizedData buildTo(NormalizedDataFileVersion input, String instanceId) { + return NormalizedData.builder().digitalTwinInstanceId(instanceId).contentPath(input.getContentPath()) + .timeObserved(input.getTimeObserved()).contentType(input.getContentType()).content(input.getContent()) + .contentJsonValue(input.getContentJsonValue()).contentJsonType(input.getContentJsonType()).build(); + } +} diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileOutput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileOutput.java new file mode 100644 index 0000000..27baa09 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileOutput.java @@ -0,0 +1,128 @@ +/*Copyright (c) 2025 Oracle and/or its affiliates. + +The Universal Permissive License (UPL), Version 1.0 + +Subject to the condition set forth below, permission is hereby granted to any +person obtaining a copy of this software, associated documentation and/or data +(collectively the "Software"), free of charge and under any and all copyright +rights in the Software, and any and all patent rights owned or freely +licensable by each licensor hereunder covering either (i) the unmodified +Software as contributed to or provided by such licensor, or (ii) the Larger +Works (as defined below), to deal in both + +(a) the Software, and +(b) any piece of software and/or hardware listed in the lrgrwrks.txt file if +one is included with the Software (each a "Larger Work" to which the Software +is contributed by such licensors), + +without restriction, including without limitation the rights to copy, create +derivative works of, display, perform, and distribute the Software and make, +use, sell, offer for sale, import, export, have made, and have sold the +Software and the Larger Work(s), and to sublicense the foregoing rights on +either these or other terms. + +This license is subject to the following condition: +The above copyright notice and either this complete permission notice or at +a minimum a reference to the UPL must be included in all copies or +substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + */ +package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter; + +import java.io.BufferedWriter; +import java.io.FileWriter; +import java.io.IOException; +import java.time.Duration; +import java.time.Instant; + +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.iotdbutils.DeviceModelInstancesCache; + +import io.micronaut.context.annotation.Property; +import io.micronaut.context.annotation.Requires; +import io.micronaut.serde.ObjectMapper; +import jakarta.inject.Inject; +import jakarta.inject.Singleton; +import lombok.extern.java.Log; + +@Singleton +@Requires(property = FileWriterProperties.RAW_DATA_FILE_OUTPUT_ENABLED, value = "true", defaultValue = "false") +@Log +public class RawDataFileOutput implements RawDataMessageHandler { + private final ObjectMapper mapper; + private final DeviceModelInstancesCache deviceModelInstancesCache; + private final int order; + private final Duration recordDuration; + private final String outputFile; + private BufferedWriter output; + private Instant endTime; + + @Inject + public RawDataFileOutput(ObjectMapper mapper, DeviceModelInstancesCache deviceModelInstancesCache, + @Property(name = FileWriterProperties.RAW_DATA_FILE_OUTPUT_ORDER) int order, + @Property(name = FileWriterProperties.RAW_DATA_FILE_OUTPUT_DURATION, defaultValue = "1h") Duration recordDuration, + @Property(name = FileWriterProperties.RAW_DATA_FILE_OUTPUT_TARGET_FILE) String outputFile) + throws IOException { + this.mapper = mapper; + this.deviceModelInstancesCache = deviceModelInstancesCache; + this.order = order; + this.recordDuration = recordDuration; + this.outputFile = outputFile; + } + + @Override + public void configure() throws Exception { + this.output = new BufferedWriter(new FileWriter(outputFile)); + this.endTime = Instant.now().plus(recordDuration); + } + + @Override + public void unconfigure() throws Exception { + this.output.close(); + this.output = null; + } + + @Override + public int getOrder() { + return order; + } + + @Override + public String getName() { + return "Raw Data file output"; + } + + @Override + public String getConfig() { + return getName() + ", order " + order + " writing to " + outputFile + " for " + recordDuration; + } + + @Override + public RawData[] processRawData(RawData input) throws Exception { + // whatever happens we just pass this on + RawData[] returnData = new RawData[1]; + returnData[0] = input; + // if the data + if (Instant.now().isAfter(endTime)) { + return returnData; + } + String deviceDisplayName = deviceModelInstancesCache + .getInstanceDisplayNameByInstanceId(input.getDigitalTwinInstanceId(), true); + RawDataFileVersion dataFileVersion = RawDataFileVersion.buildFrom(input, deviceDisplayName); + String outputString = mapper.writeValueAsString(dataFileVersion); + log.info("Saving " + outputString); + output.write(outputString); + output.newLine(); + + return returnData; + } + +} diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java new file mode 100644 index 0000000..494d126 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java @@ -0,0 +1,102 @@ +/*Copyright (c) 2026 Oracle and/or its affiliates. + +The Universal Permissive License (UPL), Version 1.0 + +Subject to the condition set forth below, permission is hereby granted to any +person obtaining a copy of this software, associated documentation and/or data +(collectively the "Software"), free of charge and under any and all copyright +rights in the Software, and any and all patent rights owned or freely +licensable by each licensor hereunder covering either (i) the unmodified +Software as contributed to or provided by such licensor, or (ii) the Larger +Works (as defined below), to deal in both + +(a) the Software, and +(b) any piece of software and/or hardware listed in the lrgrwrks.txt file if +one is included with the Software (each a "Larger Work" to which the Software +is contributed by such licensors), + +without restriction, including without limitation the rights to copy, create +derivative works of, display, perform, and distribute the Software and make, +use, sell, offer for sale, import, export, have made, and have sold the +Software and the Larger Work(s), and to sublicense the foregoing rights on +either these or other terms. + +This license is subject to the following condition: +The above copyright notice and either this complete permission notice or at +a minimum a reference to the UPL must be included in all copies or +substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + */ + +package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter; + +import java.util.Arrays; +import java.util.HashSet; + +import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.RawData; + +import io.micronaut.http.MediaType; +import io.micronaut.serde.annotation.Serdeable; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.NoArgsConstructor; +import lombok.ToString; +import lombok.experimental.SuperBuilder; +import lombok.extern.java.Log; + +@Log +@Data +@EqualsAndHashCode(callSuper = true) +@NoArgsConstructor +@SuperBuilder(toBuilder = true) +@Serdeable +public class RawDataFileVersion extends IoTDataCoreFileVersion { + // there should be a way to use the MediaType.isTextBased here + private final static HashSet STRING_OUTPUT_TYPES = new HashSet<>(Arrays.asList( + MediaType.APPLICATION_JSON.toLowerCase(), MediaType.APPLICATION_JSON_SCHEMA.toLowerCase(), + MediaType.APPLICATION_XML.toLowerCase(), MediaType.TEXT_CSV.toLowerCase(), + MediaType.TEXT_HTML.toLowerCase(), MediaType.TEXT_XML.toLowerCase(), MediaType.TEXT_JSON.toLowerCase(), + MediaType.TEXT_MARKDOWN.toLowerCase(), MediaType.TEXT_PLAIN.toLowerCase())); + private String endpoint; + // we don't want to dump raw text + @ToString.Exclude + private byte content[]; + private String contentType; + private String timeReceived; + + public String getContentString() { + if (getMediaType().isTextBased()) { + return new String(content); + } else { + return "Blob data, non next based content type of " + content.length + " bytes"; + } + } + + public MediaType getMediaType() { + if ((contentType == null) || contentType.isEmpty()) { + return MediaType.TEXT_PLAIN_TYPE; + } + return MediaType.of(contentType); + } + + // build our version, but replace the instance id with the instance display + // name, that is (hopefully) portable (if it's been set) + public static RawDataFileVersion buildFrom(RawData input, String instanceDisplayName) { + return RawDataFileVersion.builder().digitalTwinInstanceDisplayName(instanceDisplayName) + .timeReceived(input.getTimeReceived()).content(input.getContent()).contentType(input.getContentType()) + .build(); + } + + // build the RawData version from our input + public static RawData buildTo(RawDataFileVersion input, String instanceId) { + return RawData.builder().digitalTwinInstanceId(instanceId).timeReceived(input.getTimeReceived()) + .contentType(input.getContentType()).content(input.getContent()).build(); + } +} From f1ae2a944f3bda04c7e93caa1a831c45c5bb17e1 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 16:27:19 +0100 Subject: [PATCH 04/24] micronaut should now know how to handle the serdableimports needed for the OracleJsonValue --- .../filewriter/NormalizedDataFileVersion.java | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java index 13ad136..73b4426 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -39,6 +39,7 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.NormalizedData; +import io.micronaut.serde.annotation.SerdeImport; import io.micronaut.serde.annotation.Serdeable; import lombok.AllArgsConstructor; import lombok.Data; @@ -46,8 +47,23 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on import lombok.NoArgsConstructor; import lombok.experimental.SuperBuilder; import lombok.extern.java.Log; +import oracle.sql.json.OracleJsonArray; +import oracle.sql.json.OracleJsonBinary; +import oracle.sql.json.OracleJsonDate; +import oracle.sql.json.OracleJsonDecimal; +import oracle.sql.json.OracleJsonDouble; +import oracle.sql.json.OracleJsonFloat; +import oracle.sql.json.OracleJsonIntervalDS; +import oracle.sql.json.OracleJsonIntervalYM; +import oracle.sql.json.OracleJsonNumber; +import oracle.sql.json.OracleJsonObject; +import oracle.sql.json.OracleJsonString; +import oracle.sql.json.OracleJsonStructure; +import oracle.sql.json.OracleJsonTimestamp; +import oracle.sql.json.OracleJsonTimestampTZ; import oracle.sql.json.OracleJsonValue; import oracle.sql.json.OracleJsonValue.OracleJsonType; +import oracle.sql.json.OracleJsonVector; @Log @Data @@ -55,6 +71,23 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on @SuperBuilder(toBuilder = true) @NoArgsConstructor @AllArgsConstructor +@SerdeImport(OracleJsonValue.class) +@SerdeImport(OracleJsonStructure.class) +@SerdeImport(OracleJsonObject.class) +@SerdeImport(OracleJsonArray.class) +@SerdeImport(OracleJsonString.class) +@SerdeImport(OracleJsonNumber.class) +@SerdeImport(OracleJsonDecimal.class) +@SerdeImport(OracleJsonDouble.class) +@SerdeImport(OracleJsonFloat.class) +@SerdeImport(OracleJsonBinary.class) +@SerdeImport(OracleJsonDate.class) +@SerdeImport(OracleJsonTimestamp.class) +@SerdeImport(OracleJsonTimestampTZ.class) +@SerdeImport(OracleJsonIntervalDS.class) +@SerdeImport(OracleJsonIntervalYM.class) +@SerdeImport(OracleJsonVector.class) +@SerdeImport(OracleJsonType.class) @Serdeable public class NormalizedDataFileVersion extends IoTDataCoreFileVersion { private String contentPath; From edc445adc3bd8e696b54c65bd2fa161650def700 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 16:38:15 +0100 Subject: [PATCH 05/24] try to locate problem with non filled fields --- .../outputs/filewriter/IoTDataCoreFileVersion.java | 2 +- .../outputs/filewriter/NormalizedDataFileOutput.java | 2 ++ .../outputs/filewriter/NormalizedDataFileVersion.java | 2 +- 3 files changed, 4 insertions(+), 2 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/IoTDataCoreFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/IoTDataCoreFileVersion.java index 20eccdb..30b4b6f 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/IoTDataCoreFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/IoTDataCoreFileVersion.java @@ -9,7 +9,7 @@ @Log @Data @NoArgsConstructor -@SuperBuilder(toBuilder = true) +@SuperBuilder @Serdeable public abstract class IoTDataCoreFileVersion { private String digitalTwinInstanceDisplayName; diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileOutput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileOutput.java index 8b896b6..39098f9 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileOutput.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileOutput.java @@ -116,7 +116,9 @@ public NormalizedData[] processNormalizedData(NormalizedData input) throws Excep } String deviceDisplayName = deviceModelInstancesCache .getInstanceDisplayNameByInstanceId(input.getDigitalTwinInstanceId(), true); + log.info("Input NormalizedData " + input); NormalizedDataFileVersion dataFileVersion = NormalizedDataFileVersion.buildFrom(input, deviceDisplayName); + log.info("Converted NormalizedDataFileVersion " + dataFileVersion); String outputString = mapper.writeValueAsString(dataFileVersion); log.info("Saving " + outputString); output.write(outputString); diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java index 73b4426..42a6bdc 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -68,7 +68,7 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on @Log @Data @EqualsAndHashCode(callSuper = true) -@SuperBuilder(toBuilder = true) +@SuperBuilder @NoArgsConstructor @AllArgsConstructor @SerdeImport(OracleJsonValue.class) From c707e3fb5a2e4b0212b6848a133c507480ed3663 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 16:47:06 +0100 Subject: [PATCH 06/24] update to superbuilder options --- .../messagehandler/outputs/filewriter/RawDataFileVersion.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java index 494d126..d7a3135 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java @@ -55,7 +55,7 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on @Data @EqualsAndHashCode(callSuper = true) @NoArgsConstructor -@SuperBuilder(toBuilder = true) +@SuperBuilder @Serdeable public class RawDataFileVersion extends IoTDataCoreFileVersion { // there should be a way to use the MediaType.isTextBased here From da4206628dc8328eebd32e3f2624a6babaee06a8 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 17:01:49 +0100 Subject: [PATCH 07/24] remove unneeded prepared statement --- .../filters/rawdata/RawDataDeviceModelMessageFilter.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/filters/rawdata/RawDataDeviceModelMessageFilter.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/filters/rawdata/RawDataDeviceModelMessageFilter.java index bf428dd..f1f4ddd 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/filters/rawdata/RawDataDeviceModelMessageFilter.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/filters/rawdata/RawDataDeviceModelMessageFilter.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, From 745b47133120e7ee09fa4c62823697818537a858 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 17:02:25 +0100 Subject: [PATCH 08/24] try manually serializing the OracleJsonValue as micronaut can't handle it directly --- .../filewriter/NormalizedDataFileVersion.java | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java index 42a6bdc..def0f96 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -37,6 +37,8 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter; +import java.io.StringWriter; + import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.NormalizedData; import io.micronaut.serde.annotation.SerdeImport; @@ -52,7 +54,9 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on import oracle.sql.json.OracleJsonDate; import oracle.sql.json.OracleJsonDecimal; import oracle.sql.json.OracleJsonDouble; +import oracle.sql.json.OracleJsonFactory; import oracle.sql.json.OracleJsonFloat; +import oracle.sql.json.OracleJsonGenerator; import oracle.sql.json.OracleJsonIntervalDS; import oracle.sql.json.OracleJsonIntervalYM; import oracle.sql.json.OracleJsonNumber; @@ -71,6 +75,7 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on @SuperBuilder @NoArgsConstructor @AllArgsConstructor +@Serdeable @SerdeImport(OracleJsonValue.class) @SerdeImport(OracleJsonStructure.class) @SerdeImport(OracleJsonObject.class) @@ -88,22 +93,26 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on @SerdeImport(OracleJsonIntervalYM.class) @SerdeImport(OracleJsonVector.class) @SerdeImport(OracleJsonType.class) -@Serdeable public class NormalizedDataFileVersion extends IoTDataCoreFileVersion { + private final static OracleJsonFactory factory = new OracleJsonFactory(); private String contentPath; private String timeObserved; private String contentType; private String content; - private OracleJsonValue contentJsonValue; + private String contentJsonValue; private OracleJsonType contentJsonType; // build our version, but replace the instance id with the instance display // name, that is (hopefully) portable (if it's been set) public static NormalizedDataFileVersion buildFrom(NormalizedData input, String instanceDisplayName) { + StringWriter contentOuputWriter = new StringWriter(); + OracleJsonGenerator oracleJsonGenerator = NormalizedDataFileVersion.factory + .createJsonTextGenerator(contentOuputWriter); + oracleJsonGenerator.write(input.getContentJsonValue()).close(); return NormalizedDataFileVersion.builder().digitalTwinInstanceDisplayName(instanceDisplayName) .contentPath(input.getContentPath()).timeObserved(input.getTimeObserved()) .contentType(input.getContentType()).content(input.getContent()) - .contentJsonValue(input.getContentJsonValue()).contentJsonType(input.getContentJsonType()).build(); + .contentJsonValue(contentOuputWriter.toString()).contentJsonType(input.getContentJsonType()).build(); } // build the NormalizedData version from our input From 57e6737d8be061d4a7d07366842fd44363e40604 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 17:06:57 +0100 Subject: [PATCH 09/24] updated reader to match the writer --- .../outputs/filewriter/NormalizedDataFileVersion.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java index def0f96..bab934d 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -37,6 +37,7 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter; +import java.io.StringReader; import java.io.StringWriter; import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.NormalizedData; @@ -61,6 +62,7 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on import oracle.sql.json.OracleJsonIntervalYM; import oracle.sql.json.OracleJsonNumber; import oracle.sql.json.OracleJsonObject; +import oracle.sql.json.OracleJsonParser; import oracle.sql.json.OracleJsonString; import oracle.sql.json.OracleJsonStructure; import oracle.sql.json.OracleJsonTimestamp; @@ -122,8 +124,11 @@ public NormalizedData buildTo(String instanceId) { } public static NormalizedData buildTo(NormalizedDataFileVersion input, String instanceId) { + StringReader contentInputReader = new StringReader(input.getContentJsonValue()); + OracleJsonParser oracleJsonParser = NormalizedDataFileVersion.factory.createJsonTextParser(contentInputReader); + OracleJsonValue oracleJsonValue = oracleJsonParser.getValue(); return NormalizedData.builder().digitalTwinInstanceId(instanceId).contentPath(input.getContentPath()) .timeObserved(input.getTimeObserved()).contentType(input.getContentType()).content(input.getContent()) - .contentJsonValue(input.getContentJsonValue()).contentJsonType(input.getContentJsonType()).build(); + .contentJsonValue(oracleJsonValue).contentJsonType(input.getContentJsonType()).build(); } } From d95d03c7267f27507a8d7e759f4ba7163692c80d Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 17:22:38 +0100 Subject: [PATCH 10/24] more switching form SerdableImport --- .../filewriter/NormalizedDataFileVersion.java | 33 ------------------- 1 file changed, 33 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java index bab934d..5fa9d45 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -42,7 +42,6 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.NormalizedData; -import io.micronaut.serde.annotation.SerdeImport; import io.micronaut.serde.annotation.Serdeable; import lombok.AllArgsConstructor; import lombok.Data; @@ -50,26 +49,11 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on import lombok.NoArgsConstructor; import lombok.experimental.SuperBuilder; import lombok.extern.java.Log; -import oracle.sql.json.OracleJsonArray; -import oracle.sql.json.OracleJsonBinary; -import oracle.sql.json.OracleJsonDate; -import oracle.sql.json.OracleJsonDecimal; -import oracle.sql.json.OracleJsonDouble; import oracle.sql.json.OracleJsonFactory; -import oracle.sql.json.OracleJsonFloat; import oracle.sql.json.OracleJsonGenerator; -import oracle.sql.json.OracleJsonIntervalDS; -import oracle.sql.json.OracleJsonIntervalYM; -import oracle.sql.json.OracleJsonNumber; -import oracle.sql.json.OracleJsonObject; import oracle.sql.json.OracleJsonParser; -import oracle.sql.json.OracleJsonString; -import oracle.sql.json.OracleJsonStructure; -import oracle.sql.json.OracleJsonTimestamp; -import oracle.sql.json.OracleJsonTimestampTZ; import oracle.sql.json.OracleJsonValue; import oracle.sql.json.OracleJsonValue.OracleJsonType; -import oracle.sql.json.OracleJsonVector; @Log @Data @@ -78,23 +62,6 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on @NoArgsConstructor @AllArgsConstructor @Serdeable -@SerdeImport(OracleJsonValue.class) -@SerdeImport(OracleJsonStructure.class) -@SerdeImport(OracleJsonObject.class) -@SerdeImport(OracleJsonArray.class) -@SerdeImport(OracleJsonString.class) -@SerdeImport(OracleJsonNumber.class) -@SerdeImport(OracleJsonDecimal.class) -@SerdeImport(OracleJsonDouble.class) -@SerdeImport(OracleJsonFloat.class) -@SerdeImport(OracleJsonBinary.class) -@SerdeImport(OracleJsonDate.class) -@SerdeImport(OracleJsonTimestamp.class) -@SerdeImport(OracleJsonTimestampTZ.class) -@SerdeImport(OracleJsonIntervalDS.class) -@SerdeImport(OracleJsonIntervalYM.class) -@SerdeImport(OracleJsonVector.class) -@SerdeImport(OracleJsonType.class) public class NormalizedDataFileVersion extends IoTDataCoreFileVersion { private final static OracleJsonFactory factory = new OracleJsonFactory(); private String contentPath; From 669465ed30325decbe25136126e859f567f3b28c Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 17:32:49 +0100 Subject: [PATCH 11/24] try not using the superclass --- .../filewriter/NormalizedDataFileVersion.java | 11 ++++------ .../filewriter/RawDataFileVersion.java | 21 +++++++------------ 2 files changed, 11 insertions(+), 21 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java index 5fa9d45..abad2fb 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -44,12 +44,10 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on import io.micronaut.serde.annotation.Serdeable; import lombok.AllArgsConstructor; +import lombok.Builder; import lombok.Data; -import lombok.EqualsAndHashCode; import lombok.NoArgsConstructor; -import lombok.experimental.SuperBuilder; import lombok.extern.java.Log; -import oracle.sql.json.OracleJsonFactory; import oracle.sql.json.OracleJsonGenerator; import oracle.sql.json.OracleJsonParser; import oracle.sql.json.OracleJsonValue; @@ -57,13 +55,12 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on @Log @Data -@EqualsAndHashCode(callSuper = true) -@SuperBuilder +@Builder @NoArgsConstructor @AllArgsConstructor @Serdeable -public class NormalizedDataFileVersion extends IoTDataCoreFileVersion { - private final static OracleJsonFactory factory = new OracleJsonFactory(); +public class NormalizedDataFileVersion { + private String digitalTwinInstanceDisplayName; private String contentPath; private String timeObserved; private String contentType; diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java index d7a3135..b01f801 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java @@ -37,33 +37,26 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on package com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter; -import java.util.Arrays; -import java.util.HashSet; - import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.RawData; import io.micronaut.http.MediaType; import io.micronaut.serde.annotation.Serdeable; +import lombok.AllArgsConstructor; +import lombok.Builder; import lombok.Data; -import lombok.EqualsAndHashCode; import lombok.NoArgsConstructor; import lombok.ToString; -import lombok.experimental.SuperBuilder; import lombok.extern.java.Log; @Log @Data -@EqualsAndHashCode(callSuper = true) @NoArgsConstructor -@SuperBuilder +@AllArgsConstructor +@Builder @Serdeable -public class RawDataFileVersion extends IoTDataCoreFileVersion { - // there should be a way to use the MediaType.isTextBased here - private final static HashSet STRING_OUTPUT_TYPES = new HashSet<>(Arrays.asList( - MediaType.APPLICATION_JSON.toLowerCase(), MediaType.APPLICATION_JSON_SCHEMA.toLowerCase(), - MediaType.APPLICATION_XML.toLowerCase(), MediaType.TEXT_CSV.toLowerCase(), - MediaType.TEXT_HTML.toLowerCase(), MediaType.TEXT_XML.toLowerCase(), MediaType.TEXT_JSON.toLowerCase(), - MediaType.TEXT_MARKDOWN.toLowerCase(), MediaType.TEXT_PLAIN.toLowerCase())); +public class RawDataFileVersion { + + private String digitalTwinInstanceDisplayName; private String endpoint; // we don't want to dump raw text @ToString.Exclude From 9a19c010a24b1581d55ac835018839396c42aaa5 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 17:35:31 +0100 Subject: [PATCH 12/24] added factory back in --- .../outputs/filewriter/NormalizedDataFileVersion.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java index abad2fb..d670f54 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -48,6 +48,7 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on import lombok.Data; import lombok.NoArgsConstructor; import lombok.extern.java.Log; +import oracle.sql.json.OracleJsonFactory; import oracle.sql.json.OracleJsonGenerator; import oracle.sql.json.OracleJsonParser; import oracle.sql.json.OracleJsonValue; @@ -60,6 +61,7 @@ Software and the Larger Work(s), and to sublicense the foregoing rights on @AllArgsConstructor @Serdeable public class NormalizedDataFileVersion { + private final static OracleJsonFactory factory = new OracleJsonFactory(); private String digitalTwinInstanceDisplayName; private String contentPath; private String timeObserved; From 4642ebb43b6f21d74601fbaf6216fb3dcb7684f4 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Wed, 22 Jul 2026 18:35:57 +0100 Subject: [PATCH 13/24] added a reader for saved data --- .../filereader/FileDataInputMode.java | 5 + .../filereader/FileReaderProperties.java | 29 ++ .../filereader/NormalizedDataFileInput.java | 371 ++++++++++++++++++ 3 files changed, 405 insertions(+) create mode 100644 IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileDataInputMode.java create mode 100644 IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java create mode 100644 IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileDataInputMode.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileDataInputMode.java new file mode 100644 index 0000000..933b467 --- /dev/null +++ b/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 +} diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java new file mode 100644 index 0000000..5a086c8 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java @@ -0,0 +1,29 @@ +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_END = NORMALIZED_DATA_FILE_INPUT_REPLAY + ".end"; + 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 + ".source_file"; + + 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_END = RAW_DATA_FILE_INPUT_REPLAY + ".end"; + 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 + ".source_file"; +} diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java new file mode 100644 index 0000000..a4a80f7 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java @@ -0,0 +1,371 @@ +/*Copyright (c) 2026 Oracle and/or its affiliates. + +The Universal Permissive License (UPL), Version 1.0 + +Subject to the condition set forth below, permission is hereby granted to any +person obtaining a copy of this software, associated documentation and/or data +(collectively the "Software"), free of charge and under any and all copyright +rights in the Software, and any and all patent rights owned or freely +licensable by each licensor hereunder covering either (i) the unmodified +Software as contributed to or provided by such licensor, or (ii) the Larger +Works (as defined below), to deal in both + +(a) the Software, and +(b) any piece of software and/or hardware listed in the lrgrwrks.txt file if +one is included with the Software (each a "Larger Work" to which the Software +is contributed by such licensors), + +without restriction, including without limitation the rights to copy, create +derivative works of, display, perform, and distribute the Software and make, +use, sell, offer for sale, import, export, have made, and have sold the +Software and the Larger Work(s), and to sublicense the foregoing rights on +either these or other terms. + +This license is subject to the following condition: +The above copyright notice and either this complete permission notice or at +a minimum a reference to the UPL must be included in all copies or +substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + */ +package com.oracle.demo.timg.iot.iotdbjdbc.dataread.filereader; + +import java.io.BufferedReader; +import java.io.EOFException; +import java.io.FileReader; +import java.io.IOException; +import java.sql.SQLException; +import java.time.Duration; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeParseException; +import java.util.Optional; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.NormalizedData; +import com.oracle.demo.timg.iot.iotdbjdbc.dataread.IoTDBClient; +import com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.NormalizedDataMessageHandlerService; +import com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.iotdbutils.DeviceModelInstancesCache; +import com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.iotdbutils.MissingInstanceException; +import com.oracle.demo.timg.iot.iotdbjdbc.messagehandler.outputs.filewriter.NormalizedDataFileVersion; + +import io.micronaut.context.annotation.Property; +import io.micronaut.context.annotation.Requires; +import io.micronaut.serde.ObjectMapper; +import jakarta.inject.Inject; +import jakarta.inject.Singleton; +import jakarta.validation.constraints.Min; +import lombok.ToString; +import lombok.extern.java.Log; + +@Singleton +@Log +@Requires(property = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_ENABLED, value = "true", defaultValue = "false") +@Requires(property = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_ORDER) +@ToString +public class NormalizedDataFileInput implements IoTDBClient, Runnable { + // get the UTC TZ once to speed things later + private final static ZoneId UTC_TZ = ZoneId.of("UTC"); + private final static DateTimeFormatter dateTimeFormatter = DateTimeFormatter + .ofPattern("uuuu-MM-dd'T'HH:mm:ss.SSSSSSXXX"); + @ToString.Include + private boolean stopped = false; + @ToString.Include + private final int order; + @ToString.Include + private final String sourceFilename; + @ToString.Include + private Duration replayStartOffset; + @ToString.Include + private Duration replayDuration; + @ToString.Include + private ZonedDateTime replayEnd; + @ToString.Include + private final FileDataInputMode mode; + @ToString.Include + private final Duration highSpeedReplayDelay; + @ToString.Include + private ZonedDateTime startOffsetZDT; + @ToString.Include + private ZonedDateTime stopAfterZDT; + @ToString.Include + private Duration highSpeedOffset; + @ToString.Exclude + private final ObjectMapper mapper; + @ToString.Exclude + private BufferedReader inputReader; + @ToString.Exclude + private Thread currentThread; + @ToString.Exclude + private ScheduledExecutorService executor; + @ToString.Include + private NormalizedDataFileVersion nextDataToSend; + + private final NormalizedDataMessageHandlerService normalizedDataMessageHandlerService; + + private final DeviceModelInstancesCache deviceModelInstancesCache; + private ZonedDateTime nextDataToSendTimeStamp; + + @Inject + public NormalizedDataFileInput(ObjectMapper mapper, + NormalizedDataMessageHandlerService normalizedDataMessageHandlerService, + DeviceModelInstancesCache deviceModelInstancesCache, + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_ORDER) @Min(value = 0) int order, + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_SOURCE_FILE) String sourceFilename, + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_START_OFFSET, defaultValue = "0s") Duration replayStartOffset, + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_DURATION) Duration replayDuration, + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_MODE, defaultValue = "REAL_TIME") FileDataInputMode mode, + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_HIGH_SPEED_PLAYBACK_DELAY, defaultValue = "100ms") Duration highSpeedReplayDelay, + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_END) Optional replayEnd) { + this.mapper = mapper; + this.normalizedDataMessageHandlerService = normalizedDataMessageHandlerService; + this.deviceModelInstancesCache = deviceModelInstancesCache; + this.order = order; + this.sourceFilename = sourceFilename; + this.replayStartOffset = replayStartOffset; + this.replayDuration = replayDuration; + this.mode = mode; + this.highSpeedReplayDelay = highSpeedReplayDelay; + // if we have a specified end time for the replay use the current time, if now + // use what's been specified + this.replayEnd = replayEnd.orElse(ZonedDateTime.now(UTC_TZ)); + } + + @Override + public void configureDBClient(String filteringRule) throws DateTimeParseException, EOFException, IOException { + // try to locate the very first timestamp, we will need to use that to work out + // at what point in the file if any we need to stop. + // try to open the file, we need to scan it to find the line previous to the one + inputReader = new BufferedReader(new FileReader(sourceFilename)); + nextDataToSend = readNormalizedDataFileVersionFromInput(inputReader); + ZonedDateTime startZDT = getTimeObservedFromNormalizedDataFileVersion(nextDataToSend); + if (startZDT == null) { + // we can't locate the initial data, throw an exception, we can't deal with + // this, it's up to the caller to then remove us from any further processing + throw new EOFException("No data in input file, cannot determine time stamps or start point"); + } + // this is the start point based on the timestamps in the data file + startOffsetZDT = startZDT.plus(replayDuration); + // now add the replay time to the start offset time, this is also based on the + // data file timestamps + this.stopAfterZDT = startOffsetZDT.plus(replayDuration); + // if we are in REAL_TIME replay mode we will be sending based on the current + // time and then waiting for the next to send (based on the difference between + // the one we just sent and the next one we're about to send) so for that we + // don't need to do any further timestamp thinking, we're just going to scan + // forwards later on + // + // if however we are in HIGH_SPEED mode then we need to adjust" the timestamp + // of the loaded data value, based on the delta of first timestamp that will be + // sent, and the timestamp found at the replay end point (which of course could + // be in the past as well as now) relative to the end timestamp we want to + // finish with. + highSpeedOffset = Duration.between(stopAfterZDT, replayEnd); + // we're going to reset the reader as we're looking to load + // now we need to move forwards until we get to the start point, if we get null + // then we've fallen off the end of the input stream, so need to error + ZonedDateTime readZDT = startZDT; + while ((readZDT != null) && (readZDT.isBefore(startOffsetZDT))) { + nextDataToSend = readNormalizedDataFileVersionFromInput(inputReader); + readZDT = getTimeObservedFromNormalizedDataFileVersion(nextDataToSend); + } + if (nextDataToSend == null) { + throw new EOFException("Hit the end of file while moving forward to the specified start point"); + } + // to avoid doing multiple time conversions later stash the current timestamp + nextDataToSendTimeStamp = readZDT; + // OK, we're set to go, lastly setup the executors + executor = Executors.newSingleThreadScheduledExecutor(); + } + + /** + * @param inputReader + * @return + * @throws IOException + * @throws EOFException + * @throws DateTimeParseException + */ + protected ZonedDateTime readNormalizedDataFileVersionTimestampFromInput(BufferedReader inputReader) + throws IOException, DateTimeParseException { + NormalizedDataFileVersion normalizedDataFileVersion = readNormalizedDataFileVersionFromInput(inputReader); + if (normalizedDataFileVersion == null) { + return null; + } + ZonedDateTime observedZDT = getTimeObservedFromNormalizedDataFileVersion(normalizedDataFileVersion); + return observedZDT; + } + + /** + * @param normalizedDataFileVersion + * @return + * @throws DateTimeParseException + */ + protected ZonedDateTime getTimeObservedFromNormalizedDataFileVersion( + NormalizedDataFileVersion normalizedDataFileVersion) throws DateTimeParseException { + if (normalizedDataFileVersion == null) { + return null; + } + try { + return ZonedDateTime.parse(normalizedDataFileVersion.getTimeObserved(), dateTimeFormatter); + } catch (DateTimeParseException e) { + throw new DateTimeParseException("Can't parse time Obeserved", e.getParsedString(), e.getErrorIndex()); + } + } + + /** + * @param inputReader + * @return + * @throws IOException + */ + protected NormalizedDataFileVersion readNormalizedDataFileVersionFromInput(BufferedReader inputReader) + throws IOException { + String NormalizedDataFileVersionString = inputReader.readLine(); + if (NormalizedDataFileVersionString == null) { + return null; + } + log.info("Read input " + NormalizedDataFileVersionString); + // try and convert it to a normalized data line (the version we write to files) + NormalizedDataFileVersion normalizedDataFileVersion = mapper.readValue(NormalizedDataFileVersionString, + NormalizedDataFileVersion.class); + return normalizedDataFileVersion; + } + + @Override + public void startDBProcessing() throws Exception { + // if running in HIGH_SPEED mode then work out the offset between now + // start this in a separate loop to run as soon as possible + executor.execute(() -> this.run()); + } + + @Override + public void stopDBProcessing() throws Exception { + log.info("Stopping reading"); + this.stopped = true; + // interrupt the thread if it's not null + if (currentThread != null) { + log.info("Interrupting thread"); + currentThread.interrupt(); + } + } + + @Override + public void unconfigureDBClient() throws Exception { + inputReader.close(); + // stop the executors from accepting new tasks + executor.shutdown(); + // close the reader + + } + + @Override + public void run() { + // save the thread we're running in so we can interrupt it later + currentThread = Thread.currentThread(); + if (nextDataToSend == null) { + log.info("nextDataToSend is null, stopping processing"); + } + + log.info("Running a send cycle on " + nextDataToSend); + // the timestamp was extracted when the nextDataToSend was setup, but just for + // defensive reasons + if (nextDataToSendTimeStamp == null) { + log.warning("nextDataToSendTimeStamp is null, cant continue with processing"); + return; + } + // get the normalized data using the instance device name to map to the instance + // OCID + NormalizedData normalizedData = getNormalizedDataFromNormalizedDataFileVersion(nextDataToSend); + if (normalizedData == null) { + log.info( + "Programming error, this should not have happened, conversion of NormalizedDataFromNormalized to NormalizedData returned null, stopping processing"); + return; + } + // depending on the mode we need to replace the timestamp with the current time + // or work out an offset for it + ZonedDateTime timeToSet = switch (this.mode) { + case REAL_TIME -> ZonedDateTime.now(UTC_TZ); + case HIGH_SPEED -> nextDataToSendTimeStamp.plus(highSpeedOffset); + }; + normalizedData.setTimeObserved(timeToSet.format(dateTimeFormatter)); + // OK, got it all, let's send it + normalizedDataMessageHandlerService.handle(normalizedData); + // need to re-schedule for the next instance ; + NormalizedDataFileVersion followingNormalizedDataFileVersion; + try { + followingNormalizedDataFileVersion = readNormalizedDataFileVersionFromInput(inputReader); + } catch (IOException e) { + log.info("retrieving followingNormalizedDataFileVersion threw an IOException (" + e.getLocalizedMessage() + + ") will not reschedule"); + return; + } + if (followingNormalizedDataFileVersion == null) { + log.info("retrieved followingNormalizedDataFileVersion is null, cannot reschedule"); + return; + } + // get the time stamp of the next file + ZonedDateTime followingNormalizedDataFileVersionTimeObserved = getTimeObservedFromNormalizedDataFileVersion( + followingNormalizedDataFileVersion); + if (followingNormalizedDataFileVersionTimeObserved == null) { + log.info("extracted timeObserved from followingNormalizedDataFileVersion is null, cannot reschedule"); + return; + } + // work out how long until we run again + Duration delayDuration = switch (mode) { + case HIGH_SPEED -> highSpeedReplayDelay; + case REAL_TIME -> Duration.between(nextDataToSendTimeStamp, followingNormalizedDataFileVersionTimeObserved); + }; + // move the saved data along + nextDataToSend = followingNormalizedDataFileVersion; + nextDataToSendTimeStamp = followingNormalizedDataFileVersionTimeObserved; + // reschedule us to run later on + executor.schedule(this, delayDuration.toNanos(), TimeUnit.NANOSECONDS); + } + + /** + * @param instanceName + * @param normalizedData + * @return + * @throws MissingInstanceException + * @throws SQLException + */ + protected NormalizedData getNormalizedDataFromNormalizedDataFileVersion( + NormalizedDataFileVersion normalizedDataFileVersion) { + String instanceDisplayName = normalizedDataFileVersion.getDigitalTwinInstanceDisplayName(); + if (instanceDisplayName == null) { + log.info("No instance name found for " + normalizedDataFileVersion + + " won't be able to get the instanceID so can't send it"); + return null; + } + // try and get the instance id, we need this + String instanceId; + try { + instanceId = deviceModelInstancesCache.getInstanceIdByInstanceDisplayName(instanceDisplayName, true); + } catch (MissingInstanceException e) { + log.info("Can't get the instanceId for instance named " + instanceDisplayName); + return null; + } catch (SQLException e) { + log.warning("SQLException getting the instanceId, " + e.getLocalizedMessage()); + return null; + } + return normalizedDataFileVersion.buildTo(instanceId); + } + + @Override + public int getOrder() { + return order; + } + + @Override + public String getConfig() { + return this.toString(); + } +} From d2757abb2ced4470a1d8b318aede4f38bd612795 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Thu, 23 Jul 2026 12:23:20 +0100 Subject: [PATCH 14/24] added lots more log messages --- .../filereader/NormalizedDataFileInput.java | 31 ++++++++++++++++--- 1 file changed, 27 insertions(+), 4 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java index a4a80f7..fb2525f 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java @@ -123,9 +123,9 @@ public NormalizedDataFileInput(ObjectMapper mapper, @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_SOURCE_FILE) String sourceFilename, @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_START_OFFSET, defaultValue = "0s") Duration replayStartOffset, @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_DURATION) Duration replayDuration, + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_END) Optional replayEnd, @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_MODE, defaultValue = "REAL_TIME") FileDataInputMode mode, - @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_HIGH_SPEED_PLAYBACK_DELAY, defaultValue = "100ms") Duration highSpeedReplayDelay, - @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_END) Optional replayEnd) { + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_HIGH_SPEED_PLAYBACK_DELAY, defaultValue = "100ms") Duration highSpeedReplayDelay) { this.mapper = mapper; this.normalizedDataMessageHandlerService = normalizedDataMessageHandlerService; this.deviceModelInstancesCache = deviceModelInstancesCache; @@ -153,11 +153,25 @@ public void configureDBClient(String filteringRule) throws DateTimeParseExceptio // this, it's up to the caller to then remove us from any further processing throw new EOFException("No data in input file, cannot determine time stamps or start point"); } + log.info(() -> "Initial start time observed is " + startZDT.format(dateTimeFormatter)); // this is the start point based on the timestamps in the data file - startOffsetZDT = startZDT.plus(replayDuration); + startOffsetZDT = startZDT.plus(replayStartOffset); + + log.info(() -> "Start time observed after applying start offset of " + replayStartOffset + " is " + + startOffsetZDT.format(dateTimeFormatter)); // now add the replay time to the start offset time, this is also based on the // data file timestamps this.stopAfterZDT = startOffsetZDT.plus(replayDuration); + log.info(() -> "Stop time after applying offset of " + replayDuration + " to the start time observed is " + + stopAfterZDT.format(dateTimeFormatter)); + if (mode == FileDataInputMode.REAL_TIME) { + log.info(() -> "Replay is in real time, so will start with time observed of now (" + + ZonedDateTime.now().format(dateTimeFormatter)); + } else { + log.info(() -> "Replay is high speed so the time observed for the last entry uploaded is " + + replayEnd.format(dateTimeFormatter)); + } + // if we are in REAL_TIME replay mode we will be sending based on the current // time and then waiting for the next to send (based on the difference between // the one we just sent and the next one we're about to send) so for that we @@ -170,15 +184,20 @@ public void configureDBClient(String filteringRule) throws DateTimeParseExceptio // be in the past as well as now) relative to the end timestamp we want to // finish with. highSpeedOffset = Duration.between(stopAfterZDT, replayEnd); + log.info(() -> "Calculated highspeed offset from is " + highSpeedOffset + + ", this represents the time from the calculated replay " + stopAfterZDT.format(dateTimeFormatter) + + " to the specified end timeobserved of " + replayEnd.format(dateTimeFormatter)); // we're going to reset the reader as we're looking to load // now we need to move forwards until we get to the start point, if we get null // then we've fallen off the end of the input stream, so need to error ZonedDateTime readZDT = startZDT; while ((readZDT != null) && (readZDT.isBefore(startOffsetZDT))) { + log.info("Discarding entry " + nextDataToSend + " as it's before the start point"); nextDataToSend = readNormalizedDataFileVersionFromInput(inputReader); readZDT = getTimeObservedFromNormalizedDataFileVersion(nextDataToSend); } if (nextDataToSend == null) { + log.warning("Hit the end of file while moving forward to the specified start point"); throw new EOFException("Hit the end of file while moving forward to the specified start point"); } // to avoid doing multiple time conversions later stash the current timestamp @@ -274,7 +293,7 @@ public void run() { log.info("nextDataToSend is null, stopping processing"); } - log.info("Running a send cycle on " + nextDataToSend); + log.info(() -> "Running a send cycle on " + nextDataToSend); // the timestamp was extracted when the nextDataToSend was setup, but just for // defensive reasons if (nextDataToSendTimeStamp == null) { @@ -289,6 +308,7 @@ public void run() { "Programming error, this should not have happened, conversion of NormalizedDataFromNormalized to NormalizedData returned null, stopping processing"); return; } + log.finer(() -> "Extracted NormalizedData pre time adjustment is " + normalizedData); // depending on the mode we need to replace the timestamp with the current time // or work out an offset for it ZonedDateTime timeToSet = switch (this.mode) { @@ -296,6 +316,8 @@ public void run() { case HIGH_SPEED -> nextDataToSendTimeStamp.plus(highSpeedOffset); }; normalizedData.setTimeObserved(timeToSet.format(dateTimeFormatter)); + log.finer(() -> "Extracted NormalizedData tiemObserved after time adjustment is " + + normalizedData.getTimeObserved() + " Sending to message handlers"); // OK, got it all, let's send it normalizedDataMessageHandlerService.handle(normalizedData); // need to re-schedule for the next instance ; @@ -323,6 +345,7 @@ public void run() { case HIGH_SPEED -> highSpeedReplayDelay; case REAL_TIME -> Duration.between(nextDataToSendTimeStamp, followingNormalizedDataFileVersionTimeObserved); }; + log.finer(() -> "duration to next upload run is " + delayDuration + " (mode = " + mode + ")"); // move the saved data along nextDataToSend = followingNormalizedDataFileVersion; nextDataToSendTimeStamp = followingNormalizedDataFileVersionTimeObserved; From e5aa7ea6d2374fc99be83d6de0fc286bc71a3829 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Thu, 23 Jul 2026 13:03:09 +0100 Subject: [PATCH 15/24] updating logs --- .../dataread/filereader/NormalizedDataFileInput.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java index fb2525f..5d9e30a 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java @@ -308,7 +308,7 @@ public void run() { "Programming error, this should not have happened, conversion of NormalizedDataFromNormalized to NormalizedData returned null, stopping processing"); return; } - log.finer(() -> "Extracted NormalizedData pre time adjustment is " + normalizedData); + log.info(() -> "Extracted NormalizedData pre time adjustment is " + normalizedData); // depending on the mode we need to replace the timestamp with the current time // or work out an offset for it ZonedDateTime timeToSet = switch (this.mode) { @@ -316,7 +316,7 @@ public void run() { case HIGH_SPEED -> nextDataToSendTimeStamp.plus(highSpeedOffset); }; normalizedData.setTimeObserved(timeToSet.format(dateTimeFormatter)); - log.finer(() -> "Extracted NormalizedData tiemObserved after time adjustment is " + log.info(() -> "Extracted NormalizedData tiemObserved after time adjustment is " + normalizedData.getTimeObserved() + " Sending to message handlers"); // OK, got it all, let's send it normalizedDataMessageHandlerService.handle(normalizedData); @@ -345,7 +345,7 @@ public void run() { case HIGH_SPEED -> highSpeedReplayDelay; case REAL_TIME -> Duration.between(nextDataToSendTimeStamp, followingNormalizedDataFileVersionTimeObserved); }; - log.finer(() -> "duration to next upload run is " + delayDuration + " (mode = " + mode + ")"); + log.info(() -> "duration to next upload run is " + delayDuration + " (mode = " + mode + ")"); // move the saved data along nextDataToSend = followingNormalizedDataFileVersion; nextDataToSendTimeStamp = followingNormalizedDataFileVersionTimeObserved; From 1ef753e516a5bef23960eb0c6dc83d88965578bd Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Thu, 23 Jul 2026 13:12:12 +0100 Subject: [PATCH 16/24] better exception handling in run, looks like there is a problem, but need to figure where --- .../filereader/NormalizedDataFileInput.java | 128 +++++++++--------- 1 file changed, 67 insertions(+), 61 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java index 5d9e30a..1dd83ac 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java @@ -287,70 +287,76 @@ public void unconfigureDBClient() throws Exception { @Override public void run() { - // save the thread we're running in so we can interrupt it later - currentThread = Thread.currentThread(); - if (nextDataToSend == null) { - log.info("nextDataToSend is null, stopping processing"); - } - - log.info(() -> "Running a send cycle on " + nextDataToSend); - // the timestamp was extracted when the nextDataToSend was setup, but just for - // defensive reasons - if (nextDataToSendTimeStamp == null) { - log.warning("nextDataToSendTimeStamp is null, cant continue with processing"); - return; - } - // get the normalized data using the instance device name to map to the instance - // OCID - NormalizedData normalizedData = getNormalizedDataFromNormalizedDataFileVersion(nextDataToSend); - if (normalizedData == null) { - log.info( - "Programming error, this should not have happened, conversion of NormalizedDataFromNormalized to NormalizedData returned null, stopping processing"); - return; - } - log.info(() -> "Extracted NormalizedData pre time adjustment is " + normalizedData); - // depending on the mode we need to replace the timestamp with the current time - // or work out an offset for it - ZonedDateTime timeToSet = switch (this.mode) { - case REAL_TIME -> ZonedDateTime.now(UTC_TZ); - case HIGH_SPEED -> nextDataToSendTimeStamp.plus(highSpeedOffset); - }; - normalizedData.setTimeObserved(timeToSet.format(dateTimeFormatter)); - log.info(() -> "Extracted NormalizedData tiemObserved after time adjustment is " - + normalizedData.getTimeObserved() + " Sending to message handlers"); - // OK, got it all, let's send it - normalizedDataMessageHandlerService.handle(normalizedData); - // need to re-schedule for the next instance ; - NormalizedDataFileVersion followingNormalizedDataFileVersion; try { - followingNormalizedDataFileVersion = readNormalizedDataFileVersionFromInput(inputReader); - } catch (IOException e) { - log.info("retrieving followingNormalizedDataFileVersion threw an IOException (" + e.getLocalizedMessage() - + ") will not reschedule"); - return; - } - if (followingNormalizedDataFileVersion == null) { - log.info("retrieved followingNormalizedDataFileVersion is null, cannot reschedule"); - return; - } - // get the time stamp of the next file - ZonedDateTime followingNormalizedDataFileVersionTimeObserved = getTimeObservedFromNormalizedDataFileVersion( - followingNormalizedDataFileVersion); - if (followingNormalizedDataFileVersionTimeObserved == null) { - log.info("extracted timeObserved from followingNormalizedDataFileVersion is null, cannot reschedule"); + // save the thread we're running in so we can interrupt it later + currentThread = Thread.currentThread(); + if (nextDataToSend == null) { + log.info("nextDataToSend is null, stopping processing"); + } + + log.info(() -> "Running a send cycle on " + nextDataToSend); + // the timestamp was extracted when the nextDataToSend was setup, but just for + // defensive reasons + if (nextDataToSendTimeStamp == null) { + log.warning("nextDataToSendTimeStamp is null, cant continue with processing"); + return; + } + // get the normalized data using the instance device name to map to the instance + // OCID + NormalizedData normalizedData = getNormalizedDataFromNormalizedDataFileVersion(nextDataToSend); + if (normalizedData == null) { + log.info( + "Programming error, this should not have happened, conversion of NormalizedDataFromNormalized to NormalizedData returned null, stopping processing"); + return; + } + log.info(() -> "Extracted NormalizedData pre time adjustment is " + normalizedData); + // depending on the mode we need to replace the timestamp with the current time + // or work out an offset for it + ZonedDateTime timeToSet = switch (this.mode) { + case REAL_TIME -> ZonedDateTime.now(UTC_TZ); + case HIGH_SPEED -> nextDataToSendTimeStamp.plus(highSpeedOffset); + }; + normalizedData.setTimeObserved(timeToSet.format(dateTimeFormatter)); + log.info(() -> "Extracted NormalizedData tiemObserved after time adjustment is " + + normalizedData.getTimeObserved() + " Sending to message handlers"); + // OK, got it all, let's send it + normalizedDataMessageHandlerService.handle(normalizedData); + // need to re-schedule for the next instance ; + NormalizedDataFileVersion followingNormalizedDataFileVersion; + try { + followingNormalizedDataFileVersion = readNormalizedDataFileVersionFromInput(inputReader); + } catch (IOException e) { + log.info("retrieving followingNormalizedDataFileVersion threw an IOException (" + + e.getLocalizedMessage() + ") will not reschedule"); + return; + } + if (followingNormalizedDataFileVersion == null) { + log.info("retrieved followingNormalizedDataFileVersion is null, cannot reschedule"); + return; + } + // get the time stamp of the next file + ZonedDateTime followingNormalizedDataFileVersionTimeObserved = getTimeObservedFromNormalizedDataFileVersion( + followingNormalizedDataFileVersion); + if (followingNormalizedDataFileVersionTimeObserved == null) { + log.info("extracted timeObserved from followingNormalizedDataFileVersion is null, cannot reschedule"); + return; + } + // work out how long until we run again + Duration delayDuration = switch (mode) { + case HIGH_SPEED -> highSpeedReplayDelay; + case REAL_TIME -> Duration.between(nextDataToSendTimeStamp, followingNormalizedDataFileVersionTimeObserved); + }; + log.info(() -> "duration to next upload run is " + delayDuration + " (mode = " + mode + ")"); + // move the saved data along + nextDataToSend = followingNormalizedDataFileVersion; + nextDataToSendTimeStamp = followingNormalizedDataFileVersionTimeObserved; + // reschedule us to run later on + executor.schedule(this, delayDuration.toNanos(), TimeUnit.NANOSECONDS); + } catch (Exception e) { + log.severe("Exception in run, cannot continue. " + e.getLocalizedMessage()); + e.printStackTrace(); return; } - // work out how long until we run again - Duration delayDuration = switch (mode) { - case HIGH_SPEED -> highSpeedReplayDelay; - case REAL_TIME -> Duration.between(nextDataToSendTimeStamp, followingNormalizedDataFileVersionTimeObserved); - }; - log.info(() -> "duration to next upload run is " + delayDuration + " (mode = " + mode + ")"); - // move the saved data along - nextDataToSend = followingNormalizedDataFileVersion; - nextDataToSendTimeStamp = followingNormalizedDataFileVersionTimeObserved; - // reschedule us to run later on - executor.schedule(this, delayDuration.toNanos(), TimeUnit.NANOSECONDS); } /** From 92533b0a3959dc331a2585d181fdc1b3eb05e6ca Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:14:58 +0100 Subject: [PATCH 17/24] error seems to be in a called method in the cache, adding debug there --- .../iotdbutils/DeviceModelInstancesCache.java | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java index ad23b1d..a1e26d4 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java @@ -84,7 +84,7 @@ public class DeviceModelInstancesCache { private final Map instanceIdToModelId = Collections.synchronizedMap(new HashMap<>()); private final Map instanceIdToExternalKey = Collections.synchronizedMap(new HashMap<>()); - private final Map instanceIdToInstanceName = Collections.synchronizedMap(new HashMap<>()); + private final Map instanceIdToInstanceDisplayName = Collections.synchronizedMap(new HashMap<>()); private final Map instanceDisplayNameToInstanceId = Collections.synchronizedMap(new HashMap<>()); private final Map instanceIdToModelName = Collections.synchronizedMap(new HashMap<>()); private final Map externalKeyToInstanceId = Collections.synchronizedMap(new HashMap<>()); @@ -366,13 +366,15 @@ 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"); @@ -464,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 @@ -641,7 +643,7 @@ 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 " @@ -684,7 +686,7 @@ private InstanceKeyInfo loadInstanceByInstanceDisplayName(@NotNull @NotEmpty Str 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 " @@ -699,7 +701,11 @@ private InstanceKeyInfo loadInstanceByInstanceDisplayName(@NotNull @NotEmpty Str throw new MissingInstanceException("No instance found for instance id " + instanceDisplayName); } } catch (SQLException e) { - log.severe("SQLException getting existing model / instance mappings, " + e.getLocalizedMessage()); + log.severe( + "SQLException getting existing model / instance mappings in loadInstanceByInstanceDisplayName, " + + e.getLocalizedMessage() + " in class " + e.getStackTrace()[0].getFileName() + + " in method " + e.getStackTrace()[0].getMethodName() + " at line " + + e.getStackTrace()[0].getLineNumber()); throw e; } } From fdc67e24816f43ceb2425283eabac15d9e01f402 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:34:55 +0100 Subject: [PATCH 18/24] try to correctly load the json from it's string, use a fall back if needed --- .../filewriter/NormalizedDataFileVersion.java | 20 ++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java index d670f54..d68b4c3 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -92,7 +92,25 @@ public NormalizedData buildTo(String instanceId) { public static NormalizedData buildTo(NormalizedDataFileVersion input, String instanceId) { StringReader contentInputReader = new StringReader(input.getContentJsonValue()); OracleJsonParser oracleJsonParser = NormalizedDataFileVersion.factory.createJsonTextParser(contentInputReader); - OracleJsonValue oracleJsonValue = oracleJsonParser.getValue(); + + OracleJsonValue oracleJsonValue; + if (oracleJsonParser.hasNext()) { + oracleJsonParser.next(); + oracleJsonValue = oracleJsonParser.getValue(); + } else { + // try and get it from the previous text + contentInputReader = new StringReader(input.getContent()); + oracleJsonParser = NormalizedDataFileVersion.factory.createJsonTextParser(contentInputReader); + + if (oracleJsonParser.hasNext()) { + oracleJsonParser.next(); + oracleJsonValue = oracleJsonParser.getValue(); + } else { + log.warning( + "Unable to create normalized object from the stored oracle jason value or the content string"); + return null; + } + } return NormalizedData.builder().digitalTwinInstanceId(instanceId).contentPath(input.getContentPath()) .timeObserved(input.getTimeObserved()).contentType(input.getContentType()).content(input.getContent()) .contentJsonValue(oracleJsonValue).contentJsonType(input.getContentJsonType()).build(); From df2b146c875c970f6f0f1c2f5088ff5a45a3e346 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Thu, 23 Jul 2026 15:35:45 +0100 Subject: [PATCH 19/24] use correct prepared statement, DB cache now loads after construct, not startup --- .../iotdbutils/DeviceModelInstancesCache.java | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java index a1e26d4..47e9b70 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/iotdbutils/DeviceModelInstancesCache.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; @@ -117,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) { @@ -677,7 +677,7 @@ private InstanceKeyInfo loadInstanceByInstanceDisplayName(@NotNull @NotEmpty Str synchronized (selectInstanceDetailsByInstanceDisplayNamePS) { selectInstanceDetailsByInstanceDisplayNamePS.setString(1, instanceDisplayName); // get all of the results - try (ResultSet rs = selectInstanceDetailsByInstanceIdPS.executeQuery()) { + try (ResultSet rs = selectInstanceDetailsByInstanceDisplayNamePS.executeQuery()) { if (rs.next()) { String modelId = rs.getString(MODEL_ID_COLUMN_NAME); String externalKey = rs.getString(EXTERNAL_KEY_COLUMN_NAME); @@ -703,9 +703,7 @@ private InstanceKeyInfo loadInstanceByInstanceDisplayName(@NotNull @NotEmpty Str } catch (SQLException e) { log.severe( "SQLException getting existing model / instance mappings in loadInstanceByInstanceDisplayName, " - + e.getLocalizedMessage() + " in class " + e.getStackTrace()[0].getFileName() - + " in method " + e.getStackTrace()[0].getMethodName() + " at line " - + e.getStackTrace()[0].getLineNumber()); + + e.getLocalizedMessage()); throw e; } } From e4889298b3116c72765b8444636182bb1c83095d Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Thu, 23 Jul 2026 16:03:21 +0100 Subject: [PATCH 20/24] updated some property names --- .../dataread/filereader/FileReaderProperties.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java index 5a086c8..11cc400 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java @@ -9,11 +9,12 @@ public class FileReaderProperties { + ".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_END = NORMALIZED_DATA_FILE_INPUT_REPLAY + ".end"; + 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 + ".source_file"; + 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"; @@ -21,9 +22,10 @@ public class FileReaderProperties { 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_END = RAW_DATA_FILE_INPUT_REPLAY + ".end"; + 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 + ".source_file"; + public final static String RAW_DATA_FILE_INPUT_SOURCE_FILE = RAW_DATA_FILE_INPUT + ".sourcefile"; } From 4fb713496063ff94074b5cd172321f37ae1bb2dd Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Thu, 23 Jul 2026 16:04:12 +0100 Subject: [PATCH 21/24] updated the propery name, now checks for the end of data time stamp --- .../filereader/NormalizedDataFileInput.java | 27 ++++++++++++++----- .../filewriter/FileWriterProperties.java | 4 +-- 2 files changed, 22 insertions(+), 9 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java index 1dd83ac..5f975ce 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java @@ -88,7 +88,7 @@ public class NormalizedDataFileInput implements IoTDBClient, Runnable { @ToString.Include private Duration replayDuration; @ToString.Include - private ZonedDateTime replayEnd; + private ZonedDateTime replayEndTimeObserved; @ToString.Include private final FileDataInputMode mode; @ToString.Include @@ -123,7 +123,7 @@ public NormalizedDataFileInput(ObjectMapper mapper, @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_SOURCE_FILE) String sourceFilename, @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_START_OFFSET, defaultValue = "0s") Duration replayStartOffset, @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_DURATION) Duration replayDuration, - @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_END) Optional replayEnd, + @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_OUTPUT_END_TIME_OBSERVED) Optional replayEndTimeObserved, @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_MODE, defaultValue = "REAL_TIME") FileDataInputMode mode, @Property(name = FileReaderProperties.NORMALIZED_DATA_FILE_INPUT_REPLAY_HIGH_SPEED_PLAYBACK_DELAY, defaultValue = "100ms") Duration highSpeedReplayDelay) { this.mapper = mapper; @@ -137,7 +137,7 @@ public NormalizedDataFileInput(ObjectMapper mapper, this.highSpeedReplayDelay = highSpeedReplayDelay; // if we have a specified end time for the replay use the current time, if now // use what's been specified - this.replayEnd = replayEnd.orElse(ZonedDateTime.now(UTC_TZ)); + this.replayEndTimeObserved = replayEndTimeObserved.orElse(ZonedDateTime.now(UTC_TZ)); } @Override @@ -169,7 +169,7 @@ public void configureDBClient(String filteringRule) throws DateTimeParseExceptio + ZonedDateTime.now().format(dateTimeFormatter)); } else { log.info(() -> "Replay is high speed so the time observed for the last entry uploaded is " - + replayEnd.format(dateTimeFormatter)); + + replayEndTimeObserved.format(dateTimeFormatter)); } // if we are in REAL_TIME replay mode we will be sending based on the current @@ -183,10 +183,10 @@ public void configureDBClient(String filteringRule) throws DateTimeParseExceptio // sent, and the timestamp found at the replay end point (which of course could // be in the past as well as now) relative to the end timestamp we want to // finish with. - highSpeedOffset = Duration.between(stopAfterZDT, replayEnd); + highSpeedOffset = Duration.between(stopAfterZDT, replayEndTimeObserved); log.info(() -> "Calculated highspeed offset from is " + highSpeedOffset + ", this represents the time from the calculated replay " + stopAfterZDT.format(dateTimeFormatter) - + " to the specified end timeobserved of " + replayEnd.format(dateTimeFormatter)); + + " to the specified end timeobserved of " + replayEndTimeObserved.format(dateTimeFormatter)); // we're going to reset the reader as we're looking to load // now we need to move forwards until we get to the start point, if we get null // then we've fallen off the end of the input stream, so need to error @@ -341,12 +341,25 @@ public void run() { log.info("extracted timeObserved from followingNormalizedDataFileVersion is null, cannot reschedule"); return; } + // is the one just loaded BEFORE the stop time ? + if (followingNormalizedDataFileVersionTimeObserved.isAfter(stopAfterZDT)) { + log.info("the next data item just loaded has a timeObserved of " + + followingNormalizedDataFileVersionTimeObserved.format(dateTimeFormatter) + + " which is after the stop time of " + stopAfterZDT.format(dateTimeFormatter)); + log.info("Stopping replay"); + return; + } + // OK we need to send the next one, work out how long we have // work out how long until we run again Duration delayDuration = switch (mode) { case HIGH_SPEED -> highSpeedReplayDelay; case REAL_TIME -> Duration.between(nextDataToSendTimeStamp, followingNormalizedDataFileVersionTimeObserved); }; - log.info(() -> "duration to next upload run is " + delayDuration + " (mode = " + mode + ")"); + if (delayDuration.isNegative()) { + delayDuration = Duration.ZERO; + } + Duration tmpDuration = delayDuration; + log.info(() -> "duration to next upload run is " + tmpDuration + " (mode = " + mode + ")"); // move the saved data along nextDataToSend = followingNormalizedDataFileVersion; nextDataToSendTimeStamp = followingNormalizedDataFileVersionTimeObserved; diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/FileWriterProperties.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/FileWriterProperties.java index 5ef66b8..8a04b2b 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/FileWriterProperties.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/FileWriterProperties.java @@ -5,11 +5,11 @@ public class FileWriterProperties { 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 + ".target_file"; + 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 + ".target_file"; + public final static String RAW_DATA_FILE_OUTPUT_TARGET_FILE = RAW_DATA_FILE_OUTPUT + ".targetfile"; } From 09640792da0648fe8d317131921e32fcd29368da Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Thu, 23 Jul 2026 17:30:08 +0100 Subject: [PATCH 22/24] better resilience against corrupted data in input file --- .../filereader/NormalizedDataFileInput.java | 39 ++++++++++++------- 1 file changed, 25 insertions(+), 14 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java index 5f975ce..9ed337c 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java @@ -99,6 +99,11 @@ public class NormalizedDataFileInput implements IoTDBClient, Runnable { private ZonedDateTime stopAfterZDT; @ToString.Include private Duration highSpeedOffset; + @ToString.Include + private NormalizedDataFileVersion nextDataToSend; + @ToString.Include + private ZonedDateTime nextDataToSendTimeStamp; + @ToString.Exclude private final ObjectMapper mapper; @ToString.Exclude @@ -107,13 +112,10 @@ public class NormalizedDataFileInput implements IoTDBClient, Runnable { private Thread currentThread; @ToString.Exclude private ScheduledExecutorService executor; - @ToString.Include - private NormalizedDataFileVersion nextDataToSend; - + @ToString.Exclude private final NormalizedDataMessageHandlerService normalizedDataMessageHandlerService; - + @ToString.Exclude private final DeviceModelInstancesCache deviceModelInstancesCache; - private ZonedDateTime nextDataToSendTimeStamp; @Inject public NormalizedDataFileInput(ObjectMapper mapper, @@ -193,8 +195,12 @@ public void configureDBClient(String filteringRule) throws DateTimeParseExceptio ZonedDateTime readZDT = startZDT; while ((readZDT != null) && (readZDT.isBefore(startOffsetZDT))) { log.info("Discarding entry " + nextDataToSend + " as it's before the start point"); - nextDataToSend = readNormalizedDataFileVersionFromInput(inputReader); - readZDT = getTimeObservedFromNormalizedDataFileVersion(nextDataToSend); + try { + nextDataToSend = readNormalizedDataFileVersionFromInput(inputReader); + readZDT = getTimeObservedFromNormalizedDataFileVersion(nextDataToSend); + } catch (IOException e) { + log.warning("Problem getting line " + e.getLocalizedMessage() + ", skipping line anyway"); + } } if (nextDataToSend == null) { log.warning("Hit the end of file while moving forward to the specified start point"); @@ -322,13 +328,18 @@ public void run() { // OK, got it all, let's send it normalizedDataMessageHandlerService.handle(normalizedData); // need to re-schedule for the next instance ; - NormalizedDataFileVersion followingNormalizedDataFileVersion; - try { - followingNormalizedDataFileVersion = readNormalizedDataFileVersionFromInput(inputReader); - } catch (IOException e) { - log.info("retrieving followingNormalizedDataFileVersion threw an IOException (" - + e.getLocalizedMessage() + ") will not reschedule"); - return; + NormalizedDataFileVersion followingNormalizedDataFileVersion = null; + // try and get the line, if there is a json problem then an IOException will be + // thrown + while (true) { + try { + followingNormalizedDataFileVersion = readNormalizedDataFileVersionFromInput(inputReader); + break; + } catch (IOException e) { + log.info("retrieving followingNormalizedDataFileVersion threw an IOException (" + + e.getLocalizedMessage() + ") skipping this input"); + return; + } } if (followingNormalizedDataFileVersion == null) { log.info("retrieved followingNormalizedDataFileVersion is null, cannot reschedule"); From 3dfedae8477907006e78fe62b019538cae0340d3 Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Mon, 27 Jul 2026 12:56:56 +0100 Subject: [PATCH 23/24] changed to default calculation mode for time stamps, make low level debugs lambdas to improve efficiency is not logging at that level --- .../HomeAssistantMonitoredEntitySet.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/homeassistantentities/HomeAssistantMonitoredEntitySet.java b/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/homeassistantentities/HomeAssistantMonitoredEntitySet.java index c89cfac..baec351 100644 --- a/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/homeassistantentities/HomeAssistantMonitoredEntitySet.java +++ b/IoTDemoProxyGateway/src/main/java/com/oracle/demo/timg/iot/iotproxygateway/homeassistantentities/HomeAssistantMonitoredEntitySet.java @@ -80,7 +80,7 @@ public class HomeAssistantMonitoredEntitySet implements Runnable { private Duration retrievalrate = Duration.ofSeconds(10); private String devicekey; private String endpoint; - private TimestampMode timestampMode = TimestampMode.EARLIEST; + private TimestampMode timestampMode = TimestampMode.LATEST; private List monitoredentities; @ToString.Exclude @Inject @@ -198,7 +198,7 @@ private ZonedDateTime processEntity(Map payload, HomeAssistantMo gatewayStats.trackFailedHARetrieveCall(); return null; } - log.finer("Returned state string is :" + stateString); + log.finer(() -> "Returned state string is :" + stateString); HomeAssistantState state; try { state = mapper.readValue(stateString, HomeAssistantState.class); @@ -207,7 +207,7 @@ private ZonedDateTime processEntity(Map payload, HomeAssistantMo gatewayStats.trackFailedHARetrieveCall(); return null; } - log.fine("Extracted state is " + state); + log.fine(() -> "Extracted state is " + state); gatewayStats.trackSucessfullHARetrieveCall(); // make sure that we have the relevant times, even if we don't use them here // they may be needed on another pass through From 00a1354b946410887ecab66e4d3909dcf97781bc Mon Sep 17 00:00:00 2001 From: tim_graves <28924492+atimgraves@users.noreply.github.com> Date: Mon, 27 Jul 2026 12:58:17 +0100 Subject: [PATCH 24/24] make a few values final --- .../filereader/NormalizedDataFileInput.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java index 9ed337c..4e48a0e 100644 --- a/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java @@ -84,11 +84,11 @@ public class NormalizedDataFileInput implements IoTDBClient, Runnable { @ToString.Include private final String sourceFilename; @ToString.Include - private Duration replayStartOffset; + private final Duration replayStartOffset; @ToString.Include - private Duration replayDuration; + private final Duration replayDuration; @ToString.Include - private ZonedDateTime replayEndTimeObserved; + private final ZonedDateTime replayEndTimeObserved; @ToString.Include private final FileDataInputMode mode; @ToString.Include @@ -194,7 +194,7 @@ public void configureDBClient(String filteringRule) throws DateTimeParseExceptio // then we've fallen off the end of the input stream, so need to error ZonedDateTime readZDT = startZDT; while ((readZDT != null) && (readZDT.isBefore(startOffsetZDT))) { - log.info("Discarding entry " + nextDataToSend + " as it's before the start point"); + log.finer("Discarding entry " + nextDataToSend + " as it's before the start point"); try { nextDataToSend = readNormalizedDataFileVersionFromInput(inputReader); readZDT = getTimeObservedFromNormalizedDataFileVersion(nextDataToSend); @@ -300,7 +300,7 @@ public void run() { log.info("nextDataToSend is null, stopping processing"); } - log.info(() -> "Running a send cycle on " + nextDataToSend); + log.fine(() -> "Running a send cycle on " + nextDataToSend); // the timestamp was extracted when the nextDataToSend was setup, but just for // defensive reasons if (nextDataToSendTimeStamp == null) { @@ -315,7 +315,7 @@ public void run() { "Programming error, this should not have happened, conversion of NormalizedDataFromNormalized to NormalizedData returned null, stopping processing"); return; } - log.info(() -> "Extracted NormalizedData pre time adjustment is " + normalizedData); + log.finer(() -> "Extracted NormalizedData pre time adjustment is " + normalizedData); // depending on the mode we need to replace the timestamp with the current time // or work out an offset for it ZonedDateTime timeToSet = switch (this.mode) { @@ -323,7 +323,7 @@ public void run() { case HIGH_SPEED -> nextDataToSendTimeStamp.plus(highSpeedOffset); }; normalizedData.setTimeObserved(timeToSet.format(dateTimeFormatter)); - log.info(() -> "Extracted NormalizedData tiemObserved after time adjustment is " + log.finer(() -> "Extracted NormalizedData tiemObserved after time adjustment is " + normalizedData.getTimeObserved() + " Sending to message handlers"); // OK, got it all, let's send it normalizedDataMessageHandlerService.handle(normalizedData); @@ -370,7 +370,7 @@ public void run() { delayDuration = Duration.ZERO; } Duration tmpDuration = delayDuration; - log.info(() -> "duration to next upload run is " + tmpDuration + " (mode = " + mode + ")"); + log.finer(() -> "duration to next upload run is " + tmpDuration + " (mode = " + mode + ")"); // move the saved data along nextDataToSend = followingNormalizedDataFileVersion; nextDataToSendTimeStamp = followingNormalizedDataFileVersionTimeObserved;