gio/gdbusprivate.c

   1 /* GDBus - GLib D-Bus Library
   2  *
   3  * Copyright (C) 2008-2010 Red Hat, Inc.
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Lesser General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Lesser General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Lesser General
  16  * Public License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
  18  * Boston, MA 02111-1307, USA.
  19  *
  20  * Author: David Zeuthen <davidz@redhat.com>
  21  */
  22
  23 #include "config.h"
  24
  25 #include <stdlib.h>
  26 #include <string.h>
  27 #ifdef HAVE_UNISTD_H
  28 #include <unistd.h>
  29 #endif
  30
  31 #include "giotypes.h"
  32 #include "gsocket.h"
  33 #include "gdbusprivate.h"
  34 #include "gdbusmessage.h"
  35 #include "gdbuserror.h"
  36 #include "gdbusintrospection.h"
  37 #include "gasyncresult.h"
  38 #include "gsimpleasyncresult.h"
  39 #include "ginputstream.h"
  40 #include "giostream.h"
  41 #include "gsocketcontrolmessage.h"
  42 #include "gsocketconnection.h"
  43 #include "gsocketoutputstream.h"
  44
  45 #ifdef G_OS_UNIX
  46 #include "gunixfdmessage.h"
  47 #include "gunixconnection.h"
  48 #include "gunixcredentialsmessage.h"
  49 #endif
  50
  51 #ifdef G_OS_WIN32
  52 #include <windows.h>
  53 #endif
  54
  55 #include "glibintl.h"
  56
  57 /* ---------------------------------------------------------------------------------------------------- */
  58
  59 gchar *
  60 _g_dbus_hexdump (const gchar *data, gsize len, guint indent)
  61 {
  62  guint n, m;
  63  GString *ret;
  64
  65  ret = g_string_new (NULL);
  66
  67  for (n = 0; n < len; n += 16)
  68    {
  69      g_string_append_printf (ret, "%*s%04x: ", indent, "", n);
  70
  71      for (m = n; m < n + 16; m++)
  72        {
  73          if (m > n && (m%4) == 0)
  74            g_string_append_c (ret, ' ');
  75          if (m < len)
  76            g_string_append_printf (ret, "%02x ", (guchar) data[m]);
  77          else
  78            g_string_append (ret, "   ");
  79        }
  80
  81      g_string_append (ret, "   ");
  82
  83      for (m = n; m < len && m < n + 16; m++)
  84        g_string_append_c (ret, g_ascii_isprint (data[m]) ? data[m] : '.');
  85
  86      g_string_append_c (ret, '\n');
  87    }
  88
  89  return g_string_free (ret, FALSE);
  90 }
  91
  92 /* ---------------------------------------------------------------------------------------------------- */
  93
  94 /* Unfortunately ancillary messages are discarded when reading from a
  95  * socket using the GSocketInputStream abstraction. So we provide a
  96  * very GInputStream-ish API that uses GSocket in this case (very
  97  * similar to GSocketInputStream).
  98  */
  99
 100 typedef struct
 101 {
 102   GSocket *socket;
 103   GCancellable *cancellable;
 104
 105   void *buffer;
 106   gsize count;
 107
 108   GSocketControlMessage ***messages;
 109   gint *num_messages;
 110
 111   GSimpleAsyncResult *simple;
 112
 113   gboolean from_mainloop;
 114 } ReadWithControlData;
 115
 116 static void
 117 read_with_control_data_free (ReadWithControlData *data)
 118 {
 119   g_object_unref (data->socket);
 120   if (data->cancellable != NULL)
 121     g_object_unref (data->cancellable);
 122   g_object_unref (data->simple);
 123   g_free (data);
 124 }
 125
 126 static gboolean
 127 _g_socket_read_with_control_messages_ready (GSocket      *socket,
 128                                             GIOCondition  condition,
 129                                             gpointer      user_data)
 130 {
 131   ReadWithControlData *data = user_data;
 132   GError *error;
 133   gssize result;
 134   GInputVector vector;
 135
 136   error = NULL;
 137   vector.buffer = data->buffer;
 138   vector.size = data->count;
 139   result = g_socket_receive_message (data->socket,
 140                                      NULL, /* address */
 141                                      &vector,
 142                                      1,
 143                                      data->messages,
 144                                      data->num_messages,
 145                                      NULL,
 146                                      data->cancellable,
 147                                      &error);
 148   if (result >= 0)
 149     {
 150       g_simple_async_result_set_op_res_gssize (data->simple, result);
 151     }
 152   else
 153     {
 154       g_assert (error != NULL);
 155       g_simple_async_result_set_from_error (data->simple, error);
 156       g_error_free (error);
 157     }
 158
 159   if (data->from_mainloop)
 160     g_simple_async_result_complete (data->simple);
 161   else
 162     g_simple_async_result_complete_in_idle (data->simple);
 163
 164   return FALSE;
 165 }
 166
 167 static void
 168 _g_socket_read_with_control_messages (GSocket                 *socket,
 169                                       void                    *buffer,
 170                                       gsize                    count,
 171                                       GSocketControlMessage ***messages,
 172                                       gint                    *num_messages,
 173                                       gint                     io_priority,
 174                                       GCancellable            *cancellable,
 175                                       GAsyncReadyCallback      callback,
 176                                       gpointer                 user_data)
 177 {
 178   ReadWithControlData *data;
 179
 180   data = g_new0 (ReadWithControlData, 1);
 181   data->socket = g_object_ref (socket);
 182   data->cancellable = cancellable != NULL ? g_object_ref (cancellable) : NULL;
 183   data->buffer = buffer;
 184   data->count = count;
 185   data->messages = messages;
 186   data->num_messages = num_messages;
 187
 188   data->simple = g_simple_async_result_new (G_OBJECT (socket),
 189                                             callback,
 190                                             user_data,
 191                                             _g_socket_read_with_control_messages);
 192
 193   if (!g_socket_condition_check (socket, G_IO_IN))
 194     {
 195       GSource *source;
 196       data->from_mainloop = TRUE;
 197       source = g_socket_create_source (data->socket,
 198                                        G_IO_IN | G_IO_HUP | G_IO_ERR,
 199                                        cancellable);
 200       g_source_set_callback (source,
 201                              (GSourceFunc) _g_socket_read_with_control_messages_ready,
 202                              data,
 203                              (GDestroyNotify) read_with_control_data_free);
 204       g_source_attach (source, g_main_context_get_thread_default ());
 205       g_source_unref (source);
 206     }
 207   else
 208     {
 209       _g_socket_read_with_control_messages_ready (data->socket, G_IO_IN, data);
 210       read_with_control_data_free (data);
 211     }
 212 }
 213
 214 static gssize
 215 _g_socket_read_with_control_messages_finish (GSocket       *socket,
 216                                              GAsyncResult  *result,
 217                                              GError       **error)
 218 {
 219   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
 220
 221   g_return_val_if_fail (G_IS_SOCKET (socket), -1);
 222   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == _g_socket_read_with_control_messages);
 223
 224   if (g_simple_async_result_propagate_error (simple, error))
 225       return -1;
 226   else
 227     return g_simple_async_result_get_op_res_gssize (simple);
 228 }
 229
 230 /* ---------------------------------------------------------------------------------------------------- */
 231
 232 G_LOCK_DEFINE_STATIC (shared_thread_lock);
 233
 234 typedef struct
 235 {
 236   gint num_users;
 237   GThread *thread;
 238   GMainContext *context;
 239   GMainLoop *loop;
 240 } SharedThreadData;
 241
 242 static SharedThreadData *shared_thread_data = NULL;
 243
 244 static gpointer
 245 shared_thread_func (gpointer data)
 246 {
 247   g_main_context_push_thread_default (shared_thread_data->context);
 248   g_main_loop_run (shared_thread_data->loop);
 249   g_main_context_pop_thread_default (shared_thread_data->context);
 250   return NULL;
 251 }
 252
 253 typedef void (*GDBusSharedThreadFunc) (gpointer user_data);
 254
 255 typedef struct
 256 {
 257   GDBusSharedThreadFunc func;
 258   gpointer              user_data;
 259   gboolean              done;
 260 } CallerData;
 261
 262 static gboolean
 263 invoke_caller (gpointer user_data)
 264 {
 265   CallerData *data = user_data;
 266   data->func (data->user_data);
 267   data->done = TRUE;
 268   return FALSE;
 269 }
 270
 271 static void
 272 _g_dbus_shared_thread_ref (GDBusSharedThreadFunc func,
 273                            gpointer              user_data)
 274 {
 275   GError *error;
 276   GSource *idle_source;
 277   CallerData *data;
 278
 279   G_LOCK (shared_thread_lock);
 280
 281   if (shared_thread_data != NULL)
 282     {
 283       shared_thread_data->num_users += 1;
 284       goto have_thread;
 285     }
 286
 287   shared_thread_data = g_new0 (SharedThreadData, 1);
 288   shared_thread_data->num_users = 1;
 289
 290   error = NULL;
 291   shared_thread_data->context = g_main_context_new ();
 292   shared_thread_data->loop = g_main_loop_new (shared_thread_data->context, FALSE);
 293   shared_thread_data->thread = g_thread_create (shared_thread_func,
 294                                                 NULL,
 295                                                 TRUE,
 296                                                 &error);
 297   g_assert_no_error (error);
 298
 299  have_thread:
 300
 301   data = g_new0 (CallerData, 1);
 302   data->func = func;
 303   data->user_data = user_data;
 304   data->done = FALSE;
 305
 306   idle_source = g_idle_source_new ();
 307   g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
 308   g_source_set_callback (idle_source,
 309                          invoke_caller,
 310                          data,
 311                          NULL);
 312   g_source_attach (idle_source, shared_thread_data->context);
 313   g_source_unref (idle_source);
 314
 315   /* wait for the user code to run.. hmm.. probably use a condition variable instead */
 316   while (!data->done)
 317     g_thread_yield ();
 318
 319   g_free (data);
 320
 321   G_UNLOCK (shared_thread_lock);
 322 }
 323
 324 static void
 325 _g_dbus_shared_thread_unref (void)
 326 {
 327   /* TODO: actually destroy the shared thread here */
 328 #if 0
 329   G_LOCK (shared_thread_lock);
 330   g_assert (shared_thread_data != NULL);
 331   shared_thread_data->num_users -= 1;
 332   if (shared_thread_data->num_users == 0)
 333     {
 334       g_main_loop_quit (shared_thread_data->loop);
 335       //g_thread_join (shared_thread_data->thread);
 336       g_main_loop_unref (shared_thread_data->loop);
 337       g_main_context_unref (shared_thread_data->context);
 338       g_free (shared_thread_data);
 339       shared_thread_data = NULL;
 340       G_UNLOCK (shared_thread_lock);
 341     }
 342   else
 343     {
 344       G_UNLOCK (shared_thread_lock);
 345     }
 346 #endif
 347 }
 348
 349 /* ---------------------------------------------------------------------------------------------------- */
 350
 351 struct GDBusWorker
 352 {
 353   volatile gint                       ref_count;
 354
 355   gboolean                            stopped;
 356
 357   /* TODO: frozen (e.g. G_DBUS_CONNECTION_FLAGS_DELAY_MESSAGE_PROCESSING) currently
 358    * only affects messages received from the other peer (since GDBusServer is the
 359    * only user) - we might want it to affect messages sent to the other peer too?
 360    */
 361   gboolean                            frozen;
 362   GQueue                             *received_messages_while_frozen;
 363
 364   GIOStream                          *stream;
 365   GDBusCapabilityFlags                capabilities;
 366   GCancellable                       *cancellable;
 367   GDBusWorkerMessageReceivedCallback  message_received_callback;
 368   GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback;
 369   GDBusWorkerDisconnectedCallback     disconnected_callback;
 370   gpointer                            user_data;
 371
 372   GThread                            *thread;
 373
 374   /* if not NULL, stream is GSocketConnection */
 375   GSocket *socket;
 376
 377   /* used for reading */
 378   GMutex                             *read_lock;
 379   gchar                              *read_buffer;
 380   gsize                               read_buffer_allocated_size;
 381   gsize                               read_buffer_cur_size;
 382   gsize                               read_buffer_bytes_wanted;
 383   GUnixFDList                        *read_fd_list;
 384   GSocketControlMessage             **read_ancillary_messages;
 385   gint                                read_num_ancillary_messages;
 386
 387   /* used for writing */
 388   GMutex                             *write_lock;
 389   GQueue                             *write_queue;
 390   gint                                num_writes_pending;
 391   guint64                             write_num_messages_written;
 392   GList                              *write_pending_flushes;
 393 };
 394
 395 /* ---------------------------------------------------------------------------------------------------- */
 396
 397 typedef struct
 398 {
 399   GMutex *mutex;
 400   GCond *cond;
 401   guint64 number_to_wait_for;
 402   GError *error;
 403 } FlushData;
 404
 405 struct _MessageToWriteData ;
 406 typedef struct _MessageToWriteData MessageToWriteData;
 407
 408 static void message_to_write_data_free (MessageToWriteData *data);
 409
 410 static void read_message_print_transport_debug (gssize bytes_read,
 411                                                 GDBusWorker *worker);
 412
 413 static void write_message_print_transport_debug (gssize bytes_written,
 414                                                  MessageToWriteData *data);
 415
 416 /* ---------------------------------------------------------------------------------------------------- */
 417
 418 static GDBusWorker *
 419 _g_dbus_worker_ref (GDBusWorker *worker)
 420 {
 421   g_atomic_int_inc (&worker->ref_count);
 422   return worker;
 423 }
 424
 425 static void
 426 _g_dbus_worker_unref (GDBusWorker *worker)
 427 {
 428   if (g_atomic_int_dec_and_test (&worker->ref_count))
 429     {
 430       g_assert (worker->write_pending_flushes == NULL);
 431
 432       _g_dbus_shared_thread_unref ();
 433
 434       g_object_unref (worker->stream);
 435
 436       g_mutex_free (worker->read_lock);
 437       g_object_unref (worker->cancellable);
 438       if (worker->read_fd_list != NULL)
 439         g_object_unref (worker->read_fd_list);
 440
 441       g_queue_foreach (worker->received_messages_while_frozen, (GFunc) g_object_unref, NULL);
 442       g_queue_free (worker->received_messages_while_frozen);
 443
 444       g_mutex_free (worker->write_lock);
 445       g_queue_foreach (worker->write_queue, (GFunc) message_to_write_data_free, NULL);
 446       g_queue_free (worker->write_queue);
 447
 448       g_free (worker->read_buffer);
 449
 450       g_free (worker);
 451     }
 452 }
 453
 454 static void
 455 _g_dbus_worker_emit_disconnected (GDBusWorker  *worker,
 456                                   gboolean      remote_peer_vanished,
 457                                   GError       *error)
 458 {
 459   if (!worker->stopped)
 460     worker->disconnected_callback (worker, remote_peer_vanished, error, worker->user_data);
 461 }
 462
 463 static void
 464 _g_dbus_worker_emit_message_received (GDBusWorker  *worker,
 465                                       GDBusMessage *message)
 466 {
 467   if (!worker->stopped)
 468     worker->message_received_callback (worker, message, worker->user_data);
 469 }
 470
 471 static gboolean
 472 _g_dbus_worker_emit_message_about_to_be_sent (GDBusWorker  *worker,
 473                                               GDBusMessage *message)
 474 {
 475   gboolean ret;
 476   ret = FALSE;
 477   if (!worker->stopped)
 478     ret = worker->message_about_to_be_sent_callback (worker, message, worker->user_data);
 479   return ret;
 480 }
 481
 482 /* can only be called from private thread with read-lock held - takes ownership of @message */
 483 static void
 484 _g_dbus_worker_queue_or_deliver_received_message (GDBusWorker  *worker,
 485                                                   GDBusMessage *message)
 486 {
 487   if (worker->frozen || g_queue_get_length (worker->received_messages_while_frozen) > 0)
 488     {
 489       /* queue up */
 490       g_queue_push_tail (worker->received_messages_while_frozen, message);
 491     }
 492   else
 493     {
 494       /* not frozen, nor anything in queue */
 495       _g_dbus_worker_emit_message_received (worker, message);
 496       g_object_unref (message);
 497     }
 498 }
 499
 500 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 501 static gboolean
 502 unfreeze_in_idle_cb (gpointer user_data)
 503 {
 504   GDBusWorker *worker = user_data;
 505   GDBusMessage *message;
 506
 507   g_mutex_lock (worker->read_lock);
 508   if (worker->frozen)
 509     {
 510       while ((message = g_queue_pop_head (worker->received_messages_while_frozen)) != NULL)
 511         {
 512           _g_dbus_worker_emit_message_received (worker, message);
 513           g_object_unref (message);
 514         }
 515       worker->frozen = FALSE;
 516     }
 517   else
 518     {
 519       g_assert (g_queue_get_length (worker->received_messages_while_frozen) == 0);
 520     }
 521   g_mutex_unlock (worker->read_lock);
 522   return FALSE;
 523 }
 524
 525 /* can be called from any thread */
 526 void
 527 _g_dbus_worker_unfreeze (GDBusWorker *worker)
 528 {
 529   GSource *idle_source;
 530   idle_source = g_idle_source_new ();
 531   g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
 532   g_source_set_callback (idle_source,
 533                          unfreeze_in_idle_cb,
 534                          _g_dbus_worker_ref (worker),
 535                          (GDestroyNotify) _g_dbus_worker_unref);
 536   g_source_attach (idle_source, shared_thread_data->context);
 537   g_source_unref (idle_source);
 538 }
 539
 540 /* ---------------------------------------------------------------------------------------------------- */
 541
 542 static void _g_dbus_worker_do_read_unlocked (GDBusWorker *worker);
 543
 544 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 545 static void
 546 _g_dbus_worker_do_read_cb (GInputStream  *input_stream,
 547                            GAsyncResult  *res,
 548                            gpointer       user_data)
 549 {
 550   GDBusWorker *worker = user_data;
 551   GError *error;
 552   gssize bytes_read;
 553
 554   g_mutex_lock (worker->read_lock);
 555
 556   /* If already stopped, don't even process the reply */
 557   if (worker->stopped)
 558     goto out;
 559
 560   error = NULL;
 561   if (worker->socket == NULL)
 562     bytes_read = g_input_stream_read_finish (g_io_stream_get_input_stream (worker->stream),
 563                                              res,
 564                                              &error);
 565   else
 566     bytes_read = _g_socket_read_with_control_messages_finish (worker->socket,
 567                                                               res,
 568                                                               &error);
 569   if (worker->read_num_ancillary_messages > 0)
 570     {
 571       gint n;
 572       for (n = 0; n < worker->read_num_ancillary_messages; n++)
 573         {
 574           GSocketControlMessage *control_message = G_SOCKET_CONTROL_MESSAGE (worker->read_ancillary_messages[n]);
 575
 576           if (FALSE)
 577             {
 578             }
 579 #ifdef G_OS_UNIX
 580           else if (G_IS_UNIX_FD_MESSAGE (control_message))
 581             {
 582               GUnixFDMessage *fd_message;
 583               gint *fds;
 584               gint num_fds;
 585
 586               fd_message = G_UNIX_FD_MESSAGE (control_message);
 587               fds = g_unix_fd_message_steal_fds (fd_message, &num_fds);
 588               if (worker->read_fd_list == NULL)
 589                 {
 590                   worker->read_fd_list = g_unix_fd_list_new_from_array (fds, num_fds);
 591                 }
 592               else
 593                 {
 594                   gint n;
 595                   for (n = 0; n < num_fds; n++)
 596                     {
 597                       /* TODO: really want a append_steal() */
 598                       g_unix_fd_list_append (worker->read_fd_list, fds[n], NULL);
 599                       close (fds[n]);
 600                     }
 601                 }
 602               g_free (fds);
 603             }
 604           else if (G_IS_UNIX_CREDENTIALS_MESSAGE (control_message))
 605             {
 606               /* do nothing */
 607             }
 608 #endif
 609           else
 610             {
 611               if (error == NULL)
 612                 {
 613                   g_set_error (&error,
 614                                G_IO_ERROR,
 615                                G_IO_ERROR_FAILED,
 616                                "Unexpected ancillary message of type %s received from peer",
 617                                g_type_name (G_TYPE_FROM_INSTANCE (control_message)));
 618                   _g_dbus_worker_emit_disconnected (worker, TRUE, error);
 619                   g_error_free (error);
 620                   g_object_unref (control_message);
 621                   n++;
 622                   while (n < worker->read_num_ancillary_messages)
 623                     g_object_unref (worker->read_ancillary_messages[n++]);
 624                   g_free (worker->read_ancillary_messages);
 625                   goto out;
 626                 }
 627             }
 628           g_object_unref (control_message);
 629         }
 630       g_free (worker->read_ancillary_messages);
 631     }
 632
 633   if (bytes_read == -1)
 634     {
 635       _g_dbus_worker_emit_disconnected (worker, TRUE, error);
 636       g_error_free (error);
 637       goto out;
 638     }
 639
 640 #if 0
 641   g_debug ("read %d bytes (is_closed=%d blocking=%d condition=0x%02x) stream %p, %p",
 642            (gint) bytes_read,
 643            g_socket_is_closed (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream))),
 644            g_socket_get_blocking (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream))),
 645            g_socket_condition_check (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream)),
 646                                      G_IO_IN | G_IO_OUT | G_IO_HUP),
 647            worker->stream,
 648            worker);
 649 #endif
 650
 651   /* TODO: hmm, hmm... */
 652   if (bytes_read == 0)
 653     {
 654       g_set_error (&error,
 655                    G_IO_ERROR,
 656                    G_IO_ERROR_FAILED,
 657                    "Underlying GIOStream returned 0 bytes on an async read");
 658       _g_dbus_worker_emit_disconnected (worker, TRUE, error);
 659       g_error_free (error);
 660       goto out;
 661     }
 662
 663   read_message_print_transport_debug (bytes_read, worker);
 664
 665   worker->read_buffer_cur_size += bytes_read;
 666   if (worker->read_buffer_bytes_wanted == worker->read_buffer_cur_size)
 667     {
 668       /* OK, got what we asked for! */
 669       if (worker->read_buffer_bytes_wanted == 16)
 670         {
 671           gssize message_len;
 672           /* OK, got the header - determine how many more bytes are needed */
 673           error = NULL;
 674           message_len = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer,
 675                                                      16,
 676                                                      &error);
 677           if (message_len == -1)
 678             {
 679               g_warning ("_g_dbus_worker_do_read_cb: error determing bytes needed: %s", error->message);
 680               _g_dbus_worker_emit_disconnected (worker, FALSE, error);
 681               g_error_free (error);
 682               goto out;
 683             }
 684
 685           worker->read_buffer_bytes_wanted = message_len;
 686           _g_dbus_worker_do_read_unlocked (worker);
 687         }
 688       else
 689         {
 690           GDBusMessage *message;
 691           error = NULL;
 692
 693           /* TODO: use connection->priv->auth to decode the message */
 694
 695           message = g_dbus_message_new_from_blob ((guchar *) worker->read_buffer,
 696                                                   worker->read_buffer_cur_size,
 697                                                   worker->capabilities,
 698                                                   &error);
 699           if (message == NULL)
 700             {
 701               gchar *s;
 702               s = _g_dbus_hexdump (worker->read_buffer, worker->read_buffer_cur_size, 2);
 703               g_warning ("Error decoding D-Bus message of %" G_GSIZE_FORMAT " bytes\n"
 704                          "The error is: %s\n"
 705                          "The payload is as follows:\n"
 706                          "%s\n",
 707                          worker->read_buffer_cur_size,
 708                          error->message,
 709                          s);
 710               g_free (s);
 711               _g_dbus_worker_emit_disconnected (worker, FALSE, error);
 712               g_error_free (error);
 713               goto out;
 714             }
 715
 716 #ifdef G_OS_UNIX
 717           if (worker->read_fd_list != NULL)
 718             {
 719               g_dbus_message_set_unix_fd_list (message, worker->read_fd_list);
 720               worker->read_fd_list = NULL;
 721             }
 722 #endif
 723
 724           if (G_UNLIKELY (_g_dbus_debug_message ()))
 725             {
 726               gchar *s;
 727               _g_dbus_debug_print_lock ();
 728               g_print ("========================================================================\n"
 729                        "GDBus-debug:Message:\n"
 730                        "  <<<< RECEIVED D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
 731                        worker->read_buffer_cur_size);
 732               s = g_dbus_message_print (message, 2);
 733               g_print ("%s", s);
 734               g_free (s);
 735               if (G_UNLIKELY (_g_dbus_debug_payload ()))
 736                 {
 737                   s = _g_dbus_hexdump (worker->read_buffer, worker->read_buffer_cur_size, 2);
 738                   g_print ("%s\n", s);
 739                   g_free (s);
 740                 }
 741               _g_dbus_debug_print_unlock ();
 742             }
 743
 744           /* yay, got a message, go deliver it */
 745           _g_dbus_worker_queue_or_deliver_received_message (worker, message);
 746
 747           /* start reading another message! */
 748           worker->read_buffer_bytes_wanted = 0;
 749           worker->read_buffer_cur_size = 0;
 750           _g_dbus_worker_do_read_unlocked (worker);
 751         }
 752     }
 753   else
 754     {
 755       /* didn't get all the bytes we requested - so repeat the request... */
 756       _g_dbus_worker_do_read_unlocked (worker);
 757     }
 758
 759  out:
 760   g_mutex_unlock (worker->read_lock);
 761
 762   /* gives up the reference acquired when calling g_input_stream_read_async() */
 763   _g_dbus_worker_unref (worker);
 764 }
 765
 766 /* called in private thread shared by all GDBusConnection instances (with read-lock held) */
 767 static void
 768 _g_dbus_worker_do_read_unlocked (GDBusWorker *worker)
 769 {
 770   /* if bytes_wanted is zero, it means start reading a message */
 771   if (worker->read_buffer_bytes_wanted == 0)
 772     {
 773       worker->read_buffer_cur_size = 0;
 774       worker->read_buffer_bytes_wanted = 16;
 775     }
 776
 777   /* ensure we have a (big enough) buffer */
 778   if (worker->read_buffer == NULL || worker->read_buffer_bytes_wanted > worker->read_buffer_allocated_size)
 779     {
 780       /* TODO: 4096 is randomly chosen; might want a better chosen default minimum */
 781       worker->read_buffer_allocated_size = MAX (worker->read_buffer_bytes_wanted, 4096);
 782       worker->read_buffer = g_realloc (worker->read_buffer, worker->read_buffer_allocated_size);
 783     }
 784
 785   if (worker->socket == NULL)
 786     g_input_stream_read_async (g_io_stream_get_input_stream (worker->stream),
 787                                worker->read_buffer + worker->read_buffer_cur_size,
 788                                worker->read_buffer_bytes_wanted - worker->read_buffer_cur_size,
 789                                G_PRIORITY_DEFAULT,
 790                                worker->cancellable,
 791                                (GAsyncReadyCallback) _g_dbus_worker_do_read_cb,
 792                                _g_dbus_worker_ref (worker));
 793   else
 794     {
 795       worker->read_ancillary_messages = NULL;
 796       worker->read_num_ancillary_messages = 0;
 797       _g_socket_read_with_control_messages (worker->socket,
 798                                             worker->read_buffer + worker->read_buffer_cur_size,
 799                                             worker->read_buffer_bytes_wanted - worker->read_buffer_cur_size,
 800                                             &worker->read_ancillary_messages,
 801                                             &worker->read_num_ancillary_messages,
 802                                             G_PRIORITY_DEFAULT,
 803                                             worker->cancellable,
 804                                             (GAsyncReadyCallback) _g_dbus_worker_do_read_cb,
 805                                             _g_dbus_worker_ref (worker));
 806     }
 807 }
 808
 809 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 810 static void
 811 _g_dbus_worker_do_read (GDBusWorker *worker)
 812 {
 813   g_mutex_lock (worker->read_lock);
 814   _g_dbus_worker_do_read_unlocked (worker);
 815   g_mutex_unlock (worker->read_lock);
 816 }
 817
 818 /* ---------------------------------------------------------------------------------------------------- */
 819
 820 struct _MessageToWriteData
 821 {
 822   GDBusWorker  *worker;
 823   GDBusMessage *message;
 824   gchar        *blob;
 825   gsize         blob_size;
 826
 827   gsize               total_written;
 828   GSimpleAsyncResult *simple;
 829
 830 };
 831
 832 static void
 833 message_to_write_data_free (MessageToWriteData *data)
 834 {
 835   _g_dbus_worker_unref (data->worker);
 836   g_object_unref (data->message);
 837   g_free (data->blob);
 838   g_free (data);
 839 }
 840
 841 /* ---------------------------------------------------------------------------------------------------- */
 842
 843 static void write_message_continue_writing (MessageToWriteData *data);
 844
 845 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
 846 static void
 847 write_message_async_cb (GObject      *source_object,
 848                         GAsyncResult *res,
 849                         gpointer      user_data)
 850 {
 851   MessageToWriteData *data = user_data;
 852   GSimpleAsyncResult *simple;
 853   gssize bytes_written;
 854   GError *error;
 855
 856   /* Note: we can't access data->simple after calling g_async_result_complete () because the
 857    * callback can free @data and we're not completing in idle. So use a copy of the pointer.
 858    */
 859   simple = data->simple;
 860
 861   error = NULL;
 862   bytes_written = g_output_stream_write_finish (G_OUTPUT_STREAM (source_object),
 863                                                 res,
 864                                                 &error);
 865   if (bytes_written == -1)
 866     {
 867       g_simple_async_result_set_from_error (simple, error);
 868       g_error_free (error);
 869       g_simple_async_result_complete (simple);
 870       g_object_unref (simple);
 871       goto out;
 872     }
 873   g_assert (bytes_written > 0); /* zero is never returned */
 874
 875   write_message_print_transport_debug (bytes_written, data);
 876
 877   data->total_written += bytes_written;
 878   g_assert (data->total_written <= data->blob_size);
 879   if (data->total_written == data->blob_size)
 880     {
 881       g_simple_async_result_complete (simple);
 882       g_object_unref (simple);
 883       goto out;
 884     }
 885
 886   write_message_continue_writing (data);
 887
 888  out:
 889   ;
 890 }
 891
 892 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
 893 static gboolean
 894 on_socket_ready (GSocket      *socket,
 895                  GIOCondition  condition,
 896                  gpointer      user_data)
 897 {
 898   MessageToWriteData *data = user_data;
 899   write_message_continue_writing (data);
 900   return FALSE; /* remove source */
 901 }
 902
 903 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
 904 static void
 905 write_message_continue_writing (MessageToWriteData *data)
 906 {
 907   GOutputStream *ostream;
 908   GSimpleAsyncResult *simple;
 909 #ifdef G_OS_UNIX
 910   GUnixFDList *fd_list;
 911 #endif
 912
 913   /* Note: we can't access data->simple after calling g_async_result_complete () because the
 914    * callback can free @data and we're not completing in idle. So use a copy of the pointer.
 915    */
 916   simple = data->simple;
 917
 918   ostream = g_io_stream_get_output_stream (data->worker->stream);
 919 #ifdef G_OS_UNIX
 920   fd_list = g_dbus_message_get_unix_fd_list (data->message);
 921 #endif
 922
 923   g_assert (!g_output_stream_has_pending (ostream));
 924   g_assert_cmpint (data->total_written, <, data->blob_size);
 925
 926   if (FALSE)
 927     {
 928     }
 929 #ifdef G_OS_UNIX
 930   else if (G_IS_SOCKET_OUTPUT_STREAM (ostream) && data->total_written == 0)
 931     {
 932       GOutputVector vector;
 933       GSocketControlMessage *control_message;
 934       gssize bytes_written;
 935       GError *error;
 936
 937       vector.buffer = data->blob;
 938       vector.size = data->blob_size;
 939
 940       control_message = NULL;
 941       if (fd_list != NULL && g_unix_fd_list_get_length (fd_list) > 0)
 942         {
 943           if (!(data->worker->capabilities & G_DBUS_CAPABILITY_FLAGS_UNIX_FD_PASSING))
 944             {
 945               g_simple_async_result_set_error (simple,
 946                                                G_IO_ERROR,
 947                                                G_IO_ERROR_FAILED,
 948                                                "Tried sending a file descriptor but remote peer does not support this capability");
 949               g_simple_async_result_complete (simple);
 950               g_object_unref (simple);
 951               goto out;
 952             }
 953           control_message = g_unix_fd_message_new_with_fd_list (fd_list);
 954         }
 955
 956       error = NULL;
 957       bytes_written = g_socket_send_message (data->worker->socket,
 958                                              NULL, /* address */
 959                                              &vector,
 960                                              1,
 961                                              control_message != NULL ? &control_message : NULL,
 962                                              control_message != NULL ? 1 : 0,
 963                                              G_SOCKET_MSG_NONE,
 964                                              data->worker->cancellable,
 965                                              &error);
 966       if (control_message != NULL)
 967         g_object_unref (control_message);
 968
 969       if (bytes_written == -1)
 970         {
 971           /* Handle WOULD_BLOCK by waiting until there's room in the buffer */
 972           if (error->domain == G_IO_ERROR && error->code == G_IO_ERROR_WOULD_BLOCK)
 973             {
 974               GSource *source;
 975               source = g_socket_create_source (data->worker->socket,
 976                                                G_IO_OUT | G_IO_HUP | G_IO_ERR,
 977                                                data->worker->cancellable);
 978               g_source_set_callback (source,
 979                                      (GSourceFunc) on_socket_ready,
 980                                      data,
 981                                      NULL); /* GDestroyNotify */
 982               g_source_attach (source, g_main_context_get_thread_default ());
 983               g_source_unref (source);
 984               goto out;
 985             }
 986           g_simple_async_result_set_from_error (simple, error);
 987           g_error_free (error);
 988           g_simple_async_result_complete (simple);
 989           g_object_unref (simple);
 990           goto out;
 991         }
 992       g_assert (bytes_written > 0); /* zero is never returned */
 993
 994       write_message_print_transport_debug (bytes_written, data);
 995
 996       data->total_written += bytes_written;
 997       g_assert (data->total_written <= data->blob_size);
 998       if (data->total_written == data->blob_size)
 999         {
1000           g_simple_async_result_complete (simple);
1001           g_object_unref (simple);
1002           goto out;
1003         }
1004
1005       write_message_continue_writing (data);
1006     }
1007 #endif
1008   else
1009     {
1010       if (fd_list != NULL)
1011         {
1012           g_simple_async_result_set_error (simple,
1013                                            G_IO_ERROR,
1014                                            G_IO_ERROR_FAILED,
1015                                            "Tried sending a file descriptor on unsupported stream of type %s",
1016                                            g_type_name (G_TYPE_FROM_INSTANCE (ostream)));
1017           g_simple_async_result_complete (simple);
1018           g_object_unref (simple);
1019           goto out;
1020         }
1021
1022       g_output_stream_write_async (ostream,
1023                                    (const gchar *) data->blob + data->total_written,
1024                                    data->blob_size - data->total_written,
1025                                    G_PRIORITY_DEFAULT,
1026                                    data->worker->cancellable,
1027                                    write_message_async_cb,
1028                                    data);
1029     }
1030  out:
1031   ;
1032 }
1033
1034 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1035 static void
1036 write_message_async (GDBusWorker         *worker,
1037                      MessageToWriteData  *data,
1038                      GAsyncReadyCallback  callback,
1039                      gpointer             user_data)
1040 {
1041   data->simple = g_simple_async_result_new (NULL,
1042                                             callback,
1043                                             user_data,
1044                                             write_message_async);
1045   data->total_written = 0;
1046   write_message_continue_writing (data);
1047 }
1048
1049 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1050 static gboolean
1051 write_message_finish (GAsyncResult   *res,
1052                       GError        **error)
1053 {
1054   g_warn_if_fail (g_simple_async_result_get_source_tag (G_SIMPLE_ASYNC_RESULT (res)) == write_message_async);
1055   if (g_simple_async_result_propagate_error (G_SIMPLE_ASYNC_RESULT (res), error))
1056     return FALSE;
1057   else
1058     return TRUE;
1059 }
1060 /* ---------------------------------------------------------------------------------------------------- */
1061
1062 static void maybe_write_next_message (GDBusWorker *worker);
1063
1064 typedef struct
1065 {
1066   GDBusWorker *worker;
1067   GList *flushers;
1068 } FlushAsyncData;
1069
1070 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1071 static void
1072 ostream_flush_cb (GObject      *source_object,
1073                   GAsyncResult *res,
1074                   gpointer      user_data)
1075 {
1076   FlushAsyncData *data = user_data;
1077   GError *error;
1078   GList *l;
1079
1080   error = NULL;
1081   g_output_stream_flush_finish (G_OUTPUT_STREAM (source_object),
1082                                 res,
1083                                 &error);
1084
1085   if (error == NULL)
1086     {
1087       if (G_UNLIKELY (_g_dbus_debug_transport ()))
1088         {
1089           _g_dbus_debug_print_lock ();
1090           g_print ("========================================================================\n"
1091                    "GDBus-debug:Transport:\n"
1092                    "  ---- FLUSHED stream of type %s\n",
1093                    g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
1094           _g_dbus_debug_print_unlock ();
1095         }
1096     }
1097
1098   g_assert (data->flushers != NULL);
1099   for (l = data->flushers; l != NULL; l = l->next)
1100     {
1101       FlushData *f = l->data;
1102
1103       f->error = error != NULL ? g_error_copy (error) : NULL;
1104
1105       g_mutex_lock (f->mutex);
1106       g_cond_signal (f->cond);
1107       g_mutex_unlock (f->mutex);
1108     }
1109   g_list_free (data->flushers);
1110
1111   if (error != NULL)
1112     g_error_free (error);
1113
1114   /* OK, cool, finally kick off the next write */
1115   maybe_write_next_message (data->worker);
1116
1117   _g_dbus_worker_unref (data->worker);
1118   g_free (data);
1119 }
1120
1121 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1122 static void
1123 message_written (GDBusWorker *worker,
1124                  MessageToWriteData *message_data)
1125 {
1126   GList *l;
1127   GList *ll;
1128   GList *flushers;
1129
1130   /* first log the fact that we wrote a message */
1131   if (G_UNLIKELY (_g_dbus_debug_message ()))
1132     {
1133       gchar *s;
1134       _g_dbus_debug_print_lock ();
1135       g_print ("========================================================================\n"
1136                "GDBus-debug:Message:\n"
1137                "  >>>> SENT D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
1138                message_data->blob_size);
1139       s = g_dbus_message_print (message_data->message, 2);
1140       g_print ("%s", s);
1141       g_free (s);
1142       if (G_UNLIKELY (_g_dbus_debug_payload ()))
1143         {
1144           s = _g_dbus_hexdump (message_data->blob, message_data->blob_size, 2);
1145           g_print ("%s\n", s);
1146           g_free (s);
1147         }
1148       _g_dbus_debug_print_unlock ();
1149     }
1150
1151   /* then first wake up pending flushes and, if needed, flush the stream */
1152   flushers = NULL;
1153   g_mutex_lock (worker->write_lock);
1154   worker->write_num_messages_written += 1;
1155   for (l = worker->write_pending_flushes; l != NULL; l = ll)
1156     {
1157       FlushData *f = l->data;
1158       ll = l->next;
1159
1160       if (f->number_to_wait_for == worker->write_num_messages_written)
1161         {
1162           flushers = g_list_append (flushers, f);
1163           worker->write_pending_flushes = g_list_delete_link (worker->write_pending_flushes, l);
1164         }
1165     }
1166   g_mutex_unlock (worker->write_lock);
1167
1168   if (flushers != NULL)
1169     {
1170       FlushAsyncData *data;
1171       data = g_new0 (FlushAsyncData, 1);
1172       data->worker = _g_dbus_worker_ref (worker);
1173       data->flushers = flushers;
1174       /* flush the stream before writing the next message */
1175       g_output_stream_flush_async (g_io_stream_get_output_stream (worker->stream),
1176                                    G_PRIORITY_DEFAULT,
1177                                    worker->cancellable,
1178                                    ostream_flush_cb,
1179                                    data);
1180     }
1181   else
1182     {
1183       /* kick off the next write! */
1184       maybe_write_next_message (worker);
1185     }
1186 }
1187
1188 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1189 static void
1190 write_message_cb (GObject       *source_object,
1191                   GAsyncResult  *res,
1192                   gpointer       user_data)
1193 {
1194   MessageToWriteData *data = user_data;
1195   GError *error;
1196
1197   g_mutex_lock (data->worker->write_lock);
1198   data->worker->num_writes_pending -= 1;
1199   g_mutex_unlock (data->worker->write_lock);
1200
1201   error = NULL;
1202   if (!write_message_finish (res, &error))
1203     {
1204       /* TODO: handle */
1205       _g_dbus_worker_emit_disconnected (data->worker, TRUE, error);
1206       g_error_free (error);
1207     }
1208
1209   /* this function will also kick of the next write (it might need to
1210    * flush so writing the next message might happen much later
1211    * e.g. async)
1212    */
1213   message_written (data->worker, data);
1214
1215   message_to_write_data_free (data);
1216 }
1217
1218 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1219 static void
1220 maybe_write_next_message (GDBusWorker *worker)
1221 {
1222   MessageToWriteData *data;
1223
1224  write_next:
1225
1226   g_mutex_lock (worker->write_lock);
1227   data = g_queue_pop_head (worker->write_queue);
1228   if (data != NULL)
1229     worker->num_writes_pending += 1;
1230   g_mutex_unlock (worker->write_lock);
1231
1232   /* Note that write_lock is only used for protecting the @write_queue
1233    * and @num_writes_pending fields of the GDBusWorker struct ... which we
1234    * need to modify from arbitrary threads in _g_dbus_worker_send_message().
1235    *
1236    * Therefore, it's fine to drop it here when calling back into user
1237    * code and then writing the message out onto the GIOStream since this
1238    * function only runs on the worker thread.
1239    */
1240   if (data != NULL)
1241     {
1242       gboolean message_was_dropped;
1243       message_was_dropped = _g_dbus_worker_emit_message_about_to_be_sent (worker, data->message);
1244       if (G_UNLIKELY (message_was_dropped))
1245         {
1246           g_mutex_lock (worker->write_lock);
1247           worker->num_writes_pending -= 1;
1248           g_mutex_unlock (worker->write_lock);
1249           message_to_write_data_free (data);
1250           goto write_next;
1251         }
1252       else
1253         {
1254           write_message_async (worker,
1255                                data,
1256                                write_message_cb,
1257                                data);
1258         }
1259     }
1260 }
1261
1262 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1263 static gboolean
1264 write_message_in_idle_cb (gpointer user_data)
1265 {
1266   GDBusWorker *worker = user_data;
1267   if (worker->num_writes_pending == 0)
1268     maybe_write_next_message (worker);
1269   return FALSE;
1270 }
1271
1272 /* ---------------------------------------------------------------------------------------------------- */
1273
1274 /* can be called from any thread - steals blob */
1275 void
1276 _g_dbus_worker_send_message (GDBusWorker    *worker,
1277                              GDBusMessage   *message,
1278                              gchar          *blob,
1279                              gsize           blob_len)
1280 {
1281   MessageToWriteData *data;
1282
1283   g_return_if_fail (G_IS_DBUS_MESSAGE (message));
1284   g_return_if_fail (blob != NULL);
1285   g_return_if_fail (blob_len > 16);
1286
1287   data = g_new0 (MessageToWriteData, 1);
1288   data->worker = _g_dbus_worker_ref (worker);
1289   data->message = g_object_ref (message);
1290   data->blob = blob; /* steal! */
1291   data->blob_size = blob_len;
1292
1293   g_mutex_lock (worker->write_lock);
1294   g_queue_push_tail (worker->write_queue, data);
1295   if (worker->num_writes_pending == 0)
1296     {
1297       GSource *idle_source;
1298       idle_source = g_idle_source_new ();
1299       g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
1300       g_source_set_callback (idle_source,
1301                              write_message_in_idle_cb,
1302                              _g_dbus_worker_ref (worker),
1303                              (GDestroyNotify) _g_dbus_worker_unref);
1304       g_source_attach (idle_source, shared_thread_data->context);
1305       g_source_unref (idle_source);
1306     }
1307   g_mutex_unlock (worker->write_lock);
1308 }
1309
1310 /* ---------------------------------------------------------------------------------------------------- */
1311
1312 static void
1313 _g_dbus_worker_thread_begin_func (gpointer user_data)
1314 {
1315   GDBusWorker *worker = user_data;
1316
1317   worker->thread = g_thread_self ();
1318
1319   /* begin reading */
1320   _g_dbus_worker_do_read (worker);
1321 }
1322
1323 GDBusWorker *
1324 _g_dbus_worker_new (GIOStream                              *stream,
1325                     GDBusCapabilityFlags                    capabilities,
1326                     gboolean                                initially_frozen,
1327                     GDBusWorkerMessageReceivedCallback      message_received_callback,
1328                     GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback,
1329                     GDBusWorkerDisconnectedCallback         disconnected_callback,
1330                     gpointer                                user_data)
1331 {
1332   GDBusWorker *worker;
1333
1334   g_return_val_if_fail (G_IS_IO_STREAM (stream), NULL);
1335   g_return_val_if_fail (message_received_callback != NULL, NULL);
1336   g_return_val_if_fail (message_about_to_be_sent_callback != NULL, NULL);
1337   g_return_val_if_fail (disconnected_callback != NULL, NULL);
1338
1339   worker = g_new0 (GDBusWorker, 1);
1340   worker->ref_count = 1;
1341
1342   worker->read_lock = g_mutex_new ();
1343   worker->message_received_callback = message_received_callback;
1344   worker->message_about_to_be_sent_callback = message_about_to_be_sent_callback;
1345   worker->disconnected_callback = disconnected_callback;
1346   worker->user_data = user_data;
1347   worker->stream = g_object_ref (stream);
1348   worker->capabilities = capabilities;
1349   worker->cancellable = g_cancellable_new ();
1350
1351   worker->frozen = initially_frozen;
1352   worker->received_messages_while_frozen = g_queue_new ();
1353
1354   worker->write_lock = g_mutex_new ();
1355   worker->write_queue = g_queue_new ();
1356
1357   if (G_IS_SOCKET_CONNECTION (worker->stream))
1358     worker->socket = g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream));
1359
1360   _g_dbus_shared_thread_ref (_g_dbus_worker_thread_begin_func, worker);
1361
1362   return worker;
1363 }
1364
1365 /* ---------------------------------------------------------------------------------------------------- */
1366
1367 /* This can be called from any thread - frees worker - guarantees no callbacks
1368  * will ever be issued again
1369  */
1370 void
1371 _g_dbus_worker_stop (GDBusWorker *worker)
1372 {
1373   /* If we're called in the worker thread it means we are called from
1374    * a worker callback. This is fine, we just can't lock in that case since
1375    * we're already holding the lock...
1376    */
1377   if (g_thread_self () != worker->thread)
1378     g_mutex_lock (worker->read_lock);
1379   worker->stopped = TRUE;
1380   if (g_thread_self () != worker->thread)
1381     g_mutex_unlock (worker->read_lock);
1382
1383   g_cancellable_cancel (worker->cancellable);
1384   _g_dbus_worker_unref (worker);
1385 }
1386
1387 /* ---------------------------------------------------------------------------------------------------- */
1388
1389 /* can be called from any thread (except the worker thread) - blocks
1390  * calling thread until all queued outgoing messages are written and
1391  * the transport has been flushed
1392  */
1393 gboolean
1394 _g_dbus_worker_flush_sync (GDBusWorker    *worker,
1395                            GCancellable   *cancellable,
1396                            GError        **error)
1397 {
1398   gboolean ret;
1399   FlushData *data;
1400
1401   data = NULL;
1402   ret = TRUE;
1403
1404   /* if the queue is empty, there's nothing to wait for */
1405   g_mutex_lock (worker->write_lock);
1406   if (g_queue_get_length (worker->write_queue) > 0)
1407     {
1408       data = g_new0 (FlushData, 1);
1409       data->mutex = g_mutex_new ();
1410       data->cond = g_cond_new ();
1411       data->number_to_wait_for = worker->write_num_messages_written + g_queue_get_length (worker->write_queue);
1412       g_mutex_lock (data->mutex);
1413       worker->write_pending_flushes = g_list_prepend (worker->write_pending_flushes, data);
1414     }
1415   g_mutex_unlock (worker->write_lock);
1416
1417   if (data != NULL)
1418     {
1419       g_cond_wait (data->cond, data->mutex);
1420       g_mutex_unlock (data->mutex);
1421
1422       /* note:the element is removed from worker->write_pending_flushes in flush_cb() above */
1423       g_cond_free (data->cond);
1424       g_mutex_free (data->mutex);
1425       if (data->error != NULL)
1426         {
1427           ret = FALSE;
1428           g_propagate_error (error, data->error);
1429         }
1430       g_free (data);
1431     }
1432
1433   return ret;
1434 }
1435
1436 /* ---------------------------------------------------------------------------------------------------- */
1437
1438 #define G_DBUS_DEBUG_AUTHENTICATION (1<<0)
1439 #define G_DBUS_DEBUG_TRANSPORT      (1<<1)
1440 #define G_DBUS_DEBUG_MESSAGE        (1<<2)
1441 #define G_DBUS_DEBUG_PAYLOAD        (1<<3)
1442 #define G_DBUS_DEBUG_CALL           (1<<4)
1443 #define G_DBUS_DEBUG_SIGNAL         (1<<5)
1444 #define G_DBUS_DEBUG_INCOMING       (1<<6)
1445 #define G_DBUS_DEBUG_RETURN         (1<<7)
1446 #define G_DBUS_DEBUG_EMISSION       (1<<8)
1447 #define G_DBUS_DEBUG_ADDRESS        (1<<9)
1448
1449 static gint _gdbus_debug_flags = 0;
1450
1451 gboolean
1452 _g_dbus_debug_authentication (void)
1453 {
1454   _g_dbus_initialize ();
1455   return (_gdbus_debug_flags & G_DBUS_DEBUG_AUTHENTICATION) != 0;
1456 }
1457
1458 gboolean
1459 _g_dbus_debug_transport (void)
1460 {
1461   _g_dbus_initialize ();
1462   return (_gdbus_debug_flags & G_DBUS_DEBUG_TRANSPORT) != 0;
1463 }
1464
1465 gboolean
1466 _g_dbus_debug_message (void)
1467 {
1468   _g_dbus_initialize ();
1469   return (_gdbus_debug_flags & G_DBUS_DEBUG_MESSAGE) != 0;
1470 }
1471
1472 gboolean
1473 _g_dbus_debug_payload (void)
1474 {
1475   _g_dbus_initialize ();
1476   return (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD) != 0;
1477 }
1478
1479 gboolean
1480 _g_dbus_debug_call (void)
1481 {
1482   _g_dbus_initialize ();
1483   return (_gdbus_debug_flags & G_DBUS_DEBUG_CALL) != 0;
1484 }
1485
1486 gboolean
1487 _g_dbus_debug_signal (void)
1488 {
1489   _g_dbus_initialize ();
1490   return (_gdbus_debug_flags & G_DBUS_DEBUG_SIGNAL) != 0;
1491 }
1492
1493 gboolean
1494 _g_dbus_debug_incoming (void)
1495 {
1496   _g_dbus_initialize ();
1497   return (_gdbus_debug_flags & G_DBUS_DEBUG_INCOMING) != 0;
1498 }
1499
1500 gboolean
1501 _g_dbus_debug_return (void)
1502 {
1503   _g_dbus_initialize ();
1504   return (_gdbus_debug_flags & G_DBUS_DEBUG_RETURN) != 0;
1505 }
1506
1507 gboolean
1508 _g_dbus_debug_emission (void)
1509 {
1510   _g_dbus_initialize ();
1511   return (_gdbus_debug_flags & G_DBUS_DEBUG_EMISSION) != 0;
1512 }
1513
1514 gboolean
1515 _g_dbus_debug_address (void)
1516 {
1517   _g_dbus_initialize ();
1518   return (_gdbus_debug_flags & G_DBUS_DEBUG_ADDRESS) != 0;
1519 }
1520
1521 G_LOCK_DEFINE_STATIC (print_lock);
1522
1523 void
1524 _g_dbus_debug_print_lock (void)
1525 {
1526   G_LOCK (print_lock);
1527 }
1528
1529 void
1530 _g_dbus_debug_print_unlock (void)
1531 {
1532   G_UNLOCK (print_lock);
1533 }
1534
1535 /*
1536  * _g_dbus_initialize:
1537  *
1538  * Does various one-time init things such as
1539  *
1540  *  - registering the G_DBUS_ERROR error domain
1541  *  - parses the G_DBUS_DEBUG environment variable
1542  */
1543 void
1544 _g_dbus_initialize (void)
1545 {
1546   static volatile gsize initialized = 0;
1547
1548   if (g_once_init_enter (&initialized))
1549     {
1550       volatile GQuark g_dbus_error_domain;
1551       const gchar *debug;
1552
1553       g_dbus_error_domain = G_DBUS_ERROR;
1554
1555       debug = g_getenv ("G_DBUS_DEBUG");
1556       if (debug != NULL)
1557         {
1558           const GDebugKey keys[] = {
1559             { "authentication", G_DBUS_DEBUG_AUTHENTICATION },
1560             { "transport",      G_DBUS_DEBUG_TRANSPORT      },
1561             { "message",        G_DBUS_DEBUG_MESSAGE        },
1562             { "payload",        G_DBUS_DEBUG_PAYLOAD        },
1563             { "call",           G_DBUS_DEBUG_CALL           },
1564             { "signal",         G_DBUS_DEBUG_SIGNAL         },
1565             { "incoming",       G_DBUS_DEBUG_INCOMING       },
1566             { "return",         G_DBUS_DEBUG_RETURN         },
1567             { "emission",       G_DBUS_DEBUG_EMISSION       },
1568             { "address",        G_DBUS_DEBUG_ADDRESS        }
1569           };
1570
1571           _gdbus_debug_flags = g_parse_debug_string (debug, keys, G_N_ELEMENTS (keys));
1572           if (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD)
1573             _gdbus_debug_flags |= G_DBUS_DEBUG_MESSAGE;
1574         }
1575
1576       g_once_init_leave (&initialized, 1);
1577     }
1578 }
1579
1580 /* ---------------------------------------------------------------------------------------------------- */
1581
1582 GVariantType *
1583 _g_dbus_compute_complete_signature (GDBusArgInfo **args)
1584 {
1585   const GVariantType *arg_types[256];
1586   guint n;
1587
1588   if (args)
1589     for (n = 0; args[n] != NULL; n++)
1590       {
1591         /* DBus places a hard limit of 255 on signature length.
1592          * therefore number of args must be less than 256.
1593          */
1594         g_assert (n < 256);
1595
1596         arg_types[n] = G_VARIANT_TYPE (args[n]->signature);
1597
1598         if G_UNLIKELY (arg_types[n] == NULL)
1599           return NULL;
1600       }
1601   else
1602     n = 0;
1603
1604   return g_variant_type_new_tuple (arg_types, n);
1605 }
1606
1607 /* ---------------------------------------------------------------------------------------------------- */
1608
1609 #ifdef G_OS_WIN32
1610
1611 extern BOOL WINAPI ConvertSidToStringSidA (PSID Sid, LPSTR *StringSid);
1612
1613 gchar *
1614 _g_dbus_win32_get_user_sid (void)
1615 {
1616   HANDLE h;
1617   TOKEN_USER *user;
1618   DWORD token_information_len;
1619   PSID psid;
1620   gchar *sid;
1621   gchar *ret;
1622
1623   ret = NULL;
1624   user = NULL;
1625   h = INVALID_HANDLE_VALUE;
1626
1627   if (!OpenProcessToken (GetCurrentProcess (), TOKEN_QUERY, &h))
1628     {
1629       g_warning ("OpenProcessToken failed with error code %d", (gint) GetLastError ());
1630       goto out;
1631     }
1632
1633   /* Get length of buffer */
1634   token_information_len = 0;
1635   if (!GetTokenInformation (h, TokenUser, NULL, 0, &token_information_len))
1636     {
1637       if (GetLastError () != ERROR_INSUFFICIENT_BUFFER)
1638         {
1639           g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
1640           goto out;
1641         }
1642     }
1643   user = g_malloc (token_information_len);
1644   if (!GetTokenInformation (h, TokenUser, user, token_information_len, &token_information_len))
1645     {
1646       g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
1647       goto out;
1648     }
1649
1650   psid = user->User.Sid;
1651   if (!IsValidSid (psid))
1652     {
1653       g_warning ("Invalid SID");
1654       goto out;
1655     }
1656
1657   if (!ConvertSidToStringSidA (psid, &sid))
1658     {
1659       g_warning ("Invalid SID");
1660       goto out;
1661     }
1662
1663   ret = g_strdup (sid);
1664   LocalFree (sid);
1665
1666 out:
1667   g_free (user);
1668   if (h != INVALID_HANDLE_VALUE)
1669     CloseHandle (h);
1670   return ret;
1671 }
1672 #endif
1673
1674 /* ---------------------------------------------------------------------------------------------------- */
1675
1676 gchar *
1677 _g_dbus_get_machine_id (GError **error)
1678 {
1679   gchar *ret;
1680   /* TODO: use PACKAGE_LOCALSTATEDIR ? */
1681   ret = NULL;
1682   if (!g_file_get_contents ("/var/lib/dbus/machine-id",
1683                             &ret,
1684                             NULL,
1685                             error))
1686     {
1687       g_prefix_error (error, _("Unable to load /var/lib/dbus/machine-id: "));
1688     }
1689   else
1690     {
1691       /* TODO: validate value */
1692       g_strstrip (ret);
1693     }
1694   return ret;
1695 }
1696
1697 /* ---------------------------------------------------------------------------------------------------- */
1698
1699 gchar *
1700 _g_dbus_enum_to_string (GType enum_type, gint value)
1701 {
1702   gchar *ret;
1703   GEnumClass *klass;
1704   GEnumValue *enum_value;
1705
1706   klass = g_type_class_ref (enum_type);
1707   enum_value = g_enum_get_value (klass, value);
1708   if (enum_value != NULL)
1709     ret = g_strdup (enum_value->value_nick);
1710   else
1711     ret = g_strdup_printf ("unknown (value %d)", value);
1712   g_type_class_unref (klass);
1713   return ret;
1714 }
1715
1716 /* ---------------------------------------------------------------------------------------------------- */
1717
1718 static void
1719 write_message_print_transport_debug (gssize bytes_written,
1720                                      MessageToWriteData *data)
1721 {
1722   if (G_LIKELY (!_g_dbus_debug_transport ()))
1723     goto out;
1724
1725   _g_dbus_debug_print_lock ();
1726   g_print ("========================================================================\n"
1727            "GDBus-debug:Transport:\n"
1728            "  >>>> WROTE %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
1729            "       size %" G_GSIZE_FORMAT " from offset %" G_GSIZE_FORMAT " on a %s\n",
1730            bytes_written,
1731            g_dbus_message_get_serial (data->message),
1732            data->blob_size,
1733            data->total_written,
1734            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
1735   _g_dbus_debug_print_unlock ();
1736  out:
1737   ;
1738 }
1739
1740 /* ---------------------------------------------------------------------------------------------------- */
1741
1742 static void
1743 read_message_print_transport_debug (gssize bytes_read,
1744                                     GDBusWorker *worker)
1745 {
1746   gsize size;
1747   gint32 serial;
1748   gint32 message_length;
1749
1750   if (G_LIKELY (!_g_dbus_debug_transport ()))
1751     goto out;
1752
1753   size = bytes_read + worker->read_buffer_cur_size;
1754   serial = 0;
1755   message_length = 0;
1756   if (size >= 16)
1757     message_length = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer, size, NULL);
1758   if (size >= 1)
1759     {
1760       switch (worker->read_buffer[0])
1761         {
1762         case 'l':
1763           if (size >= 12)
1764             serial = GUINT32_FROM_LE (((guint32 *) worker->read_buffer)[2]);
1765           break;
1766         case 'B':
1767           if (size >= 12)
1768             serial = GUINT32_FROM_BE (((guint32 *) worker->read_buffer)[2]);
1769           break;
1770         default:
1771           /* an error will be set elsewhere if this happens */
1772           goto out;
1773         }
1774     }
1775
1776     _g_dbus_debug_print_lock ();
1777   g_print ("========================================================================\n"
1778            "GDBus-debug:Transport:\n"
1779            "  <<<< READ %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
1780            "       size %d to offset %" G_GSIZE_FORMAT " from a %s\n",
1781            bytes_read,
1782            serial,
1783            message_length,
1784            worker->read_buffer_cur_size,
1785            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker->stream))));
1786   _g_dbus_debug_print_unlock ();
1787  out:
1788   ;
1789 }