2003-01-21 Anders Carlsson <andersca@codefactory.se>

	* dbus/dbus-connection.c: (dbus_connection_send_message):
	Add a new client_serial parameter.

	(dbus_connection_send_message_with_reply):
	Remove a @todo since we've implemented the blocking function.

	(dbus_connection_send_message_with_reply_and_block):
	New function that sends a message and waits for a reply and
	then returns the reply.

	* dbus/dbus-connection.h:
	Add new functions.

	* dbus/dbus-errors.c: (dbus_result_to_string):
	* dbus/dbus-errors.h:
	Add new DBUS_RESULT.

	* dbus/dbus-message-internal.h:
	* dbus/dbus-message.c: (_dbus_message_get_reply_serial),
	(_dbus_message_set_sender), (dbus_message_write_header),
	(dbus_message_new_reply), (decode_header_data),
	(_dbus_message_loader_return_buffer), (_dbus_message_test):
	* dbus/dbus-message.h:
	Add new functions that set the reply serial and sender.
	Also marshal and demarshal them correctly and add test.

	* dbus/dbus-protocol.h:
	Add new DBUS_MESSAGE_TYPE_SENDER.

	* glib/dbus-glib.h:
	* glib/dbus-gmain.c: (watch_callback), (free_callback_data),
	(add_watch), (remove_watch), (add_timeout), (remove_timeout),
	(dbus_connection_hookup_with_g_main):
	* glib/test-dbus-glib.c: (main):
	Rewrite to use GIOChannel and remove the GSource crack.

	* test/echo-client.c: (main):
	* test/watch.c: (check_messages):
	Update for changed APIs

14 files changed, 364 insertions, 213 deletions

diff --git a/ChangeLog b/ChangeLog
index 35a7dadf..8ab9ba62 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,45 @@
+2003-01-21  Anders Carlsson  <andersca@codefactory.se>
+
+	* dbus/dbus-connection.c: (dbus_connection_send_message):
+	Add a new client_serial parameter.
+	
+	(dbus_connection_send_message_with_reply):
+	Remove a @todo since we've implemented the blocking function.
+	
+	(dbus_connection_send_message_with_reply_and_block):
+	New function that sends a message and waits for a reply and
+	then returns the reply.
+	
+	* dbus/dbus-connection.h:
+	Add new functions.
+	
+	* dbus/dbus-errors.c: (dbus_result_to_string):
+	* dbus/dbus-errors.h:
+	Add new DBUS_RESULT.
+	
+	* dbus/dbus-message-internal.h:
+	* dbus/dbus-message.c: (_dbus_message_get_reply_serial),
+	(_dbus_message_set_sender), (dbus_message_write_header),
+	(dbus_message_new_reply), (decode_header_data),
+	(_dbus_message_loader_return_buffer), (_dbus_message_test):
+	* dbus/dbus-message.h:
+	Add new functions that set the reply serial and sender.
+	Also marshal and demarshal them correctly and add test.
+	
+	* dbus/dbus-protocol.h:
+	Add new DBUS_MESSAGE_TYPE_SENDER.
+	
+	* glib/dbus-glib.h:
+	* glib/dbus-gmain.c: (watch_callback), (free_callback_data),
+	(add_watch), (remove_watch), (add_timeout), (remove_timeout),
+	(dbus_connection_hookup_with_g_main):
+	* glib/test-dbus-glib.c: (main):
+	Rewrite to use GIOChannel and remove the GSource crack.
+	
+	* test/echo-client.c: (main):
+	* test/watch.c: (check_messages):
+	Update for changed APIs
+	
 2003-01-19  Anders Carlsson  <andersca@codefactory.se>
 
 	* dbus/Makefile.am: Add dbus-timeout.[cħ]
diff --git a/dbus/dbus-connection.c b/dbus/dbus-connection.c
index 25114964..0cae7601 100644
--- a/dbus/dbus-connection.c
+++ b/dbus/dbus-connection.c
@@ -59,6 +59,9 @@
  * @{
  */
 
+/** default timeout value when waiting for a message reply */
+#define DEFAULT_TIMEOUT_VALUE (15 * 1000)
+
 /** Opaque typedef for DBusDataSlot */
 typedef struct DBusDataSlot DBusDataSlot;
 /** DBusDataSlot is used to store application data on the connection */
@@ -615,14 +618,19 @@ dbus_connection_get_is_authenticated (DBusConnection *connection)
  * 
  * @param connection the connection.
  * @param message the message to write.
+ * @param client_serial return location for client serial.
  * @param result address where result code can be placed.
  * @returns #TRUE on success.
  */
 dbus_bool_t
 dbus_connection_send_message (DBusConnection *connection,
                               DBusMessage    *message,
+			      dbus_int32_t   *client_serial,			      
                               DBusResultCode *result)
-{  
+
+{
+  dbus_int32_t serial;
+  
   if (!_dbus_list_prepend (&connection->outgoing_messages,
                            message))
     {
@@ -636,7 +644,12 @@ dbus_connection_send_message (DBusConnection *connection,
   _dbus_verbose ("Message %p added to outgoing queue, %d pending to send\n",
                  message, connection->n_outgoing);
 
-  _dbus_message_set_client_serial (message, _dbus_connection_get_next_client_serial (connection));
+  serial = _dbus_connection_get_next_client_serial (connection);
+  _dbus_message_set_client_serial (message, serial);
+
+  if (client_serial)
+    *client_serial = serial;
+  
   _dbus_message_lock (message);
   
   if (connection->n_outgoing == 1)
@@ -688,12 +701,6 @@ dbus_connection_send_message (DBusConnection *connection,
  * install a timeout. Then install a timeout which is the shortest
  * timeout of any pending reply.
  *
- * @todo implement non-reentrant "block for reply" variant.  i.e. send
- * a message, block until we get a reply, then pull reply out of
- * message queue and return it, *without dispatching any handlers for
- * any other messages* - used for non-reentrant "method calls" We can
- * block properly for this using _dbus_connection_do_iteration().
- * 
  */
 dbus_bool_t
 dbus_connection_send_message_with_reply (DBusConnection     *connection,
@@ -703,7 +710,68 @@ dbus_connection_send_message_with_reply (DBusConnection     *connection,
                                          DBusResultCode     *result)
 {
   /* FIXME */
-  return dbus_connection_send_message (connection, message, result);
+  return dbus_connection_send_message (connection, message, NULL, result);
+}
+
+/**
+ * Sends a message and blocks a certain time period while waiting for a reply.
+ * This function does not dispatch any message handlers until the main loop
+ * has been reached. This function is used to do non-reentrant "method calls."
+ *
+ * @param connection the connection
+ * @param message the message to send
+ * @param timeout_milliseconds timeout in milliseconds or -1 for default
+ * @param result return location for result code
+ * @returns the message that is the reply or #NULL with an error code if the
+ * function fails.
+ */
+DBusMessage *
+dbus_connection_send_message_with_reply_and_block (DBusConnection     *connection,
+						   DBusMessage        *message,
+						   int                 timeout_milliseconds,
+						   DBusResultCode     *result)
+{
+  dbus_int32_t client_serial;
+  DBusList *link;
+
+  if (timeout_milliseconds == -1)
+    timeout_milliseconds = DEFAULT_TIMEOUT_VALUE;
+  
+  if (!dbus_connection_send_message (connection, message, &client_serial, result))
+    return NULL;
+
+  /* Flush message queue */
+  dbus_connection_flush (connection);
+  
+  /* Now we wait... */
+  _dbus_connection_do_iteration (connection,
+				 DBUS_ITERATION_DO_READING |
+				 DBUS_ITERATION_BLOCK,
+				 timeout_milliseconds);
+
+  /* Check if we've gotten a reply */
+  link = _dbus_list_get_first_link (&connection->incoming_messages);
+
+  while (link != NULL)
+    {
+      DBusMessage *reply = link->data;
+
+      if (_dbus_message_get_reply_serial (reply) == client_serial)
+	{
+	  dbus_message_ref (message);
+
+	  if (result)
+	    *result = DBUS_RESULT_SUCCESS;
+	  
+	  return reply;
+	}
+      link = _dbus_list_get_next_link (&connection->incoming_messages, link);
+    }
+
+  if (result)
+    *result = DBUS_RESULT_NO_REPLY;
+  
+  return NULL;
 }
 
 /**
diff --git a/dbus/dbus-connection.h b/dbus/dbus-connection.h
index 2a888ec1..036d6776 100644
--- a/dbus/dbus-connection.h
+++ b/dbus/dbus-connection.h
@@ -82,14 +82,20 @@ DBusMessage*    dbus_connection_peek_message         (DBusConnection *connection
 DBusMessage*    dbus_connection_pop_message          (DBusConnection *connection);
 dbus_bool_t     dbus_connection_dispatch_message     (DBusConnection *connection);
 
-dbus_bool_t dbus_connection_send_message            (DBusConnection     *connection,
-                                                     DBusMessage        *message,
-                                                     DBusResultCode     *result);
-dbus_bool_t dbus_connection_send_message_with_reply (DBusConnection     *connection,
-                                                     DBusMessage        *message,
-                                                     DBusMessageHandler *reply_handler,
-                                                     int                 timeout_milliseconds,
-                                                     DBusResultCode     *result);
+dbus_bool_t  dbus_connection_send_message                      (DBusConnection     *connection,
+								DBusMessage        *message,
+								dbus_int32_t       *client_serial,
+								DBusResultCode     *result);
+dbus_bool_t  dbus_connection_send_message_with_reply           (DBusConnection     *connection,
+								DBusMessage        *message,
+								DBusMessageHandler *reply_handler,
+								int                 timeout_milliseconds,
+								DBusResultCode     *result);
+DBusMessage *dbus_connection_send_message_with_reply_and_block (DBusConnection     *connection,
+								DBusMessage        *message,
+								int                 timeout_milliseconds,
+								DBusResultCode     *result);
+
 
 void dbus_connection_set_disconnect_function (DBusConnection            *connection,
 					      DBusDisconnectFunction     function,
diff --git a/dbus/dbus-errors.c b/dbus/dbus-errors.c
index 223750e7..024ddd5a 100644
--- a/dbus/dbus-errors.c
+++ b/dbus/dbus-errors.c
@@ -101,6 +101,8 @@ dbus_result_to_string (DBusResultCode code)
       return "Disconnected.";
     case DBUS_RESULT_INVALID_FIELDS:
       return "Invalid fields.";
+    case DBUS_RESULT_NO_REPLY:
+      return "Did not get a reply message.";
       
       /* no default, it would break our compiler warnings */
     }
diff --git a/dbus/dbus-errors.h b/dbus/dbus-errors.h
index 4dd87677..b1013a60 100644
--- a/dbus/dbus-errors.h
+++ b/dbus/dbus-errors.h
@@ -50,7 +50,8 @@ typedef enum
   DBUS_RESULT_NO_NETWORK,      /**< Can't find the network */
   DBUS_RESULT_ADDRESS_IN_USE,  /**< Someone's already using the address */
   DBUS_RESULT_DISCONNECTED,    /**< No more connection. */
-  DBUS_RESULT_INVALID_FIELDS   /**< One or more invalid fields encountered. */
+  DBUS_RESULT_INVALID_FIELDS,  /**< One or more invalid fields encountered. */
+  DBUS_RESULT_NO_REPLY,        /**< Did not get a reply message. */
 } DBusResultCode;
 
 void        dbus_set_result       (DBusResultCode *code_address,
diff --git a/dbus/dbus-message-internal.h b/dbus/dbus-message-internal.h
index 5ec7ad9d..9507a0a4 100644
--- a/dbus/dbus-message-internal.h
+++ b/dbus/dbus-message-internal.h
@@ -34,13 +34,15 @@ void _dbus_message_get_network_data  (DBusMessage       *message,
 				      const DBusString **header,
 				      const DBusString **body);
 
-void _dbus_message_lock              (DBusMessage       *message);
+void         _dbus_message_lock              (DBusMessage  *message);
+void         _dbus_message_set_client_serial (DBusMessage  *message,
+					      dbus_int32_t  client_serial);
+void         _dbus_message_set_sender        (DBusMessage  *message,
+					      const char   *sender);
+dbus_int32_t _dbus_message_get_reply_serial  (DBusMessage  *message);
+void         _dbus_message_add_size_counter  (DBusMessage  *message,
+					      DBusCounter  *counter);
 
-void _dbus_message_set_client_serial (DBusMessage       *message,
-				      dbus_int32_t       client_serial);
-
-void _dbus_message_add_size_counter  (DBusMessage       *message,
-                                      DBusCounter       *counter);
 
 DBusMessageLoader* _dbus_message_loader_new                   (void);
 void               _dbus_message_loader_ref                   (DBusMessageLoader  *loader);
diff --git a/dbus/dbus-message.c b/dbus/dbus-message.c
index 5767bda6..e491a7db 100644
--- a/dbus/dbus-message.c
+++ b/dbus/dbus-message.c
@@ -62,6 +62,7 @@ struct DBusMessage
   
   char *name; /**< Message name. */
   char *service; /**< Message destination service. */
+  char *sender; /**< Message sender service. */
   
   dbus_int32_t client_serial; /**< Client serial or -1 if not set */
   dbus_int32_t reply_serial; /**< Reply serial or -1 if not set */
@@ -124,6 +125,35 @@ _dbus_message_set_client_serial (DBusMessage  *message,
 }
 
 /**
+ * Returns the serial that the message is
+ * a reply to.
+ *
+ * @param message the message
+ * @returns the reply serial
+ */
+dbus_int32_t
+_dbus_message_get_reply_serial  (DBusMessage *message)
+{
+  return message->client_serial;
+}
+
+/**
+ * Sets the message sender. This can only
+ * be done once on a message.
+ *
+ * @param message the message
+ * @param sender the sender
+ */
+void
+_dbus_message_set_sender (DBusMessage  *message,
+			  const char   *sender)
+{
+  _dbus_assert (message->sender == NULL);
+
+  message->sender = _dbus_strdup (sender);
+}
+
+/**
  * Adds a counter to be incremented immediately with the
  * size of this message, and decremented by the size
  * of this message when this message if finalized.
@@ -199,12 +229,24 @@ dbus_message_write_header (DBusMessage *message)
   /* Marshal reply serial */
   if (message->reply_serial != -1)
     {
+      _dbus_string_align_length (&message->header, 4);
       _dbus_string_append_len (&message->header, DBUS_HEADER_FIELD_REPLY, 4);
 
       _dbus_string_append_byte (&message->header, DBUS_TYPE_INT32);
       _dbus_marshal_int32 (&message->header, DBUS_COMPILER_BYTE_ORDER,
                            message->reply_serial);
     }
+
+  /* Marshal sender */
+  if (message->sender)
+    {
+      _dbus_string_align_length (&message->header, 4);
+      _dbus_string_append_len (&message->header, DBUS_HEADER_FIELD_SENDER, 4);
+      _dbus_string_append_byte (&message->header, DBUS_TYPE_STRING);
+
+      _dbus_marshal_string (&message->header, DBUS_COMPILER_BYTE_ORDER,
+			    message->sender);
+    }
   
   /* Fill in the length */
   _dbus_string_get_data_len (&message->header, &len_data, 4, 4);
@@ -302,6 +344,35 @@ dbus_message_new (const char *service,
   return message;
 }
 
+/**
+ * Constructs a message that is a reply to some other
+ * message. Returns #NULL if memory can't be allocated
+ * for the message.
+ *
+ * @param name the name of the message
+ * @param original_message the message which the created
+ * message is a reply to.
+ * @returns a new DBusMessage, free with dbus_message_unref()
+ * @see dbus_message_new(), dbus_message_unref()
+ */ 
+DBusMessage*
+dbus_message_new_reply (const char  *name,
+			DBusMessage *original_message)
+{
+  DBusMessage *message;
+
+  _dbus_assert (original_message->sender != NULL);
+  
+  message = dbus_message_new (original_message->sender, name);
+  
+  if (message == NULL)
+    return NULL;
+
+  message->reply_serial = original_message->client_serial;
+
+  return message;
+}
+
 
 /**
  * Increments the reference count of a DBusMessage.
@@ -1126,6 +1197,11 @@ _dbus_message_loader_get_buffer (DBusMessageLoader  *loader,
 #define DBUS_HEADER_FIELD_REPLY_AS_UINT32   \
   FOUR_CHARS_TO_UINT32 ('r', 'p', 'l', 'y')
 
+/** DBUS_HEADER_FIELD_SENDER Packed into a dbus_uint32_t */
+#define DBUS_HEADER_FIELD_SENDER_AS_UINT32  \
+  FOUR_CHARS_TO_UINT32 ('s', 'n', 'd', 'r')
+
+
 /* FIXME should be using DBusString for the stuff we demarshal.  char*
  * evil. Also, out of memory handling here seems suboptimal.
  * Should probably report it as a distinct error from "corrupt message,"
@@ -1137,8 +1213,10 @@ decode_header_data (DBusString   *data,
 		    int		  header_len,
 		    int           byte_order,
 		    dbus_int32_t *client_serial,
+		    dbus_int32_t *reply_serial,
 		    char        **service,
-		    char        **name)
+		    char        **name,
+		    char        **sender)
 {
   const char *field;
   int pos, new_pos;
@@ -1148,6 +1226,7 @@ decode_header_data (DBusString   *data,
 
   *service = NULL;
   *name = NULL;
+  *sender = NULL;
   
   /* Now handle the fields */
   while (pos < header_len)
@@ -1189,6 +1268,21 @@ decode_header_data (DBusString   *data,
 	  *name = _dbus_demarshal_string (data, byte_order, pos + 1, &new_pos);
           /* FIXME check for demarshal failure SECURITY */
           break;
+	case DBUS_HEADER_FIELD_SENDER_AS_UINT32:
+	  if (*sender != NULL)
+	    {
+	      _dbus_verbose ("%s field provided twice\n",
+			     DBUS_HEADER_FIELD_NAME);
+	      goto failed;
+	    }
+
+	  *sender = _dbus_demarshal_string (data, byte_order, pos + 1, &new_pos);
+          /* FIXME check for demarshal failure SECURITY */
+	  break;
+	case DBUS_HEADER_FIELD_REPLY_AS_UINT32:
+	  *reply_serial = _dbus_demarshal_int32 (data, byte_order, pos + 1, &new_pos);
+
+	  break;
         default:
 	  _dbus_verbose ("Ignoring an unknown header field: %c%c%c%c\n",
 			 field[0], field[1], field[2], field[3]);
@@ -1268,23 +1362,29 @@ _dbus_message_loader_return_buffer (DBusMessageLoader  *loader,
 
       if (_dbus_string_get_length (&loader->data) >= header_len + body_len)
 	{
-	  dbus_int32_t client_serial;
-	  char *service, *name;
+	  dbus_int32_t client_serial, reply_serial;
+	  char *service, *name, *sender;
 
           /* FIXME right now if this doesn't have enough memory, the
            * loader becomes corrupted. Instead we should just not
            * parse this message for now.
            */
  	  if (!decode_header_data (&loader->data, header_len, byte_order,
-				   &client_serial, &service, &name))
+				   &client_serial, &reply_serial, &service, &name, &sender))
 	    {
 	      loader->corrupted = TRUE;
 	      return;
 	    }
 
+
   	  message = dbus_message_new (service, name);
+	  message->reply_serial = reply_serial;
+	  message->client_serial = client_serial;
+	  _dbus_message_set_sender (message, sender);
+	  
           dbus_free (service);
 	  dbus_free (name);
+	  dbus_free (sender);
 	  
 	  if (message == NULL)
             break; /* ugh, postpone this I guess. */
@@ -1297,7 +1397,7 @@ _dbus_message_loader_return_buffer (DBusMessageLoader  *loader,
 
           _dbus_assert (_dbus_string_get_length (&message->header) == 0);
           _dbus_assert (_dbus_string_get_length (&message->body) == 0);
-          
+
 	  if (!_dbus_string_move_len (&loader->data, 0, header_len, &message->header, 0))
             {
               _dbus_list_remove_last (&loader->messages, message);
@@ -1486,6 +1586,8 @@ _dbus_message_test (void)
   
   message = dbus_message_new ("org.freedesktop.DBus.Test", "testMessage");
   message->client_serial = 1;
+  message->reply_serial = 0x12345678;
+
   dbus_message_append_string (message, "Test string");
   dbus_message_append_int32 (message, -0x12345678);
   dbus_message_append_uint32 (message, 0xedd1e);
@@ -1529,6 +1631,9 @@ _dbus_message_test (void)
   if (!message)
     _dbus_assert_not_reached ("received a NULL message");
 
+  if (message->reply_serial != 0x12345678)
+    _dbus_assert_not_reached ("reply serial fields differ");
+  
   message_iter_test (message);
   
   dbus_message_unref (message);
diff --git a/dbus/dbus-message.h b/dbus/dbus-message.h
index b8be6907..7d5ea21b 100644
--- a/dbus/dbus-message.h
+++ b/dbus/dbus-message.h
@@ -36,15 +36,16 @@ DBUS_BEGIN_DECLS;
 typedef struct DBusMessage DBusMessage;
 typedef struct DBusMessageIter DBusMessageIter;
 
-DBusMessage* dbus_message_new   (const char *service,
-				 const char *name);
+DBusMessage* dbus_message_new       (const char  *service,
+				     const char  *name);
+DBusMessage* dbus_message_new_reply (const char  *name,
+				     DBusMessage *original_message);
 
 void         dbus_message_ref   (DBusMessage *message);
 void         dbus_message_unref (DBusMessage *message);
 
-const char*  dbus_message_get_name (DBusMessage *message);
-const char*  dbus_message_get_service (DBusMessage *message);
-
+const char*  dbus_message_get_name         (DBusMessage *message);
+const char*  dbus_message_get_service      (DBusMessage *message);
 
 dbus_bool_t dbus_message_append_fields        (DBusMessage         *message,
                                                int                  first_field_type,
diff --git a/dbus/dbus-protocol.h b/dbus/dbus-protocol.h
index 634283c6..76ffa52c 100644
--- a/dbus/dbus-protocol.h
+++ b/dbus/dbus-protocol.h
@@ -51,7 +51,8 @@ extern "C" {
 #define DBUS_HEADER_FIELD_NAME    "name"
 #define DBUS_HEADER_FIELD_SERVICE "srvc"
 #define DBUS_HEADER_FIELD_REPLY	  "rply"
-
+#define DBUS_HEADER_FIELD_SENDER  "sndr"
+  
 #ifdef __cplusplus
 }
 #endif
diff --git a/glib/dbus-glib.h b/glib/dbus-glib.h
index 6abbc4fe..3d72e491 100644
--- a/glib/dbus-glib.h
+++ b/glib/dbus-glib.h
@@ -26,13 +26,7 @@
 #include <dbus/dbus.h>
 #include <glib.h>
 
-typedef void (*DBusMessageFunction) (DBusConnection *connection,
-                                     DBusMessage    *message,
-                                     gpointer        data);
-
-void dbus_gthread_init  (void);
-
-GSource *dbus_connection_gsource_new (DBusConnection *connection);
-
+void dbus_gthread_init                  (void);
+void dbus_connection_hookup_with_g_main (DBusConnection *connection);
 
 #endif /* DBUS_GLIB_H */
diff --git a/glib/dbus-gmain.c b/glib/dbus-gmain.c
index 7d22dbdd..6e267cdb 100644
--- a/glib/dbus-gmain.c
+++ b/glib/dbus-gmain.c
@@ -24,171 +24,122 @@
 #include "dbus-glib.h"
 #include <glib.h>
 
-typedef struct _DBusGSource DBusGSource;
-
-struct _DBusGSource
+typedef struct
 {
-  GSource source;
-
+  DBusWatch *watch;
   DBusConnection *connection;
 
-  GList *poll_fds;
-  GHashTable *watches;
-};
-
-static gboolean gdbus_connection_prepare  (GSource     *source,
-					   gint        *timeout);
-static gboolean gdbus_connection_check    (GSource     *source);
-static gboolean gdbus_connection_dispatch (GSource     *source,
-					   GSourceFunc  callback,
-					   gpointer     user_data);
-
-static GSourceFuncs dbus_funcs = {
-  gdbus_connection_prepare,
-  gdbus_connection_check,
-  gdbus_connection_dispatch,
-  NULL
-};
+  guint tag;
+} WatchCallback;
 
 static gboolean
-gdbus_connection_prepare (GSource     *source,
-			  gint        *timeout)
+watch_callback (GIOChannel *source,
+		GIOCondition condition,
+		gpointer data)
 {
-  DBusConnection *connection = ((DBusGSource *)source)->connection;
+  WatchCallback *cb = data;
+  unsigned int flags = 0;
+
+  if (condition & G_IO_IN)
+    flags |= DBUS_WATCH_READABLE;
+  if (condition & G_IO_OUT)
+    flags |= DBUS_WATCH_WRITABLE;
+  if (condition & G_IO_ERR)
+    flags |= DBUS_WATCH_ERROR;
+  if (condition & G_IO_HUP)
+    flags |= DBUS_WATCH_HANGUP;
+
+  dbus_connection_handle_watch (cb->connection,
+				cb->watch,
+				flags);
+
+  /* Dispatch messages */
+  while (dbus_connection_dispatch_message (cb->connection));
   
-  *timeout = -1;
-
-  return (dbus_connection_peek_message (connection) != NULL);
-}
-
-static gboolean
-gdbus_connection_check (GSource *source)
-{
-  DBusGSource *dbus_source = (DBusGSource *)source;
-  GList *list;
-
-  list = dbus_source->poll_fds;
-
-  while (list)
-    {
-      GPollFD *poll_fd = list->data;
-
-      if (poll_fd->revents != 0)
-	return TRUE;
-
-      list = list->next;
-    }
-
-  return FALSE;
+  return TRUE;
 }
 
-static gboolean
-gdbus_connection_dispatch (GSource     *source,
-			   GSourceFunc  callback,
-			   gpointer     user_data)
+static void
+free_callback_data (WatchCallback *cb)
 {
-   DBusGSource *dbus_source = (DBusGSource *)source;
-   DBusMessageFunction handler = (DBusMessageFunction)callback;
-   DBusMessage *message;
-   
-   GList *list;
-
-   list = dbus_source->poll_fds;
-
-   while (list)
-     {
-       GPollFD *poll_fd = list->data;
-
-       g_print ("poll_fd is: %p\n", poll_fd);
-       if (poll_fd->revents != 0)
-	 {
-	   DBusWatch *watch = g_hash_table_lookup (dbus_source->watches, poll_fd);
-	   guint condition = 0;
-
-	   if (poll_fd->revents & G_IO_IN)
-	     condition |= DBUS_WATCH_READABLE;
-	   if (poll_fd->revents & G_IO_OUT)
-	     condition |= DBUS_WATCH_WRITABLE;
-	   if (poll_fd->revents & G_IO_ERR)
-	     condition |= DBUS_WATCH_ERROR;
-	   if (poll_fd->revents & G_IO_HUP)
-	     condition |= DBUS_WATCH_HANGUP;
-	   
-	   dbus_connection_handle_watch (dbus_source->connection, watch, condition);
-	 }
-       
-       list = list->next;
-     }
-
-   while ((message = dbus_connection_pop_message (dbus_source->connection)))
-     {
-       handler (dbus_source->connection, message, user_data);
-
-       dbus_message_unref (message);
-     }
-   
-   return TRUE;
+  dbus_connection_unref (cb->connection);
+  g_free (cb);
 }
 
 static void
-gdbus_add_connection_watch (DBusWatch      *watch,
-			    DBusGSource    *source)
+add_watch (DBusWatch *watch,
+	   gpointer   data)
 {
-  GPollFD *poll_fd;
-  guint flags;
-  
-  poll_fd = g_new (GPollFD, 1);
-  poll_fd->fd = dbus_watch_get_fd (watch);
+  GIOChannel *channel;
+  DBusConnection *connection = data;
+  GIOCondition condition = 0;
+  WatchCallback *cb;
+  guint tag;
+  gint flags;
 
-  poll_fd->events = 0;
   flags = dbus_watch_get_flags (watch);
-  dbus_watch_set_data (watch, poll_fd, NULL);
+  condition = 0;
   
   if (flags & DBUS_WATCH_READABLE)
-    poll_fd->events |= G_IO_IN;
-
+    condition |= G_IO_IN;
   if (flags & DBUS_WATCH_WRITABLE)
-    poll_fd->events |= G_IO_OUT;
-
-  g_source_add_poll ((GSource *)source, poll_fd);
+    condition |= G_IO_OUT;
+  if (flags & DBUS_WATCH_ERROR)
+    condition |= G_IO_ERR;
+  if (flags & DBUS_WATCH_HANGUP)
+    condition |= G_IO_HUP;
   
-  g_print ("Add connection watch: %p!\n", watch);
-
-  source->poll_fds = g_list_prepend (source->poll_fds, poll_fd);
-  g_hash_table_insert (source->watches, poll_fd, watch);
+  channel = g_io_channel_unix_new (dbus_watch_get_fd (watch));
+  g_io_channel_set_encoding (channel, NULL, NULL);
+  g_io_channel_set_buffered (channel, FALSE);
+
+  cb = g_new0 (WatchCallback, 1);
+  cb->watch = watch;
+  cb->connection = connection;
+  dbus_connection_ref (connection);
+  
+  dbus_watch_set_data (watch, cb, (DBusFreeFunction)free_callback_data);
+  
+  tag = g_io_add_watch (channel, condition, watch_callback, cb);
+  cb->tag = tag;
 }
 
 static void
-gdbus_remove_connection_watch (DBusWatch      *watch,
-			       DBusGSource    *source)
+remove_watch (DBusWatch *watch,
+	      gpointer   data)
 {
-  GPollFD *poll_fd;
+  WatchCallback *cb;
 
-  poll_fd = dbus_watch_get_data (watch);
+  cb = dbus_watch_get_data (watch);
 
-  source->poll_fds = g_list_remove (source->poll_fds, poll_fd);
-  g_hash_table_remove (source->watches, poll_fd);
-  g_source_remove_poll ((GSource *)source, poll_fd);
+  g_source_remove (cb->tag);
   
-  g_free (poll_fd);
+  dbus_watch_set_data (watch, NULL, NULL);
 }
 
-GSource *
-dbus_connection_gsource_new (DBusConnection *connection)
+static void
+add_timeout (DBusTimeout *timeout,
+	     void        *data)
 {
-  GSource *source = g_source_new (&dbus_funcs, sizeof (DBusGSource));
-  DBusGSource *dbus_source = (DBusGSource *)source;
+}
 
-  dbus_source->watches = g_hash_table_new (NULL, NULL);
-  dbus_source->connection = connection;
-  dbus_connection_ref (dbus_source->connection);
-  
-  dbus_connection_set_watch_functions (connection,
-                                       (DBusAddWatchFunction) gdbus_add_connection_watch,
-                                       (DBusRemoveWatchFunction) gdbus_remove_connection_watch,
-                                       dbus_source,
-                                       NULL);
+static void
+remove_timeout (DBusTimeout *timeout,
+		void        *data)
+{
+}
 
+void
+dbus_connection_hookup_with_g_main (DBusConnection *connection)
+{
 
-  return source;
+  dbus_connection_set_watch_functions (connection,
+				       add_watch,
+				       remove_watch,
+				       connection, NULL);
+  dbus_connection_set_timeout_functions (connection,
+					 add_timeout,
+					 remove_timeout,
+					 NULL, NULL);
+					 
 }
diff --git a/glib/test-dbus-glib.c b/glib/test-dbus-glib.c
index cdce2899..ad147e50 100644
--- a/glib/test-dbus-glib.c
+++ b/glib/test-dbus-glib.c
@@ -1,37 +1,14 @@
 #include "dbus-glib.h"
 #include <stdio.h>
 
-GMainLoop *loop;
-
-static void
-message_handler (DBusConnection *connection,
-		 DBusMessage *message, gpointer user_data)
-{
-  static int count = 0;
-  DBusMessage *reply;
-
-  reply = dbus_message_new ("org.freedesktop.DBus.Test", "org.freedesktop.DBus.Test");
-  dbus_connection_send_message (connection,
-				reply,
-				NULL);
-  dbus_message_unref (reply);
-  count += 1;
-  
-  if (count > 100)
-    {
-      printf ("Saw %d messages, exiting\n", count);
-      g_main_loop_quit (loop);
-    }
-}
-
 int
 main (int argc, char **argv)
 {
-  GSource *source;
-  
   DBusConnection *connection;
   DBusResultCode result;
-  DBusMessage *message;
+  DBusMessage *message, *reply;
+  
+  GMainLoop *loop;
   
   loop = g_main_loop_new (NULL, FALSE);
 
@@ -43,16 +20,15 @@ main (int argc, char **argv)
       return 1;
     }
 
-  source = dbus_connection_gsource_new (connection);
-  g_source_attach (source, NULL);
-  g_source_set_callback (source, (GSourceFunc)message_handler, NULL, NULL);
-  
-  message = dbus_message_new ("org.freedesktop.DBus.Test", "org.freedesktop.DBus.Test");
-  dbus_connection_send_message (connection,
-                                message,
-                                NULL);
-  dbus_message_unref (message);
-  
+  dbus_connection_hookup_with_g_main (connection);
+
+  message = dbus_message_new ("org.freedesktop.DBus", "org.freedesktop.DBus.Hello");
+  dbus_message_append_fields (message,
+			      DBUS_TYPE_STRING, "glib-test",
+			      0);
+
+  reply = dbus_connection_send_message_with_reply_and_block (connection, message, -1, &result);
+  g_print ("reply name: %s\n", dbus_message_get_name (reply));
   
   g_main_loop_run (loop);
   
diff --git a/test/echo-client.c b/test/echo-client.c
index fe2fa713..47d7314d 100644
--- a/test/echo-client.c
+++ b/test/echo-client.c
@@ -30,6 +30,7 @@ main (int    argc,
   message = dbus_message_new ("org.freedesktop.DBus.Test", "org.freedesktop.DBus.Test");
   dbus_connection_send_message (connection,
                                 message,
+				NULL, 
                                 NULL);
   dbus_message_unref (message);
   
diff --git a/test/watch.c b/test/watch.c
index 71d9cdc2..a885a75c 100644
--- a/test/watch.c
+++ b/test/watch.c
@@ -151,6 +151,7 @@ check_messages (void)
           reply = dbus_message_new ("org.freedesktop.DBus.Test", "org.freedesktop.DBus.Test");
           dbus_connection_send_message (connection,
                                         reply,
+					NULL, 
                                         NULL);
           dbus_message_unref (reply);