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..11cc400 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/FileReaderProperties.java @@ -0,0 +1,31 @@ +package com.oracle.demo.timg.iot.iotdbjdbc.dataread.filereader; + +public class FileReaderProperties { + public final static String NORMALIZED_DATA_FILE_INPUT = "iotdatacache.filereader.normalizeddata"; + public final static String NORMALIZED_DATA_FILE_INPUT_ENABLED = NORMALIZED_DATA_FILE_INPUT + ".enabled"; + public final static String NORMALIZED_DATA_FILE_INPUT_ORDER = NORMALIZED_DATA_FILE_INPUT + ".order"; + public final static String NORMALIZED_DATA_FILE_INPUT_REPLAY = NORMALIZED_DATA_FILE_INPUT + ".replay"; + public final static String NORMALIZED_DATA_FILE_INPUT_REPLAY_DURATION = NORMALIZED_DATA_FILE_INPUT_REPLAY + + ".duration"; + public static final String NORMALIZED_DATA_FILE_INPUT_REPLAY_START_OFFSET = NORMALIZED_DATA_FILE_INPUT_REPLAY + + ".startoffset"; + public final static String NORMALIZED_DATA_FILE_INPUT_REPLAY_OUTPUT_END_TIME_OBSERVED = NORMALIZED_DATA_FILE_INPUT_REPLAY + + ".endtimeobserved"; + public final static String NORMALIZED_DATA_FILE_INPUT_REPLAY_HIGH_SPEED_PLAYBACK_DELAY = NORMALIZED_DATA_FILE_INPUT_REPLAY + + ".highspeeddelay"; + public final static String NORMALIZED_DATA_FILE_INPUT_MODE = NORMALIZED_DATA_FILE_INPUT + ".mode"; + public final static String NORMALIZED_DATA_FILE_INPUT_SOURCE_FILE = NORMALIZED_DATA_FILE_INPUT + ".sourcefile"; + + public final static String RAW_DATA_FILE_INPUT = "iotdatacache.filereader.rawdata"; + public final static String RAW_DATA_FILE_INPUT_ENABLED = RAW_DATA_FILE_INPUT + ".enabled"; + public final static String RAW_DATA_FILE_INPUT_ORDER = RAW_DATA_FILE_INPUT + ".order"; + public final static String RAW_DATA_FILE_INPUT_REPLAY = RAW_DATA_FILE_INPUT + ".replay"; + public final static String RAW_DATA_FILE_INPUT_REPLAY_DURATION = RAW_DATA_FILE_INPUT_REPLAY + ".duration"; + public static final String RAW_DATA_FILE_INPUT_REPLAY_START_OFFSET = RAW_DATA_FILE_INPUT_REPLAY + ".startoffset"; + public final static String RAW_DATA_FILE_INPUT_REPLAY_OUTPUT_END_TIME_OBSERVED = RAW_DATA_FILE_INPUT_REPLAY + + ".endtimeobserved"; + public final static String RAW_DATA_FILE_INPUT_REPLAY_HIGH_SPEED_PLAYBACK_DELAY = RAW_DATA_FILE_INPUT_REPLAY + + ".highspeeddelay"; + public final static String RAW_DATA_FILE_INPUT_MODE = RAW_DATA_FILE_INPUT + ".mode"; + public final static String RAW_DATA_FILE_INPUT_SOURCE_FILE = RAW_DATA_FILE_INPUT + ".sourcefile"; +} 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..4e48a0e --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/dataread/filereader/NormalizedDataFileInput.java @@ -0,0 +1,424 @@ +/*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 final Duration replayStartOffset; + @ToString.Include + private final Duration replayDuration; + @ToString.Include + private final ZonedDateTime replayEndTimeObserved; + @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.Include + private NormalizedDataFileVersion nextDataToSend; + @ToString.Include + private ZonedDateTime nextDataToSendTimeStamp; + + @ToString.Exclude + private final ObjectMapper mapper; + @ToString.Exclude + private BufferedReader inputReader; + @ToString.Exclude + private Thread currentThread; + @ToString.Exclude + private ScheduledExecutorService executor; + @ToString.Exclude + private final NormalizedDataMessageHandlerService normalizedDataMessageHandlerService; + @ToString.Exclude + private final DeviceModelInstancesCache deviceModelInstancesCache; + + @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_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; + 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.replayEndTimeObserved = replayEndTimeObserved.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"); + } + 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(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 " + + replayEndTimeObserved.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 + // 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, 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 " + 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 + ZonedDateTime readZDT = startZDT; + while ((readZDT != null) && (readZDT.isBefore(startOffsetZDT))) { + log.finer("Discarding entry " + nextDataToSend + " as it's before the start point"); + 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"); + 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() { + try { + // 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.fine(() -> "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.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) { + case REAL_TIME -> ZonedDateTime.now(UTC_TZ); + 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 ; + 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"); + 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; + } + // 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); + }; + if (delayDuration.isNegative()) { + delayDuration = Duration.ZERO; + } + Duration tmpDuration = delayDuration; + log.finer(() -> "duration to next upload run is " + tmpDuration + " (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; + } + } + + /** + * @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(); + } +} 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, 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..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; @@ -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; @@ -83,7 +84,8 @@ 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<>()); 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; @@ -113,9 +117,9 @@ public DeviceModelInstancesCache(DBConnectionSupplier dbConnectionSupplier, this.preloadExistingInstances = preloadExistingInstances; } - @EventListener - public void onStartup(StartupEvent event) { - log.info("Startup event received for DeviceModelInstancesCache"); + @PostConstruct + public void postConstruct() { + log.info("Post Construct event received for DeviceModelInstancesCache"); try { configure(); } catch (Exception e) { @@ -149,8 +153,10 @@ public void configure() throws Exception { // set this up so we can re-use it later if we need to query for an instance we // didn't know about log.fine("Creating prepared statements"); - selectModelIdByInstanceIdPS = connection + selectInstanceDetailsByInstanceIdPS = connection .prepareStatement(SELECT_MODEL_ID_DISPLAY_NAME_AND_EXTERNAL_KEY_BY_INSTANCE_ID); + selectInstanceDetailsByInstanceDisplayNamePS = connection + .prepareStatement(SELECT_MODEL_ID_INSTANCE_ID_AND_EXTERNAL_KEY_BY_INSTANCE_DISPLAY_NAME); selectModelNameByModelIdPS = connection.prepareStatement(SELECT_MODEL_NAME_BY_MODEL_ID); selectModelIdByModelNamePS = connection.prepareStatement(SELECT_MODEL_ID_BY_MODEL_NAME); log.fine("Prepared statements created"); @@ -360,7 +366,16 @@ private void preloadExistingInstances() throws SQLException { String instanceDisplayName = rs.getString(INSTANCE_ID_COLUMN_DISPLAY_NAME); String modelName = modelIdToModelName.get(modelIdExistingInstance); instanceIdToModelId.put(instanceIdExistingInstance, modelIdExistingInstance); - instanceIdToInstanceName.put(instanceIdExistingInstance, instanceDisplayName); + instanceIdToInstanceDisplayName.put(instanceIdExistingInstance, instanceDisplayName); + if (instanceDisplayNameToInstanceId.containsKey(instanceDisplayName)) { + log.warning("Instance display name cache already contains key " + instanceDisplayName + + " connected to instance id " + instanceDisplayNameToInstanceId.get(instanceDisplayName) + + ", duplicates are not added"); + } else { + instanceDisplayNameToInstanceId.put(instanceDisplayName, instanceIdExistingInstance); + log.info("Added instance name " + instanceDisplayName + " to instanceId " + + instanceIdExistingInstance + " mapping"); + } log.info("Added instance id " + instanceIdExistingInstance + " named " + instanceDisplayName + " to modelId " + modelIdExistingInstance + " mapping"); instanceIdToModelName.put(instanceIdExistingInstance, modelName); @@ -451,10 +466,10 @@ public String getInstanceDisplayNameByInstanceId(@NotNull @NotEmpty String insta } // do we already have the info ? note that empty string and null are valid // responses here. - if (instanceIdToInstanceName.containsKey(instanceId)) { + if (instanceIdToInstanceDisplayName.containsKey(instanceId)) { // we have the key, the model could be a string, null blank etc if one hasn't // been set, but that's still valid. - return instanceIdToInstanceName.get(instanceId); + return instanceIdToInstanceDisplayName.get(instanceId); } // we don't have a cached version // let's try and locate it @@ -474,6 +489,54 @@ public String getInstanceDisplayNameByInstanceId(@NotNull @NotEmpty String insta } } + /** + * try to get the modelId from the cache or if there is no cachedata + * + * @param instanceId the instance to locate + * @param cacheMissingResults if true and we already have looked but not found + * this then don't look again + * @return the instance display name is there is one + * @throws MissingInstanceException throws an exception if we can't locate the + * instance (either in the known missing cache + * if cacheMissingResults is true, or in the + * IoT service otherwise) + * @throws SQLException if there was a problem querying the iot + * service + */ + public String getInstanceIdByInstanceDisplayName(@NotNull @NotEmpty String instanceDisplayName, + boolean cacheMissingResults) throws MissingInstanceException, SQLException { + boolean knownMissing = foundMissingInstanceDisplayNames.contains(instanceDisplayName); + // are we looking at the cache ? + if (cacheMissingResults && knownMissing) { + // we know it's missing, and we are not checking an other time + throw new MissingInstanceException( + "No instance found in cache and not checking again for instanceDisplayName" + instanceDisplayName); + } + // do we already have the info ? note that empty string and null are valid + // responses here. + if (instanceDisplayNameToInstanceId.containsKey(instanceDisplayName)) { + // we have the key, the model could be a string, null blank etc if one hasn't + // been set, but that's still valid. + return instanceDisplayNameToInstanceId.get(instanceDisplayName); + } + // we don't have a cached version + // let's try and locate it + try { + InstanceKeyInfo ike = loadInstanceByInstanceDisplayName(instanceDisplayName); + // OK we have something, was it previously tagged as knownMissing ? if so remove + // the id + if (knownMissing) { + foundMissingInstanceDisplayNames.remove(instanceDisplayName); + } + return ike.getInstanceId(); + } catch (MissingInstanceException e) { + // cache the missing result for later use + foundMissingInstanceDisplayNames.add(instanceDisplayName); + // then throw the exception + throw e; + } + } + /** * try to get the external key from the cache or if there is no cachedata * @@ -568,10 +631,10 @@ public String getModelNameByInstanceId(@NotNull @NotEmpty String instanceId, boo */ private InstanceKeyInfo loadInstanceByInstanceId(@NotNull @NotEmpty String instanceId) throws SQLException, MissingInstanceException { - synchronized (selectModelIdByInstanceIdPS) { - selectModelIdByInstanceIdPS.setString(1, instanceId); + synchronized (selectInstanceDetailsByInstanceIdPS) { + selectInstanceDetailsByInstanceIdPS.setString(1, instanceId); // get all of the results - try (ResultSet rs = selectModelIdByInstanceIdPS.executeQuery()) { + try (ResultSet rs = selectInstanceDetailsByInstanceIdPS.executeQuery()) { if (rs.next()) { String modelId = rs.getString(MODEL_ID_COLUMN_NAME); String externalKey = rs.getString(EXTERNAL_KEY_COLUMN_NAME); @@ -580,7 +643,15 @@ private InstanceKeyInfo loadInstanceByInstanceId(@NotNull @NotEmpty String insta instanceIdToModelId.put(instanceId, modelId); instanceIdToModelName.put(instanceId, modelName); instanceIdToExternalKey.put(instanceId, externalKey); - instanceIdToInstanceName.put(instanceId, instanceDisplayName); + instanceIdToInstanceDisplayName.put(instanceId, instanceDisplayName); + if (instanceDisplayNameToInstanceId.containsKey(instanceDisplayName)) { + log.warning("Instance display name cache already contains key " + instanceDisplayName + + " connected to instance id " + + instanceDisplayNameToInstanceId.get(instanceDisplayName) + + ", duplicates are not added"); + } else { + instanceDisplayNameToInstanceId.put(instanceDisplayName, instanceId); + } externalKeyToInstanceId.put(externalKey, instanceId); return new InstanceKeyInfo(instanceId, modelId, externalKey, instanceDisplayName); } else { @@ -593,6 +664,51 @@ private InstanceKeyInfo loadInstanceByInstanceId(@NotNull @NotEmpty String insta } } + /** + * on demand load an entry in the mode details cache + * + * @param instanceId + * @throws SQLException + * @throws MissingInstanceException + * @return the located modelId, null for instances with no model + */ + private InstanceKeyInfo loadInstanceByInstanceDisplayName(@NotNull @NotEmpty String instanceDisplayName) + throws SQLException, MissingInstanceException { + synchronized (selectInstanceDetailsByInstanceDisplayNamePS) { + selectInstanceDetailsByInstanceDisplayNamePS.setString(1, instanceDisplayName); + // get all of the results + try (ResultSet rs = selectInstanceDetailsByInstanceDisplayNamePS.executeQuery()) { + if (rs.next()) { + String modelId = rs.getString(MODEL_ID_COLUMN_NAME); + String externalKey = rs.getString(EXTERNAL_KEY_COLUMN_NAME); + String instanceId = rs.getString(INSTANCE_ID_COLUMN_NAME); + String modelName = modelIdToModelName.get(modelId); + instanceIdToModelId.put(instanceId, modelId); + instanceIdToModelName.put(instanceId, modelName); + instanceIdToExternalKey.put(instanceId, externalKey); + instanceIdToInstanceDisplayName.put(instanceId, instanceDisplayName); + if (instanceDisplayNameToInstanceId.containsKey(instanceDisplayName)) { + log.warning("Instance display name cache already contains key " + instanceDisplayName + + " connected to instance id " + + instanceDisplayNameToInstanceId.get(instanceDisplayName) + + ", duplicates are not added"); + } else { + instanceDisplayNameToInstanceId.put(instanceDisplayName, instanceId); + } + externalKeyToInstanceId.put(externalKey, instanceId); + return new InstanceKeyInfo(instanceId, modelId, externalKey, instanceDisplayName); + } else { + throw new MissingInstanceException("No instance found for instance id " + instanceDisplayName); + } + } catch (SQLException e) { + log.severe( + "SQLException getting existing model / instance mappings in loadInstanceByInstanceDisplayName, " + + e.getLocalizedMessage()); + throw e; + } + } + } + @Data @AllArgsConstructor private class InstanceKeyInfo { @@ -623,6 +739,7 @@ public void unconfigure() throws Exception { foundMissingModelIds.clear(); foundMissingModelNames.clear(); foundMissingInstanceIds.clear(); + foundMissingInstanceDisplayNames.clear(); foundMissingExternalKeys.clear(); configured = false; } @@ -639,6 +756,7 @@ public String getConfig() { + modelIdToModelName + ")" + " model ids, " + modelNameToModelId.size() + " ( " + modelNameToModelId + ")" + " model names, " + foundMissingModelIds.size() + " found missing model ids" + foundMissingModelNames.size() + " found missing model names, " + foundMissingInstanceIds.size() + + " found missing instance display names, " + foundMissingInstanceDisplayNames.size() + " found missing instance ids. preloadExistingModels=" + preloadExistingModels + ", preloadExistingInstances=" + preloadExistingInstances; } 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..8a04b2b --- /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 + ".targetfile"; + + public final static String RAW_DATA_FILE_OUTPUT = "messagehandler.output.rawdata.fileoutput"; + public final static String RAW_DATA_FILE_OUTPUT_ENABLED = RAW_DATA_FILE_OUTPUT + ".enabled"; + public final static String RAW_DATA_FILE_OUTPUT_ORDER = RAW_DATA_FILE_OUTPUT + ".order"; + public final static String RAW_DATA_FILE_OUTPUT_DURATION = RAW_DATA_FILE_OUTPUT + ".duration"; + public final static String RAW_DATA_FILE_OUTPUT_TARGET_FILE = RAW_DATA_FILE_OUTPUT + ".targetfile"; +} 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..30b4b6f --- /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 +@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..39098f9 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileOutput.java @@ -0,0 +1,130 @@ +/*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); + 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); + 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..d68b4c3 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/NormalizedDataFileVersion.java @@ -0,0 +1,118 @@ +/*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.io.StringReader; +import java.io.StringWriter; + +import com.oracle.demo.timg.iot.iotdbjdbc.aqdata.NormalizedData; + +import io.micronaut.serde.annotation.Serdeable; +import lombok.AllArgsConstructor; +import lombok.Builder; +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; +import oracle.sql.json.OracleJsonValue.OracleJsonType; + +@Log +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +@Serdeable +public class NormalizedDataFileVersion { + private final static OracleJsonFactory factory = new OracleJsonFactory(); + private String digitalTwinInstanceDisplayName; + private String contentPath; + private String timeObserved; + private String contentType; + private String content; + 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(contentOuputWriter.toString()).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) { + StringReader contentInputReader = new StringReader(input.getContentJsonValue()); + OracleJsonParser oracleJsonParser = NormalizedDataFileVersion.factory.createJsonTextParser(contentInputReader); + + 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(); + } +} 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..b01f801 --- /dev/null +++ b/IoTDBJDBC/src/main/java/com/oracle/demo/timg/iot/iotdbjdbc/messagehandler/outputs/filewriter/RawDataFileVersion.java @@ -0,0 +1,95 @@ +/*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.RawData; + +import io.micronaut.http.MediaType; +import io.micronaut.serde.annotation.Serdeable; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import lombok.ToString; +import lombok.extern.java.Log; + +@Log +@Data +@NoArgsConstructor +@AllArgsConstructor +@Builder +@Serdeable +public class RawDataFileVersion { + + private String digitalTwinInstanceDisplayName; + 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(); + } +} 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 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;