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