Reader review branch

core/opendaq/reader/include/opendaq/enum_flags.h

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2022-2026 openDAQ d.o.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+template <typename Enum>
+class EnumFlags
+{
+    static_assert(std::is_enum_v<Enum>, "EnumFlags requires an enum type");
+
+public:
+    using Underlying = std::underlying_type_t<Enum>;
+
+    constexpr EnumFlags() = default;
+
+    constexpr EnumFlags(Enum value)
+        : value(static_cast<Underlying>(value))
+    {
+    }
+
+    constexpr bool empty() const
+    {
+        return value == 0;
+    }
+
+    constexpr bool contains(Enum flag) const
+    {
+        return (value & static_cast<Underlying>(flag)) != 0;
+    }
+
+    constexpr void add(Enum flag)
+    {
+        value |= static_cast<Underlying>(flag);
+    }
+
+    constexpr void remove(Enum flag)
+    {
+        value &= ~static_cast<Underlying>(flag);
+    }
+
+    constexpr void set(Enum flag, bool active)
+    {
+        if (active)
+            add(flag);
+        else
+            remove(flag);
+    }
+
+    constexpr void clear()
+    {
+        value = 0;
+    }
+
+    constexpr Underlying raw() const
+    {
+        return value;
+    }
+
+private:
+    Underlying value = 0;
+};

core/opendaq/reader/include/opendaq/multi_reader_impl.h

@@ -15,6 +15,7 @@
  */
 #pragma once
 #include <opendaq/multi_reader.h>
+
 #include <opendaq/read_info.h>
 #include <opendaq/reader_config_ptr.h>
 #include <opendaq/signal_reader.h>
@@ -184,10 +185,12 @@ class MultiReaderImpl : public ImplementationOfWeak<IMultiReader, IReaderConfig,
     Clock::duration timeout{};
     Clock::time_point startTime;
 
+    DomainInfo commonDomain;
     StringPtr readOrigin;
     RatioPtr readResolution;
     RatioPtr tickOffsetTolerance;
-    std::unique_ptr<Comparable> commonStart;
+    // std::unique_ptr<Comparable> commonStart;
+    std::unique_ptr<DomainValue> commonDomainStart;
     std::int64_t requiredCommonSampleRate = -1;
     std::int64_t commonSampleRate = -1;
     std::int32_t sampleRateDividerLcm = 1;

core/opendaq/reader/include/opendaq/queue_reader.h

@@ -0,0 +1,199 @@
+/*
+ * Copyright 2022-2026 openDAQ d.o.o.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+#include <opendaq/data_descriptor_ptr.h>
+#include <opendaq/event_packet_ptr.h>
+#include <opendaq/enum_flags.h>
+#include <opendaq/input_port_config_ptr.h>
+#include <opendaq/logger_component_ptr.h>
+#include <opendaq/sample_type.h>
+#include <opendaq/sample_reader.h>
+#include <opendaq/typed_reading_utils.h>
+
+#include <deque>
+
+BEGIN_NAMESPACE_OPENDAQ
+
+enum class SignalEventType
+{
+    NoChange = 0,
+    ValueChanged,
+    DomainChanged,
+    DomainAndValueChanged,
+    Gap
+};
+
+class SignalEvent
+{
+public:
+    SignalEvent(const EventPacketPtr& packet);
+
+    static SignalEvent syncGapEvent(Int gapDiff);
+
+    bool merge(const SignalEvent& otherEvent);
+    SignalEventType getType() const;
+    const DataDescriptorPtr& getDomainDescriptor() const;
+    const DataDescriptorPtr& getValueDescriptor() const;
+
+    EventPacketPtr toEventPacket() const;
+
+private:
+    explicit SignalEvent(Int gapDiff);
+
+    void updateType();
+private:
+    SignalEventType eventType;
+    DataDescriptorPtr domainDescriptor;
+    DataDescriptorPtr valueDescriptor;
+    Int gapDiff;
+};
+
+
+enum class AdvanceResult
+{
+    Success = 0,
+    NeedMoreData,
+    DomainChanged,
+    OvershotError,
+    Error
+};
+
+enum class QueueReaderIssue : uint32_t
+{
+    None                            = 0,
+    ValueTypesNotConvertible        = 1 << 0,
+    DomainTypesNotConvertible       = 1 << 1,
+    UnsupportedDomainRule           = 1 << 2,
+    OriginParsingFailed             = 1 << 3,
+    DomainUnitInvalid               = 1 << 4
+};
+
+class QueueReader
+{
+public:
+    explicit QueueReader(const InputPortConfigPtr& port, // Consider using Connection instead
+                 SampleType valueReadType,
+                 SampleType domainReadType,
+                 ReadMode mode,
+                 const LoggerComponentPtr& logger,
+                 bool globalIdFromSignal);
+
+public:
+    DomainInfo getDomainInfo();
+    std::unique_ptr<DomainValue> getFirstSampleDomainValue();
+    AdvanceResult advanceToDomainValue(const DomainValue* domainValue);
+    Int getSampleRate();
+
+    void dropOutdatedPacketSegments();
+
+    /**
+     * @brief Get the Available Samples in common rate equivalent
+     * 
+     * @return SizeT Available samples multiplied by the sample rate divider.
+     */
+    SizeT getAvailableSamples();
+
+    bool hasPendingEvents();
+    EventPacketPtr popFrontEvent();
+
+    bool isValid();
+
+    void domainChangeHandled();
+    void updateConnection();
+
+    void setSampleRateDivider(SizeT divider);
+    SizeT getSampleRateDivider() const;
+
+    /**
+     * @brief Read common rate equivalent samples into the buffer. There will be nativeSamples = count / sampleRateDivider
+     * samples read from the packets into the buffer.
+     * 
+     * @param buffer Buffer that has capacity of at least count / sampleRateDivider
+     * @param count Desired sample count in common rate equivalent. 
+     * @return AdvanceResult 
+     */
+    AdvanceResult read(void* valueBuffer, void* domainBuffer, SizeT* count);
+    AdvanceResult skip(SizeT* count);
+
+    /**
+     * @brief Drop the current data segment, if there are fewer than samplesInBlock samples available (common rate equivalent).
+     * 
+     * For example: If there are 3 samples available before next event, samplesInBlock=10 (this is dividerLCM in terms of multireading)
+     * and divider for the queue reader is 2, then 5 native samples are required as minimum aligned read. Since 3 < 5, we drop the 3 samples
+     * and create a synchronization GAP event in the event queue.
+     * 
+     * @param samplesInBlock Number of samples (common rate equivalent).
+     * @return true If samples were dropped.
+     * @return false If samples were not dropped - data segment is long enough or there is no event in the queue to end the segment.
+     */
+    bool dropLeftoverSegment(SizeT samplesInBlock);
+
+private:
+    void drainConnection();
+    void adoptPackets();
+    void consumeLeadingEventPackets();
+
+    SizeT getAvailableSamplesNative();
+    AdvanceResult readNative(void* valueBuffer, void* domainBuffer, SizeT* count);
+
+    void checkConnection() const;
+
+    SignalEventType addEncounteredEvent(const EventPacketPtr& packet);
+    void addToEventQueue(SignalEvent&& event);
+    void parseDomainDescriptor();
+    void parseValueDescriptor();
+    void parseCachedDescriptors();
+    size_t getNumberOfEventPacketsInQueue();
+    bool dropUntilEvent();
+
+private:
+    std::deque<PacketPtr> packets;
+    std::deque<SignalEvent> events;
+
+    SizeT readingPosition = 0;
+
+    InputPortConfigPtr port;
+    ConnectionPtr connection;
+
+    LoggerComponentPtr loggerComponent;
+
+    EnumFlags<QueueReaderIssue> issues;
+
+    ReadMode readMode;
+
+    struct TypedReadingContext
+    {
+        SampleType domainIn;
+        SampleType domainOut;
+        ReadLayout domainLayout;
+        DomainInfo domainInfo;
+        FunctionPtr domainTransform = nullptr;
+
+        SampleType valueIn;
+        SampleType valueOut;
+        ReadLayout valueLayout;
+        FunctionPtr valueTransform = nullptr;
+    };
+    TypedReadingContext typeCtx;
+
+    Int sampleRate = -1;
+    Int packetDelta{0};
+    bool domainChanged = false;
+
+    SizeT sampleRateDivider = 1;
+};
+
+END_NAMESPACE_OPENDAQ

core/opendaq/reader/include/opendaq/reader_utils.h

@@ -21,6 +21,7 @@
 #include <date/date.h>
 #include <chrono>
 #include <ostream>
+#include <optional>
 
 BEGIN_NAMESPACE_OPENDAQ
 
@@ -67,17 +68,36 @@ namespace reader
         if (epoch.find('T') == std::string::npos)
         {
             // If no time, assume Midnight UTC
-            epoch += "T00:00:00+00:00";
+            epoch += "T00:00:00+0000";
         }
-        else if (epoch[epoch.size() - 1] == 'Z')
+        else if (!epoch.empty() && epoch.back() == 'Z')
         {
             // If time-zone marked as "Zulu" (UTC) replace with offset
-            epoch = epoch.erase(epoch.size() - 1) + "+00:00";
+            epoch.pop_back();
+            epoch += "+0000";
         }
-        else if (epoch.find('+') == std::string::npos)
+        else
         {
-            // If not time-zone offset assume UTC
-            epoch += "+00:00";
+            // Search after T to ignore any '-' character in the date portion
+            const auto timeStart = epoch.find('T');
+            const auto plusPos = epoch.find('+', timeStart);
+            const auto minusPos = epoch.find('-', timeStart);
+
+            const auto offsetPos = plusPos != std::string::npos ? plusPos : minusPos;
+
+            if (offsetPos == std::string::npos)
+            {
+                // If not time-zone offset assume UTC
+                epoch += "+0000";
+            }
+            else
+            {
+                const auto colonPos = epoch.find(':', offsetPos);
+                if (colonPos != std::string::npos)
+                {
+                    epoch.erase(colonPos, 1);
+                }
+            }
         }
 
         return epoch;
@@ -94,6 +114,27 @@ namespace reader
 
         return epoch;
     }
+
+    inline std::optional<std::chrono::system_clock::time_point> tryParseEpoch(const std::string& origin)
+    {
+        std::chrono::system_clock::time_point epoch;
+
+        std::istringstream epochString(fixupIso8601(origin));
+        date::from_stream(epochString, "%FT%T%z", epoch);
+
+        if (epochString.fail())
+        {
+            return std::nullopt;
+        }
+
+        epochString >> std::ws; // strip trailing whitespace
+        if (epochString.peek() != std::istringstream::traits_type::eof())
+        {
+            return std::nullopt;
+        }
+
+        return epoch;
+    }
 
     namespace detail
     {