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 GDBusMessageFilterResult
 472 _g_dbus_worker_emit_message_about_to_be_sent (GDBusWorker  *worker,
 473                                               GDBusMessage *message)
 474 {
 475   GDBusMessageFilterResult ret;
 476   ret = G_DBUS_MESSAGE_FILTER_RESULT_NO_EFFECT;
 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 #ifdef G_OS_UNIX
1011       if (fd_list != NULL)
1012         {
1013           g_simple_async_result_set_error (simple,
1014                                            G_IO_ERROR,
1015                                            G_IO_ERROR_FAILED,
1016                                            "Tried sending a file descriptor on unsupported stream of type %s",
1017                                            g_type_name (G_TYPE_FROM_INSTANCE (ostream)));
1018           g_simple_async_result_complete (simple);
1019           g_object_unref (simple);
1020           goto out;
1021         }
1022 #endif
1023
1024       g_output_stream_write_async (ostream,
1025                                    (const gchar *) data->blob + data->total_written,
1026                                    data->blob_size - data->total_written,
1027                                    G_PRIORITY_DEFAULT,
1028                                    data->worker->cancellable,
1029                                    write_message_async_cb,
1030                                    data);
1031     }
1032  out:
1033   ;
1034 }
1035
1036 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1037 static void
1038 write_message_async (GDBusWorker         *worker,
1039                      MessageToWriteData  *data,
1040                      GAsyncReadyCallback  callback,
1041                      gpointer             user_data)
1042 {
1043   data->simple = g_simple_async_result_new (NULL,
1044                                             callback,
1045                                             user_data,
1046                                             write_message_async);
1047   data->total_written = 0;
1048   write_message_continue_writing (data);
1049 }
1050
1051 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1052 static gboolean
1053 write_message_finish (GAsyncResult   *res,
1054                       GError        **error)
1055 {
1056   g_warn_if_fail (g_simple_async_result_get_source_tag (G_SIMPLE_ASYNC_RESULT (res)) == write_message_async);
1057   if (g_simple_async_result_propagate_error (G_SIMPLE_ASYNC_RESULT (res), error))
1058     return FALSE;
1059   else
1060     return TRUE;
1061 }
1062 /* ---------------------------------------------------------------------------------------------------- */
1063
1064 static void maybe_write_next_message (GDBusWorker *worker);
1065
1066 typedef struct
1067 {
1068   GDBusWorker *worker;
1069   GList *flushers;
1070 } FlushAsyncData;
1071
1072 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1073 static void
1074 ostream_flush_cb (GObject      *source_object,
1075                   GAsyncResult *res,
1076                   gpointer      user_data)
1077 {
1078   FlushAsyncData *data = user_data;
1079   GError *error;
1080   GList *l;
1081
1082   error = NULL;
1083   g_output_stream_flush_finish (G_OUTPUT_STREAM (source_object),
1084                                 res,
1085                                 &error);
1086
1087   if (error == NULL)
1088     {
1089       if (G_UNLIKELY (_g_dbus_debug_transport ()))
1090         {
1091           _g_dbus_debug_print_lock ();
1092           g_print ("========================================================================\n"
1093                    "GDBus-debug:Transport:\n"
1094                    "  ---- FLUSHED stream of type %s\n",
1095                    g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
1096           _g_dbus_debug_print_unlock ();
1097         }
1098     }
1099
1100   g_assert (data->flushers != NULL);
1101   for (l = data->flushers; l != NULL; l = l->next)
1102     {
1103       FlushData *f = l->data;
1104
1105       f->error = error != NULL ? g_error_copy (error) : NULL;
1106
1107       g_mutex_lock (f->mutex);
1108       g_cond_signal (f->cond);
1109       g_mutex_unlock (f->mutex);
1110     }
1111   g_list_free (data->flushers);
1112
1113   if (error != NULL)
1114     g_error_free (error);
1115
1116   /* OK, cool, finally kick off the next write */
1117   maybe_write_next_message (data->worker);
1118
1119   _g_dbus_worker_unref (data->worker);
1120   g_free (data);
1121 }
1122
1123 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1124 static void
1125 message_written (GDBusWorker *worker,
1126                  MessageToWriteData *message_data)
1127 {
1128   GList *l;
1129   GList *ll;
1130   GList *flushers;
1131
1132   /* first log the fact that we wrote a message */
1133   if (G_UNLIKELY (_g_dbus_debug_message ()))
1134     {
1135       gchar *s;
1136       _g_dbus_debug_print_lock ();
1137       g_print ("========================================================================\n"
1138                "GDBus-debug:Message:\n"
1139                "  >>>> SENT D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
1140                message_data->blob_size);
1141       s = g_dbus_message_print (message_data->message, 2);
1142       g_print ("%s", s);
1143       g_free (s);
1144       if (G_UNLIKELY (_g_dbus_debug_payload ()))
1145         {
1146           s = _g_dbus_hexdump (message_data->blob, message_data->blob_size, 2);
1147           g_print ("%s\n", s);
1148           g_free (s);
1149         }
1150       _g_dbus_debug_print_unlock ();
1151     }
1152
1153   /* then first wake up pending flushes and, if needed, flush the stream */
1154   flushers = NULL;
1155   g_mutex_lock (worker->write_lock);
1156   worker->write_num_messages_written += 1;
1157   for (l = worker->write_pending_flushes; l != NULL; l = ll)
1158     {
1159       FlushData *f = l->data;
1160       ll = l->next;
1161
1162       if (f->number_to_wait_for == worker->write_num_messages_written)
1163         {
1164           flushers = g_list_append (flushers, f);
1165           worker->write_pending_flushes = g_list_delete_link (worker->write_pending_flushes, l);
1166         }
1167     }
1168   g_mutex_unlock (worker->write_lock);
1169
1170   if (flushers != NULL)
1171     {
1172       FlushAsyncData *data;
1173       data = g_new0 (FlushAsyncData, 1);
1174       data->worker = _g_dbus_worker_ref (worker);
1175       data->flushers = flushers;
1176       /* flush the stream before writing the next message */
1177       g_output_stream_flush_async (g_io_stream_get_output_stream (worker->stream),
1178                                    G_PRIORITY_DEFAULT,
1179                                    worker->cancellable,
1180                                    ostream_flush_cb,
1181                                    data);
1182     }
1183   else
1184     {
1185       /* kick off the next write! */
1186       maybe_write_next_message (worker);
1187     }
1188 }
1189
1190 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1191 static void
1192 write_message_cb (GObject       *source_object,
1193                   GAsyncResult  *res,
1194                   gpointer       user_data)
1195 {
1196   MessageToWriteData *data = user_data;
1197   GError *error;
1198
1199   g_mutex_lock (data->worker->write_lock);
1200   data->worker->num_writes_pending -= 1;
1201   g_mutex_unlock (data->worker->write_lock);
1202
1203   error = NULL;
1204   if (!write_message_finish (res, &error))
1205     {
1206       /* TODO: handle */
1207       _g_dbus_worker_emit_disconnected (data->worker, TRUE, error);
1208       g_error_free (error);
1209     }
1210
1211   /* this function will also kick of the next write (it might need to
1212    * flush so writing the next message might happen much later
1213    * e.g. async)
1214    */
1215   message_written (data->worker, data);
1216
1217   message_to_write_data_free (data);
1218 }
1219
1220 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1221 static void
1222 maybe_write_next_message (GDBusWorker *worker)
1223 {
1224   MessageToWriteData *data;
1225
1226  write_next:
1227
1228   g_mutex_lock (worker->write_lock);
1229   data = g_queue_pop_head (worker->write_queue);
1230   if (data != NULL)
1231     worker->num_writes_pending += 1;
1232   g_mutex_unlock (worker->write_lock);
1233
1234   /* Note that write_lock is only used for protecting the @write_queue
1235    * and @num_writes_pending fields of the GDBusWorker struct ... which we
1236    * need to modify from arbitrary threads in _g_dbus_worker_send_message().
1237    *
1238    * Therefore, it's fine to drop it here when calling back into user
1239    * code and then writing the message out onto the GIOStream since this
1240    * function only runs on the worker thread.
1241    */
1242   if (data != NULL)
1243     {
1244       GDBusMessageFilterResult filter_result;
1245       guchar *new_blob;
1246       gsize new_blob_size;
1247       GError *error;
1248
1249       filter_result = _g_dbus_worker_emit_message_about_to_be_sent (worker, data->message);
1250       switch (filter_result)
1251         {
1252         case G_DBUS_MESSAGE_FILTER_RESULT_NO_EFFECT:
1253           /* do nothing */
1254           break;
1255
1256         case G_DBUS_MESSAGE_FILTER_RESULT_MESSAGE_CONSUMED:
1257           /* drop message */
1258           g_mutex_lock (worker->write_lock);
1259           worker->num_writes_pending -= 1;
1260           g_mutex_unlock (worker->write_lock);
1261           message_to_write_data_free (data);
1262           goto write_next;
1263
1264         case G_DBUS_MESSAGE_FILTER_RESULT_MESSAGE_ALTERED:
1265           /* reencode altered message */
1266           error = NULL;
1267           new_blob = g_dbus_message_to_blob (data->message,
1268                                              &new_blob_size,
1269                                              worker->capabilities,
1270                                              &error);
1271           if (new_blob == NULL)
1272             {
1273               /* if filter make the GDBusMessage unencodeable, just complain on stderr and send
1274                * the old message instead
1275                */
1276               g_warning ("Error encoding GDBusMessage with serial %d altered by filter function: %s",
1277                          g_dbus_message_get_serial (data->message),
1278                          error->message);
1279               g_error_free (error);
1280             }
1281           else
1282             {
1283               g_free (data->blob);
1284               data->blob = (gchar *) new_blob;
1285               data->blob_size = new_blob_size;
1286             }
1287           break;
1288         }
1289
1290       write_message_async (worker,
1291                            data,
1292                            write_message_cb,
1293                            data);
1294     }
1295 }
1296
1297 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1298 static gboolean
1299 write_message_in_idle_cb (gpointer user_data)
1300 {
1301   GDBusWorker *worker = user_data;
1302   if (worker->num_writes_pending == 0)
1303     maybe_write_next_message (worker);
1304   return FALSE;
1305 }
1306
1307 /* ---------------------------------------------------------------------------------------------------- */
1308
1309 /* can be called from any thread - steals blob */
1310 void
1311 _g_dbus_worker_send_message (GDBusWorker    *worker,
1312                              GDBusMessage   *message,
1313                              gchar          *blob,
1314                              gsize           blob_len)
1315 {
1316   MessageToWriteData *data;
1317
1318   g_return_if_fail (G_IS_DBUS_MESSAGE (message));
1319   g_return_if_fail (blob != NULL);
1320   g_return_if_fail (blob_len > 16);
1321
1322   data = g_new0 (MessageToWriteData, 1);
1323   data->worker = _g_dbus_worker_ref (worker);
1324   data->message = g_object_ref (message);
1325   data->blob = blob; /* steal! */
1326   data->blob_size = blob_len;
1327
1328   g_mutex_lock (worker->write_lock);
1329   g_queue_push_tail (worker->write_queue, data);
1330   if (worker->num_writes_pending == 0)
1331     {
1332       GSource *idle_source;
1333       idle_source = g_idle_source_new ();
1334       g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
1335       g_source_set_callback (idle_source,
1336                              write_message_in_idle_cb,
1337                              _g_dbus_worker_ref (worker),
1338                              (GDestroyNotify) _g_dbus_worker_unref);
1339       g_source_attach (idle_source, shared_thread_data->context);
1340       g_source_unref (idle_source);
1341     }
1342   g_mutex_unlock (worker->write_lock);
1343 }
1344
1345 /* ---------------------------------------------------------------------------------------------------- */
1346
1347 static void
1348 _g_dbus_worker_thread_begin_func (gpointer user_data)
1349 {
1350   GDBusWorker *worker = user_data;
1351
1352   worker->thread = g_thread_self ();
1353
1354   /* begin reading */
1355   _g_dbus_worker_do_read (worker);
1356 }
1357
1358 GDBusWorker *
1359 _g_dbus_worker_new (GIOStream                              *stream,
1360                     GDBusCapabilityFlags                    capabilities,
1361                     gboolean                                initially_frozen,
1362                     GDBusWorkerMessageReceivedCallback      message_received_callback,
1363                     GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback,
1364                     GDBusWorkerDisconnectedCallback         disconnected_callback,
1365                     gpointer                                user_data)
1366 {
1367   GDBusWorker *worker;
1368
1369   g_return_val_if_fail (G_IS_IO_STREAM (stream), NULL);
1370   g_return_val_if_fail (message_received_callback != NULL, NULL);
1371   g_return_val_if_fail (message_about_to_be_sent_callback != NULL, NULL);
1372   g_return_val_if_fail (disconnected_callback != NULL, NULL);
1373
1374   worker = g_new0 (GDBusWorker, 1);
1375   worker->ref_count = 1;
1376
1377   worker->read_lock = g_mutex_new ();
1378   worker->message_received_callback = message_received_callback;
1379   worker->message_about_to_be_sent_callback = message_about_to_be_sent_callback;
1380   worker->disconnected_callback = disconnected_callback;
1381   worker->user_data = user_data;
1382   worker->stream = g_object_ref (stream);
1383   worker->capabilities = capabilities;
1384   worker->cancellable = g_cancellable_new ();
1385
1386   worker->frozen = initially_frozen;
1387   worker->received_messages_while_frozen = g_queue_new ();
1388
1389   worker->write_lock = g_mutex_new ();
1390   worker->write_queue = g_queue_new ();
1391
1392   if (G_IS_SOCKET_CONNECTION (worker->stream))
1393     worker->socket = g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream));
1394
1395   _g_dbus_shared_thread_ref (_g_dbus_worker_thread_begin_func, worker);
1396
1397   return worker;
1398 }
1399
1400 /* ---------------------------------------------------------------------------------------------------- */
1401
1402 /* This can be called from any thread - frees worker - guarantees no callbacks
1403  * will ever be issued again
1404  */
1405 void
1406 _g_dbus_worker_stop (GDBusWorker *worker)
1407 {
1408   /* If we're called in the worker thread it means we are called from
1409    * a worker callback. This is fine, we just can't lock in that case since
1410    * we're already holding the lock...
1411    */
1412   if (g_thread_self () != worker->thread)
1413     g_mutex_lock (worker->read_lock);
1414   worker->stopped = TRUE;
1415   if (g_thread_self () != worker->thread)
1416     g_mutex_unlock (worker->read_lock);
1417
1418   g_cancellable_cancel (worker->cancellable);
1419   _g_dbus_worker_unref (worker);
1420 }
1421
1422 /* ---------------------------------------------------------------------------------------------------- */
1423
1424 /* can be called from any thread (except the worker thread) - blocks
1425  * calling thread until all queued outgoing messages are written and
1426  * the transport has been flushed
1427  */
1428 gboolean
1429 _g_dbus_worker_flush_sync (GDBusWorker    *worker,
1430                            GCancellable   *cancellable,
1431                            GError        **error)
1432 {
1433   gboolean ret;
1434   FlushData *data;
1435
1436   data = NULL;
1437   ret = TRUE;
1438
1439   /* if the queue is empty, there's nothing to wait for */
1440   g_mutex_lock (worker->write_lock);
1441   if (g_queue_get_length (worker->write_queue) > 0)
1442     {
1443       data = g_new0 (FlushData, 1);
1444       data->mutex = g_mutex_new ();
1445       data->cond = g_cond_new ();
1446       data->number_to_wait_for = worker->write_num_messages_written + g_queue_get_length (worker->write_queue);
1447       g_mutex_lock (data->mutex);
1448       worker->write_pending_flushes = g_list_prepend (worker->write_pending_flushes, data);
1449     }
1450   g_mutex_unlock (worker->write_lock);
1451
1452   if (data != NULL)
1453     {
1454       g_cond_wait (data->cond, data->mutex);
1455       g_mutex_unlock (data->mutex);
1456
1457       /* note:the element is removed from worker->write_pending_flushes in flush_cb() above */
1458       g_cond_free (data->cond);
1459       g_mutex_free (data->mutex);
1460       if (data->error != NULL)
1461         {
1462           ret = FALSE;
1463           g_propagate_error (error, data->error);
1464         }
1465       g_free (data);
1466     }
1467
1468   return ret;
1469 }
1470
1471 /* ---------------------------------------------------------------------------------------------------- */
1472
1473 #define G_DBUS_DEBUG_AUTHENTICATION (1<<0)
1474 #define G_DBUS_DEBUG_TRANSPORT      (1<<1)
1475 #define G_DBUS_DEBUG_MESSAGE        (1<<2)
1476 #define G_DBUS_DEBUG_PAYLOAD        (1<<3)
1477 #define G_DBUS_DEBUG_CALL           (1<<4)
1478 #define G_DBUS_DEBUG_SIGNAL         (1<<5)
1479 #define G_DBUS_DEBUG_INCOMING       (1<<6)
1480 #define G_DBUS_DEBUG_RETURN         (1<<7)
1481 #define G_DBUS_DEBUG_EMISSION       (1<<8)
1482 #define G_DBUS_DEBUG_ADDRESS        (1<<9)
1483
1484 static gint _gdbus_debug_flags = 0;
1485
1486 gboolean
1487 _g_dbus_debug_authentication (void)
1488 {
1489   _g_dbus_initialize ();
1490   return (_gdbus_debug_flags & G_DBUS_DEBUG_AUTHENTICATION) != 0;
1491 }
1492
1493 gboolean
1494 _g_dbus_debug_transport (void)
1495 {
1496   _g_dbus_initialize ();
1497   return (_gdbus_debug_flags & G_DBUS_DEBUG_TRANSPORT) != 0;
1498 }
1499
1500 gboolean
1501 _g_dbus_debug_message (void)
1502 {
1503   _g_dbus_initialize ();
1504   return (_gdbus_debug_flags & G_DBUS_DEBUG_MESSAGE) != 0;
1505 }
1506
1507 gboolean
1508 _g_dbus_debug_payload (void)
1509 {
1510   _g_dbus_initialize ();
1511   return (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD) != 0;
1512 }
1513
1514 gboolean
1515 _g_dbus_debug_call (void)
1516 {
1517   _g_dbus_initialize ();
1518   return (_gdbus_debug_flags & G_DBUS_DEBUG_CALL) != 0;
1519 }
1520
1521 gboolean
1522 _g_dbus_debug_signal (void)
1523 {
1524   _g_dbus_initialize ();
1525   return (_gdbus_debug_flags & G_DBUS_DEBUG_SIGNAL) != 0;
1526 }
1527
1528 gboolean
1529 _g_dbus_debug_incoming (void)
1530 {
1531   _g_dbus_initialize ();
1532   return (_gdbus_debug_flags & G_DBUS_DEBUG_INCOMING) != 0;
1533 }
1534
1535 gboolean
1536 _g_dbus_debug_return (void)
1537 {
1538   _g_dbus_initialize ();
1539   return (_gdbus_debug_flags & G_DBUS_DEBUG_RETURN) != 0;
1540 }
1541
1542 gboolean
1543 _g_dbus_debug_emission (void)
1544 {
1545   _g_dbus_initialize ();
1546   return (_gdbus_debug_flags & G_DBUS_DEBUG_EMISSION) != 0;
1547 }
1548
1549 gboolean
1550 _g_dbus_debug_address (void)
1551 {
1552   _g_dbus_initialize ();
1553   return (_gdbus_debug_flags & G_DBUS_DEBUG_ADDRESS) != 0;
1554 }
1555
1556 G_LOCK_DEFINE_STATIC (print_lock);
1557
1558 void
1559 _g_dbus_debug_print_lock (void)
1560 {
1561   G_LOCK (print_lock);
1562 }
1563
1564 void
1565 _g_dbus_debug_print_unlock (void)
1566 {
1567   G_UNLOCK (print_lock);
1568 }
1569
1570 /*
1571  * _g_dbus_initialize:
1572  *
1573  * Does various one-time init things such as
1574  *
1575  *  - registering the G_DBUS_ERROR error domain
1576  *  - parses the G_DBUS_DEBUG environment variable
1577  */
1578 void
1579 _g_dbus_initialize (void)
1580 {
1581   static volatile gsize initialized = 0;
1582
1583   if (g_once_init_enter (&initialized))
1584     {
1585       volatile GQuark g_dbus_error_domain;
1586       const gchar *debug;
1587
1588       g_dbus_error_domain = G_DBUS_ERROR;
1589
1590       debug = g_getenv ("G_DBUS_DEBUG");
1591       if (debug != NULL)
1592         {
1593           const GDebugKey keys[] = {
1594             { "authentication", G_DBUS_DEBUG_AUTHENTICATION },
1595             { "transport",      G_DBUS_DEBUG_TRANSPORT      },
1596             { "message",        G_DBUS_DEBUG_MESSAGE        },
1597             { "payload",        G_DBUS_DEBUG_PAYLOAD        },
1598             { "call",           G_DBUS_DEBUG_CALL           },
1599             { "signal",         G_DBUS_DEBUG_SIGNAL         },
1600             { "incoming",       G_DBUS_DEBUG_INCOMING       },
1601             { "return",         G_DBUS_DEBUG_RETURN         },
1602             { "emission",       G_DBUS_DEBUG_EMISSION       },
1603             { "address",        G_DBUS_DEBUG_ADDRESS        }
1604           };
1605
1606           _gdbus_debug_flags = g_parse_debug_string (debug, keys, G_N_ELEMENTS (keys));
1607           if (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD)
1608             _gdbus_debug_flags |= G_DBUS_DEBUG_MESSAGE;
1609         }
1610
1611       g_once_init_leave (&initialized, 1);
1612     }
1613 }
1614
1615 /* ---------------------------------------------------------------------------------------------------- */
1616
1617 GVariantType *
1618 _g_dbus_compute_complete_signature (GDBusArgInfo **args)
1619 {
1620   const GVariantType *arg_types[256];
1621   guint n;
1622
1623   if (args)
1624     for (n = 0; args[n] != NULL; n++)
1625       {
1626         /* DBus places a hard limit of 255 on signature length.
1627          * therefore number of args must be less than 256.
1628          */
1629         g_assert (n < 256);
1630
1631         arg_types[n] = G_VARIANT_TYPE (args[n]->signature);
1632
1633         if G_UNLIKELY (arg_types[n] == NULL)
1634           return NULL;
1635       }
1636   else
1637     n = 0;
1638
1639   return g_variant_type_new_tuple (arg_types, n);
1640 }
1641
1642 /* ---------------------------------------------------------------------------------------------------- */
1643
1644 #ifdef G_OS_WIN32
1645
1646 extern BOOL WINAPI ConvertSidToStringSidA (PSID Sid, LPSTR *StringSid);
1647
1648 gchar *
1649 _g_dbus_win32_get_user_sid (void)
1650 {
1651   HANDLE h;
1652   TOKEN_USER *user;
1653   DWORD token_information_len;
1654   PSID psid;
1655   gchar *sid;
1656   gchar *ret;
1657
1658   ret = NULL;
1659   user = NULL;
1660   h = INVALID_HANDLE_VALUE;
1661
1662   if (!OpenProcessToken (GetCurrentProcess (), TOKEN_QUERY, &h))
1663     {
1664       g_warning ("OpenProcessToken failed with error code %d", (gint) GetLastError ());
1665       goto out;
1666     }
1667
1668   /* Get length of buffer */
1669   token_information_len = 0;
1670   if (!GetTokenInformation (h, TokenUser, NULL, 0, &token_information_len))
1671     {
1672       if (GetLastError () != ERROR_INSUFFICIENT_BUFFER)
1673         {
1674           g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
1675           goto out;
1676         }
1677     }
1678   user = g_malloc (token_information_len);
1679   if (!GetTokenInformation (h, TokenUser, user, token_information_len, &token_information_len))
1680     {
1681       g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
1682       goto out;
1683     }
1684
1685   psid = user->User.Sid;
1686   if (!IsValidSid (psid))
1687     {
1688       g_warning ("Invalid SID");
1689       goto out;
1690     }
1691
1692   if (!ConvertSidToStringSidA (psid, &sid))
1693     {
1694       g_warning ("Invalid SID");
1695       goto out;
1696     }
1697
1698   ret = g_strdup (sid);
1699   LocalFree (sid);
1700
1701 out:
1702   g_free (user);
1703   if (h != INVALID_HANDLE_VALUE)
1704     CloseHandle (h);
1705   return ret;
1706 }
1707 #endif
1708
1709 /* ---------------------------------------------------------------------------------------------------- */
1710
1711 gchar *
1712 _g_dbus_get_machine_id (GError **error)
1713 {
1714   gchar *ret;
1715   /* TODO: use PACKAGE_LOCALSTATEDIR ? */
1716   ret = NULL;
1717   if (!g_file_get_contents ("/var/lib/dbus/machine-id",
1718                             &ret,
1719                             NULL,
1720                             error))
1721     {
1722       g_prefix_error (error, _("Unable to load /var/lib/dbus/machine-id: "));
1723     }
1724   else
1725     {
1726       /* TODO: validate value */
1727       g_strstrip (ret);
1728     }
1729   return ret;
1730 }
1731
1732 /* ---------------------------------------------------------------------------------------------------- */
1733
1734 gchar *
1735 _g_dbus_enum_to_string (GType enum_type, gint value)
1736 {
1737   gchar *ret;
1738   GEnumClass *klass;
1739   GEnumValue *enum_value;
1740
1741   klass = g_type_class_ref (enum_type);
1742   enum_value = g_enum_get_value (klass, value);
1743   if (enum_value != NULL)
1744     ret = g_strdup (enum_value->value_nick);
1745   else
1746     ret = g_strdup_printf ("unknown (value %d)", value);
1747   g_type_class_unref (klass);
1748   return ret;
1749 }
1750
1751 /* ---------------------------------------------------------------------------------------------------- */
1752
1753 static void
1754 write_message_print_transport_debug (gssize bytes_written,
1755                                      MessageToWriteData *data)
1756 {
1757   if (G_LIKELY (!_g_dbus_debug_transport ()))
1758     goto out;
1759
1760   _g_dbus_debug_print_lock ();
1761   g_print ("========================================================================\n"
1762            "GDBus-debug:Transport:\n"
1763            "  >>>> WROTE %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
1764            "       size %" G_GSIZE_FORMAT " from offset %" G_GSIZE_FORMAT " on a %s\n",
1765            bytes_written,
1766            g_dbus_message_get_serial (data->message),
1767            data->blob_size,
1768            data->total_written,
1769            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
1770   _g_dbus_debug_print_unlock ();
1771  out:
1772   ;
1773 }
1774
1775 /* ---------------------------------------------------------------------------------------------------- */
1776
1777 static void
1778 read_message_print_transport_debug (gssize bytes_read,
1779                                     GDBusWorker *worker)
1780 {
1781   gsize size;
1782   gint32 serial;
1783   gint32 message_length;
1784
1785   if (G_LIKELY (!_g_dbus_debug_transport ()))
1786     goto out;
1787
1788   size = bytes_read + worker->read_buffer_cur_size;
1789   serial = 0;
1790   message_length = 0;
1791   if (size >= 16)
1792     message_length = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer, size, NULL);
1793   if (size >= 1)
1794     {
1795       switch (worker->read_buffer[0])
1796         {
1797         case 'l':
1798           if (size >= 12)
1799             serial = GUINT32_FROM_LE (((guint32 *) worker->read_buffer)[2]);
1800           break;
1801         case 'B':
1802           if (size >= 12)
1803             serial = GUINT32_FROM_BE (((guint32 *) worker->read_buffer)[2]);
1804           break;
1805         default:
1806           /* an error will be set elsewhere if this happens */
1807           goto out;
1808         }
1809     }
1810
1811     _g_dbus_debug_print_lock ();
1812   g_print ("========================================================================\n"
1813            "GDBus-debug:Transport:\n"
1814            "  <<<< READ %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
1815            "       size %d to offset %" G_GSIZE_FORMAT " from a %s\n",
1816            bytes_read,
1817            serial,
1818            message_length,
1819            worker->read_buffer_cur_size,
1820            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker->stream))));
1821   _g_dbus_debug_print_unlock ();
1822  out:
1823   ;
1824 }