refactor: use sd-bus API to get current message

src/Connection.cpp

@@ -774,6 +774,13 @@ bool Connection::arePendingMessagesInReadQueue() const
     return readQueueSize > 0;
 }
 
+Message Connection::getCurrentlyProcessedMessage() const
+{
+    auto* sdbusMsg = sdbus_->sd_bus_get_current_message(bus_.get());
+
+    return Message::Factory::create<Message>(sdbusMsg, sdbus_.get());
+}
+
 std::string Connection::composeSignalMatchFilter( const std::string &sender
                                                 , const std::string &objectPath
                                                 , const std::string &interfaceName

src/Connection.h

@@ -86,6 +86,7 @@ namespace sdbus::internal {
         void leaveEventLoop() override;
         PollData getEventLoopPollData() const override;
         bool processPendingEvent() override;
+        Message getCurrentlyProcessedMessage() const override;
 
         void addObjectManager(const std::string& objectPath) override;
         void addObjectManager(const std::string& objectPath, floating_slot_t) override;

src/ISdBus.h

@@ -87,6 +87,7 @@ namespace sdbus::internal {
         virtual int sd_bus_start(sd_bus *bus) = 0;
 
         virtual int sd_bus_process(sd_bus *bus, sd_bus_message **r) = 0;
+        virtual sd_bus_message* sd_bus_get_current_message(sd_bus *bus) = 0;
         virtual int sd_bus_get_poll_data(sd_bus *bus, PollData* data) = 0;
         virtual int sd_bus_get_n_queued_read(sd_bus *bus, uint64_t *ret) = 0;
         virtual int sd_bus_flush(sd_bus *bus) = 0;

src/Object.cpp

@@ -240,9 +240,9 @@ const std::string& Object::getObjectPath() const
     return objectPath_;
 }
 
-const Message* Object::getCurrentlyProcessedMessage() const
+Message Object::getCurrentlyProcessedMessage() const
 {
-    return m_CurrentlyProcessedMessage.load(std::memory_order_relaxed);
+    return connection_.getCurrentlyProcessedMessage();
 }
 
 Object::InterfaceData& Object::getInterface(const std::string& interfaceName)
@@ -338,12 +338,6 @@ int Object::sdbus_method_callback(sd_bus_message *sdbusMessage, void *userData,
 
     auto message = Message::Factory::create<MethodCall>(sdbusMessage, &object.connection_.getSdBusInterface());
 
-    object.m_CurrentlyProcessedMessage.store(&message, std::memory_order_relaxed);
-    SCOPE_EXIT
-    {
-        object.m_CurrentlyProcessedMessage.store(nullptr, std::memory_order_relaxed);
-    };
-
     auto& callback = interfaceData->methods[message.getMemberName()].callback;
     assert(callback);
 
@@ -396,12 +390,6 @@ int Object::sdbus_property_set_callback( sd_bus */*bus*/
 
     auto value = Message::Factory::create<PropertySetCall>(sdbusValue, &object.connection_.getSdBusInterface());
 
-    object.m_CurrentlyProcessedMessage.store(&value, std::memory_order_relaxed);
-    SCOPE_EXIT
-    {
-        object.m_CurrentlyProcessedMessage.store(nullptr, std::memory_order_relaxed);
-    };
-
     auto ok = invokeHandlerAndCatchErrors([&](){ callback(value); }, retError);
 
     return ok ? 1 : -1;

src/Object.h

@@ -35,7 +35,6 @@
 #include <vector>
 #include <functional>
 #include <memory>
-#include <atomic>
 #include <cassert>
 
 namespace sdbus::internal {
@@ -103,7 +102,7 @@ namespace sdbus::internal {
 
         sdbus::IConnection& getConnection() const override;
         const std::string& getObjectPath() const override;
-        const Message* getCurrentlyProcessedMessage() const override;
+        Message getCurrentlyProcessedMessage() const override;
 
     private:
         using InterfaceName = std::string;
@@ -178,7 +177,6 @@ namespace sdbus::internal {
         std::string objectPath_;
         std::map<InterfaceName, InterfaceData> interfaces_;
         Slot objectManagerSlot_;
-        std::atomic<const Message*> m_CurrentlyProcessedMessage{nullptr};
     };
 
 }

src/Proxy.cpp

@@ -200,9 +200,9 @@ const std::string& Proxy::getObjectPath() const
     return objectPath_;
 }
 
-const Message* Proxy::getCurrentlyProcessedMessage() const
+Message Proxy::getCurrentlyProcessedMessage() const
 {
-    return m_CurrentlyProcessedMessage.load(std::memory_order_relaxed);
+    return connection_->getCurrentlyProcessedMessage();
 }
 
 int Proxy::sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userData, sd_bus_error *retError)
@@ -225,12 +225,6 @@ int Proxy::sdbus_async_reply_handler(sd_bus_message *sdbusMessage, void *userDat
 
     auto message = Message::Factory::create<MethodReply>(sdbusMessage, &proxy.connection_->getSdBusInterface());
 
-    proxy.m_CurrentlyProcessedMessage.store(&message, std::memory_order_relaxed);
-    SCOPE_EXIT
-    {
-        proxy.m_CurrentlyProcessedMessage.store(nullptr, std::memory_order_relaxed);
-    };
-
     auto ok = invokeHandlerAndCatchErrors([&]
     {
         const auto* error = sd_bus_message_get_error(sdbusMessage);
@@ -256,12 +250,6 @@ int Proxy::sdbus_signal_handler(sd_bus_message *sdbusMessage, void *userData, sd
 
     auto message = Message::Factory::create<Signal>(sdbusMessage, &signalData->proxy.connection_->getSdBusInterface());
 
-    signalData->proxy.m_CurrentlyProcessedMessage.store(&message, std::memory_order_relaxed);
-    SCOPE_EXIT
-    {
-        signalData->proxy.m_CurrentlyProcessedMessage.store(nullptr, std::memory_order_relaxed);
-    };
-
     auto ok = invokeHandlerAndCatchErrors([&](){ signalData->callback(message); }, retError);
 
     return ok ? 0 : -1;

src/Proxy.h

@@ -35,7 +35,6 @@
 #include <map>
 #include <deque>
 #include <mutex>
-#include <atomic>
 #include <condition_variable>
 
 namespace sdbus::internal {
@@ -72,7 +71,7 @@ namespace sdbus::internal {
 
         sdbus::IConnection& getConnection() const override;
         const std::string& getObjectPath() const override;
-        const Message* getCurrentlyProcessedMessage() const override;
+        Message getCurrentlyProcessedMessage() const override;
 
     private:
         void registerSignalHandlers(sdbus::internal::IConnection& connection);
@@ -178,8 +177,6 @@ namespace sdbus::internal {
             std::mutex mutex_;
             std::deque<std::shared_ptr<CallData>> calls_;
         } pendingAsyncCalls_;
-
-        std::atomic<const Message*> m_CurrentlyProcessedMessage{nullptr};
     };
 
 }

src/SdBus.cpp

@@ -382,6 +382,11 @@ int SdBus::sd_bus_process(sd_bus *bus, sd_bus_message **r)
     return ::sd_bus_process(bus, r);
 }
 
+sd_bus_message* SdBus::sd_bus_get_current_message(sd_bus *bus)
+{
+    return ::sd_bus_get_current_message(bus);
+}
+
 int SdBus::sd_bus_get_poll_data(sd_bus *bus, PollData* data)
 {
     std::lock_guard lock(sdbusMutex_);

src/SdBus.h

@@ -79,6 +79,7 @@ public:
     virtual int sd_bus_start(sd_bus *bus) override;
 
     virtual int sd_bus_process(sd_bus *bus, sd_bus_message **r) override;
+    virtual sd_bus_message* sd_bus_get_current_message(sd_bus *bus) override;
     virtual int sd_bus_get_poll_data(sd_bus *bus, PollData* data) override;
     virtual int sd_bus_get_n_queued_read(sd_bus *bus, uint64_t *ret) override;
     virtual int sd_bus_flush(sd_bus *bus) override;