[kdbus] Add initial support for receiving messages

diff --git a/gio/gdbusprivate.c b/gio/gdbusprivate.c
index 924d833..f57bf7b 100644 (file)
--- a/gio/gdbusprivate.c
+++ b/gio/gdbusprivate.c
@@ -92,6 +92,107 @@ _g_dbus_hexdump (const gchar *data, gsize len, guint indent)
 
 /* ---------------------------------------------------------------------------------------------------- */
 
+#if defined (G_OS_UNIX) && (KDBUS_TRANSPORT)
+typedef struct
+{
+  GKdbus *kdbus;
+  GCancellable *cancellable;
+
+  GSimpleAsyncResult *simple;
+
+  gboolean from_mainloop;
+} ReadKdbusData;
+
+static void
+read_kdbus_data_free (ReadKdbusData  *data)
+{
+  g_object_unref (data->kdbus);
+  if (data->cancellable != NULL)
+    g_object_unref (data->cancellable);
+  g_object_unref (data->simple);
+  g_free (data);
+}
+
+static gboolean
+_g_kdbus_read_ready (GKdbus        *kdbus,
+                     GIOCondition   condition,
+                     gpointer       user_data)
+{
+  ReadKdbusData *data = user_data;
+  GError *error = NULL;
+  gssize result;
+
+  result = _g_kdbus_receive (data->kdbus,
+                             data->cancellable,
+                             &error);
+
+  if (result >= 0)
+    {
+      g_simple_async_result_set_op_res_gssize (data->simple, result);
+    }
+  else
+    {
+      g_assert (error != NULL);
+      g_simple_async_result_take_error (data->simple, error);
+    }
+
+  if (data->from_mainloop)
+    g_simple_async_result_complete (data->simple);
+  else
+    g_simple_async_result_complete_in_idle (data->simple);
+
+  return FALSE;
+}
+
+static void
+_g_kdbus_read (GKdbus               *kdbus,
+               GCancellable         *cancellable,
+               GAsyncReadyCallback   callback,
+               gpointer              user_data)
+{
+  ReadKdbusData *data;
+  GSource *source;
+
+  data = g_new0 (ReadKdbusData, 1);
+  data->kdbus = g_object_ref (kdbus);
+  data->cancellable = cancellable != NULL ? g_object_ref (cancellable) : NULL;
+
+  data->simple = g_simple_async_result_new (G_OBJECT (kdbus),
+                                            callback,
+                                            user_data,
+                                            _g_kdbus_read);
+  g_simple_async_result_set_check_cancellable (data->simple, cancellable);
+
+  data->from_mainloop = TRUE;
+  source = _g_kdbus_create_source (data->kdbus,
+                                   G_IO_IN,
+                                   cancellable);
+  g_source_set_callback (source,
+                         (GSourceFunc) _g_kdbus_read_ready,
+                         data,
+                         (GDestroyNotify) read_kdbus_data_free);
+  g_source_attach (source, g_main_context_get_thread_default ());
+  g_source_unref (source);
+}
+
+static gssize
+_g_kdbus_read_finish (GKdbus        *kdbus,
+                      GAsyncResult  *result,
+                      GError       **error)
+{
+  GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
+
+  g_return_val_if_fail (G_IS_KDBUS (kdbus), -1);
+  g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == _g_kdbus_read);
+
+  if (g_simple_async_result_propagate_error (simple, error))
+    return -1;
+  else
+    return g_simple_async_result_get_op_res_gssize (simple);
+}
+
+#endif /* defined (G_OS_UNIX) && (KDBUS_TRANSPORT) */
+
 /* Unfortunately ancillary messages are discarded when reading from a
  * socket using the GSocketInputStream abstraction. So we provide a
  * very GInputStream-ish API that uses GSocket in this case (very
@@ -585,7 +686,21 @@ _g_dbus_worker_do_read_cb (GInputStream  *input_stream,
     goto out;
 
   error = NULL;
-  if (worker->socket == NULL)
+  bytes_read = 0;
+
+  if (FALSE)
+    {
+    }
+#if defined (G_OS_UNIX) && (KDBUS_TRANSPORT)
+  else if (G_IS_KDBUS_CONNECTION (worker->stream))
+    {
+      bytes_read = _g_kdbus_read_finish (worker->kdbus,
+                                         res,
+                                         &error);
+      g_error ("[KDBUS] _g_dbus_worker_do_read_cb() - work in progress");
+    }
+#endif
+  else if (worker->socket == NULL)
     bytes_read = g_input_stream_read_finish (g_io_stream_get_input_stream (worker->stream),
                                              res,
                                              &error);
@@ -832,6 +947,24 @@ _g_dbus_worker_do_read_unlocked (GDBusWorker *worker)
    * true, because only failing a read causes us to signal 'closed'.
    */
 
+  /* [KDBUS]
+   * For KDBUS transport we don't  have to alloc buffer (worker->read_buffer)
+   * instead of it we use kdbus memory pool. On connection stage KDBUS client
+   * have to register a memory pool, large enough to  carry all backlog of
+   * data enqueued for the connection.
+   */
+
+#if defined (G_OS_UNIX) && (KDBUS_TRANSPORT)
+  if (G_IS_KDBUS_CONNECTION (worker->stream))
+    {
+      _g_kdbus_read(worker->kdbus,
+                    worker->cancellable,
+                    (GAsyncReadyCallback) _g_dbus_worker_do_read_cb,
+                    _g_dbus_worker_ref (worker));
+      return;
+    }
+#endif
+
   /* if bytes_wanted is zero, it means start reading a message */
   if (worker->read_buffer_bytes_wanted == 0)
     {

diff --git a/gio/gkdbus.c b/gio/gkdbus.c
index 312b34f..cfb3996 100644 (file)
--- a/gio/gkdbus.c
+++ b/gio/gkdbus.c
@@ -101,6 +101,8 @@ struct _GKdbusPrivate
   gint               fd;
 
   gchar             *kdbus_buffer;
+  struct kdbus_msg  *kmsg;
+
   gchar             *unique_name;
   guint64            unique_id;
 
@@ -1248,3 +1250,44 @@ _g_kdbus_GetConnectionUnixUser (GDBusConnection  *connection,
                                        G_BUS_CREDS_UID,
                                        error);
 }
+
+
+/**
+ * _g_kdbus_receive:
+ *
+ */
+gssize
+_g_kdbus_receive (GKdbus        *kdbus,
+                  GCancellable  *cancellable,
+                  GError       **error)
+{
+  struct kdbus_cmd_recv recv = {};
+  gssize size = 0;
+
+  if (g_cancellable_set_error_if_cancelled (cancellable, error))
+   return -1;
+
+  again:
+    if (ioctl(kdbus->priv->fd, KDBUS_CMD_MSG_RECV, &recv) < 0)
+     {
+        if (errno == EINTR || errno == EAGAIN)
+         goto again;
+
+       g_set_error (error, G_IO_ERROR, g_io_error_from_errno(errno),_("Error receiving message - KDBUS_CMD_MSG_RECV error"));
+       return -1;
+     }
+
+   kdbus->priv->kmsg = (struct kdbus_msg *)((guint8 *)kdbus->priv->kdbus_buffer + recv.offset);
+
+   if (kdbus->priv->kmsg->payload_type == KDBUS_PAYLOAD_DBUS)
+     //size = g_kdbus_decode_dbus_msg (kdbus);
+     g_print ("Standard message\n");
+   else if (kdbus->priv->kmsg->payload_type == KDBUS_PAYLOAD_KERNEL)
+     //size = g_kdbus_decode_kernel_msg (kdbus);
+     g_print ("Message from kernel\n");
+   else
+     //g_set_error
+     g_error ("Unknown payload type: %llu", kdbus->priv->kmsg->payload_type);
+
+   return size;
+}

diff --git a/gio/gkdbus.h b/gio/gkdbus.h
index b449cc6..bb6adac 100644 (file)
--- a/gio/gkdbus.h
+++ b/gio/gkdbus.h
@@ -108,6 +108,11 @@ GVariant *                              _g_kdbus_GetConnectionUnixProcessID (GDB
 GVariant *                              _g_kdbus_GetConnectionUnixUser      (GDBusConnection  *connection,
                                                                              const gchar      *name,
                                                                              GError          **error);
+
+gssize                                  _g_kdbus_receive                    (GKdbus           *kdbus,
+                                                                             GCancellable     *cancellable,
+                                                                             GError          **error);
+
 G_END_DECLS
 
 #endif /* __G_KDBUS_H__ */