gio/gdbusprivate.c

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