fix messaging subscription scoping

Author: Natsoulas

src/architecture/messaging/messaging.h

@@ -38,6 +38,9 @@ class ReadFunctor{
     messageType* payloadPointer;    //!< -- pointer to the incoming msg data
     MsgHeader *headerPointer;      //!< -- pointer to the incoming msg header
     bool initialized;               //!< -- flag indicating if the input message is connect to another message
+    void* sourceRef;               // Generic pointer to store source reference
+    void (*refIncCallback)(void*); // Function pointer for increment ref
+    void (*refDecCallback)(void*); // Function pointer for decrement ref
 
 public:
     //!< -- BSK Logging
@@ -46,10 +49,20 @@ class ReadFunctor{
 
 
     //! constructor
-    ReadFunctor() : initialized(false) {};
+    ReadFunctor() :
+        initialized(false),
+        sourceRef(nullptr),
+        refIncCallback(nullptr),
+        refDecCallback(nullptr) {}
 
     //! constructor
-    ReadFunctor(messageType* payloadPtr, MsgHeader *headerPtr) : payloadPointer(payloadPtr), headerPointer(headerPtr), initialized(true){};
+    ReadFunctor(messageType* payloadPtr, MsgHeader *headerPtr) :
+        payloadPointer(payloadPtr),
+        headerPointer(headerPtr),
+        initialized(true),
+        sourceRef(nullptr),
+        refIncCallback(nullptr),
+        refDecCallback(nullptr) {}
 
     //! constructor
     const messageType& operator()(){
@@ -174,6 +187,43 @@ class ReadFunctor{
 
     //! Recorder method description
     Recorder<messageType> recorder(uint64_t timeDiff = 0){return Recorder<messageType>(this, timeDiff);}
+
+    /*! Set source reference and callback functions for reference counting
+     * @param src Pointer to source object
+     * @param incRef Function to increment reference count
+     * @param decRef Function to decrement reference count
+     */
+    void setSourceRef(void* src, void (*incRef)(void*), void (*decRef)(void*)) {
+        if (sourceRef && refDecCallback) {
+            refDecCallback(sourceRef);
+        }
+        sourceRef = src;
+        refIncCallback = incRef;
+        refDecCallback = decRef;
+        if (sourceRef && refIncCallback) {
+            refIncCallback(sourceRef);
+        }
+    }
+
+    /*! Check if this reader is subscribed to a specific message source
+     * @param source Pointer to message source to check
+     * @return 1 if subscribed, 0 if not
+     */
+    uint8_t isSubscribedTo(const Message<messageType>* source) const {
+        if (!source) return 0;
+        return (this->payloadPointer == source->getPayloadPtr() &&
+                this->headerPointer == source->getHeaderPtr());
+    }
+
+    /*! Check if this reader is subscribed to a void pointer source
+     * @param source Void pointer to message source to check
+     * @return 1 if subscribed, 0 if not
+     */
+    uint8_t isSubscribedTo(const void* source) const {
+        const Message<messageType>* msgSource =
+            static_cast<const Message<messageType>*>(source);
+        return isSubscribedTo(msgSource);
+    }
 };
 
 /*! Write Functor */
@@ -232,6 +282,16 @@ class Message{
 
     //! Return the memory size of the payload, be careful about dynamically sized things
     uint64_t getPayloadSize() {return sizeof(messageType);};
+
+    /*! Get pointer to message payload
+     * @return Const pointer to payload data
+     */
+    const messageType* getPayloadPtr() const { return &payload; }
+
+    /*! Get pointer to message header
+     * @return Const pointer to message header
+     */
+    const MsgHeader* getHeaderPtr() const { return &header; }
 };
 
 

src/architecture/messaging/newMessaging.ih

@@ -34,38 +34,50 @@ OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 
 %template(messageType ## Reader) ReadFunctor<messageTypePayload>;
 %extend ReadFunctor<messageTypePayload> {
-        %pythoncode %{
-            def subscribeTo(self, source):
-                if type(source) == messageType:
-                    self.__subscribe_to(source)
+    void setPySourceRef(PyObject* source) {
+        if (source) {
+            self->setSourceRef(source,
+                [](void* ptr) { Py_INCREF((PyObject*)ptr); },
+                [](void* ptr) { Py_DECREF((PyObject*)ptr); }
+            );
+        }
+    }
+
+    %pythoncode %{
+        def subscribeTo(self, source):
+            if type(source) == messageType:
+                self.__subscribe_to(source)
+                self.setPySourceRef(source)
+                return
+
+            try:
+                from Basilisk.architecture.messaging.messageType ## Payload import messageType ## _C
+                if type(source) == messageType ## _C:
+                    self.__subscribe_to_C(source)
+                    self.setPySourceRef(source)
                     return
-                
-                try:
-                    from Basilisk.architecture.messaging.messageType ## Payload import messageType ## _C
-                    if type(source) == messageType ## _C:
-                        self.__subscribe_to_C(source)
-                        return
-                except ImportError:
-                    pass
-    
-                raise Exception('tried to subscribe ReadFunctor<messageTypePayload> to output message type'
-                                + str(type(source)))
-
-
-            def isSubscribedTo(self, source):
-                if type(source) == messageType:
-                    return self.__is_subscribed_to(source)
-                
-                try:
-                    from Basilisk.architecture.messaging.messageType ## Payload import messageType ## _C
-                except ImportError:
-                    return 0
-                
+            except ImportError:
+                pass
+
+            raise Exception('tried to subscribe ReadFunctor<messageTypePayload> to output message type'
+                            + str(type(source)))
+    %}
+
+    %pythoncode %{
+        def isSubscribedTo(self, source):
+            """Check if this reader is subscribed to the given source"""
+            if type(source) == messageType:
+                return self.__is_subscribed_to(source)
+
+            try:
+                from Basilisk.architecture.messaging.messageType ## Payload import messageType ## _C
                 if type(source) == messageType ## _C:
                     return self.__is_subscribed_to_C(source)
-                else:
-                    return 0
-        %}
+            except ImportError:
+                pass
+
+            return 0
+    %}
 };
 
 %template(messageType ## Writer) WriteFunctor<messageTypePayload>;
@@ -110,7 +122,7 @@ OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
                         for el, k in zip(attr, range(len(attr))):
                             self.explore_and_find_subattr(el, attr_name + "[" + str(k) + "]", content)
                     else:
-                        # The attribute is a list of common types 
+                        # The attribute is a list of common types
                         content[attr_name] = attr
             elif "Basilisk" in str(type(attr)):
                 # The attribute is a swigged BSK object