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