Message handling now performed through DBus directly.

simplebluez/src/Bluez.cpp

@@ -45,12 +45,7 @@ void Bluez::init() {
 }
 
 void Bluez::run_async() {
-    _conn->read_write();
-    SimpleDBus::Message message = _conn->pop_message();
-    while (message.is_valid()) {
-        message_forward(message);
-        message = _conn->pop_message();
-    }
+    _conn->read_write_dispatch();
 }
 
 std::vector<std::shared_ptr<Adapter>> Bluez::get_adapters() {

simpledbus/include/simpledbus/advanced/ProxyBase.h

@@ -12,6 +12,7 @@ class ProxyBase : public std::enable_shared_from_this<ProxyBase> {
     virtual ~ProxyBase();
 
     bool valid() const;
+    void invalidate();
     std::string path() const;
 
   protected:
@@ -20,6 +21,12 @@ class ProxyBase : public std::enable_shared_from_this<ProxyBase> {
     std::string _path;
     std::string _bus_name;
     std::shared_ptr<Connection> _conn;
+
+    bool _registered;
+    void register_object_path();
+    void unregister_object_path();
+
+    virtual void message_handle(Message& msg);
 };
 
 }  // namespace SimpleDBus

simpledbus/include/simpledbus/advanced/RemoteProxy.h

@@ -43,7 +43,7 @@ class RemoteProxy : public ProxyBase {
     void path_append_child(const std::string& path, std::shared_ptr<RemoteProxy> child);
 
     // ----- MESSAGE HANDLING -----
-    void message_forward(Message& msg);
+    void message_handle(Message& msg) override;
 
     // ----- CALLBACKS -----
     kvn::safe_callback<void(std::string)> on_child_created;
@@ -79,6 +79,9 @@ class RemoteProxy : public ProxyBase {
 
     std::recursive_mutex _interface_access_mutex;
     std::recursive_mutex _child_access_mutex;
+
+    // TODO: This is a hack, should be handled as a weak_ptr<RemoteProxy>
+    RemoteProxy* _parent = nullptr;
 };
 
 }  // namespace SimpleDBus

simpledbus/include/simpledbus/base/Connection.h

@@ -2,6 +2,8 @@
 
 #include <dbus/dbus.h>
 #include <mutex>
+#include <unordered_map>
+#include <functional>
 #include "Message.h"
 
 namespace SimpleDBus {
@@ -21,11 +23,15 @@ class Connection {
     void remove_match(std::string rule);
 
     void read_write();
+    void read_write_dispatch();
     Message pop_message();
 
     void send(Message& msg);
     Message send_with_reply_and_block(Message& msg);
 
+    void register_object_path(const std::string& path, std::function<void(Message&)> handler);
+    void unregister_object_path(const std::string& path);
+
     // ----- PROPERTIES -----
     std::string unique_name();
 
@@ -36,6 +42,9 @@ class Connection {
     ::DBusConnection* _conn;
 
     std::recursive_mutex _mutex;
+
+    static DBusHandlerResult static_message_handler(DBusConnection* connection, DBusMessage* message, void* user_data);
+    std::unordered_map<std::string, std::function<void(Message&)>> _message_handlers;
 };
 
 }  // namespace SimpleDBus

simpledbus/src/advanced/ProxyBase.cpp

@@ -3,10 +3,33 @@
 using namespace SimpleDBus;
 
 ProxyBase::ProxyBase(std::shared_ptr<Connection> conn, const std::string& bus_name, const std::string& path)
-    : _conn(conn), _bus_name(bus_name), _path(path), _valid(true) {}
+    : _conn(conn), _bus_name(bus_name), _path(path), _registered(false) {
+    register_object_path();
+}
 
-ProxyBase::~ProxyBase() {}
+ProxyBase::~ProxyBase() { unregister_object_path(); }
 
 bool ProxyBase::valid() const { return _valid; }
 
+void ProxyBase::invalidate() {
+    _valid = false;
+    unregister_object_path();
+}
+
 std::string ProxyBase::path() const { return _path; }
+
+void ProxyBase::register_object_path() {
+    if (!_registered) {
+        _conn->register_object_path(_path, [this](Message& msg) { this->message_handle(msg); });
+        _registered = true;
+    }
+}
+
+void ProxyBase::unregister_object_path() {
+    if (_registered) {
+        _conn->unregister_object_path(_path);
+        _registered = false;
+    }
+}
+
+void ProxyBase::message_handle(Message& msg) {}

simpledbus/src/advanced/RemoteProxy.cpp

@@ -117,6 +117,7 @@ void RemoteProxy::path_add(const std::string& path, SimpleDBus::Holder managed_i
         // If the path is a direct child of the proxy path, create a new proxy for it.
         std::shared_ptr<RemoteProxy> child = path_create(path);
         child->interfaces_load(managed_interfaces);
+        child->_parent = this;
         _children.emplace(std::make_pair(path, child));
         on_child_created(path);
     } else {
@@ -137,6 +138,7 @@ void RemoteProxy::path_add(const std::string& path, SimpleDBus::Holder managed_i
             std::shared_ptr<RemoteProxy> child = path_create(child_path);
             _children.emplace(std::make_pair(child_path, child));
             child->path_add(path, managed_interfaces);
+            child->_parent = this;
             on_child_created(child_path);
         }
     }
@@ -146,7 +148,7 @@ bool RemoteProxy::path_remove(const std::string& path, SimpleDBus::Holder option
     // `options` contains an array of strings of the interfaces that need to be removed.
 
     if (path == _path) {
-        _valid = false;
+        invalidate();
         interfaces_unload(options);
         return path_prune();
     }
@@ -212,45 +214,30 @@ void RemoteProxy::path_append_child(const std::string& path, std::shared_ptr<Rem
 }
 
 // ----- MESSAGE HANDLING -----
-void RemoteProxy::message_forward(Message& msg) {
-    // If the message is for the current proxy, then forward it to the message handler.
-    if (msg.get_path() == _path) {
-        // If the message is involves a property change, forward it to the correct interface.
-        if (msg.is_signal("org.freedesktop.DBus.Properties", "PropertiesChanged")) {
-            Holder interface_h = msg.extract();
-            std::string iface_name = interface_h.get_string();
-            msg.extract_next();
-            Holder changed_properties = msg.extract();
-            msg.extract_next();
-            Holder invalidated_properties = msg.extract();
-
-            // If the interface is not loaded, then ignore the message.
-            if (!interface_exists(iface_name)) {
-                return;
-            }
 
-            interface_get(iface_name)->signal_property_changed(changed_properties, invalidated_properties);
+void RemoteProxy::message_handle(Message& msg) {
+    if (msg.is_signal("org.freedesktop.DBus.Properties", "PropertiesChanged")) {
+        Holder interface_h = msg.extract();
+        std::string iface_name = interface_h.get_string();
+        msg.extract_next();
+        Holder changed_properties = msg.extract();
+        msg.extract_next();
+        Holder invalidated_properties = msg.extract();
 
-        } else if (interface_exists(msg.get_interface())) {
-            interface_get(msg.get_interface())->message_handle(msg);
+        // If the interface is not loaded, then ignore the message.
+        if (!interface_exists(iface_name)) {
+            return;
         }
 
-        return;
-    }
+        interface_get(iface_name)->signal_property_changed(changed_properties, invalidated_properties);
 
-    // If the message is for a child proxy or a descendant, forward it to that child proxy.
-    for (auto& [child_path, child] : _children) {
-        if (child_path == msg.get_path()) {
-            child->message_forward(msg);
-
-            if (msg.get_type() == Message::Type::SIGNAL) {
-                on_child_signal_received(child_path);
+        if (msg.get_type() == Message::Type::SIGNAL) {
+            if (_parent != nullptr) {
+                _parent->on_child_signal_received(_path);
             }
-
-            return;
-        } else if (Path::is_descendant(child_path, msg.get_path())) {
-            child->message_forward(msg);
-            return;
         }
+
+    } else if (interface_exists(msg.get_interface())) {
+        interface_get(msg.get_interface())->message_handle(msg);
     }
 }

simpledbus/src/base/Connection.cpp

@@ -109,18 +109,36 @@ void Connection::read_write() {
     dbus_connection_read_write(_conn, 0);
 }
 
+void Connection::read_write_dispatch() {
+    if (!_initialized) {
+        throw Exception::NotInitialized();
+    }
+
+    std::lock_guard<std::recursive_mutex> lock(_mutex);
+
+    // Non-blocking read of the next available message
+    dbus_connection_read_write(_conn, 0);
+
+    // Dispatch incoming messages
+    while (dbus_connection_dispatch(_conn) == DBUS_DISPATCH_DATA_REMAINS) {}
+}
+
 Message Connection::pop_message() {
     if (!_initialized) {
         throw Exception::NotInitialized();
     }
 
     std::lock_guard<std::recursive_mutex> lock(_mutex);
 
+    // IMPORTANT NOTE: Ownership of the message is transferred to the caller.
     DBusMessage* msg = dbus_connection_pop_message(_conn);
     if (msg == nullptr) {
         return Message();
     } else {
-        return Message(msg);
+        auto msg_wrapped = Message(msg);
+        // Ownership of the DBusMessage* is transferred to the Message object, we can reduce the reference count.
+        dbus_message_unref(msg);
+        return msg_wrapped;
     }
 }
 
@@ -157,6 +175,43 @@ Message Connection::send_with_reply_and_block(Message& msg) {
     return Message(msg_tmp);
 }
 
+void Connection::register_object_path(const std::string& path, std::function<void(Message&)> handler) {
+    if (!_initialized) {
+        return;
+    }
+
+    std::lock_guard<std::recursive_mutex> lock(_mutex);
+    if (_message_handlers.find(path) == _message_handlers.end()) {
+        DBusObjectPathVTable vtable = {0};
+        vtable.message_function = &Connection::static_message_handler;
+        dbus_connection_register_object_path(_conn, path.c_str(), &vtable, this);
+        _message_handlers[path] = std::move(handler);
+    }
+}
+
+void Connection::unregister_object_path(const std::string& path) {
+    std::lock_guard<std::recursive_mutex> lock(_mutex);
+    auto it = _message_handlers.find(path);
+    if (it != _message_handlers.end()) {
+        dbus_connection_unregister_object_path(_conn, path.c_str());
+        _message_handlers.erase(it);
+    }
+}
+
+DBusHandlerResult Connection::static_message_handler(DBusConnection* connection, DBusMessage* message, void* user_data) {
+    Connection* conn = static_cast<Connection*>(user_data);
+    Message msg(message);
+    std::string path = msg.get_path();
+
+    std::lock_guard<std::recursive_mutex> lock(conn->_mutex);
+    auto it = conn->_message_handlers.find(path);
+    if (it != conn->_message_handlers.end()) {
+        it->second(msg);
+    }
+
+    return DBUS_HANDLER_RESULT_HANDLED;
+}
+
 std::string Connection::unique_name() {
     if (!_initialized) {
         throw Exception::NotInitialized();

simpledbus/src/base/Message.cpp

@@ -29,6 +29,7 @@ Message::Message() : Message(nullptr) {}
 
 Message::Message(DBusMessage* msg) : _msg(msg), _iter_initialized(false), _is_extracted(false), indent(0) {
     if (is_valid()) {
+        dbus_message_ref(_msg);
         _unique_id = creation_counter++;
     } else {
         _unique_id = -1;