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