Update DelegateMQ library

Author: endurodave

DelegateMQ/DelegateMQ.h

@@ -179,13 +179,20 @@
 #elif defined(DMQ_TRANSPORT_ARM_LWIP_UDP)
     #include "predef/dispatcher/Dispatcher.h"
     #include "predef/transport/arm-lwip-udp/ArmLwipUdpTransport.h"
+#elif defined(DMQ_TRANSPORT_THREADX_UDP)
+    #include "predef/dispatcher/Dispatcher.h"
+    #include "predef/transport/threadx-udp/NetXUdpTransport.h"
+#elif defined(DMQ_TRANSPORT_ZEPHYR_UDP)
+    #include "predef/dispatcher/Dispatcher.h"
+    #include "predef/transport/zephyr-udp/ZephyrUdpTransport.h"
 #elif defined(DMQ_TRANSPORT_NONE)
     // Create a custom application-specific transport
 #else
     #error "Transport implementation not found."
 #endif
 
-#if defined(DMQ_TRANSPORT_ZEROMQ) || defined(DMQ_TRANSPORT_WIN32_UDP) || defined(DMQ_TRANSPORT_LINUX_UDP)
+#if defined(DMQ_TRANSPORT_ZEROMQ) || defined(DMQ_TRANSPORT_WIN32_UDP) || defined(DMQ_TRANSPORT_LINUX_UDP) || \
+    defined(DMQ_TRANSPORT_THREADX_UDP) || defined(DMQ_TRANSPORT_ZEPHYR_UDP)
     #include "predef/util/NetworkEngine.h"
 #endif
 
@@ -198,4 +205,4 @@
     #include "predef/util/TransportMonitor.h"
 #endif
 
-#endif
+#endif

DelegateMQ/External.cmake

@@ -117,7 +117,7 @@ if(DMQ_TRANSPORT STREQUAL "DMQ_TRANSPORT_ARM_LWIP_UDP")
     set_and_check(LWIP_INCLUDE_DIR_1 "${LWIP_ROOT}/src/include")
     set_and_check(LWIP_INCLUDE_DIR_2 "${LWIP_ROOT}/src/include/ipv4")
 
-    # Often required for lwipopts.h or arch/cc.h. 
+    # Often required for lwipopts.h or arch/cc.h.
     # NOTE: You may need to customize this depending on where your project keeps 'lwipopts.h'
     # For now, we point to the generic 'contrib' ports if present, or just the root.
     set(LWIP_INCLUDE_DIR_3 "${LWIP_ROOT}/contrib/ports/unix/port/include")
@@ -135,6 +135,25 @@ if(DMQ_TRANSPORT STREQUAL "DMQ_TRANSPORT_ARM_LWIP_UDP")
         "${LWIP_ROOT}/src/netif/*.c"
     )
 endif()
+
+# ---------------------------------------------------------------------------
+# NetX / NetX Duo library (For ThreadX)
+# ---------------------------------------------------------------------------
+if(DMQ_TRANSPORT STREQUAL "DMQ_TRANSPORT_THREADX_UDP")
+    # Adjust this path to your NetX / NetX Duo root
+    set(NETX_ROOT "${DMQ_ROOT_DIR}/../../../netxduo")
+    
+    # Verify root
+    set_and_check(NETX_INCLUDE_DIR "${NETX_ROOT}/common/inc")
+    
+    # Collect common sources
+    file(GLOB NETX_SOURCES 
+        "${NETX_ROOT}/common/src/*.c"
+        "${NETX_ROOT}/common/inc/*.h"
+    )
+    
+    # Note: Architecture-specific ports usually handled by your main ThreadX build system
+endif()
 
 # ---------------------------------------------------------------------------
 # MessagePack
@@ -180,6 +199,7 @@ if(DMQ_THREAD STREQUAL "DMQ_THREAD_FREERTOS")
         "${FREERTOS_ROOT_DIR}/FreeRTOS/Source/*.c"
         "${FREERTOS_ROOT_DIR}/FreeRTOS/Source/include/*.h"
     )
+    
     list(APPEND FREERTOS_SOURCES 
         "${FREERTOS_ROOT_DIR}/FreeRTOS-Plus/Source/FreeRTOS-Plus-Trace/trcKernelPort.c"
         "${FREERTOS_ROOT_DIR}/FreeRTOS-Plus/Source/FreeRTOS-Plus-Trace/trcSnapshotRecorder.c"
@@ -212,11 +232,10 @@ endif()
 # Zephyr
 # ---------------------------------------------------------------------------
 if(DMQ_THREAD STREQUAL "DMQ_THREAD_ZEPHYR")
-    # Zephyr is a build system, not just a library. 
+    # Zephyr is a build system, not just a library.
     # We typically do NOT manually glob source files here.
     # The application's main CMakeLists.txt must call `find_package(Zephyr)` 
     # which sets up the include paths and kernel linking automatically.
-
     # Optional: Just verify the root directory exists if you want to be safe
     set_and_check(ZEPHYR_ROOT_DIR "${DMQ_ROOT_DIR}/../../../zephyr")
 

DelegateMQ/Predef.cmake

@@ -89,6 +89,12 @@ elseif (DMQ_TRANSPORT STREQUAL "DMQ_TRANSPORT_SERIAL_PORT")
 elseif (DMQ_TRANSPORT STREQUAL "DMQ_TRANSPORT_ARM_LWIP_UDP")
     add_compile_definitions(DMQ_TRANSPORT_ARM_LWIP_UDP)
     file(GLOB TRANSPORT_SOURCES "${DMQ_ROOT_DIR}/predef/transport/arm-lwip-udp/*.h")
+elseif (DMQ_TRANSPORT STREQUAL "DMQ_TRANSPORT_THREADX_UDP")
+    add_compile_definitions(DMQ_TRANSPORT_THREADX_UDP)
+    file(GLOB TRANSPORT_SOURCES "${DMQ_ROOT_DIR}/predef/transport/threadx-udp/*.h")
+elseif (DMQ_TRANSPORT STREQUAL "DMQ_TRANSPORT_ZEPHYR_UDP")
+    add_compile_definitions(DMQ_TRANSPORT_ZEPHYR_UDP)
+    file(GLOB TRANSPORT_SOURCES "${DMQ_ROOT_DIR}/predef/transport/zephyr-udp/*.h")
 else()
     message(FATAL_ERROR "Must set DMQ_TRANSPORT option.")
 endif()
@@ -144,4 +150,4 @@ list(APPEND DMQ_PREDEF_SOURCES ${UTIL_SOURCES})
 file(GLOB DMQ_LIB_SOURCES
     "${DMQ_ROOT_DIR}/*.h"
     "${DMQ_ROOT_DIR}/delegate/*.h"
-)
+)

DelegateMQ/delegate/DelegateOpt.h

@@ -13,6 +13,9 @@
 #elif defined(DMQ_THREAD_THREADX)
     #include "predef/util/ThreadXClock.h"
     #include "predef/util/ThreadXMutex.h"
+#elif defined(DMQ_THREAD_ZEPHYR)
+    #include "predef/util/ZephyrClock.h"
+    #include "predef/util/ZephyrMutex.h"
 #elif defined(DMQ_THREAD_CMSIS_RTOS2)
     #include "predef/util/CmsisRtos2Clock.h"
     #include "predef/util/CmsisRtos2Mutex.h"
@@ -146,4 +149,4 @@ namespace dmq
     #define LOG_ERROR(...)   do {} while(0)
 #endif
 
-#endif
+#endif // _DELEGATE_OPT_H

DelegateMQ/predef/os/freertos/Thread.cpp

@@ -22,6 +22,11 @@ Thread::Thread(const std::string& threadName) : THREAD_NAME(threadName)
 Thread::~Thread()
 {
     ExitThread();
+    // Safe to delete semaphore now that the thread is definitely gone
+    if (m_exitSem) {
+        vSemaphoreDelete(m_exitSem);
+        m_exitSem = nullptr;
+    }
 }
 
 //----------------------------------------------------------------------------
@@ -31,12 +36,16 @@ bool Thread::CreateThread()
 {
     if (!m_thread)
     {
-        // 1. Create the Queue first
+        // 1. Create Exit Synchronization Semaphore
+        m_exitSem = xSemaphoreCreateBinary();
+        ASSERT_TRUE(m_exitSem != nullptr);
+
+        // 2. Create the Queue
         // Holds pointers to ThreadMsg objects (heap allocated)
         m_queue = xQueueCreate(20, sizeof(ThreadMsg*));
         ASSERT_TRUE(m_queue != nullptr);
 
-        // 2. Create the Task
+        // 3. Create the Task
         BaseType_t xReturn = xTaskCreate(
             (TaskFunction_t)&Thread::Process,
             THREAD_NAME.c_str(),
@@ -58,12 +67,26 @@ void Thread::ExitThread()
     if (m_queue) {
         // Send exit message
         ThreadMsg* msg = new ThreadMsg(MSG_EXIT_THREAD);
-        if (xQueueSend(m_queue, &msg, 100) != pdPASS) {
+        // Wait 100ms to send
+        if (xQueueSend(m_queue, &msg, pdMS_TO_TICKS(100)) != pdPASS) {
             delete msg; // Failed to send, clean up
         }
-        // Note: We don't join/wait here because FreeRTOS tasks 
-        // delete themselves asynchronously.
-        m_thread = nullptr; // Detach
+
+        // Wait for the thread to actually finish to avoid Use-After-Free.
+        // We only wait if we are NOT the thread itself (prevent deadlock).
+        if (xTaskGetCurrentTaskHandle() != m_thread && m_exitSem) {
+            xSemaphoreTake(m_exitSem, portMAX_DELAY);
+        }
+
+        // Now safe to clean up resources
+        // Delete Queue
+        if (m_queue) {
+            vQueueDelete(m_queue);
+            m_queue = nullptr;
+        }
+
+        // Note: m_thread handle is invalid after the task deletes itself
+        m_thread = nullptr;
     }
 }
 
@@ -99,7 +122,7 @@ void Thread::DispatchDelegate(std::shared_ptr<dmq::DelegateMsg> msg)
     {
         // 3. Handle failure: Delete to prevent memory leak
         delete threadMsg;
-        printf("Error: Thread '%s' queue full! Delegate dropped.\n", THREAD_NAME.c_str());
+        // printf("Error: Thread '%s' queue full! Delegate dropped.\n", THREAD_NAME.c_str());
     }
 }
 
@@ -110,6 +133,7 @@ void Thread::Process(void* instance)
 {
     Thread* thread = static_cast<Thread*>(instance);
     ASSERT_TRUE(thread != nullptr);
+
     thread->Run();
 
     // Self-delete when Run() returns (ExitThread called)
@@ -146,6 +170,10 @@ void Thread::Run()
             case MSG_EXIT_THREAD:
             {
                 delete msg;
+                // Signal ExitThread() that we are done
+                if (m_exitSem) {
+                    xSemaphoreGive(m_exitSem);
+                }
                 return; // Breaks loop, Process() calls vTaskDelete
             }
 

DelegateMQ/predef/os/freertos/Thread.h

@@ -26,6 +26,7 @@
 #include "FreeRTOS.h"
 #include "task.h"
 #include "queue.h"
+#include "semphr.h"
 #include <string>
 #include <memory>
 
@@ -71,6 +72,8 @@ class Thread : public dmq::IThread
 
     TaskHandle_t m_thread = nullptr;
     QueueHandle_t m_queue = nullptr;
+    SemaphoreHandle_t m_exitSem = nullptr; // Synchronization for safe destruction
+
     const std::string THREAD_NAME;
 };
 

DelegateMQ/predef/serialize/bitsery/Serializer.h

@@ -9,34 +9,50 @@
 /// to a remote. Bitsery provides fast, compact binary serialization.
 
 #include "delegate/ISerializer.h"
+
+// Core Bitsery
 #include <bitsery/bitsery.h>
 #include <bitsery/adapter/stream.h>
-#include <bitsery/ext/std_tuple.h>
+
+// Common Traits (Include these so standard types work out of the box)
+#include <bitsery/traits/string.h>
+#include <bitsery/traits/vector.h>
+#include <bitsery/traits/list.h>
+
 #include <sstream>
 #include <iostream>
 #include <type_traits>
 
-// Type trait to check if a type is const
-template <typename T>
-using is_const_type = std::is_const<std::remove_reference_t<T>>;
-
 template <class R>
 struct Serializer; // Not defined
 
 template<class RetType, class... Args>
 class Serializer<RetType(Args...)> : public dmq::ISerializer<RetType(Args...)>
 {
 public:
+    // Bitsery Adapters for std::ostream / std::istream
     using OutputAdapter = bitsery::OutputStreamAdapter;
     using InputAdapter = bitsery::InputStreamAdapter;
 
-    virtual std::ostream& Write(std::ostream& os, Args... args) override {
+    // Write: Changed 'Args... args' to 'const Args&... args' for efficiency
+    virtual std::ostream& Write(std::ostream& os, const Args&... args) override {
         try {
-            os.seekp(0);
-            bitsery::Serializer<OutputAdapter> writer{ os }; 
+            // Reset stream position.
+            os.seekp(0, std::ios::beg);
 
-            // Serialize each argument using fold expression
+            // Clear stringstreams explicitly to avoid appending new data to old data.
+            // DelegateMQ often reuses the stream object.
+            if (auto* ss = dynamic_cast<std::ostringstream*>(&os)) {
+                ss->str("");
+            }
+
+            // Construct the adapter properly passing the stream
+            bitsery::Serializer<OutputAdapter> writer{ OutputAdapter{os} };
+
+            // Serialize each argument using C++17 fold expression
             (writer.object(args), ...);
+
+            // Ensure buffer is flushed to the stream
             writer.adapter().flush();
         }
         catch (const std::exception& e) {
@@ -48,11 +64,16 @@ class Serializer<RetType(Args...)> : public dmq::ISerializer<RetType(Args...)>
 
     virtual std::istream& Read(std::istream& is, Args&... args) override {
         try {
-            bitsery::Deserializer<InputAdapter> reader{ is }; 
+            // Construct the adapter properly passing the stream
+            bitsery::Deserializer<InputAdapter> reader{ InputAdapter{is} };
 
             // Deserialize each argument using fold expression
             (reader.object(args), ...);
 
+            // Optional: Check for deserialization errors
+            if (reader.adapter().error() != bitsery::ReaderError::NoError) {
+                throw std::runtime_error("Bitsery reported a read error");
+            }
         }
         catch (const std::exception& e) {
             std::cerr << "Bitsery deserialize error: " << e.what() << std::endl;
@@ -62,4 +83,4 @@ class Serializer<RetType(Args...)> : public dmq::ISerializer<RetType(Args...)>
     }
 };
 
-#endif
+#endif // SERIALIZER_H