Add support for transports to process incoming messages

This is used in AF_BUS case, so that the org.freedesktop.DBus.AF_BUS's Forwarded
can be caught and thus have an up-to-date cache of well-known-names<->AF_BUS addresses,
which is used to send messages directly to its destination when we have the address,
instead of sending multicast messages.

6 files changed, 102 insertions, 5 deletions

diff --git a/dbus/dbus-transport-afbus.c b/dbus/dbus-transport-afbus.c
index d920f5df..fddc33ac 100644
--- a/dbus/dbus-transport-afbus.c
+++ b/dbus/dbus-transport-afbus.c
@@ -25,10 +25,13 @@
 #include <errno.h>
 #include <sys/socket.h>
 #include "dbus-connection-internal.h"
+#include "dbus-hash.h"
 #include "dbus-transport-afbus.h"
 #include "dbus-transport-socket.h"
 #include "dbus-watch.h"
 
+static DBusHashTable *wkn_addresses_cache = NULL;
+
 /**
  * @defgroup DBusTransportAfbus AF_BUS-based DBusTransport implementation
  * @ingroup DBusInternals
@@ -86,10 +89,14 @@ get_write_destination (int                  fd,
 
   if (destination != NULL && strlen (destination) > 0)
     {
+      dbus_uint64_t peer = 0x0000000000000000;
+
       /* a message for a peer */
-      /* FIXME: it should check the cache, and send to the correct address
-       *        for debugging purposes, just send messages to the daemon */
-      sock->sbus_addr.s_addr = 0x0000000000000000;
+      if (wkn_addresses_cache != NULL)
+        peer = (dbus_uint64_t) _dbus_hash_table_lookup_string (wkn_addresses_cache, destination);
+
+      sock->sbus_addr.s_addr = peer;
+
       return TRUE;
     }
   else
@@ -204,10 +211,53 @@ afbus_authenticated (DBusTransportSocket *socket_transport)
     }
 }
 
+static void
+afbus_message_received (DBusTransportSocket *socket_transport,
+                        DBusMessage         *message)
+{
+  if (dbus_message_get_type (message) == DBUS_MESSAGE_TYPE_SIGNAL)
+    {
+      const char *path, *interface, *member;
+
+      path = dbus_message_get_path (message);
+      interface = dbus_message_get_interface (message);
+      member = dbus_message_get_member (message);
+
+      if (strcmp (path, DBUS_PATH_AFBUS) == 0
+          && strcmp (interface, DBUS_INTERFACE_AFBUS) == 0
+          && strcmp (member, "Forwarded") == 0)
+        {
+          char *wkn;
+          dbus_uint64_t peer;
+
+          /* Update the cache */
+          if (wkn_addresses_cache == NULL)
+            {
+              wkn_addresses_cache = _dbus_hash_table_new (DBUS_HASH_STRING,
+                                                          dbus_free,
+                                                          NULL);
+              if (wkn_addresses_cache == NULL)
+                return;
+            }
+
+          dbus_message_get_args (message, NULL,
+                                 DBUS_TYPE_STRING, &wkn,
+                                 DBUS_TYPE_UINT64, &peer,
+                                 DBUS_TYPE_INVALID);
+          _dbus_hash_table_insert_string (wkn_addresses_cache,
+                                          _dbus_strdup (wkn),
+                                          (void *) peer);
+
+          printf ("AF_BUS.Forwarded for %s!\n", wkn);
+        }
+    }
+}
+
 static const DBusTransportSocketVTable afbus_vtable = {
   afbus_write_socket,
   afbus_write_socket_two,
-  afbus_authenticated
+  afbus_authenticated,
+  afbus_message_received
 };
 
 /**
diff --git a/dbus/dbus-transport-protected.h b/dbus/dbus-transport-protected.h
index 44b9d785..9becbdb3 100644
--- a/dbus/dbus-transport-protected.h
+++ b/dbus/dbus-transport-protected.h
@@ -69,6 +69,10 @@ struct DBusTransportVTable
   dbus_bool_t (* get_socket_fd) (DBusTransport *transport,
                                  int           *fd_p);
   /**< Get socket file descriptor */
+
+  void        (* process_incoming_message) (DBusTransport *transport,
+                                            DBusMessage   *message);
+  /**> Method to allow transport to filter messages */
 };
 
 /**
diff --git a/dbus/dbus-transport-socket.c b/dbus/dbus-transport-socket.c
index 0fea35d8..3192c8b3 100644
--- a/dbus/dbus-transport-socket.c
+++ b/dbus/dbus-transport-socket.c
@@ -1251,6 +1251,19 @@ socket_get_socket_fd (DBusTransport *transport,
   return TRUE;
 }
 
+static void
+socket_process_incoming_message (DBusTransport *transport,
+                                 DBusMessage   *message)
+{
+  DBusTransportSocket *socket_transport = (DBusTransportSocket*) transport;
+
+  if (socket_transport->vtable->message_received != NULL)
+    {
+      (* socket_transport->vtable->message_received) (socket_transport,
+                                                      message);
+    }
+}
+
 static const DBusTransportVTable socket_vtable = {
   socket_finalize,
   socket_handle_watch,
@@ -1258,7 +1271,8 @@ static const DBusTransportVTable socket_vtable = {
   socket_connection_set,
   socket_do_iteration,
   socket_live_messages_changed,
-  socket_get_socket_fd
+  socket_get_socket_fd,
+  socket_process_incoming_message
 };
 
 /**
diff --git a/dbus/dbus-transport-socket.h b/dbus/dbus-transport-socket.h
index 52b5f88b..ed7fdcc3 100644
--- a/dbus/dbus-transport-socket.h
+++ b/dbus/dbus-transport-socket.h
@@ -59,6 +59,10 @@ struct DBusTransportSocketVTable
 
   void (* authenticated)    (DBusTransportSocket *socket_transport);
   /**< Notification of authentication successful */
+
+  void (* message_received) (DBusTransportSocket *socket_transport,
+                             DBusMessage         *message);
+  /**< Notification of a message received */
 };
 
 /**
diff --git a/dbus/dbus-transport.c b/dbus/dbus-transport.c
index c28b186a..1e9accfd 100644
--- a/dbus/dbus-transport.c
+++ b/dbus/dbus-transport.c
@@ -987,6 +987,28 @@ _dbus_transport_do_iteration (DBusTransport  *transport,
   _dbus_verbose ("end\n");
 }
 
+/**
+ * Allows transports to process incoming messages
+ *
+ * @param transport the transport.
+ * @param message the message to be filtered
+ */
+void
+_dbus_transport_process_incoming_message (DBusTransport *transport,
+                                          DBusMessage   *message)
+{
+  _dbus_verbose ("Transport message filtering\n");
+
+  if (transport->vtable->process_incoming_message != NULL)
+    {
+      _dbus_transport_ref (transport);
+      (* transport->vtable->process_incoming_message) (transport, message);
+      _dbus_transport_unref (transport);
+    }
+
+  _dbus_verbose ("end\n");
+}
+
 static dbus_bool_t
 recover_unused_bytes (DBusTransport *transport)
 {
@@ -1141,6 +1163,7 @@ _dbus_transport_queue_messages (DBusTransport *transport)
       _dbus_assert (link != NULL);
       
       message = link->data;
+      _dbus_transport_process_incoming_message (transport, message);
       
       _dbus_verbose ("queueing received message %p\n", message);
 
diff --git a/dbus/dbus-transport.h b/dbus/dbus-transport.h
index 4b821517..194b2c6c 100644
--- a/dbus/dbus-transport.h
+++ b/dbus/dbus-transport.h
@@ -52,6 +52,8 @@ dbus_bool_t        _dbus_transport_set_connection         (DBusTransport
 void               _dbus_transport_do_iteration           (DBusTransport              *transport,
                                                            unsigned int                flags,
                                                            int                         timeout_milliseconds);
+void               _dbus_transport_process_incoming_message (DBusTransport              *transport,
+                                                             DBusMessage                *message);
 DBusDispatchStatus _dbus_transport_get_dispatch_status    (DBusTransport              *transport);
 dbus_bool_t        _dbus_transport_queue_messages         (DBusTransport              *transport);