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, see <http://www.gnu.org/licenses/>.
  17  *
  18  * Author: David Zeuthen <davidz@redhat.com>
  19  */
  20
  21 #include "config.h"
  22
  23 #include <stdlib.h>
  24 #include <string.h>
  25
  26 #include "giotypes.h"
  27 #include "gsocket.h"
  28 #include "gdbusprivate.h"
  29 #include "gdbusmessage.h"
  30 #include "gdbusconnection.h"
  31 #include "gdbusproxy.h"
  32 #include "gdbuserror.h"
  33 #include "gdbusintrospection.h"
  34 #include "gtask.h"
  35 #include "ginputstream.h"
  36 #include "gmemoryinputstream.h"
  37 #include "giostream.h"
  38 #include "glib/gstdio.h"
  39 #include "gsocketcontrolmessage.h"
  40 #include "gsocketconnection.h"
  41 #include "gsocketoutputstream.h"
  42
  43 #ifdef G_OS_UNIX
  44 #include "gunixfdmessage.h"
  45 #include "gunixconnection.h"
  46 #include "gunixcredentialsmessage.h"
  47 #endif
  48
  49 #ifdef G_OS_WIN32
  50 #include <windows.h>
  51 #endif
  52
  53 #include "glibintl.h"
  54
  55 static gboolean _g_dbus_worker_do_initial_read (gpointer data);
  56 static void schedule_pending_close (GDBusWorker *worker);
  57
  58 /* ---------------------------------------------------------------------------------------------------- */
  59
  60 gchar *
  61 _g_dbus_hexdump (const gchar *data, gsize len, guint indent)
  62 {
  63  guint n, m;
  64  GString *ret;
  65
  66  ret = g_string_new (NULL);
  67
  68  for (n = 0; n < len; n += 16)
  69    {
  70      g_string_append_printf (ret, "%*s%04x: ", indent, "", n);
  71
  72      for (m = n; m < n + 16; m++)
  73        {
  74          if (m > n && (m%4) == 0)
  75            g_string_append_c (ret, ' ');
  76          if (m < len)
  77            g_string_append_printf (ret, "%02x ", (guchar) data[m]);
  78          else
  79            g_string_append (ret, "   ");
  80        }
  81
  82      g_string_append (ret, "   ");
  83
  84      for (m = n; m < len && m < n + 16; m++)
  85        g_string_append_c (ret, g_ascii_isprint (data[m]) ? data[m] : '.');
  86
  87      g_string_append_c (ret, '\n');
  88    }
  89
  90  return g_string_free (ret, FALSE);
  91 }
  92
  93 /* ---------------------------------------------------------------------------------------------------- */
  94
  95 /* Unfortunately ancillary messages are discarded when reading from a
  96  * socket using the GSocketInputStream abstraction. So we provide a
  97  * very GInputStream-ish API that uses GSocket in this case (very
  98  * similar to GSocketInputStream).
  99  */
 100
 101 typedef struct
 102 {
 103   void *buffer;
 104   gsize count;
 105
 106   GSocketControlMessage ***messages;
 107   gint *num_messages;
 108 } ReadWithControlData;
 109
 110 static void
 111 read_with_control_data_free (ReadWithControlData *data)
 112 {
 113   g_slice_free (ReadWithControlData, data);
 114 }
 115
 116 static gboolean
 117 _g_socket_read_with_control_messages_ready (GSocket      *socket,
 118                                             GIOCondition  condition,
 119                                             gpointer      user_data)
 120 {
 121   GTask *task = user_data;
 122   ReadWithControlData *data = g_task_get_task_data (task);
 123   GError *error;
 124   gssize result;
 125   GInputVector vector;
 126
 127   error = NULL;
 128   vector.buffer = data->buffer;
 129   vector.size = data->count;
 130   result = g_socket_receive_message (socket,
 131                                      NULL, /* address */
 132                                      &vector,
 133                                      1,
 134                                      data->messages,
 135                                      data->num_messages,
 136                                      NULL,
 137                                      g_task_get_cancellable (task),
 138                                      &error);
 139
 140   if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
 141     {
 142       g_error_free (error);
 143       return TRUE;
 144     }
 145
 146   g_assert (result >= 0 || error != NULL);
 147   if (result >= 0)
 148     g_task_return_int (task, result);
 149   else
 150     g_task_return_error (task, error);
 151   g_object_unref (task);
 152
 153   return FALSE;
 154 }
 155
 156 static void
 157 _g_socket_read_with_control_messages (GSocket                 *socket,
 158                                       void                    *buffer,
 159                                       gsize                    count,
 160                                       GSocketControlMessage ***messages,
 161                                       gint                    *num_messages,
 162                                       gint                     io_priority,
 163                                       GCancellable            *cancellable,
 164                                       GAsyncReadyCallback      callback,
 165                                       gpointer                 user_data)
 166 {
 167   GTask *task;
 168   ReadWithControlData *data;
 169   GSource *source;
 170
 171   data = g_slice_new0 (ReadWithControlData);
 172   data->buffer = buffer;
 173   data->count = count;
 174   data->messages = messages;
 175   data->num_messages = num_messages;
 176
 177   task = g_task_new (socket, cancellable, callback, user_data);
 178   g_task_set_source_tag (task, _g_socket_read_with_control_messages);
 179   g_task_set_task_data (task, data, (GDestroyNotify) read_with_control_data_free);
 180
 181   if (g_socket_condition_check (socket, G_IO_IN))
 182     {
 183       if (!_g_socket_read_with_control_messages_ready (socket, G_IO_IN, task))
 184         return;
 185     }
 186
 187   source = g_socket_create_source (socket,
 188                                    G_IO_IN | G_IO_HUP | G_IO_ERR,
 189                                    cancellable);
 190   g_task_attach_source (task, source, (GSourceFunc) _g_socket_read_with_control_messages_ready);
 191   g_source_unref (source);
 192 }
 193
 194 static gssize
 195 _g_socket_read_with_control_messages_finish (GSocket       *socket,
 196                                              GAsyncResult  *result,
 197                                              GError       **error)
 198 {
 199   g_return_val_if_fail (G_IS_SOCKET (socket), -1);
 200   g_return_val_if_fail (g_task_is_valid (result, socket), -1);
 201
 202   return g_task_propagate_int (G_TASK (result), error);
 203 }
 204
 205 /* ---------------------------------------------------------------------------------------------------- */
 206
 207 /* Work-around for https://bugzilla.gnome.org/show_bug.cgi?id=674885
 208    and see also the original https://bugzilla.gnome.org/show_bug.cgi?id=627724  */
 209
 210 static GPtrArray *ensured_classes = NULL;
 211
 212 static void
 213 ensure_type (GType gtype)
 214 {
 215   g_ptr_array_add (ensured_classes, g_type_class_ref (gtype));
 216 }
 217
 218 static void
 219 release_required_types (void)
 220 {
 221   g_ptr_array_foreach (ensured_classes, (GFunc) g_type_class_unref, NULL);
 222   g_ptr_array_unref (ensured_classes);
 223   ensured_classes = NULL;
 224 }
 225
 226 static void
 227 ensure_required_types (void)
 228 {
 229   g_assert (ensured_classes == NULL);
 230   ensured_classes = g_ptr_array_new ();
 231   ensure_type (G_TYPE_TASK);
 232   ensure_type (G_TYPE_MEMORY_INPUT_STREAM);
 233   ensure_type (G_TYPE_DBUS_CONNECTION);
 234   ensure_type (G_TYPE_DBUS_PROXY);
 235 }
 236 /* ---------------------------------------------------------------------------------------------------- */
 237
 238 typedef struct
 239 {
 240   volatile gint refcount;
 241   GThread *thread;
 242   GMainContext *context;
 243   GMainLoop *loop;
 244 } SharedThreadData;
 245
 246 static gpointer
 247 gdbus_shared_thread_func (gpointer user_data)
 248 {
 249   SharedThreadData *data = user_data;
 250
 251   g_main_context_push_thread_default (data->context);
 252   g_main_loop_run (data->loop);
 253   g_main_context_pop_thread_default (data->context);
 254
 255   release_required_types ();
 256
 257   return NULL;
 258 }
 259
 260 /* ---------------------------------------------------------------------------------------------------- */
 261
 262 static SharedThreadData *
 263 _g_dbus_shared_thread_ref (void)
 264 {
 265   static gsize shared_thread_data = 0;
 266   SharedThreadData *ret;
 267
 268   if (g_once_init_enter (&shared_thread_data))
 269     {
 270       SharedThreadData *data;
 271
 272       data = g_new0 (SharedThreadData, 1);
 273       data->refcount = 0;
 274
 275       data->context = g_main_context_new ();
 276       data->loop = g_main_loop_new (data->context, FALSE);
 277       data->thread = g_thread_new ("gdbus",
 278                                    gdbus_shared_thread_func,
 279                                    data);
 280       /* We can cast between gsize and gpointer safely */
 281       g_once_init_leave (&shared_thread_data, (gsize) data);
 282     }
 283
 284   ret = (SharedThreadData*) shared_thread_data;
 285   g_atomic_int_inc (&ret->refcount);
 286   return ret;
 287 }
 288
 289 static void
 290 _g_dbus_shared_thread_unref (SharedThreadData *data)
 291 {
 292   /* TODO: actually destroy the shared thread here */
 293 #if 0
 294   g_assert (data != NULL);
 295   if (g_atomic_int_dec_and_test (&data->refcount))
 296     {
 297       g_main_loop_quit (data->loop);
 298       //g_thread_join (data->thread);
 299       g_main_loop_unref (data->loop);
 300       g_main_context_unref (data->context);
 301     }
 302 #endif
 303 }
 304
 305 /* ---------------------------------------------------------------------------------------------------- */
 306
 307 typedef enum {
 308     PENDING_NONE = 0,
 309     PENDING_WRITE,
 310     PENDING_FLUSH,
 311     PENDING_CLOSE
 312 } OutputPending;
 313
 314 struct GDBusWorker
 315 {
 316   volatile gint                       ref_count;
 317
 318   SharedThreadData                   *shared_thread_data;
 319
 320   /* really a boolean, but GLib 2.28 lacks atomic boolean ops */
 321   volatile gint                       stopped;
 322
 323   /* TODO: frozen (e.g. G_DBUS_CONNECTION_FLAGS_DELAY_MESSAGE_PROCESSING) currently
 324    * only affects messages received from the other peer (since GDBusServer is the
 325    * only user) - we might want it to affect messages sent to the other peer too?
 326    */
 327   gboolean                            frozen;
 328   GDBusCapabilityFlags                capabilities;
 329   GQueue                             *received_messages_while_frozen;
 330
 331   GIOStream                          *stream;
 332   GCancellable                       *cancellable;
 333   GDBusWorkerMessageReceivedCallback  message_received_callback;
 334   GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback;
 335   GDBusWorkerDisconnectedCallback     disconnected_callback;
 336   gpointer                            user_data;
 337
 338   /* if not NULL, stream is GSocketConnection */
 339   GSocket *socket;
 340
 341   /* used for reading */
 342   GMutex                              read_lock;
 343   gchar                              *read_buffer;
 344   gsize                               read_buffer_allocated_size;
 345   gsize                               read_buffer_cur_size;
 346   gsize                               read_buffer_bytes_wanted;
 347   GUnixFDList                        *read_fd_list;
 348   GSocketControlMessage             **read_ancillary_messages;
 349   gint                                read_num_ancillary_messages;
 350
 351   /* Whether an async write, flush or close, or none of those, is pending.
 352    * Only the worker thread may change its value, and only with the write_lock.
 353    * Other threads may read its value when holding the write_lock.
 354    * The worker thread may read its value at any time.
 355    */
 356   OutputPending                       output_pending;
 357   /* used for writing */
 358   GMutex                              write_lock;
 359   /* queue of MessageToWriteData, protected by write_lock */
 360   GQueue                             *write_queue;
 361   /* protected by write_lock */
 362   guint64                             write_num_messages_written;
 363   /* number of messages we'd written out last time we flushed;
 364    * protected by write_lock
 365    */
 366   guint64                             write_num_messages_flushed;
 367   /* list of FlushData, protected by write_lock */
 368   GList                              *write_pending_flushes;
 369   /* list of CloseData, protected by write_lock */
 370   GList                              *pending_close_attempts;
 371   /* no lock - only used from the worker thread */
 372   gboolean                            close_expected;
 373 };
 374
 375 static void _g_dbus_worker_unref (GDBusWorker *worker);
 376
 377 /* ---------------------------------------------------------------------------------------------------- */
 378
 379 typedef struct
 380 {
 381   GMutex  mutex;
 382   GCond   cond;
 383   guint64 number_to_wait_for;
 384   GError *error;
 385 } FlushData;
 386
 387 struct _MessageToWriteData ;
 388 typedef struct _MessageToWriteData MessageToWriteData;
 389
 390 static void message_to_write_data_free (MessageToWriteData *data);
 391
 392 static void read_message_print_transport_debug (gssize bytes_read,
 393                                                 GDBusWorker *worker);
 394
 395 static void write_message_print_transport_debug (gssize bytes_written,
 396                                                  MessageToWriteData *data);
 397
 398 typedef struct {
 399     GDBusWorker *worker;
 400     GTask *task;
 401 } CloseData;
 402
 403 static void close_data_free (CloseData *close_data)
 404 {
 405   g_clear_object (&close_data->task);
 406
 407   _g_dbus_worker_unref (close_data->worker);
 408   g_slice_free (CloseData, close_data);
 409 }
 410
 411 /* ---------------------------------------------------------------------------------------------------- */
 412
 413 static GDBusWorker *
 414 _g_dbus_worker_ref (GDBusWorker *worker)
 415 {
 416   g_atomic_int_inc (&worker->ref_count);
 417   return worker;
 418 }
 419
 420 static void
 421 _g_dbus_worker_unref (GDBusWorker *worker)
 422 {
 423   if (g_atomic_int_dec_and_test (&worker->ref_count))
 424     {
 425       g_assert (worker->write_pending_flushes == NULL);
 426
 427       _g_dbus_shared_thread_unref (worker->shared_thread_data);
 428
 429       g_object_unref (worker->stream);
 430
 431       g_mutex_clear (&worker->read_lock);
 432       g_object_unref (worker->cancellable);
 433       if (worker->read_fd_list != NULL)
 434         g_object_unref (worker->read_fd_list);
 435
 436       g_queue_free_full (worker->received_messages_while_frozen, (GDestroyNotify) g_object_unref);
 437       g_mutex_clear (&worker->write_lock);
 438       g_queue_free_full (worker->write_queue, (GDestroyNotify) message_to_write_data_free);
 439       g_free (worker->read_buffer);
 440
 441       g_free (worker);
 442     }
 443 }
 444
 445 static void
 446 _g_dbus_worker_emit_disconnected (GDBusWorker  *worker,
 447                                   gboolean      remote_peer_vanished,
 448                                   GError       *error)
 449 {
 450   if (!g_atomic_int_get (&worker->stopped))
 451     worker->disconnected_callback (worker, remote_peer_vanished, error, worker->user_data);
 452 }
 453
 454 static void
 455 _g_dbus_worker_emit_message_received (GDBusWorker  *worker,
 456                                       GDBusMessage *message)
 457 {
 458   if (!g_atomic_int_get (&worker->stopped))
 459     worker->message_received_callback (worker, message, worker->user_data);
 460 }
 461
 462 static GDBusMessage *
 463 _g_dbus_worker_emit_message_about_to_be_sent (GDBusWorker  *worker,
 464                                               GDBusMessage *message)
 465 {
 466   GDBusMessage *ret;
 467   if (!g_atomic_int_get (&worker->stopped))
 468     ret = worker->message_about_to_be_sent_callback (worker, message, worker->user_data);
 469   else
 470     ret = message;
 471   return ret;
 472 }
 473
 474 /* can only be called from private thread with read-lock held - takes ownership of @message */
 475 static void
 476 _g_dbus_worker_queue_or_deliver_received_message (GDBusWorker  *worker,
 477                                                   GDBusMessage *message)
 478 {
 479   if (worker->frozen || g_queue_get_length (worker->received_messages_while_frozen) > 0)
 480     {
 481       /* queue up */
 482       g_queue_push_tail (worker->received_messages_while_frozen, message);
 483     }
 484   else
 485     {
 486       /* not frozen, nor anything in queue */
 487       _g_dbus_worker_emit_message_received (worker, message);
 488       g_object_unref (message);
 489     }
 490 }
 491
 492 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 493 static gboolean
 494 unfreeze_in_idle_cb (gpointer user_data)
 495 {
 496   GDBusWorker *worker = user_data;
 497   GDBusMessage *message;
 498
 499   g_mutex_lock (&worker->read_lock);
 500   if (worker->frozen)
 501     {
 502       while ((message = g_queue_pop_head (worker->received_messages_while_frozen)) != NULL)
 503         {
 504           _g_dbus_worker_emit_message_received (worker, message);
 505           g_object_unref (message);
 506         }
 507       worker->frozen = FALSE;
 508     }
 509   else
 510     {
 511       g_assert (g_queue_get_length (worker->received_messages_while_frozen) == 0);
 512     }
 513   g_mutex_unlock (&worker->read_lock);
 514   return FALSE;
 515 }
 516
 517 /* can be called from any thread */
 518 void
 519 _g_dbus_worker_unfreeze (GDBusWorker *worker)
 520 {
 521   GSource *idle_source;
 522   idle_source = g_idle_source_new ();
 523   g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
 524   g_source_set_callback (idle_source,
 525                          unfreeze_in_idle_cb,
 526                          _g_dbus_worker_ref (worker),
 527                          (GDestroyNotify) _g_dbus_worker_unref);
 528   g_source_set_name (idle_source, "[gio] unfreeze_in_idle_cb");
 529   g_source_attach (idle_source, worker->shared_thread_data->context);
 530   g_source_unref (idle_source);
 531 }
 532
 533 /* ---------------------------------------------------------------------------------------------------- */
 534
 535 static void _g_dbus_worker_do_read_unlocked (GDBusWorker *worker);
 536
 537 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 538 static void
 539 _g_dbus_worker_do_read_cb (GInputStream  *input_stream,
 540                            GAsyncResult  *res,
 541                            gpointer       user_data)
 542 {
 543   GDBusWorker *worker = user_data;
 544   GError *error;
 545   gssize bytes_read;
 546
 547   g_mutex_lock (&worker->read_lock);
 548
 549   /* If already stopped, don't even process the reply */
 550   if (g_atomic_int_get (&worker->stopped))
 551     goto out;
 552
 553   error = NULL;
 554   if (worker->socket == NULL)
 555     bytes_read = g_input_stream_read_finish (g_io_stream_get_input_stream (worker->stream),
 556                                              res,
 557                                              &error);
 558   else
 559     bytes_read = _g_socket_read_with_control_messages_finish (worker->socket,
 560                                                               res,
 561                                                               &error);
 562   if (worker->read_num_ancillary_messages > 0)
 563     {
 564       gint n;
 565       for (n = 0; n < worker->read_num_ancillary_messages; n++)
 566         {
 567           GSocketControlMessage *control_message = G_SOCKET_CONTROL_MESSAGE (worker->read_ancillary_messages[n]);
 568
 569           if (FALSE)
 570             {
 571             }
 572 #ifdef G_OS_UNIX
 573           else if (G_IS_UNIX_FD_MESSAGE (control_message))
 574             {
 575               GUnixFDMessage *fd_message;
 576               gint *fds;
 577               gint num_fds;
 578
 579               fd_message = G_UNIX_FD_MESSAGE (control_message);
 580               fds = g_unix_fd_message_steal_fds (fd_message, &num_fds);
 581               if (worker->read_fd_list == NULL)
 582                 {
 583                   worker->read_fd_list = g_unix_fd_list_new_from_array (fds, num_fds);
 584                 }
 585               else
 586                 {
 587                   gint n;
 588                   for (n = 0; n < num_fds; n++)
 589                     {
 590                       /* TODO: really want a append_steal() */
 591                       g_unix_fd_list_append (worker->read_fd_list, fds[n], NULL);
 592                       (void) g_close (fds[n], NULL);
 593                     }
 594                 }
 595               g_free (fds);
 596             }
 597           else if (G_IS_UNIX_CREDENTIALS_MESSAGE (control_message))
 598             {
 599               /* do nothing */
 600             }
 601 #endif
 602           else
 603             {
 604               if (error == NULL)
 605                 {
 606                   g_set_error (&error,
 607                                G_IO_ERROR,
 608                                G_IO_ERROR_FAILED,
 609                                "Unexpected ancillary message of type %s received from peer",
 610                                g_type_name (G_TYPE_FROM_INSTANCE (control_message)));
 611                   _g_dbus_worker_emit_disconnected (worker, TRUE, error);
 612                   g_error_free (error);
 613                   g_object_unref (control_message);
 614                   n++;
 615                   while (n < worker->read_num_ancillary_messages)
 616                     g_object_unref (worker->read_ancillary_messages[n++]);
 617                   g_free (worker->read_ancillary_messages);
 618                   goto out;
 619                 }
 620             }
 621           g_object_unref (control_message);
 622         }
 623       g_free (worker->read_ancillary_messages);
 624     }
 625
 626   if (bytes_read == -1)
 627     {
 628       if (G_UNLIKELY (_g_dbus_debug_transport ()))
 629         {
 630           _g_dbus_debug_print_lock ();
 631           g_print ("========================================================================\n"
 632                    "GDBus-debug:Transport:\n"
 633                    "  ---- READ ERROR on stream of type %s:\n"
 634                    "  ---- %s %d: %s\n",
 635                    g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker->stream))),
 636                    g_quark_to_string (error->domain), error->code,
 637                    error->message);
 638           _g_dbus_debug_print_unlock ();
 639         }
 640
 641       /* Every async read that uses this callback uses worker->cancellable
 642        * as its GCancellable. worker->cancellable gets cancelled if and only
 643        * if the GDBusConnection tells us to close (either via
 644        * _g_dbus_worker_stop, which is called on last-unref, or directly),
 645        * so a cancelled read must mean our connection was closed locally.
 646        *
 647        * If we're closing, other errors are possible - notably,
 648        * G_IO_ERROR_CLOSED can be seen if we close the stream with an async
 649        * read in-flight. It seems sensible to treat all read errors during
 650        * closing as an expected thing that doesn't trip exit-on-close.
 651        *
 652        * Because close_expected can't be set until we get into the worker
 653        * thread, but the cancellable is signalled sooner (from another
 654        * thread), we do still need to check the error.
 655        */
 656       if (worker->close_expected ||
 657           g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED))
 658         _g_dbus_worker_emit_disconnected (worker, FALSE, NULL);
 659       else
 660         _g_dbus_worker_emit_disconnected (worker, TRUE, error);
 661
 662       g_error_free (error);
 663       goto out;
 664     }
 665
 666 #if 0
 667   g_debug ("read %d bytes (is_closed=%d blocking=%d condition=0x%02x) stream %p, %p",
 668            (gint) bytes_read,
 669            g_socket_is_closed (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream))),
 670            g_socket_get_blocking (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream))),
 671            g_socket_condition_check (g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream)),
 672                                      G_IO_IN | G_IO_OUT | G_IO_HUP),
 673            worker->stream,
 674            worker);
 675 #endif
 676
 677   /* TODO: hmm, hmm... */
 678   if (bytes_read == 0)
 679     {
 680       g_set_error (&error,
 681                    G_IO_ERROR,
 682                    G_IO_ERROR_FAILED,
 683                    "Underlying GIOStream returned 0 bytes on an async read");
 684       _g_dbus_worker_emit_disconnected (worker, TRUE, error);
 685       g_error_free (error);
 686       goto out;
 687     }
 688
 689   read_message_print_transport_debug (bytes_read, worker);
 690
 691   worker->read_buffer_cur_size += bytes_read;
 692   if (worker->read_buffer_bytes_wanted == worker->read_buffer_cur_size)
 693     {
 694       /* OK, got what we asked for! */
 695       if (worker->read_buffer_bytes_wanted == 16)
 696         {
 697           gssize message_len;
 698           /* OK, got the header - determine how many more bytes are needed */
 699           error = NULL;
 700           message_len = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer,
 701                                                      16,
 702                                                      &error);
 703           if (message_len == -1)
 704             {
 705               g_warning ("_g_dbus_worker_do_read_cb: error determining bytes needed: %s", error->message);
 706               _g_dbus_worker_emit_disconnected (worker, FALSE, error);
 707               g_error_free (error);
 708               goto out;
 709             }
 710
 711           worker->read_buffer_bytes_wanted = message_len;
 712           _g_dbus_worker_do_read_unlocked (worker);
 713         }
 714       else
 715         {
 716           GDBusMessage *message;
 717           error = NULL;
 718
 719           /* TODO: use connection->priv->auth to decode the message */
 720
 721           message = g_dbus_message_new_from_blob ((guchar *) worker->read_buffer,
 722                                                   worker->read_buffer_cur_size,
 723                                                   worker->capabilities,
 724                                                   &error);
 725           if (message == NULL)
 726             {
 727               gchar *s;
 728               s = _g_dbus_hexdump (worker->read_buffer, worker->read_buffer_cur_size, 2);
 729               g_warning ("Error decoding D-Bus message of %" G_GSIZE_FORMAT " bytes\n"
 730                          "The error is: %s\n"
 731                          "The payload is as follows:\n"
 732                          "%s\n",
 733                          worker->read_buffer_cur_size,
 734                          error->message,
 735                          s);
 736               g_free (s);
 737               _g_dbus_worker_emit_disconnected (worker, FALSE, error);
 738               g_error_free (error);
 739               goto out;
 740             }
 741
 742 #ifdef G_OS_UNIX
 743           if (worker->read_fd_list != NULL)
 744             {
 745               g_dbus_message_set_unix_fd_list (message, worker->read_fd_list);
 746               g_object_unref (worker->read_fd_list);
 747               worker->read_fd_list = NULL;
 748             }
 749 #endif
 750
 751           if (G_UNLIKELY (_g_dbus_debug_message ()))
 752             {
 753               gchar *s;
 754               _g_dbus_debug_print_lock ();
 755               g_print ("========================================================================\n"
 756                        "GDBus-debug:Message:\n"
 757                        "  <<<< RECEIVED D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
 758                        worker->read_buffer_cur_size);
 759               s = g_dbus_message_print (message, 2);
 760               g_print ("%s", s);
 761               g_free (s);
 762               if (G_UNLIKELY (_g_dbus_debug_payload ()))
 763                 {
 764                   s = _g_dbus_hexdump (worker->read_buffer, worker->read_buffer_cur_size, 2);
 765                   g_print ("%s\n", s);
 766                   g_free (s);
 767                 }
 768               _g_dbus_debug_print_unlock ();
 769             }
 770
 771           /* yay, got a message, go deliver it */
 772           _g_dbus_worker_queue_or_deliver_received_message (worker, message);
 773
 774           /* start reading another message! */
 775           worker->read_buffer_bytes_wanted = 0;
 776           worker->read_buffer_cur_size = 0;
 777           _g_dbus_worker_do_read_unlocked (worker);
 778         }
 779     }
 780   else
 781     {
 782       /* didn't get all the bytes we requested - so repeat the request... */
 783       _g_dbus_worker_do_read_unlocked (worker);
 784     }
 785
 786  out:
 787   g_mutex_unlock (&worker->read_lock);
 788
 789   /* gives up the reference acquired when calling g_input_stream_read_async() */
 790   _g_dbus_worker_unref (worker);
 791
 792   /* check if there is any pending close */
 793   schedule_pending_close (worker);
 794 }
 795
 796 /* called in private thread shared by all GDBusConnection instances (with read-lock held) */
 797 static void
 798 _g_dbus_worker_do_read_unlocked (GDBusWorker *worker)
 799 {
 800   /* Note that we do need to keep trying to read even if close_expected is
 801    * true, because only failing a read causes us to signal 'closed'.
 802    */
 803
 804   /* if bytes_wanted is zero, it means start reading a message */
 805   if (worker->read_buffer_bytes_wanted == 0)
 806     {
 807       worker->read_buffer_cur_size = 0;
 808       worker->read_buffer_bytes_wanted = 16;
 809     }
 810
 811   /* ensure we have a (big enough) buffer */
 812   if (worker->read_buffer == NULL || worker->read_buffer_bytes_wanted > worker->read_buffer_allocated_size)
 813     {
 814       /* TODO: 4096 is randomly chosen; might want a better chosen default minimum */
 815       worker->read_buffer_allocated_size = MAX (worker->read_buffer_bytes_wanted, 4096);
 816       worker->read_buffer = g_realloc (worker->read_buffer, worker->read_buffer_allocated_size);
 817     }
 818
 819   if (worker->socket == NULL)
 820     g_input_stream_read_async (g_io_stream_get_input_stream (worker->stream),
 821                                worker->read_buffer + worker->read_buffer_cur_size,
 822                                worker->read_buffer_bytes_wanted - worker->read_buffer_cur_size,
 823                                G_PRIORITY_DEFAULT,
 824                                worker->cancellable,
 825                                (GAsyncReadyCallback) _g_dbus_worker_do_read_cb,
 826                                _g_dbus_worker_ref (worker));
 827   else
 828     {
 829       worker->read_ancillary_messages = NULL;
 830       worker->read_num_ancillary_messages = 0;
 831       _g_socket_read_with_control_messages (worker->socket,
 832                                             worker->read_buffer + worker->read_buffer_cur_size,
 833                                             worker->read_buffer_bytes_wanted - worker->read_buffer_cur_size,
 834                                             &worker->read_ancillary_messages,
 835                                             &worker->read_num_ancillary_messages,
 836                                             G_PRIORITY_DEFAULT,
 837                                             worker->cancellable,
 838                                             (GAsyncReadyCallback) _g_dbus_worker_do_read_cb,
 839                                             _g_dbus_worker_ref (worker));
 840     }
 841 }
 842
 843 /* called in private thread shared by all GDBusConnection instances (without read-lock held) */
 844 static gboolean
 845 _g_dbus_worker_do_initial_read (gpointer data)
 846 {
 847   GDBusWorker *worker = data;
 848   g_mutex_lock (&worker->read_lock);
 849   _g_dbus_worker_do_read_unlocked (worker);
 850   g_mutex_unlock (&worker->read_lock);
 851   return FALSE;
 852 }
 853
 854 /* ---------------------------------------------------------------------------------------------------- */
 855
 856 struct _MessageToWriteData
 857 {
 858   GDBusWorker  *worker;
 859   GDBusMessage *message;
 860   gchar        *blob;
 861   gsize         blob_size;
 862
 863   gsize         total_written;
 864   GTask        *task;
 865 };
 866
 867 static void
 868 message_to_write_data_free (MessageToWriteData *data)
 869 {
 870   _g_dbus_worker_unref (data->worker);
 871   if (data->message)
 872     g_object_unref (data->message);
 873   g_free (data->blob);
 874   g_slice_free (MessageToWriteData, data);
 875 }
 876
 877 /* ---------------------------------------------------------------------------------------------------- */
 878
 879 static void write_message_continue_writing (MessageToWriteData *data);
 880
 881 /* called in private thread shared by all GDBusConnection instances
 882  *
 883  * write-lock is not held on entry
 884  * output_pending is PENDING_WRITE on entry
 885  */
 886 static void
 887 write_message_async_cb (GObject      *source_object,
 888                         GAsyncResult *res,
 889                         gpointer      user_data)
 890 {
 891   MessageToWriteData *data = user_data;
 892   GTask *task;
 893   gssize bytes_written;
 894   GError *error;
 895
 896   /* Note: we can't access data->task after calling g_task_return_* () because the
 897    * callback can free @data and we're not completing in idle. So use a copy of the pointer.
 898    */
 899   task = data->task;
 900
 901   error = NULL;
 902   bytes_written = g_output_stream_write_finish (G_OUTPUT_STREAM (source_object),
 903                                                 res,
 904                                                 &error);
 905   if (bytes_written == -1)
 906     {
 907       g_task_return_error (task, error);
 908       g_object_unref (task);
 909       goto out;
 910     }
 911   g_assert (bytes_written > 0); /* zero is never returned */
 912
 913   write_message_print_transport_debug (bytes_written, data);
 914
 915   data->total_written += bytes_written;
 916   g_assert (data->total_written <= data->blob_size);
 917   if (data->total_written == data->blob_size)
 918     {
 919       g_task_return_boolean (task, TRUE);
 920       g_object_unref (task);
 921       goto out;
 922     }
 923
 924   write_message_continue_writing (data);
 925
 926  out:
 927   ;
 928 }
 929
 930 /* called in private thread shared by all GDBusConnection instances
 931  *
 932  * write-lock is not held on entry
 933  * output_pending is PENDING_WRITE on entry
 934  */
 935 #ifdef G_OS_UNIX
 936 static gboolean
 937 on_socket_ready (GSocket      *socket,
 938                  GIOCondition  condition,
 939                  gpointer      user_data)
 940 {
 941   MessageToWriteData *data = user_data;
 942   write_message_continue_writing (data);
 943   return FALSE; /* remove source */
 944 }
 945 #endif
 946
 947 /* called in private thread shared by all GDBusConnection instances
 948  *
 949  * write-lock is not held on entry
 950  * output_pending is PENDING_WRITE on entry
 951  */
 952 static void
 953 write_message_continue_writing (MessageToWriteData *data)
 954 {
 955   GOutputStream *ostream;
 956 #ifdef G_OS_UNIX
 957   GTask *task;
 958   GUnixFDList *fd_list;
 959 #endif
 960
 961 #ifdef G_OS_UNIX
 962   /* Note: we can't access data->task after calling g_task_return_* () because the
 963    * callback can free @data and we're not completing in idle. So use a copy of the pointer.
 964    */
 965   task = data->task;
 966 #endif
 967
 968   ostream = g_io_stream_get_output_stream (data->worker->stream);
 969 #ifdef G_OS_UNIX
 970   fd_list = g_dbus_message_get_unix_fd_list (data->message);
 971 #endif
 972
 973   g_assert (!g_output_stream_has_pending (ostream));
 974   g_assert_cmpint (data->total_written, <, data->blob_size);
 975
 976   if (FALSE)
 977     {
 978     }
 979 #ifdef G_OS_UNIX
 980   else if (G_IS_SOCKET_OUTPUT_STREAM (ostream) && data->total_written == 0)
 981     {
 982       GOutputVector vector;
 983       GSocketControlMessage *control_message;
 984       gssize bytes_written;
 985       GError *error;
 986
 987       vector.buffer = data->blob;
 988       vector.size = data->blob_size;
 989
 990       control_message = NULL;
 991       if (fd_list != NULL && g_unix_fd_list_get_length (fd_list) > 0)
 992         {
 993           if (!(data->worker->capabilities & G_DBUS_CAPABILITY_FLAGS_UNIX_FD_PASSING))
 994             {
 995               g_task_return_new_error (task,
 996                                        G_IO_ERROR,
 997                                        G_IO_ERROR_FAILED,
 998                                        "Tried sending a file descriptor but remote peer does not support this capability");
 999               g_object_unref (task);
1000               goto out;
1001             }
1002           control_message = g_unix_fd_message_new_with_fd_list (fd_list);
1003         }
1004
1005       error = NULL;
1006       bytes_written = g_socket_send_message (data->worker->socket,
1007                                              NULL, /* address */
1008                                              &vector,
1009                                              1,
1010                                              control_message != NULL ? &control_message : NULL,
1011                                              control_message != NULL ? 1 : 0,
1012                                              G_SOCKET_MSG_NONE,
1013                                              data->worker->cancellable,
1014                                              &error);
1015       if (control_message != NULL)
1016         g_object_unref (control_message);
1017
1018       if (bytes_written == -1)
1019         {
1020           /* Handle WOULD_BLOCK by waiting until there's room in the buffer */
1021           if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
1022             {
1023               GSource *source;
1024               source = g_socket_create_source (data->worker->socket,
1025                                                G_IO_OUT | G_IO_HUP | G_IO_ERR,
1026                                                data->worker->cancellable);
1027               g_source_set_callback (source,
1028                                      (GSourceFunc) on_socket_ready,
1029                                      data,
1030                                      NULL); /* GDestroyNotify */
1031               g_source_attach (source, g_main_context_get_thread_default ());
1032               g_source_unref (source);
1033               g_error_free (error);
1034               goto out;
1035             }
1036           g_task_return_error (task, error);
1037           g_object_unref (task);
1038           goto out;
1039         }
1040       g_assert (bytes_written > 0); /* zero is never returned */
1041
1042       write_message_print_transport_debug (bytes_written, data);
1043
1044       data->total_written += bytes_written;
1045       g_assert (data->total_written <= data->blob_size);
1046       if (data->total_written == data->blob_size)
1047         {
1048           g_task_return_boolean (task, TRUE);
1049           g_object_unref (task);
1050           goto out;
1051         }
1052
1053       write_message_continue_writing (data);
1054     }
1055 #endif
1056   else
1057     {
1058 #ifdef G_OS_UNIX
1059       if (fd_list != NULL)
1060         {
1061           g_task_return_new_error (task,
1062                                    G_IO_ERROR,
1063                                    G_IO_ERROR_FAILED,
1064                                    "Tried sending a file descriptor on unsupported stream of type %s",
1065                                    g_type_name (G_TYPE_FROM_INSTANCE (ostream)));
1066           g_object_unref (task);
1067           goto out;
1068         }
1069 #endif
1070
1071       g_output_stream_write_async (ostream,
1072                                    (const gchar *) data->blob + data->total_written,
1073                                    data->blob_size - data->total_written,
1074                                    G_PRIORITY_DEFAULT,
1075                                    data->worker->cancellable,
1076                                    write_message_async_cb,
1077                                    data);
1078     }
1079 #ifdef G_OS_UNIX
1080  out:
1081 #endif
1082   ;
1083 }
1084
1085 /* called in private thread shared by all GDBusConnection instances
1086  *
1087  * write-lock is not held on entry
1088  * output_pending is PENDING_WRITE on entry
1089  */
1090 static void
1091 write_message_async (GDBusWorker         *worker,
1092                      MessageToWriteData  *data,
1093                      GAsyncReadyCallback  callback,
1094                      gpointer             user_data)
1095 {
1096   data->task = g_task_new (NULL, NULL, callback, user_data);
1097   g_task_set_source_tag (data->task, write_message_async);
1098   data->total_written = 0;
1099   write_message_continue_writing (data);
1100 }
1101
1102 /* called in private thread shared by all GDBusConnection instances (with write-lock held) */
1103 static gboolean
1104 write_message_finish (GAsyncResult   *res,
1105                       GError        **error)
1106 {
1107   g_return_val_if_fail (g_task_is_valid (res, NULL), FALSE);
1108
1109   return g_task_propagate_boolean (G_TASK (res), error);
1110 }
1111 /* ---------------------------------------------------------------------------------------------------- */
1112
1113 static void continue_writing (GDBusWorker *worker);
1114
1115 typedef struct
1116 {
1117   GDBusWorker *worker;
1118   GList *flushers;
1119 } FlushAsyncData;
1120
1121 static void
1122 flush_data_list_complete (const GList  *flushers,
1123                           const GError *error)
1124 {
1125   const GList *l;
1126
1127   for (l = flushers; l != NULL; l = l->next)
1128     {
1129       FlushData *f = l->data;
1130
1131       f->error = error != NULL ? g_error_copy (error) : NULL;
1132
1133       g_mutex_lock (&f->mutex);
1134       g_cond_signal (&f->cond);
1135       g_mutex_unlock (&f->mutex);
1136     }
1137 }
1138
1139 /* called in private thread shared by all GDBusConnection instances
1140  *
1141  * write-lock is not held on entry
1142  * output_pending is PENDING_FLUSH on entry
1143  */
1144 static void
1145 ostream_flush_cb (GObject      *source_object,
1146                   GAsyncResult *res,
1147                   gpointer      user_data)
1148 {
1149   FlushAsyncData *data = user_data;
1150   GError *error;
1151
1152   error = NULL;
1153   g_output_stream_flush_finish (G_OUTPUT_STREAM (source_object),
1154                                 res,
1155                                 &error);
1156
1157   if (error == NULL)
1158     {
1159       if (G_UNLIKELY (_g_dbus_debug_transport ()))
1160         {
1161           _g_dbus_debug_print_lock ();
1162           g_print ("========================================================================\n"
1163                    "GDBus-debug:Transport:\n"
1164                    "  ---- FLUSHED stream of type %s\n",
1165                    g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
1166           _g_dbus_debug_print_unlock ();
1167         }
1168     }
1169
1170   g_assert (data->flushers != NULL);
1171   flush_data_list_complete (data->flushers, error);
1172   g_list_free (data->flushers);
1173
1174   if (error != NULL)
1175     g_error_free (error);
1176
1177   /* Make sure we tell folks that we don't have additional
1178      flushes pending */
1179   g_mutex_lock (&data->worker->write_lock);
1180   data->worker->write_num_messages_flushed = data->worker->write_num_messages_written;
1181   g_assert (data->worker->output_pending == PENDING_FLUSH);
1182   data->worker->output_pending = PENDING_NONE;
1183   g_mutex_unlock (&data->worker->write_lock);
1184
1185   /* OK, cool, finally kick off the next write */
1186   continue_writing (data->worker);
1187
1188   _g_dbus_worker_unref (data->worker);
1189   g_free (data);
1190 }
1191
1192 /* called in private thread shared by all GDBusConnection instances
1193  *
1194  * write-lock is not held on entry
1195  * output_pending is PENDING_FLUSH on entry
1196  */
1197 static void
1198 start_flush (FlushAsyncData *data)
1199 {
1200   g_output_stream_flush_async (g_io_stream_get_output_stream (data->worker->stream),
1201                                G_PRIORITY_DEFAULT,
1202                                data->worker->cancellable,
1203                                ostream_flush_cb,
1204                                data);
1205 }
1206
1207 /* called in private thread shared by all GDBusConnection instances
1208  *
1209  * write-lock is held on entry
1210  * output_pending is PENDING_NONE on entry
1211  */
1212 static void
1213 message_written_unlocked (GDBusWorker *worker,
1214                           MessageToWriteData *message_data)
1215 {
1216   if (G_UNLIKELY (_g_dbus_debug_message ()))
1217     {
1218       gchar *s;
1219       _g_dbus_debug_print_lock ();
1220       g_print ("========================================================================\n"
1221                "GDBus-debug:Message:\n"
1222                "  >>>> SENT D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
1223                message_data->blob_size);
1224       s = g_dbus_message_print (message_data->message, 2);
1225       g_print ("%s", s);
1226       g_free (s);
1227       if (G_UNLIKELY (_g_dbus_debug_payload ()))
1228         {
1229           s = _g_dbus_hexdump (message_data->blob, message_data->blob_size, 2);
1230           g_print ("%s\n", s);
1231           g_free (s);
1232         }
1233       _g_dbus_debug_print_unlock ();
1234     }
1235
1236   worker->write_num_messages_written += 1;
1237 }
1238
1239 /* called in private thread shared by all GDBusConnection instances
1240  *
1241  * write-lock is held on entry
1242  * output_pending is PENDING_NONE on entry
1243  *
1244  * Returns: non-%NULL, setting @output_pending, if we need to flush now
1245  */
1246 static FlushAsyncData *
1247 prepare_flush_unlocked (GDBusWorker *worker)
1248 {
1249   GList *l;
1250   GList *ll;
1251   GList *flushers;
1252
1253   flushers = NULL;
1254   for (l = worker->write_pending_flushes; l != NULL; l = ll)
1255     {
1256       FlushData *f = l->data;
1257       ll = l->next;
1258
1259       if (f->number_to_wait_for == worker->write_num_messages_written)
1260         {
1261           flushers = g_list_append (flushers, f);
1262           worker->write_pending_flushes = g_list_delete_link (worker->write_pending_flushes, l);
1263         }
1264     }
1265   if (flushers != NULL)
1266     {
1267       g_assert (worker->output_pending == PENDING_NONE);
1268       worker->output_pending = PENDING_FLUSH;
1269     }
1270
1271   if (flushers != NULL)
1272     {
1273       FlushAsyncData *data;
1274
1275       data = g_new0 (FlushAsyncData, 1);
1276       data->worker = _g_dbus_worker_ref (worker);
1277       data->flushers = flushers;
1278       return data;
1279     }
1280
1281   return NULL;
1282 }
1283
1284 /* called in private thread shared by all GDBusConnection instances
1285  *
1286  * write-lock is not held on entry
1287  * output_pending is PENDING_WRITE on entry
1288  */
1289 static void
1290 write_message_cb (GObject       *source_object,
1291                   GAsyncResult  *res,
1292                   gpointer       user_data)
1293 {
1294   MessageToWriteData *data = user_data;
1295   GError *error;
1296
1297   g_mutex_lock (&data->worker->write_lock);
1298   g_assert (data->worker->output_pending == PENDING_WRITE);
1299   data->worker->output_pending = PENDING_NONE;
1300
1301   error = NULL;
1302   if (!write_message_finish (res, &error))
1303     {
1304       g_mutex_unlock (&data->worker->write_lock);
1305
1306       /* TODO: handle */
1307       _g_dbus_worker_emit_disconnected (data->worker, TRUE, error);
1308       g_error_free (error);
1309
1310       g_mutex_lock (&data->worker->write_lock);
1311     }
1312
1313   message_written_unlocked (data->worker, data);
1314
1315   g_mutex_unlock (&data->worker->write_lock);
1316
1317   continue_writing (data->worker);
1318
1319   message_to_write_data_free (data);
1320 }
1321
1322 /* called in private thread shared by all GDBusConnection instances
1323  *
1324  * write-lock is not held on entry
1325  * output_pending is PENDING_CLOSE on entry
1326  */
1327 static void
1328 iostream_close_cb (GObject      *source_object,
1329                    GAsyncResult *res,
1330                    gpointer      user_data)
1331 {
1332   GDBusWorker *worker = user_data;
1333   GError *error = NULL;
1334   GList *pending_close_attempts, *pending_flush_attempts;
1335   GQueue *send_queue;
1336
1337   g_io_stream_close_finish (worker->stream, res, &error);
1338
1339   g_mutex_lock (&worker->write_lock);
1340
1341   pending_close_attempts = worker->pending_close_attempts;
1342   worker->pending_close_attempts = NULL;
1343
1344   pending_flush_attempts = worker->write_pending_flushes;
1345   worker->write_pending_flushes = NULL;
1346
1347   send_queue = worker->write_queue;
1348   worker->write_queue = g_queue_new ();
1349
1350   g_assert (worker->output_pending == PENDING_CLOSE);
1351   worker->output_pending = PENDING_NONE;
1352
1353   g_mutex_unlock (&worker->write_lock);
1354
1355   while (pending_close_attempts != NULL)
1356     {
1357       CloseData *close_data = pending_close_attempts->data;
1358
1359       pending_close_attempts = g_list_delete_link (pending_close_attempts,
1360                                                    pending_close_attempts);
1361
1362       if (close_data->task != NULL)
1363         {
1364           if (error != NULL)
1365             g_task_return_error (close_data->task, g_error_copy (error));
1366           else
1367             g_task_return_boolean (close_data->task, TRUE);
1368         }
1369
1370       close_data_free (close_data);
1371     }
1372
1373   g_clear_error (&error);
1374
1375   /* all messages queued for sending are discarded */
1376   g_queue_free_full (send_queue, (GDestroyNotify) message_to_write_data_free);
1377   /* all queued flushes fail */
1378   error = g_error_new (G_IO_ERROR, G_IO_ERROR_CANCELLED,
1379                        _("Operation was cancelled"));
1380   flush_data_list_complete (pending_flush_attempts, error);
1381   g_list_free (pending_flush_attempts);
1382   g_clear_error (&error);
1383
1384   _g_dbus_worker_unref (worker);
1385 }
1386
1387 /* called in private thread shared by all GDBusConnection instances
1388  *
1389  * write-lock is not held on entry
1390  * output_pending must be PENDING_NONE on entry
1391  */
1392 static void
1393 continue_writing (GDBusWorker *worker)
1394 {
1395   MessageToWriteData *data;
1396   FlushAsyncData *flush_async_data;
1397
1398  write_next:
1399   /* we mustn't try to write two things at once */
1400   g_assert (worker->output_pending == PENDING_NONE);
1401
1402   g_mutex_lock (&worker->write_lock);
1403
1404   data = NULL;
1405   flush_async_data = NULL;
1406
1407   /* if we want to close the connection, that takes precedence */
1408   if (worker->pending_close_attempts != NULL)
1409     {
1410       GInputStream *input = g_io_stream_get_input_stream (worker->stream);
1411
1412       if (!g_input_stream_has_pending (input))
1413         {
1414           worker->close_expected = TRUE;
1415           worker->output_pending = PENDING_CLOSE;
1416
1417           g_io_stream_close_async (worker->stream, G_PRIORITY_DEFAULT,
1418                                    NULL, iostream_close_cb,
1419                                    _g_dbus_worker_ref (worker));
1420         }
1421     }
1422   else
1423     {
1424       flush_async_data = prepare_flush_unlocked (worker);
1425
1426       if (flush_async_data == NULL)
1427         {
1428           data = g_queue_pop_head (worker->write_queue);
1429
1430           if (data != NULL)
1431             worker->output_pending = PENDING_WRITE;
1432         }
1433     }
1434
1435   g_mutex_unlock (&worker->write_lock);
1436
1437   /* Note that write_lock is only used for protecting the @write_queue
1438    * and @output_pending fields of the GDBusWorker struct ... which we
1439    * need to modify from arbitrary threads in _g_dbus_worker_send_message().
1440    *
1441    * Therefore, it's fine to drop it here when calling back into user
1442    * code and then writing the message out onto the GIOStream since this
1443    * function only runs on the worker thread.
1444    */
1445
1446   if (flush_async_data != NULL)
1447     {
1448       start_flush (flush_async_data);
1449       g_assert (data == NULL);
1450     }
1451   else if (data != NULL)
1452     {
1453       GDBusMessage *old_message;
1454       guchar *new_blob;
1455       gsize new_blob_size;
1456       GError *error;
1457
1458       old_message = data->message;
1459       data->message = _g_dbus_worker_emit_message_about_to_be_sent (worker, data->message);
1460       if (data->message == old_message)
1461         {
1462           /* filters had no effect - do nothing */
1463         }
1464       else if (data->message == NULL)
1465         {
1466           /* filters dropped message */
1467           g_mutex_lock (&worker->write_lock);
1468           worker->output_pending = PENDING_NONE;
1469           g_mutex_unlock (&worker->write_lock);
1470           message_to_write_data_free (data);
1471           goto write_next;
1472         }
1473       else
1474         {
1475           /* filters altered the message -> reencode */
1476           error = NULL;
1477           new_blob = g_dbus_message_to_blob (data->message,
1478                                              &new_blob_size,
1479                                              worker->capabilities,
1480                                              &error);
1481           if (new_blob == NULL)
1482             {
1483               /* if filter make the GDBusMessage unencodeable, just complain on stderr and send
1484                * the old message instead
1485                */
1486               g_warning ("Error encoding GDBusMessage with serial %d altered by filter function: %s",
1487                          g_dbus_message_get_serial (data->message),
1488                          error->message);
1489               g_error_free (error);
1490             }
1491           else
1492             {
1493               g_free (data->blob);
1494               data->blob = (gchar *) new_blob;
1495               data->blob_size = new_blob_size;
1496             }
1497         }
1498
1499       write_message_async (worker,
1500                            data,
1501                            write_message_cb,
1502                            data);
1503     }
1504 }
1505
1506 /* called in private thread shared by all GDBusConnection instances
1507  *
1508  * write-lock is not held on entry
1509  * output_pending may be anything
1510  */
1511 static gboolean
1512 continue_writing_in_idle_cb (gpointer user_data)
1513 {
1514   GDBusWorker *worker = user_data;
1515
1516   /* Because this is the worker thread, we can read this struct member
1517    * without holding the lock: no other thread ever modifies it.
1518    */
1519   if (worker->output_pending == PENDING_NONE)
1520     continue_writing (worker);
1521
1522   return FALSE;
1523 }
1524
1525 /*
1526  * @write_data: (transfer full) (nullable):
1527  * @flush_data: (transfer full) (nullable):
1528  * @close_data: (transfer full) (nullable):
1529  *
1530  * Can be called from any thread
1531  *
1532  * write_lock is held on entry
1533  * output_pending may be anything
1534  */
1535 static void
1536 schedule_writing_unlocked (GDBusWorker        *worker,
1537                            MessageToWriteData *write_data,
1538                            FlushData          *flush_data,
1539                            CloseData          *close_data)
1540 {
1541   if (write_data != NULL)
1542     g_queue_push_tail (worker->write_queue, write_data);
1543
1544   if (flush_data != NULL)
1545     worker->write_pending_flushes = g_list_prepend (worker->write_pending_flushes, flush_data);
1546
1547   if (close_data != NULL)
1548     worker->pending_close_attempts = g_list_prepend (worker->pending_close_attempts,
1549                                                      close_data);
1550
1551   /* If we had output pending, the next bit of output will happen
1552    * automatically when it finishes, so we only need to do this
1553    * if nothing was pending.
1554    *
1555    * The idle callback will re-check that output_pending is still
1556    * PENDING_NONE, to guard against output starting before the idle.
1557    */
1558   if (worker->output_pending == PENDING_NONE)
1559     {
1560       GSource *idle_source;
1561       idle_source = g_idle_source_new ();
1562       g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
1563       g_source_set_callback (idle_source,
1564                              continue_writing_in_idle_cb,
1565                              _g_dbus_worker_ref (worker),
1566                              (GDestroyNotify) _g_dbus_worker_unref);
1567       g_source_set_name (idle_source, "[gio] continue_writing_in_idle_cb");
1568       g_source_attach (idle_source, worker->shared_thread_data->context);
1569       g_source_unref (idle_source);
1570     }
1571 }
1572
1573 static void
1574 schedule_pending_close (GDBusWorker *worker)
1575 {
1576   g_mutex_lock (&worker->write_lock);
1577   if (worker->pending_close_attempts)
1578     schedule_writing_unlocked (worker, NULL, NULL, NULL);
1579   g_mutex_unlock (&worker->write_lock);
1580 }
1581
1582 /* ---------------------------------------------------------------------------------------------------- */
1583
1584 /* can be called from any thread - steals blob
1585  *
1586  * write_lock is not held on entry
1587  * output_pending may be anything
1588  */
1589 void
1590 _g_dbus_worker_send_message (GDBusWorker    *worker,
1591                              GDBusMessage   *message,
1592                              gchar          *blob,
1593                              gsize           blob_len)
1594 {
1595   MessageToWriteData *data;
1596
1597   g_return_if_fail (G_IS_DBUS_MESSAGE (message));
1598   g_return_if_fail (blob != NULL);
1599   g_return_if_fail (blob_len > 16);
1600
1601   data = g_slice_new0 (MessageToWriteData);
1602   data->worker = _g_dbus_worker_ref (worker);
1603   data->message = g_object_ref (message);
1604   data->blob = blob; /* steal! */
1605   data->blob_size = blob_len;
1606
1607   g_mutex_lock (&worker->write_lock);
1608   schedule_writing_unlocked (worker, data, NULL, NULL);
1609   g_mutex_unlock (&worker->write_lock);
1610 }
1611
1612 /* ---------------------------------------------------------------------------------------------------- */
1613
1614 GDBusWorker *
1615 _g_dbus_worker_new (GIOStream                              *stream,
1616                     GDBusCapabilityFlags                    capabilities,
1617                     gboolean                                initially_frozen,
1618                     GDBusWorkerMessageReceivedCallback      message_received_callback,
1619                     GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback,
1620                     GDBusWorkerDisconnectedCallback         disconnected_callback,
1621                     gpointer                                user_data)
1622 {
1623   GDBusWorker *worker;
1624   GSource *idle_source;
1625
1626   g_return_val_if_fail (G_IS_IO_STREAM (stream), NULL);
1627   g_return_val_if_fail (message_received_callback != NULL, NULL);
1628   g_return_val_if_fail (message_about_to_be_sent_callback != NULL, NULL);
1629   g_return_val_if_fail (disconnected_callback != NULL, NULL);
1630
1631   worker = g_new0 (GDBusWorker, 1);
1632   worker->ref_count = 1;
1633
1634   g_mutex_init (&worker->read_lock);
1635   worker->message_received_callback = message_received_callback;
1636   worker->message_about_to_be_sent_callback = message_about_to_be_sent_callback;
1637   worker->disconnected_callback = disconnected_callback;
1638   worker->user_data = user_data;
1639   worker->stream = g_object_ref (stream);
1640   worker->capabilities = capabilities;
1641   worker->cancellable = g_cancellable_new ();
1642   worker->output_pending = PENDING_NONE;
1643
1644   worker->frozen = initially_frozen;
1645   worker->received_messages_while_frozen = g_queue_new ();
1646
1647   g_mutex_init (&worker->write_lock);
1648   worker->write_queue = g_queue_new ();
1649
1650   if (G_IS_SOCKET_CONNECTION (worker->stream))
1651     worker->socket = g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream));
1652
1653   worker->shared_thread_data = _g_dbus_shared_thread_ref ();
1654
1655   /* begin reading */
1656   idle_source = g_idle_source_new ();
1657   g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
1658   g_source_set_callback (idle_source,
1659                          _g_dbus_worker_do_initial_read,
1660                          _g_dbus_worker_ref (worker),
1661                          (GDestroyNotify) _g_dbus_worker_unref);
1662   g_source_set_name (idle_source, "[gio] _g_dbus_worker_do_initial_read");
1663   g_source_attach (idle_source, worker->shared_thread_data->context);
1664   g_source_unref (idle_source);
1665
1666   return worker;
1667 }
1668
1669 /* ---------------------------------------------------------------------------------------------------- */
1670
1671 /* can be called from any thread
1672  *
1673  * write_lock is not held on entry
1674  * output_pending may be anything
1675  */
1676 void
1677 _g_dbus_worker_close (GDBusWorker         *worker,
1678                       GTask               *task)
1679 {
1680   CloseData *close_data;
1681
1682   close_data = g_slice_new0 (CloseData);
1683   close_data->worker = _g_dbus_worker_ref (worker);
1684   close_data->task = (task == NULL ? NULL : g_object_ref (task));
1685
1686   /* Don't set worker->close_expected here - we're in the wrong thread.
1687    * It'll be set before the actual close happens.
1688    */
1689   g_cancellable_cancel (worker->cancellable);
1690   g_mutex_lock (&worker->write_lock);
1691   schedule_writing_unlocked (worker, NULL, NULL, close_data);
1692   g_mutex_unlock (&worker->write_lock);
1693 }
1694
1695 /* This can be called from any thread - frees worker. Note that
1696  * callbacks might still happen if called from another thread than the
1697  * worker - use your own synchronization primitive in the callbacks.
1698  *
1699  * write_lock is not held on entry
1700  * output_pending may be anything
1701  */
1702 void
1703 _g_dbus_worker_stop (GDBusWorker *worker)
1704 {
1705   g_atomic_int_set (&worker->stopped, TRUE);
1706
1707   /* Cancel any pending operations and schedule a close of the underlying I/O
1708    * stream in the worker thread
1709    */
1710   _g_dbus_worker_close (worker, NULL);
1711
1712   /* _g_dbus_worker_close holds a ref until after an idle in the worker
1713    * thread has run, so we no longer need to unref in an idle like in
1714    * commit 322e25b535
1715    */
1716   _g_dbus_worker_unref (worker);
1717 }
1718
1719 /* ---------------------------------------------------------------------------------------------------- */
1720
1721 /* can be called from any thread (except the worker thread) - blocks
1722  * calling thread until all queued outgoing messages are written and
1723  * the transport has been flushed
1724  *
1725  * write_lock is not held on entry
1726  * output_pending may be anything
1727  */
1728 gboolean
1729 _g_dbus_worker_flush_sync (GDBusWorker    *worker,
1730                            GCancellable   *cancellable,
1731                            GError        **error)
1732 {
1733   gboolean ret;
1734   FlushData *data;
1735   guint64 pending_writes;
1736
1737   data = NULL;
1738   ret = TRUE;
1739
1740   g_mutex_lock (&worker->write_lock);
1741
1742   /* if the queue is empty, no write is in-flight and we haven't written
1743    * anything since the last flush, then there's nothing to wait for
1744    */
1745   pending_writes = g_queue_get_length (worker->write_queue);
1746
1747   /* if a write is in-flight, we shouldn't be satisfied until the first
1748    * flush operation that follows it
1749    */
1750   if (worker->output_pending == PENDING_WRITE)
1751     pending_writes += 1;
1752
1753   if (pending_writes > 0 ||
1754       worker->write_num_messages_written != worker->write_num_messages_flushed)
1755     {
1756       data = g_new0 (FlushData, 1);
1757       g_mutex_init (&data->mutex);
1758       g_cond_init (&data->cond);
1759       data->number_to_wait_for = worker->write_num_messages_written + pending_writes;
1760       g_mutex_lock (&data->mutex);
1761
1762       schedule_writing_unlocked (worker, NULL, data, NULL);
1763     }
1764   g_mutex_unlock (&worker->write_lock);
1765
1766   if (data != NULL)
1767     {
1768       g_cond_wait (&data->cond, &data->mutex);
1769       g_mutex_unlock (&data->mutex);
1770
1771       /* note:the element is removed from worker->write_pending_flushes in flush_cb() above */
1772       g_cond_clear (&data->cond);
1773       g_mutex_clear (&data->mutex);
1774       if (data->error != NULL)
1775         {
1776           ret = FALSE;
1777           g_propagate_error (error, data->error);
1778         }
1779       g_free (data);
1780     }
1781
1782   return ret;
1783 }
1784
1785 /* ---------------------------------------------------------------------------------------------------- */
1786
1787 #define G_DBUS_DEBUG_AUTHENTICATION (1<<0)
1788 #define G_DBUS_DEBUG_TRANSPORT      (1<<1)
1789 #define G_DBUS_DEBUG_MESSAGE        (1<<2)
1790 #define G_DBUS_DEBUG_PAYLOAD        (1<<3)
1791 #define G_DBUS_DEBUG_CALL           (1<<4)
1792 #define G_DBUS_DEBUG_SIGNAL         (1<<5)
1793 #define G_DBUS_DEBUG_INCOMING       (1<<6)
1794 #define G_DBUS_DEBUG_RETURN         (1<<7)
1795 #define G_DBUS_DEBUG_EMISSION       (1<<8)
1796 #define G_DBUS_DEBUG_ADDRESS        (1<<9)
1797
1798 static gint _gdbus_debug_flags = 0;
1799
1800 gboolean
1801 _g_dbus_debug_authentication (void)
1802 {
1803   _g_dbus_initialize ();
1804   return (_gdbus_debug_flags & G_DBUS_DEBUG_AUTHENTICATION) != 0;
1805 }
1806
1807 gboolean
1808 _g_dbus_debug_transport (void)
1809 {
1810   _g_dbus_initialize ();
1811   return (_gdbus_debug_flags & G_DBUS_DEBUG_TRANSPORT) != 0;
1812 }
1813
1814 gboolean
1815 _g_dbus_debug_message (void)
1816 {
1817   _g_dbus_initialize ();
1818   return (_gdbus_debug_flags & G_DBUS_DEBUG_MESSAGE) != 0;
1819 }
1820
1821 gboolean
1822 _g_dbus_debug_payload (void)
1823 {
1824   _g_dbus_initialize ();
1825   return (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD) != 0;
1826 }
1827
1828 gboolean
1829 _g_dbus_debug_call (void)
1830 {
1831   _g_dbus_initialize ();
1832   return (_gdbus_debug_flags & G_DBUS_DEBUG_CALL) != 0;
1833 }
1834
1835 gboolean
1836 _g_dbus_debug_signal (void)
1837 {
1838   _g_dbus_initialize ();
1839   return (_gdbus_debug_flags & G_DBUS_DEBUG_SIGNAL) != 0;
1840 }
1841
1842 gboolean
1843 _g_dbus_debug_incoming (void)
1844 {
1845   _g_dbus_initialize ();
1846   return (_gdbus_debug_flags & G_DBUS_DEBUG_INCOMING) != 0;
1847 }
1848
1849 gboolean
1850 _g_dbus_debug_return (void)
1851 {
1852   _g_dbus_initialize ();
1853   return (_gdbus_debug_flags & G_DBUS_DEBUG_RETURN) != 0;
1854 }
1855
1856 gboolean
1857 _g_dbus_debug_emission (void)
1858 {
1859   _g_dbus_initialize ();
1860   return (_gdbus_debug_flags & G_DBUS_DEBUG_EMISSION) != 0;
1861 }
1862
1863 gboolean
1864 _g_dbus_debug_address (void)
1865 {
1866   _g_dbus_initialize ();
1867   return (_gdbus_debug_flags & G_DBUS_DEBUG_ADDRESS) != 0;
1868 }
1869
1870 G_LOCK_DEFINE_STATIC (print_lock);
1871
1872 void
1873 _g_dbus_debug_print_lock (void)
1874 {
1875   G_LOCK (print_lock);
1876 }
1877
1878 void
1879 _g_dbus_debug_print_unlock (void)
1880 {
1881   G_UNLOCK (print_lock);
1882 }
1883
1884 /**
1885  * _g_dbus_initialize:
1886  *
1887  * Does various one-time init things such as
1888  *
1889  *  - registering the G_DBUS_ERROR error domain
1890  *  - parses the G_DBUS_DEBUG environment variable
1891  */
1892 void
1893 _g_dbus_initialize (void)
1894 {
1895   static volatile gsize initialized = 0;
1896
1897   if (g_once_init_enter (&initialized))
1898     {
1899       volatile GQuark g_dbus_error_domain;
1900       const gchar *debug;
1901
1902       g_dbus_error_domain = G_DBUS_ERROR;
1903       (g_dbus_error_domain); /* To avoid -Wunused-but-set-variable */
1904
1905       debug = g_getenv ("G_DBUS_DEBUG");
1906       if (debug != NULL)
1907         {
1908           const GDebugKey keys[] = {
1909             { "authentication", G_DBUS_DEBUG_AUTHENTICATION },
1910             { "transport",      G_DBUS_DEBUG_TRANSPORT      },
1911             { "message",        G_DBUS_DEBUG_MESSAGE        },
1912             { "payload",        G_DBUS_DEBUG_PAYLOAD        },
1913             { "call",           G_DBUS_DEBUG_CALL           },
1914             { "signal",         G_DBUS_DEBUG_SIGNAL         },
1915             { "incoming",       G_DBUS_DEBUG_INCOMING       },
1916             { "return",         G_DBUS_DEBUG_RETURN         },
1917             { "emission",       G_DBUS_DEBUG_EMISSION       },
1918             { "address",        G_DBUS_DEBUG_ADDRESS        }
1919           };
1920
1921           _gdbus_debug_flags = g_parse_debug_string (debug, keys, G_N_ELEMENTS (keys));
1922           if (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD)
1923             _gdbus_debug_flags |= G_DBUS_DEBUG_MESSAGE;
1924         }
1925
1926       /* Work-around for https://bugzilla.gnome.org/show_bug.cgi?id=627724 */
1927       ensure_required_types ();
1928
1929       g_once_init_leave (&initialized, 1);
1930     }
1931 }
1932
1933 /* ---------------------------------------------------------------------------------------------------- */
1934
1935 GVariantType *
1936 _g_dbus_compute_complete_signature (GDBusArgInfo **args)
1937 {
1938   const GVariantType *arg_types[256];
1939   guint n;
1940
1941   if (args)
1942     for (n = 0; args[n] != NULL; n++)
1943       {
1944         /* DBus places a hard limit of 255 on signature length.
1945          * therefore number of args must be less than 256.
1946          */
1947         g_assert (n < 256);
1948
1949         arg_types[n] = G_VARIANT_TYPE (args[n]->signature);
1950
1951         if G_UNLIKELY (arg_types[n] == NULL)
1952           return NULL;
1953       }
1954   else
1955     n = 0;
1956
1957   return g_variant_type_new_tuple (arg_types, n);
1958 }
1959
1960 /* ---------------------------------------------------------------------------------------------------- */
1961
1962 #ifdef G_OS_WIN32
1963
1964 extern BOOL WINAPI ConvertSidToStringSidA (PSID Sid, LPSTR *StringSid);
1965
1966 gchar *
1967 _g_dbus_win32_get_user_sid (void)
1968 {
1969   HANDLE h;
1970   TOKEN_USER *user;
1971   DWORD token_information_len;
1972   PSID psid;
1973   gchar *sid;
1974   gchar *ret;
1975
1976   ret = NULL;
1977   user = NULL;
1978   h = INVALID_HANDLE_VALUE;
1979
1980   if (!OpenProcessToken (GetCurrentProcess (), TOKEN_QUERY, &h))
1981     {
1982       g_warning ("OpenProcessToken failed with error code %d", (gint) GetLastError ());
1983       goto out;
1984     }
1985
1986   /* Get length of buffer */
1987   token_information_len = 0;
1988   if (!GetTokenInformation (h, TokenUser, NULL, 0, &token_information_len))
1989     {
1990       if (GetLastError () != ERROR_INSUFFICIENT_BUFFER)
1991         {
1992           g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
1993           goto out;
1994         }
1995     }
1996   user = g_malloc (token_information_len);
1997   if (!GetTokenInformation (h, TokenUser, user, token_information_len, &token_information_len))
1998     {
1999       g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
2000       goto out;
2001     }
2002
2003   psid = user->User.Sid;
2004   if (!IsValidSid (psid))
2005     {
2006       g_warning ("Invalid SID");
2007       goto out;
2008     }
2009
2010   if (!ConvertSidToStringSidA (psid, &sid))
2011     {
2012       g_warning ("Invalid SID");
2013       goto out;
2014     }
2015
2016   ret = g_strdup (sid);
2017   LocalFree (sid);
2018
2019 out:
2020   g_free (user);
2021   if (h != INVALID_HANDLE_VALUE)
2022     CloseHandle (h);
2023   return ret;
2024 }
2025 #endif
2026
2027 /* ---------------------------------------------------------------------------------------------------- */
2028
2029 gchar *
2030 _g_dbus_get_machine_id (GError **error)
2031 {
2032 #ifdef G_OS_WIN32
2033   HW_PROFILE_INFOA info;
2034   char *src, *dest, *res;
2035   int i;
2036
2037   if (!GetCurrentHwProfileA (&info))
2038     {
2039       char *message = g_win32_error_message (GetLastError ());
2040       g_set_error (error,
2041                    G_IO_ERROR,
2042                    G_IO_ERROR_FAILED,
2043                    _("Unable to get Hardware profile: %s"), message);
2044       g_free (message);
2045       return NULL;
2046     }
2047
2048   /* Form: {12340001-4980-1920-6788-123456789012} */
2049   src = &info.szHwProfileGuid[0];
2050
2051   res = g_malloc (32+1);
2052   dest = res;
2053
2054   src++; /* Skip { */
2055   for (i = 0; i < 8; i++)
2056     *dest++ = *src++;
2057   src++; /* Skip - */
2058   for (i = 0; i < 4; i++)
2059     *dest++ = *src++;
2060   src++; /* Skip - */
2061   for (i = 0; i < 4; i++)
2062     *dest++ = *src++;
2063   src++; /* Skip - */
2064   for (i = 0; i < 4; i++)
2065     *dest++ = *src++;
2066   src++; /* Skip - */
2067   for (i = 0; i < 12; i++)
2068     *dest++ = *src++;
2069   *dest = 0;
2070
2071   return res;
2072 #else
2073   gchar *ret;
2074   GError *first_error;
2075   /* TODO: use PACKAGE_LOCALSTATEDIR ? */
2076   ret = NULL;
2077   first_error = NULL;
2078   if (!g_file_get_contents ("/var/lib/dbus/machine-id",
2079                             &ret,
2080                             NULL,
2081                             &first_error) &&
2082       !g_file_get_contents ("/etc/machine-id",
2083                             &ret,
2084                             NULL,
2085                             NULL))
2086     {
2087       g_propagate_prefixed_error (error, first_error,
2088                                   _("Unable to load /var/lib/dbus/machine-id or /etc/machine-id: "));
2089     }
2090   else
2091     {
2092       /* ignore the error from the first try, if any */
2093       g_clear_error (&first_error);
2094       /* TODO: validate value */
2095       g_strstrip (ret);
2096     }
2097   return ret;
2098 #endif
2099 }
2100
2101 /* ---------------------------------------------------------------------------------------------------- */
2102
2103 gchar *
2104 _g_dbus_enum_to_string (GType enum_type, gint value)
2105 {
2106   gchar *ret;
2107   GEnumClass *klass;
2108   GEnumValue *enum_value;
2109
2110   klass = g_type_class_ref (enum_type);
2111   enum_value = g_enum_get_value (klass, value);
2112   if (enum_value != NULL)
2113     ret = g_strdup (enum_value->value_nick);
2114   else
2115     ret = g_strdup_printf ("unknown (value %d)", value);
2116   g_type_class_unref (klass);
2117   return ret;
2118 }
2119
2120 /* ---------------------------------------------------------------------------------------------------- */
2121
2122 static void
2123 write_message_print_transport_debug (gssize bytes_written,
2124                                      MessageToWriteData *data)
2125 {
2126   if (G_LIKELY (!_g_dbus_debug_transport ()))
2127     goto out;
2128
2129   _g_dbus_debug_print_lock ();
2130   g_print ("========================================================================\n"
2131            "GDBus-debug:Transport:\n"
2132            "  >>>> WROTE %" G_GSSIZE_FORMAT " bytes of message with serial %d and\n"
2133            "       size %" G_GSIZE_FORMAT " from offset %" G_GSIZE_FORMAT " on a %s\n",
2134            bytes_written,
2135            g_dbus_message_get_serial (data->message),
2136            data->blob_size,
2137            data->total_written,
2138            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
2139   _g_dbus_debug_print_unlock ();
2140  out:
2141   ;
2142 }
2143
2144 /* ---------------------------------------------------------------------------------------------------- */
2145
2146 static void
2147 read_message_print_transport_debug (gssize bytes_read,
2148                                     GDBusWorker *worker)
2149 {
2150   gsize size;
2151   gint32 serial;
2152   gint32 message_length;
2153
2154   if (G_LIKELY (!_g_dbus_debug_transport ()))
2155     goto out;
2156
2157   size = bytes_read + worker->read_buffer_cur_size;
2158   serial = 0;
2159   message_length = 0;
2160   if (size >= 16)
2161     message_length = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer, size, NULL);
2162   if (size >= 1)
2163     {
2164       switch (worker->read_buffer[0])
2165         {
2166         case 'l':
2167           if (size >= 12)
2168             serial = GUINT32_FROM_LE (((guint32 *) worker->read_buffer)[2]);
2169           break;
2170         case 'B':
2171           if (size >= 12)
2172             serial = GUINT32_FROM_BE (((guint32 *) worker->read_buffer)[2]);
2173           break;
2174         default:
2175           /* an error will be set elsewhere if this happens */
2176           goto out;
2177         }
2178     }
2179
2180     _g_dbus_debug_print_lock ();
2181   g_print ("========================================================================\n"
2182            "GDBus-debug:Transport:\n"
2183            "  <<<< READ %" G_GSSIZE_FORMAT " bytes of message with serial %d and\n"
2184            "       size %d to offset %" G_GSIZE_FORMAT " from a %s\n",
2185            bytes_read,
2186            serial,
2187            message_length,
2188            worker->read_buffer_cur_size,
2189            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker->stream))));
2190   _g_dbus_debug_print_unlock ();
2191  out:
2192   ;
2193 }
2194
2195 /* ---------------------------------------------------------------------------------------------------- */
2196
2197 gboolean
2198 _g_signal_accumulator_false_handled (GSignalInvocationHint *ihint,
2199                                      GValue                *return_accu,
2200                                      const GValue          *handler_return,
2201                                      gpointer               dummy)
2202 {
2203   gboolean continue_emission;
2204   gboolean signal_return;
2205
2206   signal_return = g_value_get_boolean (handler_return);
2207   g_value_set_boolean (return_accu, signal_return);
2208   continue_emission = signal_return;
2209
2210   return continue_emission;
2211 }