Updated DelegateMQ library

Author: endurodave

DelegateMQ/DelegateMQ.h

@@ -250,6 +250,7 @@
 #if defined(DMQ_DATABUS)
     #include "extras/databus/DataBus.h"
     #include "extras/databus/Participant.h"
+    #include "extras/databus/DeadlineSubscription.h"
 #endif
 
 #endif

DelegateMQ/delegate/DelegateOpt.h

@@ -225,6 +225,7 @@ namespace dmq
     #include "extras/allocator/xsstream.h"
     #include "extras/allocator/stl_allocator.h"
     #include "extras/allocator/xnew.h"
+    #include "extras/allocator/xmake_shared.h"
 #else
     #include <string>
     #include <list>

DelegateMQ/extras/allocator/README.md

@@ -0,0 +1,64 @@
+# Allocator Suite
+
+A robust, fixed-block memory allocator suite for C++ applications, designed for performance and deterministic behavior in resource-constrained or real-time systems.
+
+## Features
+
+- **Fixed-Block Allocation**: Prevents heap fragmentation by using pre-allocated pools of equal-sized blocks.
+- **C++ Class Integration**: `DECLARE_ALLOCATOR` and `IMPLEMENT_ALLOCATOR` macros provide easy integration with custom classes.
+- **XALLOCATOR Macro**: Overloads `operator new` and `operator delete` for seamless use.
+- **STL Compatibility**: `stl_allocator<T>` allows STL containers (`std::vector`, `std::list`, `std::map`, etc.) to use the fixed-block pool.
+- **Smart Pointer Support**: `xmake_shared<T>` creates `std::shared_ptr` with both the object and control block allocated from the pool.
+- **C API**: `xmalloc`, `xfree`, and `xrealloc` functions for C-style memory management.
+
+## Components
+
+- **Allocator**: The core engine for fixed-block management.
+- **AllocatorPool**: A template-based pool for specific types.
+- **stl_allocator**: STL-compliant allocator wrapper.
+- **xmake_shared**: Helper for pool-allocated shared pointers.
+- **xnew / xdelete**: Helper templates for placement new/destroy in the pool.
+
+## Basic Usage
+
+### Using XALLOCATOR in a class
+
+```cpp
+class MyClass {
+    XALLOCATOR
+public:
+    int data;
+};
+
+// Now 'new MyClass' uses the fixed-block allocator
+MyClass* obj = new MyClass();
+delete obj;
+```
+
+### Using STL containers
+
+```cpp
+#include "xlist.h"
+
+// xlist is a typedef for std::list using stl_allocator
+xlist<int> myList;
+myList.push_back(10);
+```
+
+### Using xmake_shared
+
+```cpp
+auto sp = xmake_shared<MyClass>(args...);
+```
+
+## Memory Alignment
+
+The `xallocator` implementation ensures that all allocated blocks are correctly aligned for the target architecture:
+
+- **Power-of-Two Rounding**: All requested block sizes are rounded up to the next power of two (8, 16, 32, 64, etc.). This inherently satisfies the natural alignment requirements of all standard C++ types (`int`, `double`, `long long`, etc.) and modern SIMD/AVX vector instructions.
+- **Header Alignment**: A `BLOCK_HEADER_SIZE` (16 bytes on 64-bit, 8 bytes on 32-bit) is added to each block to store metadata. Both the raw block and the user's data pointer remain aligned on 8 or 16-byte boundaries.
+- **Efficiency vs. Safety**: This strategy prioritizes "automatic" alignment and safety over maximum memory density. While it may result in some internal fragmentation, it ensures that memory is always safe for any data type without manual alignment configuration.
+
+## Integration with DelegateMQ
+
+To enable the allocator suite within DelegateMQ, define `DMQ_ALLOCATOR` in your build configuration. This will cause DelegateMQ's internal containers and remote delegate marshalling to use the fixed-block pools instead of the standard heap.

DelegateMQ/extras/allocator/stl_allocator.h

@@ -69,11 +69,4 @@ inline bool operator==(const stl_allocator<T>&, const stl_allocator<U>&) { retur
 template <typename T, typename U>
 inline bool operator!=(const stl_allocator<T>&, const stl_allocator<U>&) { return false; }
 
-// Create a shared_ptr with both the object and control block allocated from the fixed-block pool
-template <typename T, typename... Args>
-inline std::shared_ptr<T> xmake_shared(Args&&... args)
-{
-    return std::allocate_shared<T>(stl_allocator<T>(), std::forward<Args>(args)...);
-}
-
 #endif 

DelegateMQ/extras/allocator/xmake_shared.h

@@ -0,0 +1,16 @@
+#ifndef _XMAKE_SHARED_H
+#define _XMAKE_SHARED_H
+
+#include "stl_allocator.h"
+#include <memory>
+#include <utility>
+
+/// @brief Create a shared_ptr with both the object and control block allocated 
+///        from the fixed-block pool.
+template <typename T, typename... Args>
+inline std::shared_ptr<T> xmake_shared(Args&&... args)
+{
+    return std::allocate_shared<T>(stl_allocator<T>(), std::forward<Args>(args)...);
+}
+
+#endif // _XMAKE_SHARED_H

DelegateMQ/extras/databus/DataBus.h

@@ -15,6 +15,7 @@
 #include <vector>
 #include <functional>
 #include <typeindex>
+#include <atomic>
 
 namespace dmq {
 
@@ -121,16 +122,34 @@ class DataBus {
     template <typename T, typename F>
     dmq::ScopedConnection InternalSubscribe(const std::string& topic, F&& func, dmq::IThread* thread, QoS qos) {
         SignalPtr<T> signal;
-        
+
         // Performance Note: Forced std::function construction for internal type management.
         std::function<void(T)> typedFunc = std::forward<F>(func);
+
+        // Wrap with min separation rate limiter if requested. Each subscriber gets its
+        // own independent last-delivery timestamp, so different subscribers on the same
+        // topic can have different (or no) rate limits without affecting each other.
+        if (qos.minSeparation.has_value()) {
+            auto minSepRep = qos.minSeparation.value().count();
+            auto lastDeliveryRep = std::make_shared<std::atomic<int64_t>>(0);
+            auto inner = std::move(typedFunc);
+            typedFunc = [inner = std::move(inner), minSepRep, lastDeliveryRep](T data) {
+                auto nowRep = static_cast<int64_t>(dmq::Clock::now().time_since_epoch().count());
+                auto lastRep = lastDeliveryRep->load(std::memory_order_relaxed);
+                if (nowRep - lastRep >= minSepRep) {
+                    lastDeliveryRep->store(nowRep, std::memory_order_relaxed);
+                    inner(data);
+                }
+            };
+        }
+
         T* cachedValPtr = nullptr;
         T cachedVal;
         dmq::ScopedConnection conn;
 
         {
             std::lock_guard<dmq::RecursiveMutex> lock(m_mutex);
-            
+
             // 1. Enable LVC if requested (persists for topic lifetime until ResetForTesting)
             if (qos.lastValueCache) {
                 m_topicQos[topic].lastValueCache = true;
@@ -154,17 +173,25 @@ class DataBus {
             if (qos.lastValueCache) {
                 auto it = m_lastValues.find(topic);
                 if (it != m_lastValues.end()) {
-                    cachedVal = *std::static_pointer_cast<T>(it->second);
-                    cachedValPtr = &cachedVal;
+                    // Check lifespan: skip delivery if the cached value is too old
+                    bool expired = false;
+                    if (qos.lifespan.has_value()) {
+                        auto age = dmq::Clock::now() - it->second.timestamp;
+                        expired = (age > qos.lifespan.value());
+                    }
+                    if (!expired) {
+                        cachedVal = *std::static_pointer_cast<T>(it->second.value);
+                        cachedValPtr = &cachedVal;
+                    }
                 }
             }
         }
 
-        // 5. Dispatch LVC outside the lock to prevent deadlocks. 
+        // 5. Dispatch LVC outside the lock to prevent deadlocks.
         // IMPORTANT: Because this happens after releasing the lock, a high-frequency
         // publisher on another thread could have already sent a new value to the
         // connected signal. The subscriber might receive the fresh value FIRST,
-        // followed by this stale LVC value. Users of LVC should ensure their 
+        // followed by this stale LVC value. Users of LVC should ensure their
         // logic handles potential out-of-order state arrival.
         if (cachedValPtr) {
             if (thread) {
@@ -207,7 +234,7 @@ class DataBus {
             // by any subscriber, it remains active for that topic until ResetForTesting().
             auto itQos = m_topicQos.find(topic);
             if (itQos != m_topicQos.end() && itQos->second.lastValueCache) {
-                m_lastValues[topic] = std::make_shared<T>(data);
+                m_lastValues[topic] = LvcEntry{ std::make_shared<T>(data), now };
             }
 
             // 3. Prepare monitor data
@@ -337,12 +364,17 @@ class DataBus {
         return signal;
     }
 
+    struct LvcEntry {
+        std::shared_ptr<void> value;
+        dmq::TimePoint timestamp;
+    };
+
     dmq::RecursiveMutex m_mutex;
     std::unordered_map<std::string, std::shared_ptr<void>> m_signals;
     std::unordered_map<std::string, std::type_index> m_typeIndices;
     std::vector<std::shared_ptr<Participant>> m_participants;
     std::unordered_map<std::string, std::shared_ptr<void>> m_serializers;
-    std::unordered_map<std::string, std::shared_ptr<void>> m_lastValues;
+    std::unordered_map<std::string, LvcEntry> m_lastValues;
     std::unordered_map<std::string, QoS> m_topicQos;
     std::unordered_map<std::string, std::shared_ptr<void>> m_stringifiers;
     dmq::Signal<void(const SpyPacket&)> m_monitorSignal;

DelegateMQ/extras/databus/DataBusQos.h

@@ -1,11 +1,25 @@
 #ifndef DMQ_DATABUSQOS_H
 #define DMQ_DATABUSQOS_H
 
+#include "DelegateMQ.h"
+#include <optional>
+
 namespace dmq {
 
-// Quality of Service settings for a topic.
+// Quality of Service settings for a topic subscription.
 struct QoS {
-    bool lastValueCache = false; // If true, new subscribers receive the last published value.
+    // If true, new subscribers receive the last published value immediately on subscribe.
+    bool lastValueCache = false;
+
+    // Maximum age of a cached LVC value. If the cached value is older than this duration
+    // when a new subscriber connects, it is considered stale and not delivered.
+    // Only meaningful when lastValueCache = true.
+    std::optional<dmq::Duration> lifespan;
+
+    // Minimum time between deliveries to this subscriber. Publishes that arrive faster
+    // than this interval are silently dropped for this subscriber only. Other subscribers
+    // with a different (or no) minSeparation are unaffected.
+    std::optional<dmq::Duration> minSeparation;
 };
 
 } // namespace dmq

DelegateMQ/extras/databus/DeadlineSubscription.h

@@ -0,0 +1,122 @@
+#ifndef DMQ_DEADLINE_SUBSCRIPTION_H
+#define DMQ_DEADLINE_SUBSCRIPTION_H
+
+/// @file DeadlineSubscription.h
+/// @see https://github.com/DelegateMQ/DelegateMQ
+/// David Lafreniere, 2025.
+///
+/// @brief RAII helper that combines a DataBus subscription with a deadline timer.
+///
+/// @details
+/// `DeadlineSubscription<T>` monitors a DataBus topic and fires a user callback
+/// if no message arrives within a configurable deadline window. It is built
+/// entirely from existing DelegateMQ primitives — `DataBus::Subscribe`, `Timer`,
+/// and `ScopedConnection` — and adds no library-internal mechanism.
+///
+/// **How it works:**
+/// A `Timer` is started at construction time. Every incoming message resets the
+/// timer. If the timer fires before the next message arrives, the `onMissed`
+/// callback is invoked. Both the data handler and the deadline callback are
+/// dispatched to the same optional worker thread.
+///
+/// **Lifetime:**
+/// The object is non-copyable and non-movable. All resources — the DataBus
+/// connection, the timer expiry connection, and the timer itself — are released
+/// automatically when the object is destroyed. Destruction is always clean:
+/// members are destroyed in reverse declaration order, so the DataBus connection
+/// disconnects before the timer is torn down.
+///
+/// **`Timer::ProcessTimers()` requirement:**
+/// The deadline timer fires only when `Timer::ProcessTimers()` is called. On
+/// platforms with a running `Thread`, this is typically driven by the thread's
+/// internal timer. On bare-metal targets, call `ProcessTimers()` from the main
+/// super-loop or a SysTick handler. If `ProcessTimers()` is not called, the
+/// deadline callback silently never fires.
+///
+/// **`onMissed` callback context:**
+/// - With a `thread` argument: the callback is dispatched asynchronously to
+///   that thread, matching the delivery context of the data handler.
+/// - Without a `thread` argument: the callback fires synchronously on whatever
+///   thread calls `Timer::ProcessTimers()`. On bare-metal this may be an ISR —
+///   keep the callback short and non-blocking.
+///
+/// **Usage:**
+/// @code
+/// dmq::DeadlineSubscription<SensorData> m_watch{
+///     "sensor/temp",
+///     std::chrono::milliseconds(500),
+///     [](const SensorData& d) { /* handle data */ },
+///     []()                    { /* deadline missed — sensor silent */ },
+///     &m_workerThread
+/// };
+/// @endcode
+
+#include "DelegateMQ.h"
+#include "extras/databus/DataBus.h"
+#include "extras/util/Timer.h"
+#include <functional>
+#include <string>
+
+namespace dmq {
+
+template <typename T>
+class DeadlineSubscription {
+public:
+    /// Construct a deadline-monitored DataBus subscription.
+    ///
+    /// @param topic     DataBus topic to subscribe to.
+    /// @param deadline  Maximum allowed interval between deliveries. Must be > 0.
+    /// @param handler   Called on each data delivery.
+    /// @param onMissed  Called when no delivery arrives within the deadline window.
+    /// @param thread    Optional worker thread for both callbacks. If nullptr,
+    ///                  handler fires on the publisher's thread and onMissed fires
+    ///                  on the Timer::ProcessTimers() thread.
+    DeadlineSubscription(
+        const std::string& topic,
+        dmq::Duration deadline,
+        std::function<void(const T&)> handler,
+        std::function<void()> onMissed,
+        dmq::IThread* thread = nullptr)
+        : m_deadline(deadline)
+    {
+        // Connect onMissed to the timer expiry signal, dispatching to thread if provided
+        if (thread) {
+            m_timerConn = m_timer.OnExpired.Connect(
+                MakeDelegate(std::move(onMissed), *thread));
+        } else {
+            m_timerConn = m_timer.OnExpired.Connect(
+                MakeDelegate(std::move(onMissed)));
+        }
+
+        // Arm the timer immediately. It fires if no delivery arrives within deadline.
+        m_timer.Start(m_deadline, false);
+
+        // Subscribe and reset the timer on every delivery
+        m_conn = DataBus::Subscribe<T>(topic,
+            [this, h = std::move(handler)](const T& data) {
+                m_timer.Start(m_deadline, false); // reset deadline window
+                h(data);
+            }, thread);
+    }
+
+    ~DeadlineSubscription() = default;
+
+    DeadlineSubscription(const DeadlineSubscription&) = delete;
+    DeadlineSubscription& operator=(const DeadlineSubscription&) = delete;
+    DeadlineSubscription(DeadlineSubscription&&) = delete;
+    DeadlineSubscription& operator=(DeadlineSubscription&&) = delete;
+
+private:
+    // Declaration order controls destruction order (reverse).
+    // m_conn disconnects first (no more timer resets via the data lambda),
+    // then m_timerConn disconnects (onMissed removed from timer signal),
+    // then m_timer destructs (removed from global timer list).
+    dmq::Duration m_deadline;
+    Timer m_timer;
+    dmq::ScopedConnection m_timerConn;
+    dmq::ScopedConnection m_conn;
+};
+
+} // namespace dmq
+
+#endif // DMQ_DEADLINE_SUBSCRIPTION_H