mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2026-02-10 04:27:48 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

subscribe: rework some more

Browse source 
Track senders in the subscribe object and aggregate events from all
connected clients. This allows each client to get a complete view of all
the objects of pulsevideo. With all the source objects available, we can
then to the selection of the source on each client.

Remove the CreatSourceOutput on the Client1 object but let the client
select a good source and call CreateSourceOutput directly on the source.
This avoid going through the server to get a connection and the client
can just as well select a source.
Add a state property to the source and make it such that it can do async
state changes.

Remove the source provider object, each client can now directly see the
objects of another clients so there is no need for intermediate objects
in the server.
This commit is contained in:
Wim Taymans 2015-04-27 09:05:14 +02:00
parent 89c7955f3d
commit 417cd76f3e
17 changed files with 603 additions and 388 deletions
Download patch file Download diff file Expand all files Collapse all files

106
src/client/pv-context.c
View file

@ -40,8 +40,11 @@ struct _PvContextPrivate
PvClient1 *client;
PvSubscribe *internal_subscribe;
PvSubscribe *subscribe;
GList *sources;
GDBusObjectManagerServer *server_manager;
};
@ -51,6 +54,14 @@ struct _PvContextPrivate
G_DEFINE_TYPE (PvContext, pv_context, G_TYPE_OBJECT);
static void subscription_state (GObject *object, GParamSpec *pspec, gpointer user_data);
static void
subscription_cb (PvSubscribe *subscribe,
PvSubscriptionEvent event,
PvSubscriptionFlags flags,
GDBusProxy *object,
gpointer user_data);
enum
{
PROP_0,
@ -223,6 +234,10 @@ pv_context_init (PvContext * context)
priv->state = PV_CONTEXT_STATE_UNCONNECTED;
priv->server_manager = g_dbus_object_manager_server_new (PV_DBUS_OBJECT_PREFIX);
priv->internal_subscribe = pv_subscribe_new ();
g_object_set (priv->internal_subscribe, "subscription-mask", PV_SUBSCRIPTION_FLAGS_ALL, NULL);
g_signal_connect (priv->internal_subscribe, "subscription-event", (GCallback) subscription_cb, context);
g_signal_connect (priv->internal_subscribe, "notify::state", (GCallback) subscription_state, context);
}
/**
@ -302,15 +317,6 @@ on_daemon_connected (GObject *source_object,
{
PvContext *context = user_data;
PvContextPrivate *priv = context->priv;
GError *error = NULL;
priv->daemon = pv_daemon1_proxy_new_finish (res, &error);
if (priv->daemon == NULL) {
context_set_state (context, PV_CONTEXT_STATE_ERROR);
g_error ("failed to get daemon: %s", error->message);
g_clear_error (&error);
return;
}
context_set_state (context, PV_CONTEXT_STATE_REGISTERING);
@ -329,6 +335,57 @@ on_daemon_connected (GObject *source_object,
}
}
static void
subscription_cb (PvSubscribe *subscribe,
PvSubscriptionEvent event,
PvSubscriptionFlags flags,
GDBusProxy *object,
gpointer user_data)
{
PvContext *context = user_data;
PvContextPrivate *priv = context->priv;
g_print ("got event %d %d\n", event, flags);
switch (flags) {
case PV_SUBSCRIPTION_FLAGS_DAEMON:
priv->daemon = PV_DAEMON1 (object);
break;
case PV_SUBSCRIPTION_FLAGS_CLIENT:
break;
case PV_SUBSCRIPTION_FLAGS_SOURCE:
priv->sources = g_list_prepend (priv->sources, object);
break;
case PV_SUBSCRIPTION_FLAGS_SOURCE_OUTPUT:
break;
}
}
static void
subscription_state (GObject *object,
GParamSpec *pspec,
gpointer user_data)
{
PvContext *context = user_data;
PvSubscriptionState state;
g_object_get (object, "state", &state, NULL);
g_print ("got subscription state %d\n", state);
switch (state) {
case PV_SUBSCRIPTION_STATE_READY:
on_daemon_connected (NULL, NULL, context);
break;
default:
break;
}
}
static void
on_name_appeared (GDBusConnection *connection,
@ -343,17 +400,11 @@ on_name_appeared (GDBusConnection *connection,
g_dbus_object_manager_server_set_connection (priv->server_manager, connection);
if (priv->subscribe) {
g_object_set (priv->subscribe, "connection", priv->connection, NULL);
g_object_set (priv->subscribe, "service", name, NULL);
g_object_set (priv->subscribe, "connection", priv->connection,
"service", name, NULL);
}
pv_daemon1_proxy_new (priv->connection,
G_DBUS_PROXY_FLAGS_NONE,
name,
PV_DBUS_OBJECT_SERVER,
NULL,
on_daemon_connected,
user_data);
g_object_set (priv->internal_subscribe, "connection", priv->connection,
"service", name, NULL);
}
static void
@ -370,6 +421,8 @@ on_name_vanished (GDBusConnection *connection,
if (priv->subscribe)
g_object_set (priv->subscribe, "connection", connection, NULL);
g_object_set (priv->internal_subscribe, "connection", connection, NULL);
if (priv->flags & PV_CONTEXT_FLAGS_NOFAIL) {
context_set_state (context, PV_CONTEXT_STATE_CONNECTING);
} else {
@ -586,3 +639,18 @@ pv_context_get_client_proxy (PvContext *context)
return G_DBUS_PROXY (context->priv->client);
}
GDBusProxy *
pv_context_find_source (PvContext *context, const gchar *name, GVariant *props)
{
PvContextPrivate *priv;
g_return_val_if_fail (PV_IS_CONTEXT (context), NULL);
priv = context->priv;
if (priv->sources == NULL)
return NULL;
return priv->sources->data;
}

3
src/client/pv-context.h
View file

@ -109,9 +109,12 @@ gboolean pv_context_register_source (PvContext *context, PvSource
gboolean pv_context_unregister_source (PvContext *context, PvSource *source);
PvContextState pv_context_get_state (PvContext *context);
GDBusConnection * pv_context_get_connection (PvContext *context);
GDBusProxy * pv_context_get_client_proxy (PvContext *context);
GDBusProxy * pv_context_find_source (PvContext *context, const gchar *name, GVariant *props);
G_END_DECLS

84
src/client/pv-source.c
View file

@ -21,6 +21,7 @@
#include "client/pulsevideo.h"
#include "client/pv-source.h"
#include "client/pv-enumtypes.h"
#include "dbus/org-pulsevideo.h"
@ -31,10 +32,11 @@
struct _PvSourcePrivate
{
GDBusObjectManagerServer *server_manager;
PvSource1 *iface;
gchar *object_path;
gchar *name;
gboolean suspended;
PvSourceState state;
GVariant *properties;
};
@ -46,7 +48,7 @@ enum
PROP_MANAGER,
PROP_OBJECT_PATH,
PROP_NAME,
PROP_SUSPENDED,
PROP_STATE,
PROP_PROPERTIES
};
@ -72,8 +74,8 @@ pv_source_get_property (GObject *_object,
g_value_set_string (value, priv->name);
break;
case PROP_SUSPENDED:
g_value_set_boolean (value, priv->suspended);
case PROP_STATE:
g_value_set_enum (value, priv->state);
break;
case PROP_PROPERTIES:
@ -168,20 +170,22 @@ source_register_object (PvSource *source)
PvObjectSkeleton *skel;
skel = pv_object_skeleton_new (PV_DBUS_OBJECT_SOURCE);
{
PvSource1 *iface;
iface = pv_source1_skeleton_new ();
g_object_set (iface, "name", priv->name,
"suspended", priv->suspended,
"properties", priv->properties,
NULL);
g_signal_connect (iface, "handle-create-source-output", (GCallback) handle_create_source_output, source);
g_signal_connect (iface, "handle-get-capabilities", (GCallback) handle_get_capabilities, source);
pv_object_skeleton_set_source1 (skel, iface);
g_object_unref (iface);
}
priv->iface = pv_source1_skeleton_new ();
g_object_set (priv->iface, "name", priv->name,
"state", priv->state,
"properties", priv->properties,
NULL);
g_signal_connect (priv->iface, "handle-create-source-output",
(GCallback) handle_create_source_output,
source);
g_signal_connect (priv->iface, "handle-get-capabilities",
(GCallback) handle_get_capabilities,
source);
pv_object_skeleton_set_source1 (skel, priv->iface);
g_dbus_object_manager_server_export_uniquely (priv->server_manager, G_DBUS_OBJECT_SKELETON (skel));
g_object_unref (skel);
g_free (priv->object_path);
priv->object_path = g_strdup (g_dbus_object_get_object_path (G_DBUS_OBJECT (skel)));
@ -195,6 +199,7 @@ source_unregister_object (PvSource *source)
PvSourcePrivate *priv = source->priv;
g_dbus_object_manager_server_unexport (priv->server_manager, priv->object_path);
g_clear_object (&priv->iface);
}
static void
@ -272,13 +277,14 @@ pv_source_class_init (PvSourceClass * klass)
G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class,
PROP_SUSPENDED,
g_param_spec_boolean ("suspended",
"Suspended",
"The suspended state of the source",
FALSE,
G_PARAM_READABLE |
G_PARAM_STATIC_STRINGS));
PROP_STATE,
g_param_spec_enum ("state",
"State",
"The state of the source",
PV_TYPE_SOURCE_STATE,
PV_SOURCE_STATE_INIT,
G_PARAM_READABLE |
G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class,
PROP_PROPERTIES,
@ -339,7 +345,7 @@ pv_source_get_capabilities (PvSource *source, GVariant *props)
}
gboolean
pv_source_suspend (PvSource *source)
pv_source_set_state (PvSource *source, PvSourceState state)
{
PvSourceClass *klass;
gboolean res;
@ -348,33 +354,31 @@ pv_source_suspend (PvSource *source)
klass = PV_SOURCE_GET_CLASS (source);
if (klass->suspend)
res = klass->suspend (source);
if (klass->set_state)
res = klass->set_state (source, state);
else
res = FALSE;
return res;
}
gboolean
pv_source_resume (PvSource *source)
void
pv_source_update_state (PvSource *source, PvSourceState state)
{
PvSourceClass *klass;
gboolean res;
PvSourcePrivate *priv;
g_return_val_if_fail (PV_IS_SOURCE (source), FALSE);
g_return_if_fail (PV_IS_SOURCE (source));
priv = source->priv;
klass = PV_SOURCE_GET_CLASS (source);
if (klass->resume)
res = klass->resume (source);
else
res = FALSE;
return res;
if (priv->state != state) {
priv->state = state;
g_print ("source changed state %d\n", state);
if (priv->iface)
pv_source1_set_state (priv->iface, state);
g_object_notify (G_OBJECT (source), "state");
}
}
PvSourceOutput *
pv_source_create_source_output (PvSource *source, GVariant *props, const gchar *prefix)
{

33
src/client/pv-source.h
View file

@ -40,6 +40,26 @@ typedef struct _PvSourcePrivate PvSourcePrivate;
#define PV_SOURCE_CAST(obj) ((PvSource*)(obj))
#define PV_SOURCE_CLASS_CAST(klass) ((PvSourceClass*)(klass))
/**
* PvSourceState:
* @PV_SOURCE_STATE_ERROR: the source is in error
* @PV_SOURCE_STATE_SUSPENDED: the source is suspended, the device might
* be closed
* @PV_SOURCE_STATE_INIT: the source is initializing, it opens the device
* and gets the device capabilities
* @PV_SOURCE_STATE_IDLE: the source is running but there is no active
* source-output
* @PV_SOURCE_STATE_RUNNING: the source is running.
*
* The different source states
*/
typedef enum {
PV_SOURCE_STATE_ERROR = 0,
PV_SOURCE_STATE_SUSPENDED = 1,
PV_SOURCE_STATE_INIT = 2,
PV_SOURCE_STATE_IDLE = 3,
PV_SOURCE_STATE_RUNNING = 4,
} PvSourceState;
/**
* PvSource:
*
@ -53,16 +73,19 @@ struct _PvSource {
/**
* PvSourceClass:
* @get_capabilities: called to get a list of supported formats from the source
* @set_state: called to change the current state of the source
* @create_source_output: called to create a new source-output object
* @release_source_output: called to release a source-output object
*
* Pulsevideo source object class.
*/
struct _PvSourceClass {
GObjectClass parent_class;
GVariant * (*get_capabilities) (PvSource *source, GVariant *props);
GVariant * (*get_capabilities) (PvSource *source, GVariant *props);
gboolean (*suspend) (PvSource *source);
gboolean (*resume) (PvSource *source);
gboolean (*set_state) (PvSource *source, PvSourceState);
PvSourceOutput * (*create_source_output) (PvSource *source, GVariant *props, const gchar *prefix);
gboolean (*release_source_output) (PvSource *source, PvSourceOutput *output);
@ -75,8 +98,8 @@ void pv_source_set_manager (PvSource *source, GDBusObject
GVariant * pv_source_get_capabilities (PvSource *source, GVariant *props);
gboolean pv_source_suspend (PvSource *source);
gboolean pv_source_resume (PvSource *source);
gboolean pv_source_set_state (PvSource *source, PvSourceState state);
void pv_source_update_state (PvSource *source, PvSourceState state);
PvSourceOutput * pv_source_create_source_output (PvSource *source, GVariant *props, const gchar *prefix);
gboolean pv_source_release_source_output (PvSource *source, PvSourceOutput *output);

52
src/client/pv-stream.c
View file

@ -33,8 +33,8 @@ struct _PvStreamPrivate
gchar *target;
PvStreamState state;
gchar *source_output_sender;
gchar *source_output_path;
PvSource1 *source;
PvSourceOutput1 *source_output;
GSocket *socket;
@ -319,19 +319,16 @@ on_source_output_created (GObject *source_object,
PvStreamPrivate *priv = stream->priv;
PvContext *context = priv->context;
GError *error = NULL;
PvClient1 *proxy;
proxy = PV_CLIENT1 (pv_context_get_client_proxy (priv->context));
if (!pv_client1_call_create_source_output_finish (proxy,
&priv->source_output_sender, &priv->source_output_path, res, &error))
if (!pv_source1_call_create_source_output_finish (priv->source,
&priv->source_output_path, res, &error))
goto create_failed;
g_print ("got source-output %s %s\n", priv->source_output_sender, priv->source_output_path);
g_print ("got source-output %s\n", priv->source_output_path);
pv_source_output1_proxy_new (pv_context_get_connection (context),
G_DBUS_PROXY_FLAGS_NONE,
priv->source_output_sender,
g_dbus_proxy_get_name (G_DBUS_PROXY (priv->source)),
priv->source_output_path,
NULL,
on_source_output1_proxy,
@ -348,27 +345,6 @@ create_failed:
}
}
static gboolean
create_source_output (PvStream *stream)
{
PvStreamPrivate *priv = stream->priv;
GVariantBuilder builder;
PvClient1 *proxy;
g_variant_builder_init (&builder, G_VARIANT_TYPE ("a{sv}"));
g_variant_builder_add (&builder, "{sv}", "name", g_variant_new_string ("hello"));
proxy = PV_CLIENT1 (pv_context_get_client_proxy (priv->context));
pv_client1_call_create_source_output (proxy,
priv->target ? priv->target : "/", /* const gchar *arg_source */
g_variant_builder_end (&builder), /* GVariant *arg_props */
NULL, /* GCancellable *cancellable */
on_source_output_created,
stream);
return TRUE;
}
static void
on_source_output_removed (GObject *source_object,
GAsyncResult *res,
@ -385,7 +361,6 @@ on_source_output_removed (GObject *source_object,
g_clear_error (&error);
return;
}
g_clear_pointer (&priv->source_output_sender, g_free);
g_clear_pointer (&priv->source_output_path, g_free);
g_clear_object (&priv->source_output);
}
@ -418,6 +393,7 @@ pv_stream_connect_capture (PvStream *stream,
PvStreamFlags flags)
{
PvStreamPrivate *priv;
GVariantBuilder builder;
PvContext *context;
g_return_val_if_fail (PV_IS_STREAM (stream), FALSE);
@ -429,7 +405,21 @@ pv_stream_connect_capture (PvStream *stream,
stream_set_state (stream, PV_STREAM_STATE_CONNECTING);
return create_source_output (stream);
priv->source = PV_SOURCE1 (pv_context_find_source (context, priv->target, NULL));
if (priv->source == NULL) {
g_warning ("can't find source");
return FALSE;
}
g_variant_builder_init (&builder, G_VARIANT_TYPE ("a{sv}"));
g_variant_builder_add (&builder, "{sv}", "name", g_variant_new_string ("hello"));
pv_source1_call_create_source_output (priv->source,
g_variant_builder_end (&builder), /* GVariant *arg_props */
NULL, /* GCancellable *cancellable */
on_source_output_created,
stream);
return TRUE;
}
/**

282
src/client/pv-subscribe.c
View file

@ -26,12 +26,16 @@
struct _PvSubscribePrivate
{
PvSubscriptionState state;
GDBusConnection *connection;
gchar *service;
PvSubscriptionFlags subscription_mask;
GDBusObjectManager *client_manager;
guint pending_subscribes;
GHashTable *senders;
};
@ -45,7 +49,8 @@ enum
PROP_0,
PROP_CONNECTION,
PROP_SERVICE,
PROP_SUBSCRIPTION_MASK
PROP_SUBSCRIPTION_MASK,
PROP_STATE,
};
enum
@ -56,6 +61,161 @@ enum
static guint signals[LAST_SIGNAL] = { 0 };
typedef struct {
PvSubscribe *subscribe;
gchar *sender;
guint id;
PvSubscribe *sender_subscribe;
GList *clients;
} SenderData;
static void
notify_subscription (PvSubscribe *subscribe,
GDBusObject *object,
GDBusInterface *interface,
PvSubscriptionEvent event);
static void
on_sender_subscription_event (PvSubscribe *sender_subscribe,
PvSubscriptionEvent event,
PvSubscriptionFlags flags,
GDBusProxy *object,
gpointer user_data)
{
SenderData *data = user_data;
PvSubscribe *subscribe = data->subscribe;
g_signal_emit (subscribe,
signals[SIGNAL_SUBSCRIPTION_EVENT],
0,
event,
flags,
object);
}
static void
subscription_set_state (PvSubscribe *subscribe, PvSubscriptionState state)
{
PvSubscribePrivate *priv = subscribe->priv;
if (state != priv->state) {
priv->state = state;
g_object_notify (G_OBJECT (subscribe), "state");
}
}
static void
on_sender_subscription_state (GObject *object,
GParamSpec *pspec,
gpointer user_data)
{
SenderData *data = user_data;
PvSubscribe *subscribe = data->subscribe;
PvSubscribePrivate *priv = subscribe->priv;
PvSubscriptionState state;
g_object_get (object, "state", &state, NULL);
switch (state) {
case PV_SUBSCRIPTION_STATE_READY:
if (--priv->pending_subscribes == 0)
subscription_set_state (subscribe, state);
break;
case PV_SUBSCRIPTION_STATE_ERROR:
subscription_set_state (subscribe, state);
break;
default:
break;
}
}
static void
client_name_appeared_handler (GDBusConnection *connection,
const gchar *name,
const gchar *name_owner,
gpointer user_data)
{
SenderData *data = user_data;
/* subscribe to Source events. We want to be notified when this new
* sender add/change/remove sources and outputs */
data->sender_subscribe = pv_subscribe_new ();
g_object_set (data->sender_subscribe, "service", data->sender,
"subscription-mask", PV_SUBSCRIPTION_FLAGS_ALL,
"connection", connection,
NULL);
g_signal_connect (data->sender_subscribe,
"subscription-event",
(GCallback) on_sender_subscription_event,
data);
g_signal_connect (data->sender_subscribe,
"notify::state",
(GCallback) on_sender_subscription_state,
data);
}
static void
remove_client (PvClient1 *client, SenderData *data)
{
g_signal_emit (data->subscribe,
signals[SIGNAL_SUBSCRIPTION_EVENT],
0,
PV_SUBSCRIPTION_EVENT_REMOVE,
PV_SUBSCRIPTION_FLAGS_CLIENT,
client);
}
static void
client_name_vanished_handler (GDBusConnection *connection,
const gchar *name,
gpointer user_data)
{
SenderData *data = user_data;
PvSubscribePrivate *priv = data->subscribe->priv;
g_print ("vanished client %s\n", name);
g_list_foreach (data->clients, (GFunc) remove_client, data);
g_hash_table_remove (priv->senders, data->sender);
if (data->sender_subscribe)
g_object_unref (data->sender_subscribe);
g_free (data->sender);
g_bus_unwatch_name (data->id);
g_free (data);
}
static SenderData *
sender_data_new (PvSubscribe *subscribe, const gchar *sender)
{
PvSubscribePrivate *priv = subscribe->priv;
SenderData *data;
g_print ("watch name %s\n", sender);
data = g_new0 (SenderData, 1);
data->subscribe = subscribe;
data->sender = g_strdup (sender);
data->id = g_bus_watch_name_on_connection (priv->connection,
sender,
G_BUS_NAME_WATCHER_FLAGS_NONE,
client_name_appeared_handler,
client_name_vanished_handler,
data,
NULL);
g_hash_table_insert (priv->senders, data->sender, data);
priv->pending_subscribes++;
return data;
}
static void
notify_subscription (PvSubscribe *subscribe,
GDBusObject *object,
@ -64,29 +224,81 @@ notify_subscription (PvSubscribe *subscribe,
{
PvSubscribePrivate *priv = subscribe->priv;
if (priv->subscription_mask & PV_SUBSCRIPTION_FLAGS_CLIENT) {
if ((interface == NULL && pv_object_peek_client1 (PV_OBJECT (object))) ||
PV_IS_CLIENT1_PROXY (interface))
if (priv->subscription_mask & PV_SUBSCRIPTION_FLAGS_DAEMON) {
PvDaemon1 *daemon;
if (interface == NULL)
daemon = pv_object_peek_daemon1 (PV_OBJECT (object));
else if (PV_IS_DAEMON1_PROXY (interface))
daemon = PV_DAEMON1 (interface);
else
daemon = NULL;
if (daemon) {
g_signal_emit (subscribe, signals[SIGNAL_SUBSCRIPTION_EVENT], 0, event,
PV_SUBSCRIPTION_FLAGS_CLIENT, object);
PV_SUBSCRIPTION_FLAGS_DAEMON, daemon);
}
}
if (priv->subscription_mask & PV_SUBSCRIPTION_FLAGS_SOURCE_PROVIDER) {
if ((interface == NULL && pv_object_peek_source_provider1 (PV_OBJECT (object))) ||
PV_IS_SOURCE_PROVIDER1_PROXY (interface))
g_signal_emit (subscribe, signals[SIGNAL_SUBSCRIPTION_EVENT], 0, event,
PV_SUBSCRIPTION_FLAGS_SOURCE_PROVIDER, object);
if (priv->subscription_mask & PV_SUBSCRIPTION_FLAGS_CLIENT) {
PvClient1 *client;
if (interface == NULL)
client = pv_object_peek_client1 (PV_OBJECT (object));
else if (PV_IS_CLIENT1_PROXY (interface))
client = PV_CLIENT1 (interface);
else
client = NULL;
if (client) {
const gchar *sender;
SenderData *data;
sender = pv_client1_get_name (client);
data = g_hash_table_lookup (priv->senders, sender);
if (data == NULL && event != PV_SUBSCRIPTION_EVENT_REMOVE) {
data = sender_data_new (subscribe, sender);
}
if (data) {
if (event == PV_SUBSCRIPTION_EVENT_NEW)
data->clients = g_list_prepend (data->clients, client);
else if (event == PV_SUBSCRIPTION_EVENT_REMOVE)
data->clients = g_list_remove (data->clients, client);
g_signal_emit (subscribe, signals[SIGNAL_SUBSCRIPTION_EVENT], 0, event,
PV_SUBSCRIPTION_FLAGS_CLIENT, client);
}
}
}
if (priv->subscription_mask & PV_SUBSCRIPTION_FLAGS_SOURCE) {
if ((interface == NULL && pv_object_peek_source1 (PV_OBJECT (object))) ||
PV_IS_SOURCE1_PROXY (interface))
PvSource1 *source;
if (interface == NULL)
source = pv_object_peek_source1 (PV_OBJECT (object));
else if (PV_IS_SOURCE1_PROXY (interface))
source = PV_SOURCE1 (interface);
else
source = NULL;
if (source) {
g_signal_emit (subscribe, signals[SIGNAL_SUBSCRIPTION_EVENT], 0, event,
PV_SUBSCRIPTION_FLAGS_SOURCE, object);
PV_SUBSCRIPTION_FLAGS_SOURCE, source);
}
}
if (priv->subscription_mask & PV_SUBSCRIPTION_FLAGS_SOURCE_OUTPUT) {
if ((interface == NULL && pv_object_peek_source_output1 (PV_OBJECT (object))) ||
PV_IS_SOURCE_OUTPUT1_PROXY (interface))
PvSourceOutput1 *output;
if (interface == NULL)
output = pv_object_peek_source_output1 (PV_OBJECT (object));
else if PV_IS_SOURCE_OUTPUT1_PROXY (interface)
output = PV_SOURCE_OUTPUT1 (interface);
else
output = NULL;
if (output) {
g_signal_emit (subscribe, signals[SIGNAL_SUBSCRIPTION_EVENT], 0, event,
PV_SUBSCRIPTION_FLAGS_SOURCE_OUTPUT, object);
PV_SUBSCRIPTION_FLAGS_SOURCE_OUTPUT, output);
}
}
}
@ -184,13 +396,20 @@ on_client_manager_ready (GObject *source_object,
if (priv->client_manager == NULL)
goto manager_error;
g_print ("client manager %s %s\n",
g_dbus_object_manager_client_get_name (G_DBUS_OBJECT_MANAGER_CLIENT (priv->client_manager)),
g_dbus_object_manager_client_get_name_owner (G_DBUS_OBJECT_MANAGER_CLIENT (priv->client_manager)));
connect_client_signals (subscribe);
objects = g_dbus_object_manager_get_objects (G_DBUS_OBJECT_MANAGER (priv->client_manager));
for (walk = objects; walk ; walk = g_list_next (walk)) {
on_client_manager_object_added (G_DBUS_OBJECT_MANAGER (priv->client_manager),
walk->data,
subscribe);
}
connect_client_signals (subscribe);
if (--priv->pending_subscribes == 0)
subscription_set_state (subscribe, PV_SUBSCRIPTION_STATE_READY);
return;
@ -199,6 +418,7 @@ manager_error:
{
g_warning ("could not create client manager: %s", error->message);
g_clear_error (&error);
subscription_set_state (subscribe, PV_SUBSCRIPTION_STATE_ERROR);
return;
}
}
@ -208,6 +428,9 @@ install_subscription (PvSubscribe *subscribe)
{
PvSubscribePrivate *priv = subscribe->priv;
subscription_set_state (subscribe, PV_SUBSCRIPTION_STATE_CONNECTING);
g_print ("new client manager for %s\n", priv->service);
pv_object_manager_client_new (priv->connection,
G_DBUS_OBJECT_MANAGER_CLIENT_FLAGS_NONE,
priv->service,
@ -215,6 +438,7 @@ install_subscription (PvSubscribe *subscribe)
NULL,
on_client_manager_ready,
subscribe);
priv->pending_subscribes++;
}
static void
@ -223,6 +447,7 @@ uninstall_subscription (PvSubscribe *subscribe)
PvSubscribePrivate *priv = subscribe->priv;
g_clear_object (&priv->client_manager);
subscription_set_state (subscribe, PV_SUBSCRIPTION_STATE_UNCONNECTED);
}
static void
@ -247,6 +472,10 @@ pv_subscribe_get_property (GObject *_object,
g_value_set_flags (value, priv->subscription_mask);
break;
case PROP_STATE:
g_value_set_enum (value, priv->state);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (subscribe, prop_id, pspec);
break;
@ -296,6 +525,7 @@ pv_subscribe_finalize (GObject * object)
g_free (priv->service);
g_object_unref (priv->client_manager);
g_hash_table_unref (priv->senders);
G_OBJECT_CLASS (pv_subscribe_parent_class)->finalize (object);
}
@ -352,6 +582,20 @@ pv_subscribe_class_init (PvSubscribeClass * klass)
0,
G_PARAM_READWRITE |
G_PARAM_STATIC_STRINGS));
/**
* PvSubscribe:state
*
* The state of the subscription
*/
g_object_class_install_property (gobject_class,
PROP_STATE,
g_param_spec_enum ("state",
"State",
"The state",
PV_TYPE_SUBSCRIPTION_STATE,
PV_SUBSCRIPTION_STATE_UNCONNECTED,
G_PARAM_READABLE |
G_PARAM_STATIC_STRINGS));
/**
* PvSubscribe:subscription-event
* @subscribe: The #PvSubscribe emitting the signal.
@ -372,7 +616,7 @@ pv_subscribe_class_init (PvSubscribeClass * klass)
3,
PV_TYPE_SUBSCRIPTION_EVENT,
PV_TYPE_SUBSCRIPTION_FLAGS,
G_TYPE_DBUS_OBJECT_PROXY);
G_TYPE_DBUS_PROXY);
}
static void
@ -381,6 +625,8 @@ pv_subscribe_init (PvSubscribe * subscribe)
PvSubscribePrivate *priv = subscribe->priv = PV_SUBSCRIBE_GET_PRIVATE (subscribe);
priv->service = g_strdup (PV_DBUS_SERVICE);
priv->senders = g_hash_table_new (g_str_hash, g_str_equal);
priv->state = PV_SUBSCRIPTION_STATE_UNCONNECTED;
}
/**

11
src/client/pv-subscribe.h
View file

@ -38,8 +38,15 @@ typedef struct _PvSubscribeClass PvSubscribeClass;
typedef struct _PvSubscribePrivate PvSubscribePrivate;
typedef enum {
PV_SUBSCRIPTION_FLAGS_CLIENT = (1 << 0),
PV_SUBSCRIPTION_FLAGS_SOURCE_PROVIDER = (1 << 1),
PV_SUBSCRIPTION_STATE_UNCONNECTED = 0,
PV_SUBSCRIPTION_STATE_CONNECTING = 1,
PV_SUBSCRIPTION_STATE_READY = 2,
PV_SUBSCRIPTION_STATE_ERROR = 3,
} PvSubscriptionState;
typedef enum {
PV_SUBSCRIPTION_FLAGS_DAEMON = (1 << 0),
PV_SUBSCRIPTION_FLAGS_CLIENT = (1 << 1),
PV_SUBSCRIPTION_FLAGS_SOURCE = (1 << 2),
PV_SUBSCRIPTION_FLAGS_SOURCE_OUTPUT = (1 << 3),
} PvSubscriptionFlags;