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               g_object_unref (worker->read_fd_list);
 721               worker->read_fd_list = NULL;
 722             }
 723 #endif
 724
 725           if (G_UNLIKELY (_g_dbus_debug_message ()))
 726             {
 727               gchar *s;
 728               _g_dbus_debug_print_lock ();
 729               g_print ("========================================================================\n"
 730                        "GDBus-debug:Message:\n"
 731                        "  <<<< RECEIVED D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
 732                        worker->read_buffer_cur_size);
 733               s = g_dbus_message_print (message, 2);
 734               g_print ("%s", s);
 735               g_free (s);
 736               if (G_UNLIKELY (_g_dbus_debug_payload ()))
 737                 {
 738                   s = _g_dbus_hexdump (worker->read_buffer, worker->read_buffer_cur_size, 2);
 739                   g_print ("%s\n", s);
 740                   g_free (s);
 741                 }
 742               _g_dbus_debug_print_unlock ();
 743             }
 744
 745           /* yay, got a message, go deliver it */
 746           _g_dbus_worker_queue_or_deliver_received_message (worker, message);
 747
 748           /* start reading another message! */
 749           worker->read_buffer_bytes_wanted = 0;
 750           worker->read_buffer_cur_size = 0;
 751           _g_dbus_worker_do_read_unlocked (worker);
 752         }
 753     }
 754   else
 755     {
 756       /* didn't get all the bytes we requested - so repeat the request... */
 757       _g_dbus_worker_do_read_unlocked (worker);
 758     }
 759
 760  out:
 761   g_mutex_unlock (worker->read_lock);
 762
 763   /* gives up the reference acquired when calling g_input_stream_read_async() */
 764   _g_dbus_worker_unref (worker);
 765 }
 766
 767 /* called in private thread shared by all GDBusConnection instances (with read-lock held) */
 768 static void
 769 _g_dbus_worker_do_read_unlocked (GDBusWorker *worker)
 770 {
 771   /* if bytes_wanted is zero, it means start reading a message */
 772   if (worker->read_buffer_bytes_wanted == 0)
 773     {
 774       worker->read_buffer_cur_size = 0;
 775       worker->read_buffer_bytes_wanted = 16;
 776     }
 777
 778   /* ensure we have a (big enough) buffer */
 779   if (worker->read_buffer == NULL || worker->read_buffer_bytes_wanted > worker->read_buffer_allocated_size)
 780     {
 781       /* TODO: 4096 is randomly chosen; might want a better chosen default minimum */
 782       worker->read_buffer_allocated_size = MAX (worker->read_buffer_bytes_wanted, 4096);
 783       worker->read_buffer = g_realloc (worker->read_buffer, worker->read_buffer_allocated_size);
 784     }
 785
 786   if (worker->socket == NULL)
 787     g_input_stream_read_async (g_io_stream_get_input_stream (worker->stream),
 788                                worker->read_buffer + worker->read_buffer_cur_size,
 789                                worker->read_buffer_bytes_wanted - worker->read_buffer_cur_size,
 790                                G_PRIORITY_DEFAULT,
 791                                worker->cancellable,
 792                                (GAsyncReadyCallback) _g_dbus_worker_do_read_cb,
 793                                _g_dbus_worker_ref (worker));
 794   else
 795     {
 796       worker->read_ancillary_messages = NULL;
 797       worker->read_num_ancillary_messages = 0;
 798       _g_socket_read_with_control_messages (worker->socket,
 799                                             worker->read_buffer + worker->read_buffer_cur_size,
 800                                             worker->read_buffer_bytes_wanted - worker->read_buffer_cur_size,
 801                                             &worker->read_ancillary_messages,
 802                                             &worker->read_num_ancillary_messages,
 803                                             G_PRIORITY_DEFAULT,
 804                                             worker->cancellable,
 805                                             (GAsyncReadyCallback) _g_dbus_worker_do_read_cb,
 806                                             _g_dbus_worker_ref (worker));
 807     }
 808 }
 809
 810 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 811 static void
 812 _g_dbus_worker_do_read (GDBusWorker *worker)
 813 {
 814   g_mutex_lock (worker->read_lock);
 815   _g_dbus_worker_do_read_unlocked (worker);
 816   g_mutex_unlock (worker->read_lock);
 817 }
 818
 819 /* ---------------------------------------------------------------------------------------------------- */
 820
 821 struct _MessageToWriteData
 822 {
 823   GDBusWorker  *worker;
 824   GDBusMessage *message;
 825   gchar        *blob;
 826   gsize         blob_size;
 827
 828   gsize               total_written;
 829   GSimpleAsyncResult *simple;
 830
 831 };
 832
 833 static void
 834 message_to_write_data_free (MessageToWriteData *data)
 835 {
 836   _g_dbus_worker_unref (data->worker);
 837   g_object_unref (data->message);
 838   g_free (data->blob);
 839   g_free (data);
 840 }
 841
 842 /* ---------------------------------------------------------------------------------------------------- */
 843
 844 static void write_message_continue_writing (MessageToWriteData *data);
 845
 846 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
 847 static void
 848 write_message_async_cb (GObject      *source_object,
 849                         GAsyncResult *res,
 850                         gpointer      user_data)
 851 {
 852   MessageToWriteData *data = user_data;
 853   GSimpleAsyncResult *simple;
 854   gssize bytes_written;
 855   GError *error;
 856
 857   /* Note: we can't access data->simple after calling g_async_result_complete () because the
 858    * callback can free @data and we're not completing in idle. So use a copy of the pointer.
 859    */
 860   simple = data->simple;
 861
 862   error = NULL;
 863   bytes_written = g_output_stream_write_finish (G_OUTPUT_STREAM (source_object),
 864                                                 res,
 865                                                 &error);
 866   if (bytes_written == -1)
 867     {
 868       g_simple_async_result_set_from_error (simple, error);
 869       g_error_free (error);
 870       g_simple_async_result_complete (simple);
 871       g_object_unref (simple);
 872       goto out;
 873     }
 874   g_assert (bytes_written > 0); /* zero is never returned */
 875
 876   write_message_print_transport_debug (bytes_written, data);
 877
 878   data->total_written += bytes_written;
 879   g_assert (data->total_written <= data->blob_size);
 880   if (data->total_written == data->blob_size)
 881     {
 882       g_simple_async_result_complete (simple);
 883       g_object_unref (simple);
 884       goto out;
 885     }
 886
 887   write_message_continue_writing (data);
 888
 889  out:
 890   ;
 891 }
 892
 893 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
 894 static gboolean
 895 on_socket_ready (GSocket      *socket,
 896                  GIOCondition  condition,
 897                  gpointer      user_data)
 898 {
 899   MessageToWriteData *data = user_data;
 900   write_message_continue_writing (data);
 901   return FALSE; /* remove source */
 902 }
 903
 904 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
 905 static void
 906 write_message_continue_writing (MessageToWriteData *data)
 907 {
 908   GOutputStream *ostream;
 909   GSimpleAsyncResult *simple;
 910 #ifdef G_OS_UNIX
 911   GUnixFDList *fd_list;
 912 #endif
 913
 914   /* Note: we can't access data->simple after calling g_async_result_complete () because the
 915    * callback can free @data and we're not completing in idle. So use a copy of the pointer.
 916    */
 917   simple = data->simple;
 918
 919   ostream = g_io_stream_get_output_stream (data->worker->stream);
 920 #ifdef G_OS_UNIX
 921   fd_list = g_dbus_message_get_unix_fd_list (data->message);
 922 #endif
 923
 924   g_assert (!g_output_stream_has_pending (ostream));
 925   g_assert_cmpint (data->total_written, <, data->blob_size);
 926
 927   if (FALSE)
 928     {
 929     }
 930 #ifdef G_OS_UNIX
 931   else if (G_IS_SOCKET_OUTPUT_STREAM (ostream) && data->total_written == 0)
 932     {
 933       GOutputVector vector;
 934       GSocketControlMessage *control_message;
 935       gssize bytes_written;
 936       GError *error;
 937
 938       vector.buffer = data->blob;
 939       vector.size = data->blob_size;
 940
 941       control_message = NULL;
 942       if (fd_list != NULL && g_unix_fd_list_get_length (fd_list) > 0)
 943         {
 944           if (!(data->worker->capabilities & G_DBUS_CAPABILITY_FLAGS_UNIX_FD_PASSING))
 945             {
 946               g_simple_async_result_set_error (simple,
 947                                                G_IO_ERROR,
 948                                                G_IO_ERROR_FAILED,
 949                                                "Tried sending a file descriptor but remote peer does not support this capability");
 950               g_simple_async_result_complete (simple);
 951               g_object_unref (simple);
 952               goto out;
 953             }
 954           control_message = g_unix_fd_message_new_with_fd_list (fd_list);
 955         }
 956
 957       error = NULL;
 958       bytes_written = g_socket_send_message (data->worker->socket,
 959                                              NULL, /* address */
 960                                              &vector,
 961                                              1,
 962                                              control_message != NULL ? &control_message : NULL,
 963                                              control_message != NULL ? 1 : 0,
 964                                              G_SOCKET_MSG_NONE,
 965                                              data->worker->cancellable,
 966                                              &error);
 967       if (control_message != NULL)
 968         g_object_unref (control_message);
 969
 970       if (bytes_written == -1)
 971         {
 972           /* Handle WOULD_BLOCK by waiting until there's room in the buffer */
 973           if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
 974             {
 975               GSource *source;
 976               source = g_socket_create_source (data->worker->socket,
 977                                                G_IO_OUT | G_IO_HUP | G_IO_ERR,
 978                                                data->worker->cancellable);
 979               g_source_set_callback (source,
 980                                      (GSourceFunc) on_socket_ready,
 981                                      data,
 982                                      NULL); /* GDestroyNotify */
 983               g_source_attach (source, g_main_context_get_thread_default ());
 984               g_source_unref (source);
 985               g_error_free (error);
 986               goto out;
 987             }
 988           g_simple_async_result_set_from_error (simple, error);
 989           g_error_free (error);
 990           g_simple_async_result_complete (simple);
 991           g_object_unref (simple);
 992           goto out;
 993         }
 994       g_assert (bytes_written > 0); /* zero is never returned */
 995
 996       write_message_print_transport_debug (bytes_written, data);
 997
 998       data->total_written += bytes_written;
 999       g_assert (data->total_written <= data->blob_size);
1000       if (data->total_written == data->blob_size)
1001         {
1002           g_simple_async_result_complete (simple);
1003           g_object_unref (simple);
1004           goto out;
1005         }
1006
1007       write_message_continue_writing (data);
1008     }
1009 #endif
1010   else
1011     {
1012 #ifdef G_OS_UNIX
1013       if (fd_list != NULL)
1014         {
1015           g_simple_async_result_set_error (simple,
1016                                            G_IO_ERROR,
1017                                            G_IO_ERROR_FAILED,
1018                                            "Tried sending a file descriptor on unsupported stream of type %s",
1019                                            g_type_name (G_TYPE_FROM_INSTANCE (ostream)));
1020           g_simple_async_result_complete (simple);
1021           g_object_unref (simple);
1022           goto out;
1023         }
1024 #endif
1025
1026       g_output_stream_write_async (ostream,
1027                                    (const gchar *) data->blob + data->total_written,
1028                                    data->blob_size - data->total_written,
1029                                    G_PRIORITY_DEFAULT,
1030                                    data->worker->cancellable,
1031                                    write_message_async_cb,
1032                                    data);
1033     }
1034  out:
1035   ;
1036 }
1037
1038 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1039 static void
1040 write_message_async (GDBusWorker         *worker,
1041                      MessageToWriteData  *data,
1042                      GAsyncReadyCallback  callback,
1043                      gpointer             user_data)
1044 {
1045   data->simple = g_simple_async_result_new (NULL,
1046                                             callback,
1047                                             user_data,
1048                                             write_message_async);
1049   data->total_written = 0;
1050   write_message_continue_writing (data);
1051 }
1052
1053 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1054 static gboolean
1055 write_message_finish (GAsyncResult   *res,
1056                       GError        **error)
1057 {
1058   g_warn_if_fail (g_simple_async_result_get_source_tag (G_SIMPLE_ASYNC_RESULT (res)) == write_message_async);
1059   if (g_simple_async_result_propagate_error (G_SIMPLE_ASYNC_RESULT (res), error))
1060     return FALSE;
1061   else
1062     return TRUE;
1063 }
1064 /* ---------------------------------------------------------------------------------------------------- */
1065
1066 static void maybe_write_next_message (GDBusWorker *worker);
1067
1068 typedef struct
1069 {
1070   GDBusWorker *worker;
1071   GList *flushers;
1072 } FlushAsyncData;
1073
1074 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1075 static void
1076 ostream_flush_cb (GObject      *source_object,
1077                   GAsyncResult *res,
1078                   gpointer      user_data)
1079 {
1080   FlushAsyncData *data = user_data;
1081   GError *error;
1082   GList *l;
1083
1084   error = NULL;
1085   g_output_stream_flush_finish (G_OUTPUT_STREAM (source_object),
1086                                 res,
1087                                 &error);
1088
1089   if (error == NULL)
1090     {
1091       if (G_UNLIKELY (_g_dbus_debug_transport ()))
1092         {
1093           _g_dbus_debug_print_lock ();
1094           g_print ("========================================================================\n"
1095                    "GDBus-debug:Transport:\n"
1096                    "  ---- FLUSHED stream of type %s\n",
1097                    g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
1098           _g_dbus_debug_print_unlock ();
1099         }
1100     }
1101
1102   g_assert (data->flushers != NULL);
1103   for (l = data->flushers; l != NULL; l = l->next)
1104     {
1105       FlushData *f = l->data;
1106
1107       f->error = error != NULL ? g_error_copy (error) : NULL;
1108
1109       g_mutex_lock (f->mutex);
1110       g_cond_signal (f->cond);
1111       g_mutex_unlock (f->mutex);
1112     }
1113   g_list_free (data->flushers);
1114
1115   if (error != NULL)
1116     g_error_free (error);
1117
1118   /* OK, cool, finally kick off the next write */
1119   maybe_write_next_message (data->worker);
1120
1121   _g_dbus_worker_unref (data->worker);
1122   g_free (data);
1123 }
1124
1125 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1126 static void
1127 message_written (GDBusWorker *worker,
1128                  MessageToWriteData *message_data)
1129 {
1130   GList *l;
1131   GList *ll;
1132   GList *flushers;
1133
1134   /* first log the fact that we wrote a message */
1135   if (G_UNLIKELY (_g_dbus_debug_message ()))
1136     {
1137       gchar *s;
1138       _g_dbus_debug_print_lock ();
1139       g_print ("========================================================================\n"
1140                "GDBus-debug:Message:\n"
1141                "  >>>> SENT D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
1142                message_data->blob_size);
1143       s = g_dbus_message_print (message_data->message, 2);
1144       g_print ("%s", s);
1145       g_free (s);
1146       if (G_UNLIKELY (_g_dbus_debug_payload ()))
1147         {
1148           s = _g_dbus_hexdump (message_data->blob, message_data->blob_size, 2);
1149           g_print ("%s\n", s);
1150           g_free (s);
1151         }
1152       _g_dbus_debug_print_unlock ();
1153     }
1154
1155   /* then first wake up pending flushes and, if needed, flush the stream */
1156   flushers = NULL;
1157   g_mutex_lock (worker->write_lock);
1158   worker->write_num_messages_written += 1;
1159   for (l = worker->write_pending_flushes; l != NULL; l = ll)
1160     {
1161       FlushData *f = l->data;
1162       ll = l->next;
1163
1164       if (f->number_to_wait_for == worker->write_num_messages_written)
1165         {
1166           flushers = g_list_append (flushers, f);
1167           worker->write_pending_flushes = g_list_delete_link (worker->write_pending_flushes, l);
1168         }
1169     }
1170   g_mutex_unlock (worker->write_lock);
1171
1172   if (flushers != NULL)
1173     {
1174       FlushAsyncData *data;
1175       data = g_new0 (FlushAsyncData, 1);
1176       data->worker = _g_dbus_worker_ref (worker);
1177       data->flushers = flushers;
1178       /* flush the stream before writing the next message */
1179       g_output_stream_flush_async (g_io_stream_get_output_stream (worker->stream),
1180                                    G_PRIORITY_DEFAULT,
1181                                    worker->cancellable,
1182                                    ostream_flush_cb,
1183                                    data);
1184     }
1185   else
1186     {
1187       /* kick off the next write! */
1188       maybe_write_next_message (worker);
1189     }
1190 }
1191
1192 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1193 static void
1194 write_message_cb (GObject       *source_object,
1195                   GAsyncResult  *res,
1196                   gpointer       user_data)
1197 {
1198   MessageToWriteData *data = user_data;
1199   GError *error;
1200
1201   g_mutex_lock (data->worker->write_lock);
1202   data->worker->num_writes_pending -= 1;
1203   g_mutex_unlock (data->worker->write_lock);
1204
1205   error = NULL;
1206   if (!write_message_finish (res, &error))
1207     {
1208       /* TODO: handle */
1209       _g_dbus_worker_emit_disconnected (data->worker, TRUE, error);
1210       g_error_free (error);
1211     }
1212
1213   /* this function will also kick of the next write (it might need to
1214    * flush so writing the next message might happen much later
1215    * e.g. async)
1216    */
1217   message_written (data->worker, data);
1218
1219   message_to_write_data_free (data);
1220 }
1221
1222 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1223 static void
1224 maybe_write_next_message (GDBusWorker *worker)
1225 {
1226   MessageToWriteData *data;
1227
1228  write_next:
1229
1230   g_mutex_lock (worker->write_lock);
1231   data = g_queue_pop_head (worker->write_queue);
1232   if (data != NULL)
1233     worker->num_writes_pending += 1;
1234   g_mutex_unlock (worker->write_lock);
1235
1236   /* Note that write_lock is only used for protecting the @write_queue
1237    * and @num_writes_pending fields of the GDBusWorker struct ... which we
1238    * need to modify from arbitrary threads in _g_dbus_worker_send_message().
1239    *
1240    * Therefore, it's fine to drop it here when calling back into user
1241    * code and then writing the message out onto the GIOStream since this
1242    * function only runs on the worker thread.
1243    */
1244   if (data != NULL)
1245     {
1246       GDBusMessageFilterResult filter_result;
1247       guchar *new_blob;
1248       gsize new_blob_size;
1249       GError *error;
1250
1251       filter_result = _g_dbus_worker_emit_message_about_to_be_sent (worker, data->message);
1252       switch (filter_result)
1253         {
1254         case G_DBUS_MESSAGE_FILTER_RESULT_NO_EFFECT:
1255           /* do nothing */
1256           break;
1257
1258         case G_DBUS_MESSAGE_FILTER_RESULT_MESSAGE_CONSUMED:
1259           /* drop message */
1260           g_mutex_lock (worker->write_lock);
1261           worker->num_writes_pending -= 1;
1262           g_mutex_unlock (worker->write_lock);
1263           message_to_write_data_free (data);
1264           goto write_next;
1265
1266         case G_DBUS_MESSAGE_FILTER_RESULT_MESSAGE_ALTERED:
1267           /* reencode altered message */
1268           error = NULL;
1269           new_blob = g_dbus_message_to_blob (data->message,
1270                                              &new_blob_size,
1271                                              worker->capabilities,
1272                                              &error);
1273           if (new_blob == NULL)
1274             {
1275               /* if filter make the GDBusMessage unencodeable, just complain on stderr and send
1276                * the old message instead
1277                */
1278               g_warning ("Error encoding GDBusMessage with serial %d altered by filter function: %s",
1279                          g_dbus_message_get_serial (data->message),
1280                          error->message);
1281               g_error_free (error);
1282             }
1283           else
1284             {
1285               g_free (data->blob);
1286               data->blob = (gchar *) new_blob;
1287               data->blob_size = new_blob_size;
1288             }
1289           break;
1290         }
1291
1292       write_message_async (worker,
1293                            data,
1294                            write_message_cb,
1295                            data);
1296     }
1297 }
1298
1299 /* called in private thread shared by all GDBusConnection instances (without write-lock held) */
1300 static gboolean
1301 write_message_in_idle_cb (gpointer user_data)
1302 {
1303   GDBusWorker *worker = user_data;
1304   if (worker->num_writes_pending == 0)
1305     maybe_write_next_message (worker);
1306   return FALSE;
1307 }
1308
1309 /* ---------------------------------------------------------------------------------------------------- */
1310
1311 /* can be called from any thread - steals blob */
1312 void
1313 _g_dbus_worker_send_message (GDBusWorker    *worker,
1314                              GDBusMessage   *message,
1315                              gchar          *blob,
1316                              gsize           blob_len)
1317 {
1318   MessageToWriteData *data;
1319
1320   g_return_if_fail (G_IS_DBUS_MESSAGE (message));
1321   g_return_if_fail (blob != NULL);
1322   g_return_if_fail (blob_len > 16);
1323
1324   data = g_new0 (MessageToWriteData, 1);
1325   data->worker = _g_dbus_worker_ref (worker);
1326   data->message = g_object_ref (message);
1327   data->blob = blob; /* steal! */
1328   data->blob_size = blob_len;
1329
1330   g_mutex_lock (worker->write_lock);
1331   g_queue_push_tail (worker->write_queue, data);
1332   if (worker->num_writes_pending == 0)
1333     {
1334       GSource *idle_source;
1335       idle_source = g_idle_source_new ();
1336       g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
1337       g_source_set_callback (idle_source,
1338                              write_message_in_idle_cb,
1339                              _g_dbus_worker_ref (worker),
1340                              (GDestroyNotify) _g_dbus_worker_unref);
1341       g_source_attach (idle_source, shared_thread_data->context);
1342       g_source_unref (idle_source);
1343     }
1344   g_mutex_unlock (worker->write_lock);
1345 }
1346
1347 /* ---------------------------------------------------------------------------------------------------- */
1348
1349 static void
1350 _g_dbus_worker_thread_begin_func (gpointer user_data)
1351 {
1352   GDBusWorker *worker = user_data;
1353
1354   worker->thread = g_thread_self ();
1355
1356   /* begin reading */
1357   _g_dbus_worker_do_read (worker);
1358 }
1359
1360 GDBusWorker *
1361 _g_dbus_worker_new (GIOStream                              *stream,
1362                     GDBusCapabilityFlags                    capabilities,
1363                     gboolean                                initially_frozen,
1364                     GDBusWorkerMessageReceivedCallback      message_received_callback,
1365                     GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback,
1366                     GDBusWorkerDisconnectedCallback         disconnected_callback,
1367                     gpointer                                user_data)
1368 {
1369   GDBusWorker *worker;
1370
1371   g_return_val_if_fail (G_IS_IO_STREAM (stream), NULL);
1372   g_return_val_if_fail (message_received_callback != NULL, NULL);
1373   g_return_val_if_fail (message_about_to_be_sent_callback != NULL, NULL);
1374   g_return_val_if_fail (disconnected_callback != NULL, NULL);
1375
1376   worker = g_new0 (GDBusWorker, 1);
1377   worker->ref_count = 1;
1378
1379   worker->read_lock = g_mutex_new ();
1380   worker->message_received_callback = message_received_callback;
1381   worker->message_about_to_be_sent_callback = message_about_to_be_sent_callback;
1382   worker->disconnected_callback = disconnected_callback;
1383   worker->user_data = user_data;
1384   worker->stream = g_object_ref (stream);
1385   worker->capabilities = capabilities;
1386   worker->cancellable = g_cancellable_new ();
1387
1388   worker->frozen = initially_frozen;
1389   worker->received_messages_while_frozen = g_queue_new ();
1390
1391   worker->write_lock = g_mutex_new ();
1392   worker->write_queue = g_queue_new ();
1393
1394   if (G_IS_SOCKET_CONNECTION (worker->stream))
1395     worker->socket = g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream));
1396
1397   _g_dbus_shared_thread_ref (_g_dbus_worker_thread_begin_func, worker);
1398
1399   return worker;
1400 }
1401
1402 /* ---------------------------------------------------------------------------------------------------- */
1403
1404 /* This can be called from any thread - frees worker - guarantees no callbacks
1405  * will ever be issued again
1406  */
1407 void
1408 _g_dbus_worker_stop (GDBusWorker *worker)
1409 {
1410   /* If we're called in the worker thread it means we are called from
1411    * a worker callback. This is fine, we just can't lock in that case since
1412    * we're already holding the lock...
1413    */
1414   if (g_thread_self () != worker->thread)
1415     g_mutex_lock (worker->read_lock);
1416   worker->stopped = TRUE;
1417   if (g_thread_self () != worker->thread)
1418     g_mutex_unlock (worker->read_lock);
1419
1420   g_cancellable_cancel (worker->cancellable);
1421   _g_dbus_worker_unref (worker);
1422 }
1423
1424 /* ---------------------------------------------------------------------------------------------------- */
1425
1426 /* can be called from any thread (except the worker thread) - blocks
1427  * calling thread until all queued outgoing messages are written and
1428  * the transport has been flushed
1429  */
1430 gboolean
1431 _g_dbus_worker_flush_sync (GDBusWorker    *worker,
1432                            GCancellable   *cancellable,
1433                            GError        **error)
1434 {
1435   gboolean ret;
1436   FlushData *data;
1437
1438   data = NULL;
1439   ret = TRUE;
1440
1441   /* if the queue is empty, there's nothing to wait for */
1442   g_mutex_lock (worker->write_lock);
1443   if (g_queue_get_length (worker->write_queue) > 0)
1444     {
1445       data = g_new0 (FlushData, 1);
1446       data->mutex = g_mutex_new ();
1447       data->cond = g_cond_new ();
1448       data->number_to_wait_for = worker->write_num_messages_written + g_queue_get_length (worker->write_queue);
1449       g_mutex_lock (data->mutex);
1450       worker->write_pending_flushes = g_list_prepend (worker->write_pending_flushes, data);
1451     }
1452   g_mutex_unlock (worker->write_lock);
1453
1454   if (data != NULL)
1455     {
1456       g_cond_wait (data->cond, data->mutex);
1457       g_mutex_unlock (data->mutex);
1458
1459       /* note:the element is removed from worker->write_pending_flushes in flush_cb() above */
1460       g_cond_free (data->cond);
1461       g_mutex_free (data->mutex);
1462       if (data->error != NULL)
1463         {
1464           ret = FALSE;
1465           g_propagate_error (error, data->error);
1466         }
1467       g_free (data);
1468     }
1469
1470   return ret;
1471 }
1472
1473 /* ---------------------------------------------------------------------------------------------------- */
1474
1475 #define G_DBUS_DEBUG_AUTHENTICATION (1<<0)
1476 #define G_DBUS_DEBUG_TRANSPORT      (1<<1)
1477 #define G_DBUS_DEBUG_MESSAGE        (1<<2)
1478 #define G_DBUS_DEBUG_PAYLOAD        (1<<3)
1479 #define G_DBUS_DEBUG_CALL           (1<<4)
1480 #define G_DBUS_DEBUG_SIGNAL         (1<<5)
1481 #define G_DBUS_DEBUG_INCOMING       (1<<6)
1482 #define G_DBUS_DEBUG_RETURN         (1<<7)
1483 #define G_DBUS_DEBUG_EMISSION       (1<<8)
1484 #define G_DBUS_DEBUG_ADDRESS        (1<<9)
1485
1486 static gint _gdbus_debug_flags = 0;
1487
1488 gboolean
1489 _g_dbus_debug_authentication (void)
1490 {
1491   _g_dbus_initialize ();
1492   return (_gdbus_debug_flags & G_DBUS_DEBUG_AUTHENTICATION) != 0;
1493 }
1494
1495 gboolean
1496 _g_dbus_debug_transport (void)
1497 {
1498   _g_dbus_initialize ();
1499   return (_gdbus_debug_flags & G_DBUS_DEBUG_TRANSPORT) != 0;
1500 }
1501
1502 gboolean
1503 _g_dbus_debug_message (void)
1504 {
1505   _g_dbus_initialize ();
1506   return (_gdbus_debug_flags & G_DBUS_DEBUG_MESSAGE) != 0;
1507 }
1508
1509 gboolean
1510 _g_dbus_debug_payload (void)
1511 {
1512   _g_dbus_initialize ();
1513   return (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD) != 0;
1514 }
1515
1516 gboolean
1517 _g_dbus_debug_call (void)
1518 {
1519   _g_dbus_initialize ();
1520   return (_gdbus_debug_flags & G_DBUS_DEBUG_CALL) != 0;
1521 }
1522
1523 gboolean
1524 _g_dbus_debug_signal (void)
1525 {
1526   _g_dbus_initialize ();
1527   return (_gdbus_debug_flags & G_DBUS_DEBUG_SIGNAL) != 0;
1528 }
1529
1530 gboolean
1531 _g_dbus_debug_incoming (void)
1532 {
1533   _g_dbus_initialize ();
1534   return (_gdbus_debug_flags & G_DBUS_DEBUG_INCOMING) != 0;
1535 }
1536
1537 gboolean
1538 _g_dbus_debug_return (void)
1539 {
1540   _g_dbus_initialize ();
1541   return (_gdbus_debug_flags & G_DBUS_DEBUG_RETURN) != 0;
1542 }
1543
1544 gboolean
1545 _g_dbus_debug_emission (void)
1546 {
1547   _g_dbus_initialize ();
1548   return (_gdbus_debug_flags & G_DBUS_DEBUG_EMISSION) != 0;
1549 }
1550
1551 gboolean
1552 _g_dbus_debug_address (void)
1553 {
1554   _g_dbus_initialize ();
1555   return (_gdbus_debug_flags & G_DBUS_DEBUG_ADDRESS) != 0;
1556 }
1557
1558 G_LOCK_DEFINE_STATIC (print_lock);
1559
1560 void
1561 _g_dbus_debug_print_lock (void)
1562 {
1563   G_LOCK (print_lock);
1564 }
1565
1566 void
1567 _g_dbus_debug_print_unlock (void)
1568 {
1569   G_UNLOCK (print_lock);
1570 }
1571
1572 /*
1573  * _g_dbus_initialize:
1574  *
1575  * Does various one-time init things such as
1576  *
1577  *  - registering the G_DBUS_ERROR error domain
1578  *  - parses the G_DBUS_DEBUG environment variable
1579  */
1580 void
1581 _g_dbus_initialize (void)
1582 {
1583   static volatile gsize initialized = 0;
1584
1585   if (g_once_init_enter (&initialized))
1586     {
1587       volatile GQuark g_dbus_error_domain;
1588       const gchar *debug;
1589
1590       g_dbus_error_domain = G_DBUS_ERROR;
1591
1592       debug = g_getenv ("G_DBUS_DEBUG");
1593       if (debug != NULL)
1594         {
1595           const GDebugKey keys[] = {
1596             { "authentication", G_DBUS_DEBUG_AUTHENTICATION },
1597             { "transport",      G_DBUS_DEBUG_TRANSPORT      },
1598             { "message",        G_DBUS_DEBUG_MESSAGE        },
1599             { "payload",        G_DBUS_DEBUG_PAYLOAD        },
1600             { "call",           G_DBUS_DEBUG_CALL           },
1601             { "signal",         G_DBUS_DEBUG_SIGNAL         },
1602             { "incoming",       G_DBUS_DEBUG_INCOMING       },
1603             { "return",         G_DBUS_DEBUG_RETURN         },
1604             { "emission",       G_DBUS_DEBUG_EMISSION       },
1605             { "address",        G_DBUS_DEBUG_ADDRESS        }
1606           };
1607
1608           _gdbus_debug_flags = g_parse_debug_string (debug, keys, G_N_ELEMENTS (keys));
1609           if (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD)
1610             _gdbus_debug_flags |= G_DBUS_DEBUG_MESSAGE;
1611         }
1612
1613       g_once_init_leave (&initialized, 1);
1614     }
1615 }
1616
1617 /* ---------------------------------------------------------------------------------------------------- */
1618
1619 GVariantType *
1620 _g_dbus_compute_complete_signature (GDBusArgInfo **args)
1621 {
1622   const GVariantType *arg_types[256];
1623   guint n;
1624
1625   if (args)
1626     for (n = 0; args[n] != NULL; n++)
1627       {
1628         /* DBus places a hard limit of 255 on signature length.
1629          * therefore number of args must be less than 256.
1630          */
1631         g_assert (n < 256);
1632
1633         arg_types[n] = G_VARIANT_TYPE (args[n]->signature);
1634
1635         if G_UNLIKELY (arg_types[n] == NULL)
1636           return NULL;
1637       }
1638   else
1639     n = 0;
1640
1641   return g_variant_type_new_tuple (arg_types, n);
1642 }
1643
1644 /* ---------------------------------------------------------------------------------------------------- */
1645
1646 #ifdef G_OS_WIN32
1647
1648 extern BOOL WINAPI ConvertSidToStringSidA (PSID Sid, LPSTR *StringSid);
1649
1650 gchar *
1651 _g_dbus_win32_get_user_sid (void)
1652 {
1653   HANDLE h;
1654   TOKEN_USER *user;
1655   DWORD token_information_len;
1656   PSID psid;
1657   gchar *sid;
1658   gchar *ret;
1659
1660   ret = NULL;
1661   user = NULL;
1662   h = INVALID_HANDLE_VALUE;
1663
1664   if (!OpenProcessToken (GetCurrentProcess (), TOKEN_QUERY, &h))
1665     {
1666       g_warning ("OpenProcessToken failed with error code %d", (gint) GetLastError ());
1667       goto out;
1668     }
1669
1670   /* Get length of buffer */
1671   token_information_len = 0;
1672   if (!GetTokenInformation (h, TokenUser, NULL, 0, &token_information_len))
1673     {
1674       if (GetLastError () != ERROR_INSUFFICIENT_BUFFER)
1675         {
1676           g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
1677           goto out;
1678         }
1679     }
1680   user = g_malloc (token_information_len);
1681   if (!GetTokenInformation (h, TokenUser, user, token_information_len, &token_information_len))
1682     {
1683       g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
1684       goto out;
1685     }
1686
1687   psid = user->User.Sid;
1688   if (!IsValidSid (psid))
1689     {
1690       g_warning ("Invalid SID");
1691       goto out;
1692     }
1693
1694   if (!ConvertSidToStringSidA (psid, &sid))
1695     {
1696       g_warning ("Invalid SID");
1697       goto out;
1698     }
1699
1700   ret = g_strdup (sid);
1701   LocalFree (sid);
1702
1703 out:
1704   g_free (user);
1705   if (h != INVALID_HANDLE_VALUE)
1706     CloseHandle (h);
1707   return ret;
1708 }
1709 #endif
1710
1711 /* ---------------------------------------------------------------------------------------------------- */
1712
1713 gchar *
1714 _g_dbus_get_machine_id (GError **error)
1715 {
1716   gchar *ret;
1717   /* TODO: use PACKAGE_LOCALSTATEDIR ? */
1718   ret = NULL;
1719   if (!g_file_get_contents ("/var/lib/dbus/machine-id",
1720                             &ret,
1721                             NULL,
1722                             error))
1723     {
1724       g_prefix_error (error, _("Unable to load /var/lib/dbus/machine-id: "));
1725     }
1726   else
1727     {
1728       /* TODO: validate value */
1729       g_strstrip (ret);
1730     }
1731   return ret;
1732 }
1733
1734 /* ---------------------------------------------------------------------------------------------------- */
1735
1736 gchar *
1737 _g_dbus_enum_to_string (GType enum_type, gint value)
1738 {
1739   gchar *ret;
1740   GEnumClass *klass;
1741   GEnumValue *enum_value;
1742
1743   klass = g_type_class_ref (enum_type);
1744   enum_value = g_enum_get_value (klass, value);
1745   if (enum_value != NULL)
1746     ret = g_strdup (enum_value->value_nick);
1747   else
1748     ret = g_strdup_printf ("unknown (value %d)", value);
1749   g_type_class_unref (klass);
1750   return ret;
1751 }
1752
1753 /* ---------------------------------------------------------------------------------------------------- */
1754
1755 static void
1756 write_message_print_transport_debug (gssize bytes_written,
1757                                      MessageToWriteData *data)
1758 {
1759   if (G_LIKELY (!_g_dbus_debug_transport ()))
1760     goto out;
1761
1762   _g_dbus_debug_print_lock ();
1763   g_print ("========================================================================\n"
1764            "GDBus-debug:Transport:\n"
1765            "  >>>> WROTE %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
1766            "       size %" G_GSIZE_FORMAT " from offset %" G_GSIZE_FORMAT " on a %s\n",
1767            bytes_written,
1768            g_dbus_message_get_serial (data->message),
1769            data->blob_size,
1770            data->total_written,
1771            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
1772   _g_dbus_debug_print_unlock ();
1773  out:
1774   ;
1775 }
1776
1777 /* ---------------------------------------------------------------------------------------------------- */
1778
1779 static void
1780 read_message_print_transport_debug (gssize bytes_read,
1781                                     GDBusWorker *worker)
1782 {
1783   gsize size;
1784   gint32 serial;
1785   gint32 message_length;
1786
1787   if (G_LIKELY (!_g_dbus_debug_transport ()))
1788     goto out;
1789
1790   size = bytes_read + worker->read_buffer_cur_size;
1791   serial = 0;
1792   message_length = 0;
1793   if (size >= 16)
1794     message_length = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer, size, NULL);
1795   if (size >= 1)
1796     {
1797       switch (worker->read_buffer[0])
1798         {
1799         case 'l':
1800           if (size >= 12)
1801             serial = GUINT32_FROM_LE (((guint32 *) worker->read_buffer)[2]);
1802           break;
1803         case 'B':
1804           if (size >= 12)
1805             serial = GUINT32_FROM_BE (((guint32 *) worker->read_buffer)[2]);
1806           break;
1807         default:
1808           /* an error will be set elsewhere if this happens */
1809           goto out;
1810         }
1811     }
1812
1813     _g_dbus_debug_print_lock ();
1814   g_print ("========================================================================\n"
1815            "GDBus-debug:Transport:\n"
1816            "  <<<< READ %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
1817            "       size %d to offset %" G_GSIZE_FORMAT " from a %s\n",
1818            bytes_read,
1819            serial,
1820            message_length,
1821            worker->read_buffer_cur_size,
1822            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker->stream))));
1823   _g_dbus_debug_print_unlock ();
1824  out:
1825   ;
1826 }