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 (without 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_NONE on entry
1226  */
1227 static void
1228 message_written (GDBusWorker *worker,
1229                  MessageToWriteData *message_data)
1230 {
1231   GList *l;
1232   GList *ll;
1233   GList *flushers;
1234
1235   /* first log the fact that we wrote a message */
1236   if (G_UNLIKELY (_g_dbus_debug_message ()))
1237     {
1238       gchar *s;
1239       _g_dbus_debug_print_lock ();
1240       g_print ("========================================================================\n"
1241                "GDBus-debug:Message:\n"
1242                "  >>>> SENT D-Bus message (%" G_GSIZE_FORMAT " bytes)\n",
1243                message_data->blob_size);
1244       s = g_dbus_message_print (message_data->message, 2);
1245       g_print ("%s", s);
1246       g_free (s);
1247       if (G_UNLIKELY (_g_dbus_debug_payload ()))
1248         {
1249           s = _g_dbus_hexdump (message_data->blob, message_data->blob_size, 2);
1250           g_print ("%s\n", s);
1251           g_free (s);
1252         }
1253       _g_dbus_debug_print_unlock ();
1254     }
1255
1256   /* then first wake up pending flushes and, if needed, flush the stream */
1257   flushers = NULL;
1258   g_mutex_lock (&worker->write_lock);
1259   worker->write_num_messages_written += 1;
1260   for (l = worker->write_pending_flushes; l != NULL; l = ll)
1261     {
1262       FlushData *f = l->data;
1263       ll = l->next;
1264
1265       if (f->number_to_wait_for == worker->write_num_messages_written)
1266         {
1267           flushers = g_list_append (flushers, f);
1268           worker->write_pending_flushes = g_list_delete_link (worker->write_pending_flushes, l);
1269         }
1270     }
1271   if (flushers != NULL)
1272     {
1273       g_assert (worker->output_pending == PENDING_NONE);
1274       worker->output_pending = PENDING_FLUSH;
1275     }
1276   g_mutex_unlock (&worker->write_lock);
1277
1278   if (flushers != NULL)
1279     {
1280       FlushAsyncData *data;
1281       data = g_new0 (FlushAsyncData, 1);
1282       data->worker = _g_dbus_worker_ref (worker);
1283       data->flushers = flushers;
1284       /* flush the stream before writing the next message */
1285       g_output_stream_flush_async (g_io_stream_get_output_stream (worker->stream),
1286                                    G_PRIORITY_DEFAULT,
1287                                    worker->cancellable,
1288                                    ostream_flush_cb,
1289                                    data);
1290     }
1291   else
1292     {
1293       /* kick off the next write! */
1294       continue_writing (worker);
1295     }
1296 }
1297
1298 /* called in private thread shared by all GDBusConnection instances
1299  *
1300  * write-lock is not held on entry
1301  * output_pending is PENDING_WRITE on entry
1302  */
1303 static void
1304 write_message_cb (GObject       *source_object,
1305                   GAsyncResult  *res,
1306                   gpointer       user_data)
1307 {
1308   MessageToWriteData *data = user_data;
1309   GError *error;
1310
1311   g_mutex_lock (&data->worker->write_lock);
1312   g_assert (data->worker->output_pending == PENDING_WRITE);
1313   data->worker->output_pending = PENDING_NONE;
1314   g_mutex_unlock (&data->worker->write_lock);
1315
1316   error = NULL;
1317   if (!write_message_finish (res, &error))
1318     {
1319       /* TODO: handle */
1320       _g_dbus_worker_emit_disconnected (data->worker, TRUE, error);
1321       g_error_free (error);
1322     }
1323
1324   /* this function will also kick of the next write (it might need to
1325    * flush so writing the next message might happen much later
1326    * e.g. async)
1327    */
1328   message_written (data->worker, data);
1329
1330   message_to_write_data_free (data);
1331 }
1332
1333 /* called in private thread shared by all GDBusConnection instances
1334  *
1335  * write-lock is not held on entry
1336  * output_pending is PENDING_CLOSE on entry
1337  */
1338 static void
1339 iostream_close_cb (GObject      *source_object,
1340                    GAsyncResult *res,
1341                    gpointer      user_data)
1342 {
1343   GDBusWorker *worker = user_data;
1344   GError *error = NULL;
1345   GList *pending_close_attempts, *pending_flush_attempts;
1346   GQueue *send_queue;
1347
1348   g_io_stream_close_finish (worker->stream, res, &error);
1349
1350   g_mutex_lock (&worker->write_lock);
1351
1352   pending_close_attempts = worker->pending_close_attempts;
1353   worker->pending_close_attempts = NULL;
1354
1355   pending_flush_attempts = worker->write_pending_flushes;
1356   worker->write_pending_flushes = NULL;
1357
1358   send_queue = worker->write_queue;
1359   worker->write_queue = g_queue_new ();
1360
1361   g_assert (worker->output_pending == PENDING_CLOSE);
1362   worker->output_pending = PENDING_NONE;
1363
1364   g_mutex_unlock (&worker->write_lock);
1365
1366   while (pending_close_attempts != NULL)
1367     {
1368       CloseData *close_data = pending_close_attempts->data;
1369
1370       pending_close_attempts = g_list_delete_link (pending_close_attempts,
1371                                                    pending_close_attempts);
1372
1373       if (close_data->result != NULL)
1374         {
1375           if (error != NULL)
1376             g_simple_async_result_set_from_error (close_data->result, error);
1377
1378           /* this must be in an idle because the result is likely to be
1379            * intended for another thread
1380            */
1381           g_simple_async_result_complete_in_idle (close_data->result);
1382         }
1383
1384       close_data_free (close_data);
1385     }
1386
1387   g_clear_error (&error);
1388
1389   /* all messages queued for sending are discarded */
1390   g_queue_foreach (send_queue, (GFunc) message_to_write_data_free, NULL);
1391   g_queue_free (send_queue);
1392
1393   /* all queued flushes fail */
1394   error = g_error_new (G_IO_ERROR, G_IO_ERROR_CANCELLED,
1395                        _("Operation was cancelled"));
1396   flush_data_list_complete (pending_flush_attempts, error);
1397   g_list_free (pending_flush_attempts);
1398   g_clear_error (&error);
1399
1400   _g_dbus_worker_unref (worker);
1401 }
1402
1403 /* called in private thread shared by all GDBusConnection instances
1404  *
1405  * write-lock is not held on entry
1406  * output_pending must be PENDING_NONE on entry
1407  */
1408 static void
1409 continue_writing (GDBusWorker *worker)
1410 {
1411   MessageToWriteData *data;
1412
1413  write_next:
1414   /* we mustn't try to write two things at once */
1415   g_assert (worker->output_pending == PENDING_NONE);
1416
1417   g_mutex_lock (&worker->write_lock);
1418
1419   /* if we want to close the connection, that takes precedence */
1420   if (worker->pending_close_attempts != NULL)
1421     {
1422       worker->close_expected = TRUE;
1423       worker->output_pending = PENDING_CLOSE;
1424
1425       g_io_stream_close_async (worker->stream, G_PRIORITY_DEFAULT,
1426                                NULL, iostream_close_cb,
1427                                _g_dbus_worker_ref (worker));
1428       data = NULL;
1429     }
1430   else
1431     {
1432       data = g_queue_pop_head (worker->write_queue);
1433
1434       if (data != NULL)
1435         worker->output_pending = PENDING_WRITE;
1436     }
1437
1438   g_mutex_unlock (&worker->write_lock);
1439
1440   /* Note that write_lock is only used for protecting the @write_queue
1441    * and @output_pending fields of the GDBusWorker struct ... which we
1442    * need to modify from arbitrary threads in _g_dbus_worker_send_message().
1443    *
1444    * Therefore, it's fine to drop it here when calling back into user
1445    * code and then writing the message out onto the GIOStream since this
1446    * function only runs on the worker thread.
1447    */
1448   if (data != NULL)
1449     {
1450       GDBusMessage *old_message;
1451       guchar *new_blob;
1452       gsize new_blob_size;
1453       GError *error;
1454
1455       old_message = data->message;
1456       data->message = _g_dbus_worker_emit_message_about_to_be_sent (worker, data->message);
1457       if (data->message == old_message)
1458         {
1459           /* filters had no effect - do nothing */
1460         }
1461       else if (data->message == NULL)
1462         {
1463           /* filters dropped message */
1464           g_mutex_lock (&worker->write_lock);
1465           worker->output_pending = PENDING_NONE;
1466           g_mutex_unlock (&worker->write_lock);
1467           message_to_write_data_free (data);
1468           goto write_next;
1469         }
1470       else
1471         {
1472           /* filters altered the message -> reencode */
1473           error = NULL;
1474           new_blob = g_dbus_message_to_blob (data->message,
1475                                              &new_blob_size,
1476                                              worker->capabilities,
1477                                              &error);
1478           if (new_blob == NULL)
1479             {
1480               /* if filter make the GDBusMessage unencodeable, just complain on stderr and send
1481                * the old message instead
1482                */
1483               g_warning ("Error encoding GDBusMessage with serial %d altered by filter function: %s",
1484                          g_dbus_message_get_serial (data->message),
1485                          error->message);
1486               g_error_free (error);
1487             }
1488           else
1489             {
1490               g_free (data->blob);
1491               data->blob = (gchar *) new_blob;
1492               data->blob_size = new_blob_size;
1493             }
1494         }
1495
1496       write_message_async (worker,
1497                            data,
1498                            write_message_cb,
1499                            data);
1500     }
1501 }
1502
1503 /* called in private thread shared by all GDBusConnection instances
1504  *
1505  * write-lock is not held on entry
1506  * output_pending may be anything
1507  */
1508 static gboolean
1509 continue_writing_in_idle_cb (gpointer user_data)
1510 {
1511   GDBusWorker *worker = user_data;
1512
1513   /* Because this is the worker thread, we can read this struct member
1514    * without holding the lock: no other thread ever modifies it.
1515    */
1516   if (worker->output_pending == PENDING_NONE)
1517     continue_writing (worker);
1518
1519   return FALSE;
1520 }
1521
1522 /*
1523  * @write_data: (transfer full) (allow-none):
1524  * @close_data: (transfer full) (allow-none):
1525  *
1526  * Can be called from any thread
1527  *
1528  * write_lock is held on entry
1529  * output_pending may be anything
1530  */
1531 static void
1532 schedule_writing_unlocked (GDBusWorker        *worker,
1533                            MessageToWriteData *write_data,
1534                            CloseData          *close_data)
1535 {
1536   if (write_data != NULL)
1537     g_queue_push_tail (worker->write_queue, write_data);
1538
1539   if (close_data != NULL)
1540     worker->pending_close_attempts = g_list_prepend (worker->pending_close_attempts,
1541                                                      close_data);
1542
1543   if (worker->output_pending == PENDING_NONE)
1544     {
1545       GSource *idle_source;
1546       idle_source = g_idle_source_new ();
1547       g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
1548       g_source_set_callback (idle_source,
1549                              continue_writing_in_idle_cb,
1550                              _g_dbus_worker_ref (worker),
1551                              (GDestroyNotify) _g_dbus_worker_unref);
1552       g_source_attach (idle_source, worker->shared_thread_data->context);
1553       g_source_unref (idle_source);
1554     }
1555 }
1556
1557 /* ---------------------------------------------------------------------------------------------------- */
1558
1559 /* can be called from any thread - steals blob
1560  *
1561  * write_lock is not held on entry
1562  * output_pending may be anything
1563  */
1564 void
1565 _g_dbus_worker_send_message (GDBusWorker    *worker,
1566                              GDBusMessage   *message,
1567                              gchar          *blob,
1568                              gsize           blob_len)
1569 {
1570   MessageToWriteData *data;
1571
1572   g_return_if_fail (G_IS_DBUS_MESSAGE (message));
1573   g_return_if_fail (blob != NULL);
1574   g_return_if_fail (blob_len > 16);
1575
1576   data = g_new0 (MessageToWriteData, 1);
1577   data->worker = _g_dbus_worker_ref (worker);
1578   data->message = g_object_ref (message);
1579   data->blob = blob; /* steal! */
1580   data->blob_size = blob_len;
1581
1582   g_mutex_lock (&worker->write_lock);
1583   schedule_writing_unlocked (worker, data, NULL);
1584   g_mutex_unlock (&worker->write_lock);
1585 }
1586
1587 /* ---------------------------------------------------------------------------------------------------- */
1588
1589 GDBusWorker *
1590 _g_dbus_worker_new (GIOStream                              *stream,
1591                     GDBusCapabilityFlags                    capabilities,
1592                     gboolean                                initially_frozen,
1593                     GDBusWorkerMessageReceivedCallback      message_received_callback,
1594                     GDBusWorkerMessageAboutToBeSentCallback message_about_to_be_sent_callback,
1595                     GDBusWorkerDisconnectedCallback         disconnected_callback,
1596                     gpointer                                user_data)
1597 {
1598   GDBusWorker *worker;
1599   GSource *idle_source;
1600
1601   g_return_val_if_fail (G_IS_IO_STREAM (stream), NULL);
1602   g_return_val_if_fail (message_received_callback != NULL, NULL);
1603   g_return_val_if_fail (message_about_to_be_sent_callback != NULL, NULL);
1604   g_return_val_if_fail (disconnected_callback != NULL, NULL);
1605
1606   worker = g_new0 (GDBusWorker, 1);
1607   worker->ref_count = 1;
1608
1609   g_mutex_init (&worker->read_lock);
1610   worker->message_received_callback = message_received_callback;
1611   worker->message_about_to_be_sent_callback = message_about_to_be_sent_callback;
1612   worker->disconnected_callback = disconnected_callback;
1613   worker->user_data = user_data;
1614   worker->stream = g_object_ref (stream);
1615   worker->capabilities = capabilities;
1616   worker->cancellable = g_cancellable_new ();
1617   worker->output_pending = PENDING_NONE;
1618
1619   worker->frozen = initially_frozen;
1620   worker->received_messages_while_frozen = g_queue_new ();
1621
1622   g_mutex_init (&worker->write_lock);
1623   worker->write_queue = g_queue_new ();
1624
1625   if (G_IS_SOCKET_CONNECTION (worker->stream))
1626     worker->socket = g_socket_connection_get_socket (G_SOCKET_CONNECTION (worker->stream));
1627
1628   worker->shared_thread_data = _g_dbus_shared_thread_ref ();
1629
1630   /* begin reading */
1631   idle_source = g_idle_source_new ();
1632   g_source_set_priority (idle_source, G_PRIORITY_DEFAULT);
1633   g_source_set_callback (idle_source,
1634                          _g_dbus_worker_do_initial_read,
1635                          _g_dbus_worker_ref (worker),
1636                          (GDestroyNotify) _g_dbus_worker_unref);
1637   g_source_attach (idle_source, worker->shared_thread_data->context);
1638   g_source_unref (idle_source);
1639
1640   return worker;
1641 }
1642
1643 /* ---------------------------------------------------------------------------------------------------- */
1644
1645 /* can be called from any thread
1646  *
1647  * write_lock is not held on entry
1648  * output_pending may be anything
1649  */
1650 void
1651 _g_dbus_worker_close (GDBusWorker         *worker,
1652                       GCancellable        *cancellable,
1653                       GSimpleAsyncResult  *result)
1654 {
1655   CloseData *close_data;
1656
1657   close_data = g_slice_new0 (CloseData);
1658   close_data->worker = _g_dbus_worker_ref (worker);
1659   close_data->cancellable =
1660       (cancellable == NULL ? NULL : g_object_ref (cancellable));
1661   close_data->result = (result == NULL ? NULL : g_object_ref (result));
1662
1663   /* Don't set worker->close_expected here - we're in the wrong thread.
1664    * It'll be set before the actual close happens.
1665    */
1666   g_cancellable_cancel (worker->cancellable);
1667   g_mutex_lock (&worker->write_lock);
1668   schedule_writing_unlocked (worker, NULL, close_data);
1669   g_mutex_unlock (&worker->write_lock);
1670 }
1671
1672 /* This can be called from any thread - frees worker. Note that
1673  * callbacks might still happen if called from another thread than the
1674  * worker - use your own synchronization primitive in the callbacks.
1675  *
1676  * write_lock is not held on entry
1677  * output_pending may be anything
1678  */
1679 void
1680 _g_dbus_worker_stop (GDBusWorker *worker)
1681 {
1682   g_atomic_int_set (&worker->stopped, TRUE);
1683
1684   /* Cancel any pending operations and schedule a close of the underlying I/O
1685    * stream in the worker thread
1686    */
1687   _g_dbus_worker_close (worker, NULL, NULL);
1688
1689   /* _g_dbus_worker_close holds a ref until after an idle in the the worker
1690    * thread has run, so we no longer need to unref in an idle like in
1691    * commit 322e25b535
1692    */
1693   _g_dbus_worker_unref (worker);
1694 }
1695
1696 /* ---------------------------------------------------------------------------------------------------- */
1697
1698 /* can be called from any thread (except the worker thread) - blocks
1699  * calling thread until all queued outgoing messages are written and
1700  * the transport has been flushed
1701  *
1702  * write_lock is not held on entry
1703  * output_pending may be anything
1704  */
1705 gboolean
1706 _g_dbus_worker_flush_sync (GDBusWorker    *worker,
1707                            GCancellable   *cancellable,
1708                            GError        **error)
1709 {
1710   gboolean ret;
1711   FlushData *data;
1712
1713   data = NULL;
1714   ret = TRUE;
1715
1716   /* if the queue is empty, there's nothing to wait for */
1717   g_mutex_lock (&worker->write_lock);
1718   if (g_queue_get_length (worker->write_queue) > 0)
1719     {
1720       data = g_new0 (FlushData, 1);
1721       g_mutex_init (&data->mutex);
1722       g_cond_init (&data->cond);
1723       data->number_to_wait_for = worker->write_num_messages_written + g_queue_get_length (worker->write_queue);
1724       g_mutex_lock (&data->mutex);
1725       worker->write_pending_flushes = g_list_prepend (worker->write_pending_flushes, data);
1726     }
1727   g_mutex_unlock (&worker->write_lock);
1728
1729   if (data != NULL)
1730     {
1731       g_cond_wait (&data->cond, &data->mutex);
1732       g_mutex_unlock (&data->mutex);
1733
1734       /* note:the element is removed from worker->write_pending_flushes in flush_cb() above */
1735       g_cond_clear (&data->cond);
1736       g_mutex_clear (&data->mutex);
1737       if (data->error != NULL)
1738         {
1739           ret = FALSE;
1740           g_propagate_error (error, data->error);
1741         }
1742       g_free (data);
1743     }
1744
1745   return ret;
1746 }
1747
1748 /* ---------------------------------------------------------------------------------------------------- */
1749
1750 #define G_DBUS_DEBUG_AUTHENTICATION (1<<0)
1751 #define G_DBUS_DEBUG_TRANSPORT      (1<<1)
1752 #define G_DBUS_DEBUG_MESSAGE        (1<<2)
1753 #define G_DBUS_DEBUG_PAYLOAD        (1<<3)
1754 #define G_DBUS_DEBUG_CALL           (1<<4)
1755 #define G_DBUS_DEBUG_SIGNAL         (1<<5)
1756 #define G_DBUS_DEBUG_INCOMING       (1<<6)
1757 #define G_DBUS_DEBUG_RETURN         (1<<7)
1758 #define G_DBUS_DEBUG_EMISSION       (1<<8)
1759 #define G_DBUS_DEBUG_ADDRESS        (1<<9)
1760
1761 static gint _gdbus_debug_flags = 0;
1762
1763 gboolean
1764 _g_dbus_debug_authentication (void)
1765 {
1766   _g_dbus_initialize ();
1767   return (_gdbus_debug_flags & G_DBUS_DEBUG_AUTHENTICATION) != 0;
1768 }
1769
1770 gboolean
1771 _g_dbus_debug_transport (void)
1772 {
1773   _g_dbus_initialize ();
1774   return (_gdbus_debug_flags & G_DBUS_DEBUG_TRANSPORT) != 0;
1775 }
1776
1777 gboolean
1778 _g_dbus_debug_message (void)
1779 {
1780   _g_dbus_initialize ();
1781   return (_gdbus_debug_flags & G_DBUS_DEBUG_MESSAGE) != 0;
1782 }
1783
1784 gboolean
1785 _g_dbus_debug_payload (void)
1786 {
1787   _g_dbus_initialize ();
1788   return (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD) != 0;
1789 }
1790
1791 gboolean
1792 _g_dbus_debug_call (void)
1793 {
1794   _g_dbus_initialize ();
1795   return (_gdbus_debug_flags & G_DBUS_DEBUG_CALL) != 0;
1796 }
1797
1798 gboolean
1799 _g_dbus_debug_signal (void)
1800 {
1801   _g_dbus_initialize ();
1802   return (_gdbus_debug_flags & G_DBUS_DEBUG_SIGNAL) != 0;
1803 }
1804
1805 gboolean
1806 _g_dbus_debug_incoming (void)
1807 {
1808   _g_dbus_initialize ();
1809   return (_gdbus_debug_flags & G_DBUS_DEBUG_INCOMING) != 0;
1810 }
1811
1812 gboolean
1813 _g_dbus_debug_return (void)
1814 {
1815   _g_dbus_initialize ();
1816   return (_gdbus_debug_flags & G_DBUS_DEBUG_RETURN) != 0;
1817 }
1818
1819 gboolean
1820 _g_dbus_debug_emission (void)
1821 {
1822   _g_dbus_initialize ();
1823   return (_gdbus_debug_flags & G_DBUS_DEBUG_EMISSION) != 0;
1824 }
1825
1826 gboolean
1827 _g_dbus_debug_address (void)
1828 {
1829   _g_dbus_initialize ();
1830   return (_gdbus_debug_flags & G_DBUS_DEBUG_ADDRESS) != 0;
1831 }
1832
1833 G_LOCK_DEFINE_STATIC (print_lock);
1834
1835 void
1836 _g_dbus_debug_print_lock (void)
1837 {
1838   G_LOCK (print_lock);
1839 }
1840
1841 void
1842 _g_dbus_debug_print_unlock (void)
1843 {
1844   G_UNLOCK (print_lock);
1845 }
1846
1847 /*
1848  * _g_dbus_initialize:
1849  *
1850  * Does various one-time init things such as
1851  *
1852  *  - registering the G_DBUS_ERROR error domain
1853  *  - parses the G_DBUS_DEBUG environment variable
1854  */
1855 void
1856 _g_dbus_initialize (void)
1857 {
1858   static volatile gsize initialized = 0;
1859
1860   if (g_once_init_enter (&initialized))
1861     {
1862       volatile GQuark g_dbus_error_domain;
1863       const gchar *debug;
1864
1865       g_dbus_error_domain = G_DBUS_ERROR;
1866       (g_dbus_error_domain); /* To avoid -Wunused-but-set-variable */
1867
1868       debug = g_getenv ("G_DBUS_DEBUG");
1869       if (debug != NULL)
1870         {
1871           const GDebugKey keys[] = {
1872             { "authentication", G_DBUS_DEBUG_AUTHENTICATION },
1873             { "transport",      G_DBUS_DEBUG_TRANSPORT      },
1874             { "message",        G_DBUS_DEBUG_MESSAGE        },
1875             { "payload",        G_DBUS_DEBUG_PAYLOAD        },
1876             { "call",           G_DBUS_DEBUG_CALL           },
1877             { "signal",         G_DBUS_DEBUG_SIGNAL         },
1878             { "incoming",       G_DBUS_DEBUG_INCOMING       },
1879             { "return",         G_DBUS_DEBUG_RETURN         },
1880             { "emission",       G_DBUS_DEBUG_EMISSION       },
1881             { "address",        G_DBUS_DEBUG_ADDRESS        }
1882           };
1883
1884           _gdbus_debug_flags = g_parse_debug_string (debug, keys, G_N_ELEMENTS (keys));
1885           if (_gdbus_debug_flags & G_DBUS_DEBUG_PAYLOAD)
1886             _gdbus_debug_flags |= G_DBUS_DEBUG_MESSAGE;
1887         }
1888
1889       g_once_init_leave (&initialized, 1);
1890     }
1891 }
1892
1893 /* ---------------------------------------------------------------------------------------------------- */
1894
1895 GVariantType *
1896 _g_dbus_compute_complete_signature (GDBusArgInfo **args)
1897 {
1898   const GVariantType *arg_types[256];
1899   guint n;
1900
1901   if (args)
1902     for (n = 0; args[n] != NULL; n++)
1903       {
1904         /* DBus places a hard limit of 255 on signature length.
1905          * therefore number of args must be less than 256.
1906          */
1907         g_assert (n < 256);
1908
1909         arg_types[n] = G_VARIANT_TYPE (args[n]->signature);
1910
1911         if G_UNLIKELY (arg_types[n] == NULL)
1912           return NULL;
1913       }
1914   else
1915     n = 0;
1916
1917   return g_variant_type_new_tuple (arg_types, n);
1918 }
1919
1920 /* ---------------------------------------------------------------------------------------------------- */
1921
1922 #ifdef G_OS_WIN32
1923
1924 extern BOOL WINAPI ConvertSidToStringSidA (PSID Sid, LPSTR *StringSid);
1925
1926 gchar *
1927 _g_dbus_win32_get_user_sid (void)
1928 {
1929   HANDLE h;
1930   TOKEN_USER *user;
1931   DWORD token_information_len;
1932   PSID psid;
1933   gchar *sid;
1934   gchar *ret;
1935
1936   ret = NULL;
1937   user = NULL;
1938   h = INVALID_HANDLE_VALUE;
1939
1940   if (!OpenProcessToken (GetCurrentProcess (), TOKEN_QUERY, &h))
1941     {
1942       g_warning ("OpenProcessToken failed with error code %d", (gint) GetLastError ());
1943       goto out;
1944     }
1945
1946   /* Get length of buffer */
1947   token_information_len = 0;
1948   if (!GetTokenInformation (h, TokenUser, NULL, 0, &token_information_len))
1949     {
1950       if (GetLastError () != ERROR_INSUFFICIENT_BUFFER)
1951         {
1952           g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
1953           goto out;
1954         }
1955     }
1956   user = g_malloc (token_information_len);
1957   if (!GetTokenInformation (h, TokenUser, user, token_information_len, &token_information_len))
1958     {
1959       g_warning ("GetTokenInformation() failed with error code %d", (gint) GetLastError ());
1960       goto out;
1961     }
1962
1963   psid = user->User.Sid;
1964   if (!IsValidSid (psid))
1965     {
1966       g_warning ("Invalid SID");
1967       goto out;
1968     }
1969
1970   if (!ConvertSidToStringSidA (psid, &sid))
1971     {
1972       g_warning ("Invalid SID");
1973       goto out;
1974     }
1975
1976   ret = g_strdup (sid);
1977   LocalFree (sid);
1978
1979 out:
1980   g_free (user);
1981   if (h != INVALID_HANDLE_VALUE)
1982     CloseHandle (h);
1983   return ret;
1984 }
1985 #endif
1986
1987 /* ---------------------------------------------------------------------------------------------------- */
1988
1989 gchar *
1990 _g_dbus_get_machine_id (GError **error)
1991 {
1992   gchar *ret;
1993   GError *first_error;
1994   /* TODO: use PACKAGE_LOCALSTATEDIR ? */
1995   ret = NULL;
1996   first_error = NULL;
1997   if (!g_file_get_contents ("/var/lib/dbus/machine-id",
1998                             &ret,
1999                             NULL,
2000                             &first_error) &&
2001       !g_file_get_contents ("/etc/machine-id",
2002                             &ret,
2003                             NULL,
2004                             NULL))
2005     {
2006       g_propagate_prefixed_error (error, first_error,
2007                                   _("Unable to load /var/lib/dbus/machine-id or /etc/machine-id: "));
2008     }
2009   else
2010     {
2011       /* ignore the error from the first try, if any */
2012       g_clear_error (&first_error);
2013       /* TODO: validate value */
2014       g_strstrip (ret);
2015     }
2016   return ret;
2017 }
2018
2019 /* ---------------------------------------------------------------------------------------------------- */
2020
2021 gchar *
2022 _g_dbus_enum_to_string (GType enum_type, gint value)
2023 {
2024   gchar *ret;
2025   GEnumClass *klass;
2026   GEnumValue *enum_value;
2027
2028   klass = g_type_class_ref (enum_type);
2029   enum_value = g_enum_get_value (klass, value);
2030   if (enum_value != NULL)
2031     ret = g_strdup (enum_value->value_nick);
2032   else
2033     ret = g_strdup_printf ("unknown (value %d)", value);
2034   g_type_class_unref (klass);
2035   return ret;
2036 }
2037
2038 /* ---------------------------------------------------------------------------------------------------- */
2039
2040 static void
2041 write_message_print_transport_debug (gssize bytes_written,
2042                                      MessageToWriteData *data)
2043 {
2044   if (G_LIKELY (!_g_dbus_debug_transport ()))
2045     goto out;
2046
2047   _g_dbus_debug_print_lock ();
2048   g_print ("========================================================================\n"
2049            "GDBus-debug:Transport:\n"
2050            "  >>>> WROTE %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
2051            "       size %" G_GSIZE_FORMAT " from offset %" G_GSIZE_FORMAT " on a %s\n",
2052            bytes_written,
2053            g_dbus_message_get_serial (data->message),
2054            data->blob_size,
2055            data->total_written,
2056            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_output_stream (data->worker->stream))));
2057   _g_dbus_debug_print_unlock ();
2058  out:
2059   ;
2060 }
2061
2062 /* ---------------------------------------------------------------------------------------------------- */
2063
2064 static void
2065 read_message_print_transport_debug (gssize bytes_read,
2066                                     GDBusWorker *worker)
2067 {
2068   gsize size;
2069   gint32 serial;
2070   gint32 message_length;
2071
2072   if (G_LIKELY (!_g_dbus_debug_transport ()))
2073     goto out;
2074
2075   size = bytes_read + worker->read_buffer_cur_size;
2076   serial = 0;
2077   message_length = 0;
2078   if (size >= 16)
2079     message_length = g_dbus_message_bytes_needed ((guchar *) worker->read_buffer, size, NULL);
2080   if (size >= 1)
2081     {
2082       switch (worker->read_buffer[0])
2083         {
2084         case 'l':
2085           if (size >= 12)
2086             serial = GUINT32_FROM_LE (((guint32 *) worker->read_buffer)[2]);
2087           break;
2088         case 'B':
2089           if (size >= 12)
2090             serial = GUINT32_FROM_BE (((guint32 *) worker->read_buffer)[2]);
2091           break;
2092         default:
2093           /* an error will be set elsewhere if this happens */
2094           goto out;
2095         }
2096     }
2097
2098     _g_dbus_debug_print_lock ();
2099   g_print ("========================================================================\n"
2100            "GDBus-debug:Transport:\n"
2101            "  <<<< READ %" G_GSIZE_FORMAT " bytes of message with serial %d and\n"
2102            "       size %d to offset %" G_GSIZE_FORMAT " from a %s\n",
2103            bytes_read,
2104            serial,
2105            message_length,
2106            worker->read_buffer_cur_size,
2107            g_type_name (G_TYPE_FROM_INSTANCE (g_io_stream_get_input_stream (worker->stream))));
2108   _g_dbus_debug_print_unlock ();
2109  out:
2110   ;
2111 }
2112
2113 /* ---------------------------------------------------------------------------------------------------- */
2114
2115 gboolean
2116 _g_signal_accumulator_false_handled (GSignalInvocationHint *ihint,
2117                                      GValue                *return_accu,
2118                                      const GValue          *handler_return,
2119                                      gpointer               dummy)
2120 {
2121   gboolean continue_emission;
2122   gboolean signal_return;
2123
2124   signal_return = g_value_get_boolean (handler_return);
2125   g_value_set_boolean (return_accu, signal_return);
2126   continue_emission = signal_return;
2127
2128   return continue_emission;
2129 }