Update DelegateMQ library

Author: endurodave

DelegateMQ/DelegateMQ.h

@@ -88,7 +88,8 @@
     defined(DMQ_THREAD_THREADX) || \
     defined(DMQ_THREAD_ZEPHYR) || \
     defined(DMQ_THREAD_CMSIS_RTOS2) || \
-    defined(DMQ_THREAD_QT)
+    defined(DMQ_THREAD_QT) || \
+    defined(DMQ_THREAD_NONE)
     #include "delegate/MulticastDelegateSafe.h"
     #include "delegate/UnicastDelegateSafe.h"
 #endif
@@ -253,4 +254,5 @@
     #include "extras/databus/DeadlineSubscription.h"
 #endif
 
+
 #endif

DelegateMQ/delegate/DelegateOpt.h

@@ -116,22 +116,22 @@ namespace dmq
 
 #elif defined(DMQ_THREAD_FREERTOS)
     // Use the custom FreeRTOS wrapper
-    using Clock = dmq::FreeRTOSClock;
+    using Clock = dmq::os::FreeRTOSClock;
 
 #elif defined(DMQ_THREAD_THREADX)
     // Use the custom ThreadX wrapper
-    using Clock = dmq::ThreadXClock;
+    using Clock = dmq::os::ThreadXClock;
 
 #elif defined(DMQ_THREAD_ZEPHYR)
     // Use the custom Zephyr wrapper
-    using Clock = dmq::ZephyrClock;
+    using Clock = dmq::os::ZephyrClock;
 
 #elif defined(DMQ_THREAD_CMSIS_RTOS2)
-    using Clock = dmq::CmsisRtos2Clock;
+    using Clock = dmq::os::CmsisRtos2Clock;
 
 #else
     // Assuming implemented the 'g_ticks' variable
-    using Clock = dmq::BareMetalClock;
+    using Clock = dmq::os::BareMetalClock;
 #endif
 
     // --- GENERIC TYPES ---
@@ -151,31 +151,31 @@ namespace dmq
 
 #elif defined(DMQ_THREAD_FREERTOS)
     // Use the custom FreeRTOS wrapper
-    using Mutex = dmq::FreeRTOSMutex;
-    using RecursiveMutex = dmq::FreeRTOSRecursiveMutex;
-    using ConditionVariable = dmq::FreeRTOSConditionVariable;
+    using Mutex = dmq::os::FreeRTOSMutex;
+    using RecursiveMutex = dmq::os::FreeRTOSRecursiveMutex;
+    using ConditionVariable = dmq::os::FreeRTOSConditionVariable;
     template<typename T> using LockGuard = PortableLockGuard<T>;
     template<typename T> using UniqueLock = std::unique_lock<T>;
     #define DMQ_HAS_CV
 
 #elif defined(DMQ_THREAD_THREADX)
     // Use the custom ThreadX wrapper
-    using Mutex = dmq::ThreadXMutex;
-    using RecursiveMutex = dmq::ThreadXRecursiveMutex;
-    using ConditionVariable = dmq::ThreadXConditionVariable;
+    using Mutex = dmq::os::ThreadXMutex;
+    using RecursiveMutex = dmq::os::ThreadXRecursiveMutex;
+    using ConditionVariable = dmq::os::ThreadXConditionVariable;
     template<typename T> using LockGuard = PortableLockGuard<T>;
     template<typename T> using UniqueLock = std::unique_lock<T>;
     #define DMQ_HAS_CV
 
 #elif defined(DMQ_THREAD_ZEPHYR)
     // Use the custom Zephyr wrapper
-    using Mutex = dmq::ZephyrMutex;
-    using RecursiveMutex = dmq::ZephyrRecursiveMutex;
+    using Mutex = dmq::os::ZephyrMutex;
+    using RecursiveMutex = dmq::os::ZephyrRecursiveMutex;
     template<typename T> using LockGuard = PortableLockGuard<T>;
 
 #elif defined(DMQ_THREAD_CMSIS_RTOS2)
-    using Mutex = dmq::CmsisRtos2Mutex;
-    using RecursiveMutex = dmq::CmsisRtos2RecursiveMutex;
+    using Mutex = dmq::os::CmsisRtos2Mutex;
+    using RecursiveMutex = dmq::os::CmsisRtos2RecursiveMutex;
     template<typename T> using LockGuard = PortableLockGuard<T>;
 
 #else
@@ -237,35 +237,37 @@ namespace dmq
     #undef XALLOCATOR
     #define XALLOCATOR
 
-    // Use default std::allocator for dynamic storage allocation
-    template <typename T, typename Alloc = std::allocator<T>>
-    class xlist : public std::list<T, Alloc> {
-    public:
-        using std::list<T, Alloc>::list; // Inherit constructors
-        using std::list<T, Alloc>::operator=;
-    };
-
-    typedef std::basic_ostringstream<char, std::char_traits<char>> xostringstream;
-    typedef std::basic_stringstream<char, std::char_traits<char>> xstringstream;
-
-    typedef std::string xstring;
-
-    // Fallback xmake_shared — uses std::make_shared when fixed-block allocator is disabled
-    template <typename T, typename... Args>
-    inline std::shared_ptr<T> xmake_shared(Args&&... args)
-    {
-        return std::make_shared<T>(std::forward<Args>(args)...);
-    }
-
-    // Fallback xnew/xdelete — use standard new/delete when fixed-block allocator is disabled
-    template<typename T, typename... Args>
-    inline T* xnew(Args&&... args) {
-        return new(std::nothrow) T(std::forward<Args>(args)...);
-    }
-
-    template<typename T>
-    inline void xdelete(T* p) {
-        delete p;
+    namespace dmq {
+        // Use default std::allocator for dynamic storage allocation
+        template <typename T, typename Alloc = std::allocator<T>>
+        class xlist : public std::list<T, Alloc> {
+        public:
+            using std::list<T, Alloc>::list; // Inherit constructors
+            using std::list<T, Alloc>::operator=;
+        };
+
+        typedef std::basic_ostringstream<char, std::char_traits<char>> xostringstream;
+        typedef std::basic_stringstream<char, std::char_traits<char>> xstringstream;
+
+        typedef std::string xstring;
+
+        // Fallback xmake_shared — uses std::make_shared when fixed-block allocator is disabled
+        template <typename T, typename... Args>
+        inline std::shared_ptr<T> xmake_shared(Args&&... args)
+        {
+            return std::make_shared<T>(std::forward<Args>(args)...);
+        }
+
+        // Fallback xnew/xdelete — use standard new/delete when fixed-block allocator is disabled
+        template<typename T, typename... Args>
+        inline T* xnew(Args&&... args) {
+            return new(std::nothrow) T(std::forward<Args>(args)...);
+        }
+
+        template<typename T>
+        inline void xdelete(T* p) {
+            delete p;
+        }
     }
 #endif
 

DelegateMQ/extras/allocator/xlist.h

@@ -4,9 +4,11 @@
 #include "stl_allocator.h"
 #include <list>
 
-// xlist uses a fix-block memory allocator
-template <typename T, typename Alloc = stl_allocator<T>>
-using xlist = std::list<T, Alloc>;
+namespace dmq {
+    // xlist uses a fix-block memory allocator
+    template <typename T, typename Alloc = stl_allocator<T>>
+    using xlist = std::list<T, Alloc>;
+}
 
 #if 0  // Deprecated
 template<class _Ty,

DelegateMQ/extras/allocator/xnew.h

@@ -18,24 +18,26 @@
 
 #include "xallocator.h"
 
-/// @brief Allocate and construct T in the fixed-block pool.
-/// @return Pointer to the constructed object, or nullptr if xmalloc fails.
-template<typename T, typename... Args>
-inline T* xnew(Args&&... args)
-{
-    void* mem = xmalloc(sizeof(T));
-    if (!mem) return nullptr;
-    return ::new(mem) T(std::forward<Args>(args)...);
-}
+namespace dmq {
+    /// @brief Allocate and construct T in the fixed-block pool.
+    /// @return Pointer to the constructed object, or nullptr if xmalloc fails.
+    template<typename T, typename... Args>
+    inline T* xnew(Args&&... args)
+    {
+        void* mem = xmalloc(sizeof(T));
+        if (!mem) return nullptr;
+        return ::new(mem) T(std::forward<Args>(args)...);
+    }
 
-/// @brief Destroy T and return its memory to the fixed-block pool.
-template<typename T>
-inline void xdelete(T* p)
-{
-    if (p)
+    /// @brief Destroy T and return its memory to the fixed-block pool.
+    template<typename T>
+    inline void xdelete(T* p)
     {
-        p->~T();
-        xfree(const_cast<void*>(static_cast<const void*>(p)));
+        if (p)
+        {
+            p->~T();
+            xfree(const_cast<void*>(static_cast<const void*>(p)));
+        }
     }
 }
 

DelegateMQ/extras/allocator/xsstream.h

@@ -4,8 +4,10 @@
 #include "stl_allocator.h"
 #include <sstream>
 
-typedef std::basic_stringstream<char, std::char_traits<char>, stl_allocator<char> > xstringstream;
-typedef std::basic_ostringstream<char, std::char_traits<char>, stl_allocator<char> > xostringstream;
+namespace dmq {
+    typedef std::basic_stringstream<char, std::char_traits<char>, stl_allocator<char> > xstringstream;
+    typedef std::basic_ostringstream<char, std::char_traits<char>, stl_allocator<char> > xostringstream;
+}
 // xwstringstream/xwostringstream are unused by DelegateMQ. Uncomment if wide string support is needed by the application.
 // typedef std::basic_stringstream<wchar_t, std::char_traits<wchar_t>, stl_allocator<wchar_t> > xwstringstream;
 // typedef std::basic_ostringstream<wchar_t, std::char_traits<wchar_t>, stl_allocator<wchar_t> > xwostringstream;

DelegateMQ/extras/allocator/xstring.h

@@ -4,7 +4,9 @@
 #include "stl_allocator.h"
 #include <string>
 
-typedef std::basic_string<char, std::char_traits<char>, stl_allocator<char> > xstring;
+namespace dmq {
+    typedef std::basic_string<char, std::char_traits<char>, stl_allocator<char> > xstring;
+}
 // xwstring is unused by DelegateMQ. Uncomment if wide string support is needed by the application.
 // typedef std::basic_string<wchar_t, std::char_traits<wchar_t>, stl_allocator<wchar_t> > xwstring;
 

DelegateMQ/extras/databus/DataBus.h

@@ -1,7 +1,11 @@
 #ifndef DMQ_DATABUS_H
 #define DMQ_DATABUS_H
 
-#include "DelegateMQ.h"
+#include "delegate/Signal.h"
+#include "delegate/DelegateRemote.h"
+#include "delegate/DelegateAsync.h"
+#include "delegate/IThread.h"
+#include "delegate/DelegateOpt.h"
 #include "Participant.h"
 #include "DataBusQos.h"
 #include "SpyPacket.h"
@@ -17,7 +21,7 @@
 #include <typeindex>
 #include <atomic>
 
-namespace dmq {
+namespace dmq::databus {
 
 // The DataBus is a central registry for topic-based communication.
 // It allows components to publish and subscribe to data topics identified by strings.
@@ -81,7 +85,13 @@ class DataBus {
         GetInstance().InternalRegisterStringifier<T>(topic, std::move(func));
     }
 
+    // Enable/Disable Last Value Cache (LVC) for a topic.
+    static void LastValueCache(const std::string& topic, bool enabled) {
+        GetInstance().InternalLastValueCache(topic, enabled);
+    }
+
     // Subscribe to all bus traffic (topic and stringified value).
+    // Monitor all traffic on the DataBus.
     // NOTE: priority is only applied when thread != nullptr; passing a non-default
     // priority without a thread is a programming error and triggers FaultHandler.
     static dmq::ScopedConnection Monitor(std::function<void(const SpyPacket&)> func, dmq::IThread* thread = nullptr, dmq::Priority priority = dmq::Priority::NORMAL) {
@@ -99,12 +109,22 @@ class DataBus {
         return instance.m_monitorSignal.Connect(dmq::MakeDelegate(std::move(func)));
     }
 
+    /// Fired when a message is published but has no local or remote subscribers.
+    static dmq::ScopedConnection SubscribeUnhandled(std::function<void(const std::string& topic)> func) {
+        return GetInstance().m_unhandledSignal.Connect(dmq::MakeDelegate(std::move(func)));
+    }
+
     // Reset the DataBus (mostly for testing).
     static void ResetForTesting() {
         GetInstance().InternalReset();
     }
 
 private:
+    void InternalLastValueCache(const std::string& topic, bool enabled) {
+        std::lock_guard<dmq::RecursiveMutex> lock(m_mutex);
+        m_topicQos[topic].lastValueCache = enabled;
+    }
+
     DataBus() = default;
     ~DataBus() = default;
 
@@ -225,7 +245,7 @@ class DataBus {
             // Must be first — before any writes — so a mismatch never corrupts LVC.
             auto itType = m_typeIndices.find(topic);
             if (itType != m_typeIndices.end() && itType->second != std::type_index(typeid(T))) {
-                ::FaultHandler(__FILE__, (unsigned short)__LINE__);
+                ::dmq::util::FaultHandler(__FILE__, (unsigned short)__LINE__);
                 return;
             }
 
@@ -271,16 +291,24 @@ class DataBus {
         }
 
         // 7. Local distribution
+        bool handled = false;
         if (signal) {
             (*signal)(data);
+            handled = true;
         }
 
         // 8. Remote distribution using the snapshot
         if (serializer) {
             for (auto& participant : participantsSnapshot) {
                 participant->Send<T>(topic, data, *serializer);
+                handled = true;
             }
         }
+
+        // 9. Notify if no one received the message
+        if (!handled) {
+            m_unhandledSignal(topic);
+        }
     }
 
     void InternalAddParticipant(std::shared_ptr<Participant> participant) {
@@ -295,7 +323,7 @@ class DataBus {
         auto itType = m_typeIndices.find(topic);
         if (itType != m_typeIndices.end()) {
             if (itType->second != std::type_index(typeid(T))) {
-                ::FaultHandler(__FILE__, (unsigned short)__LINE__);
+                ::dmq::util::FaultHandler(__FILE__, (unsigned short)__LINE__);
                 return;
             }
         } else {
@@ -314,7 +342,7 @@ class DataBus {
         auto itType = m_typeIndices.find(topic);
         if (itType != m_typeIndices.end()) {
             if (itType->second != std::type_index(typeid(T))) {
-                ::FaultHandler(__FILE__, (unsigned short)__LINE__);
+                ::dmq::util::FaultHandler(__FILE__, (unsigned short)__LINE__);
                 return;
             }
         } else {
@@ -347,7 +375,7 @@ class DataBus {
         if (itType != m_typeIndices.end()) {
             if (itType->second != std::type_index(typeid(T))) {
                 // Runtime Type Safety: Catch same topic string used with different types
-                ::FaultHandler(__FILE__, (unsigned short)__LINE__);
+                ::dmq::util::FaultHandler(__FILE__, (unsigned short)__LINE__);
                 return nullptr; 
             }
         } else {
@@ -378,9 +406,11 @@ class DataBus {
     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;
+    dmq::Signal<void(const std::string& topic)> m_unhandledSignal;
 };
 
-} // namespace dmq
+} // namespace dmq::databus
+
 
 #endif // DMQ_DATABUS_H
 

DelegateMQ/extras/databus/DataBusQos.h

@@ -1,10 +1,10 @@
 #ifndef DMQ_DATABUSQOS_H
 #define DMQ_DATABUSQOS_H
 
-#include "DelegateMQ.h"
+#include "delegate/DelegateOpt.h"
 #include <optional>
 
-namespace dmq {
+namespace dmq::databus {
 
 // Quality of Service settings for a topic subscription.
 struct QoS {
@@ -22,6 +22,7 @@ struct QoS {
     std::optional<dmq::Duration> minSeparation;
 };
 
-} // namespace dmq
+} // namespace dmq::databus
+
 
 #endif // DMQ_DATABUSQOS_H