mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-01 22:58:50 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Work on node creation

Browse source 
Implements the basics of the PORT_UPDATE control message
Add the ports to the proxy node with whe PORT_UPDATE control message.
Let the proxy node check the events and create dbus objects based on
added/removed ports.
This commit is contained in:
Wim Taymans 2016-08-05 19:46:37 +02:00
parent ac5d22ec79
commit de53315f6e
17 changed files with 350 additions and 215 deletions
Download patch file Download diff file Expand all files Collapse all files

53
doc/design.txt
View file

@ -391,6 +391,59 @@ communication channel
+----+ +----+
Client Proxy
| INIT
node-update |
-------------------------------------->|
port-update |
-------------------------------------->|
state-change CONFIGURE | CONFIGURE
-------------------------------------->|
|<--- enum-ports
|<--- enum-formats
|<--- add-port
|<--- remove-port
set-property |<--- set-property
<--------------------------------------|
set-format |<--- set-format
<--------------------------------------|
|
port-update |
-------------------------------------->|
state-change READY | READY
-------------------------------------->|
|<--- port memory requirements
add-mem |<--- use-buffers
<--------------------------------------|
remove-mem |
<--------------------------------------|
add-buffer |
<--------------------------------------|
remove-buffer |
<--------------------------------------|
|
pause |<--- stop
<--------------------------------------|
state-change PAUSED | PAUSED
-------------------------------------->|
|
play |<--- start
<--------------------------------------|
state-change STREAMING | STREAMING
-------------------------------------->|
|
need-input |
<------------------------------------->|
have-output |
<------------------------------------->|
process-buffer |
<------------------------------------->|
reuse-buffer |
<------------------------------------->|
1) Update config C->S INIT 1) Update config C->S INIT
node-update node-update

96
pinos/client/stream.c
View file

@ -74,6 +74,7 @@ struct _PinosStreamPrivate
GPtrArray *possible_formats; GPtrArray *possible_formats;
SpaFormat *format; SpaFormat *format;
SpaPortInfo port_info;
PinosStreamFlags flags; PinosStreamFlags flags;
@ -512,17 +513,28 @@ pinos_stream_get_error (PinosStream *stream)
return stream->priv->error; return stream->priv->error;
} }
static void
control_builder_init (PinosStream *stream, SpaControlBuilder *builder)
{
PinosStreamPrivate *priv = stream->priv;
spa_control_builder_init_into (builder,
priv->send_data,
MAX_BUFFER_SIZE,
priv->send_fds,
MAX_FDS);
}
static void static void
send_need_input (PinosStream *stream, uint32_t port_id, uint32_t buffer_id) send_need_input (PinosStream *stream, uint32_t port_id, uint32_t buffer_id)
{ {
PinosStreamPrivate *priv = stream->priv; PinosStreamPrivate *priv = stream->priv;
SpaControlBuilder builder; SpaControlBuilder builder;
SpaControl control; SpaControl control;
guint8 buffer[64];
SpaControlCmdNeedInput ni; SpaControlCmdNeedInput ni;
SpaControlCmdReuseBuffer rb; SpaControlCmdReuseBuffer rb;
spa_control_builder_init_into (&builder, buffer, 64, NULL, 0); control_builder_init (stream, &builder);
if (buffer_id != SPA_ID_INVALID) { if (buffer_id != SPA_ID_INVALID) {
rb.port_id = port_id; rb.port_id = port_id;
rb.buffer_id = buffer_id; rb.buffer_id = buffer_id;
@ -586,7 +598,6 @@ parse_control (PinosStream *stream,
SpaControlBuilder builder; SpaControlBuilder builder;
SpaControl control; SpaControl control;
SpaControlCmdStateChange sc; SpaControlCmdStateChange sc;
guint8 buffer[64];
if (spa_control_iter_parse_cmd (&it, &p) < 0) if (spa_control_iter_parse_cmd (&it, &p) < 0)
break; break;
@ -598,10 +609,11 @@ parse_control (PinosStream *stream,
spa_debug_format (p.format); spa_debug_format (p.format);
g_object_notify (G_OBJECT (stream), "format"); g_object_notify (G_OBJECT (stream), "format");
control_builder_init (stream, &builder);
/* FIXME send update port status */ /* FIXME send update port status */
/* send state-change */ /* send state-change */
spa_control_builder_init_into (&builder, buffer, sizeof(buffer), NULL, 0);
sc.state = SPA_NODE_STATE_READY; sc.state = SPA_NODE_STATE_READY;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_STATE_CHANGE, &sc); spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_STATE_CHANGE, &sc);
spa_control_builder_end (&builder, &control); spa_control_builder_end (&builder, &control);
@ -852,6 +864,44 @@ unhandle_socket (PinosStream *stream)
} }
} }
static void
do_node_init (PinosStream *stream)
{
PinosStreamPrivate *priv = stream->priv;
SpaControlCmdNodeUpdate nu;
SpaControlCmdPortUpdate pu;
SpaControlBuilder builder;
SpaControl control;
control_builder_init (stream, &builder);
nu.change_mask = SPA_CONTROL_CMD_NODE_UPDATE_MAX_INPUTS |
SPA_CONTROL_CMD_NODE_UPDATE_MAX_OUTPUTS;
nu.max_input_ports = 1;
nu.max_output_ports = 0;
nu.props = NULL;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_NODE_UPDATE, &nu);
pu.port_id = 0;
pu.change_mask = SPA_CONTROL_CMD_PORT_UPDATE_DIRECTION |
SPA_CONTROL_CMD_PORT_UPDATE_POSSIBLE_FORMATS |
SPA_CONTROL_CMD_PORT_UPDATE_INFO;
pu.direction = priv->direction;
pu.n_possible_formats = priv->possible_formats->len;
pu.possible_formats = (const SpaFormat **)priv->possible_formats->pdata;
pu.props = NULL;
pu.info = &priv->port_info;
priv->port_info.flags = SPA_PORT_INFO_FLAG_NONE |
SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_PORT_UPDATE, &pu);
spa_control_builder_end (&builder, &control);
if (spa_control_write (&control, priv->fd) < 0)
g_warning ("stream %p: error writing control", stream);
spa_control_clear (&control);
}
static void static void
on_node_proxy (GObject *source_object, on_node_proxy (GObject *source_object,
@ -869,6 +919,8 @@ on_node_proxy (GObject *source_object,
if (priv->node == NULL) if (priv->node == NULL)
goto node_failed; goto node_failed;
do_node_init (stream);
stream_set_state (stream, PINOS_STREAM_STATE_READY, NULL); stream_set_state (stream, PINOS_STREAM_STATE_READY, NULL);
g_object_unref (stream); g_object_unref (stream);
@ -951,25 +1003,12 @@ do_connect (PinosStream *stream)
{ {
PinosStreamPrivate *priv = stream->priv; PinosStreamPrivate *priv = stream->priv;
PinosContext *context = priv->context; PinosContext *context = priv->context;
GVariantBuilder b;
GVariant *ports;
g_variant_builder_init (&b, G_VARIANT_TYPE ("a(uusa{sv}s)"));
g_variant_builder_open (&b, G_VARIANT_TYPE ("(uusa{sv}s)"));
g_variant_builder_add (&b, "u", priv->direction);
g_variant_builder_add (&b, "u", 0);
g_variant_builder_add (&b, "s", "");
g_variant_builder_add_value (&b, pinos_properties_to_variant (priv->properties));
g_variant_builder_add (&b, "s", priv->path == NULL ? "" : priv->path);
g_variant_builder_close (&b);
ports = g_variant_builder_end (&b);
g_dbus_proxy_call (context->priv->daemon, g_dbus_proxy_call (context->priv->daemon,
"CreateClientNode", "CreateClientNode",
g_variant_new ("(s@a{sv}@a(uusa{sv}s))", g_variant_new ("(s@a{sv})",
"client-node", "client-node",
pinos_properties_to_variant (priv->properties), pinos_properties_to_variant (priv->properties)),
ports),
G_DBUS_CALL_FLAGS_NONE, G_DBUS_CALL_FLAGS_NONE,
-1, -1,
NULL, /* GCancellable *cancellable */ NULL, /* GCancellable *cancellable */
@ -1025,34 +1064,17 @@ pinos_stream_connect (PinosStream *stream,
return TRUE; return TRUE;
} }
static void
control_builder_init (PinosStream *stream, SpaControlBuilder *builder)
{
PinosStreamPrivate *priv = stream->priv;
spa_control_builder_init_into (builder,
priv->send_data,
MAX_BUFFER_SIZE,
priv->send_fds,
MAX_FDS);
}
static gboolean static gboolean
do_start (PinosStream *stream) do_start (PinosStream *stream)
{ {
PinosStreamPrivate *priv = stream->priv; PinosStreamPrivate *priv = stream->priv;
SpaControlBuilder builder; SpaControlBuilder builder;
SpaControlCmdPortUpdate pu;
SpaControlCmdStateChange sc; SpaControlCmdStateChange sc;
SpaControl control; SpaControl control;
handle_socket (stream, priv->fd); handle_socket (stream, priv->fd);
control_builder_init (stream, &builder); control_builder_init (stream, &builder);
pu.port_id = 0;
pu.change_mask = 0;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_PORT_UPDATE, &pu);
sc.state = SPA_NODE_STATE_CONFIGURE; sc.state = SPA_NODE_STATE_CONFIGURE;
spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_STATE_CHANGE, &sc); spa_control_builder_add_cmd (&builder, SPA_CONTROL_CMD_STATE_CHANGE, &sc);
spa_control_builder_end (&builder, &control); spa_control_builder_end (&builder, &control);
@ -1060,6 +1082,8 @@ do_start (PinosStream *stream)
if (spa_control_write (&control, priv->fd) < 0) if (spa_control_write (&control, priv->fd) < 0)
g_warning ("stream %p: failed to write control", stream); g_warning ("stream %p: failed to write control", stream);
spa_control_clear (&control);
g_object_unref (stream); g_object_unref (stream);
return FALSE; return FALSE;

4
pinos/dbus/org.pinos.xml
View file

@ -41,15 +41,13 @@
<!-- CreateClientNode: <!-- CreateClientNode:
@name: the name of the node @name: the name of the node
@Properties: extra properties @Properties: extra properties
@ports: the port descriptions
@node: the Node1 object path @node: the Node1 object path
Create a new Node with given name and properties and ports. Create a new Node
--> -->
<method name='CreateClientNode'> <method name='CreateClientNode'>
<arg type='s' name='name' direction='in' /> <arg type='s' name='name' direction='in' />
<arg type='a{sv}' name='properties' direction='in'/> <arg type='a{sv}' name='properties' direction='in'/>
<arg type='a(uusa{sv}s)' name='ports' direction='in'/>
<arg type='o' name='node' direction='out'/> <arg type='o' name='node' direction='out'/>
<arg type='h' name='fd' direction='out'/> <arg type='h' name='fd' direction='out'/>
</method> </method>

68
pinos/modules/spa/spa-v4l2-source.c
View file

@ -38,14 +38,6 @@
#define PINOS_SPA_V4L2_SOURCE_GET_PRIVATE(obj) \ #define PINOS_SPA_V4L2_SOURCE_GET_PRIVATE(obj) \
(G_TYPE_INSTANCE_GET_PRIVATE ((obj), PINOS_TYPE_SPA_V4L2_SOURCE, PinosSpaV4l2SourcePrivate)) (G_TYPE_INSTANCE_GET_PRIVATE ((obj), PINOS_TYPE_SPA_V4L2_SOURCE, PinosSpaV4l2SourcePrivate))
typedef struct {
PinosSpaV4l2Source *source;
guint id;
gboolean have_format;
PinosPort *port;
} SourcePortData;
struct _PinosSpaV4l2SourcePrivate struct _PinosSpaV4l2SourcePrivate
{ {
SpaPollFd fds[16]; SpaPollFd fds[16];
@ -56,8 +48,6 @@ struct _PinosSpaV4l2SourcePrivate
pthread_t thread; pthread_t thread;
GBytes *format; GBytes *format;
GList *ports;
}; };
enum { enum {
@ -158,11 +148,12 @@ on_source_event (SpaNode *node, SpaEvent *event, void *user_data)
if ((res = spa_node_port_pull_output (node, 1, info)) < 0) if ((res = spa_node_port_pull_output (node, 1, info)) < 0)
g_debug ("spa-v4l2-source %p: got pull error %d, %d", this, res, info[0].status); g_debug ("spa-v4l2-source %p: got pull error %d, %d", this, res, info[0].status);
for (walk = priv->ports; walk; walk = g_list_next (walk)) { walk = pinos_node_get_ports (PINOS_NODE (this));
SourcePortData *data = walk->data; for (; walk; walk = g_list_next (walk)) {
PinosPort *port = walk->data;
GError *error = NULL; GError *error = NULL;
if (!pinos_port_send_buffer (data->port, info[0].buffer_id, &error)) { if (!pinos_port_send_buffer (port, info[0].buffer_id, &error)) {
g_debug ("send failed: %s", error->message); g_debug ("send failed: %s", error->message);
g_clear_error (&error); g_clear_error (&error);
} }
@ -309,8 +300,7 @@ set_property (GObject *object,
static gboolean static gboolean
on_activate (PinosPort *port, gpointer user_data) on_activate (PinosPort *port, gpointer user_data)
{ {
SourcePortData *data = user_data; PinosSpaV4l2Source *source = user_data;
PinosSpaV4l2Source *source = data->source;
pinos_node_report_busy (PINOS_NODE (source)); pinos_node_report_busy (PINOS_NODE (source));
@ -320,39 +310,17 @@ on_activate (PinosPort *port, gpointer user_data)
static void static void
on_deactivate (PinosPort *port, gpointer user_data) on_deactivate (PinosPort *port, gpointer user_data)
{ {
SourcePortData *data = user_data; PinosSpaV4l2Source *source = user_data;
PinosSpaV4l2Source *source = data->source;
pinos_node_report_idle (PINOS_NODE (source)); pinos_node_report_idle (PINOS_NODE (source));
} }
static void
free_source_port_data (SourcePortData *data)
{
g_slice_free (SourcePortData, data);
}
static gboolean static gboolean
remove_port (PinosNode *node, remove_port (PinosNode *node,
guint id) guint id)
{ {
PinosSpaV4l2Source *source = PINOS_SPA_V4L2_SOURCE (node); return PINOS_NODE_CLASS (pinos_spa_v4l2_source_parent_class)
PinosSpaV4l2SourcePrivate *priv = source->priv; ->remove_port (node, id);
GList *walk;
for (walk = priv->ports; walk; walk = g_list_next (walk)) {
SourcePortData *data = walk->data;
if (data->id == id) {
free_source_port_data (data);
priv->ports = g_list_delete_link (priv->ports, walk);
break;
}
}
if (priv->ports == NULL)
pinos_node_report_idle (node);
return TRUE;
} }
static void static void
@ -417,26 +385,18 @@ add_port (PinosNode *node,
GError **error) GError **error)
{ {
PinosSpaV4l2Source *source = PINOS_SPA_V4L2_SOURCE (node); PinosSpaV4l2Source *source = PINOS_SPA_V4L2_SOURCE (node);
PinosSpaV4l2SourcePrivate *priv = source->priv; PinosPort *port;
SourcePortData *data;
data = g_slice_new0 (SourcePortData); port = PINOS_NODE_CLASS (pinos_spa_v4l2_source_parent_class)
data->source = source;
data->id = id;
data->have_format = FALSE;
data->port = PINOS_NODE_CLASS (pinos_spa_v4l2_source_parent_class)
->add_port (node, direction, id, error); ->add_port (node, direction, id, error);
pinos_port_set_received_cb (data->port, on_received_buffer, on_received_event, node, NULL); pinos_port_set_received_cb (port, on_received_buffer, on_received_event, node, NULL);
g_debug ("connecting signals"); g_debug ("connecting signals");
g_signal_connect (data->port, "activate", (GCallback) on_activate, data); g_signal_connect (port, "activate", (GCallback) on_activate, source);
g_signal_connect (data->port, "deactivate", (GCallback) on_deactivate, data); g_signal_connect (port, "deactivate", (GCallback) on_deactivate, source);
priv->ports = g_list_append (priv->ports, data); return port;
return data->port;
} }
static void static void

44
pinos/server/client-node.c
View file

@ -38,6 +38,7 @@
#include "pinos/server/daemon.h" #include "pinos/server/daemon.h"
#include "pinos/server/client-node.h" #include "pinos/server/client-node.h"
#include "pinos/server/utils.h" #include "pinos/server/utils.h"
#include "pinos/server/link.h"
#include "pinos/dbus/org-pinos.h" #include "pinos/dbus/org-pinos.h"
@ -174,8 +175,6 @@ static gboolean
on_received_buffer (PinosPort *port, uint32_t buffer_id, GError **error, gpointer user_data) on_received_buffer (PinosPort *port, uint32_t buffer_id, GError **error, gpointer user_data)
{ {
PinosNode *node = user_data; PinosNode *node = user_data;
PinosClientNode *this = PINOS_CLIENT_NODE (node);
PinosClientNodePrivate *priv = this->priv;
SpaResult res; SpaResult res;
SpaInputInfo info[1]; SpaInputInfo info[1];
@ -195,8 +194,6 @@ static gboolean
on_received_event (PinosPort *port, SpaEvent *event, GError **error, gpointer user_data) on_received_event (PinosPort *port, SpaEvent *event, GError **error, gpointer user_data)
{ {
PinosNode *node = user_data; PinosNode *node = user_data;
PinosClientNode *this = PINOS_CLIENT_NODE (node);
PinosClientNodePrivate *priv = this->priv;
SpaResult res; SpaResult res;
if ((res = spa_node_port_push_event (node->node, port->id, event)) < 0) if ((res = spa_node_port_push_event (node->node, port->id, event)) < 0)
@ -271,6 +268,45 @@ on_node_event (SpaNode *node, SpaEvent *event, void *user_data)
PinosClientNodePrivate *priv = this->priv; PinosClientNodePrivate *priv = this->priv;
switch (event->type) { switch (event->type) {
case SPA_EVENT_TYPE_PORT_ADDED:
{
SpaEventPortAdded *pa = event->data;
PinosPort *port, *target;
PinosLink *link;
PinosNode *pnode = PINOS_NODE (this);
GError *error = NULL;
port = PINOS_NODE_CLASS (pinos_client_node_parent_class)->add_port (pnode,
pa->direction,
event->port_id,
&error);
if (port == NULL) {
g_warning ("proxy %p: can't create port: %s", this, error->message);
g_clear_error (&error);
break;
}
pinos_port_set_received_cb (port, on_received_buffer, on_received_event, this, NULL);
target = pinos_daemon_find_port (pinos_node_get_daemon (pnode),
pinos_direction_reverse (pa->direction),
"/org/pinos/node_8",
NULL,
NULL,
&error);
if (target == NULL) {
g_warning ("proxy %p: can't find port target: %s", this, error->message);
g_clear_error (&error);
break;
}
link = pinos_link_new (pinos_node_get_daemon (pnode), port, target, NULL);
break;
}
case SPA_EVENT_TYPE_PORT_REMOVED:
{
break;
}
case SPA_EVENT_TYPE_STATE_CHANGE: case SPA_EVENT_TYPE_STATE_CHANGE:
{ {
SpaEventStateChange *sc = event->data; SpaEventStateChange *sc = event->data;

53
pinos/server/daemon.c
View file

@ -187,7 +187,6 @@ handle_create_client_node (PinosDaemon1 *interface,
GDBusMethodInvocation *invocation, GDBusMethodInvocation *invocation,
const gchar *arg_name, const gchar *arg_name,
GVariant *arg_properties, GVariant *arg_properties,
GVariant *arg_ports,
gpointer user_data) gpointer user_data)
{ {
PinosDaemon *daemon = user_data; PinosDaemon *daemon = user_data;
@ -210,56 +209,11 @@ handle_create_client_node (PinosDaemon1 *interface,
arg_name, arg_name,
props); props);
client = sender_get_client (daemon, sender);
{
GVariantIter it;
guint direction;
guint id;
const gchar *format_str, *target_node;
GVariant *port_props;
g_variant_iter_init (&it, arg_ports);
while (g_variant_iter_loop (&it, "(uu&s@a{sv}&s)", &direction, &id, &format_str, &port_props, &target_node)) {
PinosPort *port, *target;
PinosProperties *pprops;
GBytes *formats;
PinosLink *link;
port = pinos_node_add_port (node, direction, id, &error);
if (port == NULL)
goto add_failed;
pprops = pinos_properties_from_variant (port_props);
formats = g_bytes_new (format_str, strlen (format_str) + 1);
g_object_set (port, "possible-formats", formats,
"properties", port_props,
NULL);
target = pinos_daemon_find_port (daemon,
pinos_direction_reverse (direction),
target_node,
pprops,
formats,
&error);
if (target == NULL) {
g_warning ("daemon %p: can't find port target: %s", daemon, error->message);
g_clear_error (&error);
continue;
}
// pinos_client_add_object (client, G_OBJECT (target));
link = pinos_link_new (daemon, port, target, NULL);
pinos_client_add_object (client, G_OBJECT (link));
}
}
socket = pinos_client_node_get_socket_pair (PINOS_CLIENT_NODE (node), &error); socket = pinos_client_node_get_socket_pair (PINOS_CLIENT_NODE (node), &error);
if (socket == NULL) if (socket == NULL)
goto no_socket; goto no_socket;
client = sender_get_client (daemon, sender);
pinos_client_add_object (client, G_OBJECT (node)); pinos_client_add_object (client, G_OBJECT (node));
object_path = pinos_node_get_object_path (PINOS_NODE (node)); object_path = pinos_node_get_object_path (PINOS_NODE (node));
@ -279,11 +233,6 @@ no_socket:
g_debug ("daemon %p: could not create node %s", daemon, error->message); g_debug ("daemon %p: could not create node %s", daemon, error->message);
goto exit_error; goto exit_error;
} }
add_failed:
{
g_debug ("daemon %p: failed to add port %s", daemon, error->message);
goto exit_error;
}
exit_error: exit_error:
{ {
g_dbus_method_invocation_return_gerror (invocation, error); g_dbus_method_invocation_return_gerror (invocation, error);

31
pinos/server/node.c
View file

@ -81,6 +81,12 @@ node_set_state (PinosNode *node,
return FALSE; return FALSE;
} }
static void
do_remove_port (PinosPort *port, PinosNode *node)
{
pinos_node_remove_port (node, port->id);
}
static PinosPort * static PinosPort *
node_add_port (PinosNode *node, node_add_port (PinosNode *node,
PinosDirection direction, PinosDirection direction,
@ -96,6 +102,10 @@ node_add_port (PinosNode *node,
"direction", direction, "direction", direction,
"id", id, "id", id,
NULL); NULL);
if (port) {
g_hash_table_insert (priv->ports, GUINT_TO_POINTER (id), port);
g_signal_connect (port, "remove", (GCallback) do_remove_port, node);
}
return port; return port;
} }
@ -103,6 +113,9 @@ static gboolean
node_remove_port (PinosNode *node, node_remove_port (PinosNode *node,
guint id) guint id)
{ {
PinosNodePrivate *priv = node->priv;
g_debug ("node %p: removed port %u", node, id);
g_hash_table_remove (priv->ports, GUINT_TO_POINTER (id));
return TRUE; return TRUE;
} }
@ -670,12 +683,6 @@ pinos_node_remove (PinosNode *node)
g_signal_emit (node, signals[SIGNAL_REMOVE], 0, NULL); g_signal_emit (node, signals[SIGNAL_REMOVE], 0, NULL);
} }
static void
do_remove_port (PinosPort *port, PinosNode *node)
{
pinos_node_remove_port (node, port->id);
}
/** /**
* pinos_node_add_port: * pinos_node_add_port:
* @node: a #PinosNode * @node: a #PinosNode
@ -693,11 +700,9 @@ pinos_node_add_port (PinosNode *node,
GError **error) GError **error)
{ {
PinosNodeClass *klass; PinosNodeClass *klass;
PinosNodePrivate *priv;
PinosPort *port; PinosPort *port;
g_return_val_if_fail (PINOS_IS_NODE (node), NULL); g_return_val_if_fail (PINOS_IS_NODE (node), NULL);
priv = node->priv;
klass = PINOS_NODE_GET_CLASS (node); klass = PINOS_NODE_GET_CLASS (node);
if (!klass->add_port) { if (!klass->add_port) {
@ -708,10 +713,6 @@ pinos_node_add_port (PinosNode *node,
g_debug ("node %p: add port", node); g_debug ("node %p: add port", node);
port = klass->add_port (node, direction, id, error); port = klass->add_port (node, direction, id, error);
if (port) {
g_hash_table_insert (priv->ports, GUINT_TO_POINTER (id), port);
g_signal_connect (port, "remove", (GCallback) do_remove_port, node);
}
return port; return port;
} }
@ -727,12 +728,10 @@ pinos_node_add_port (PinosNode *node,
gboolean gboolean
pinos_node_remove_port (PinosNode *node, guint id) pinos_node_remove_port (PinosNode *node, guint id)
{ {
PinosNodePrivate *priv;
PinosNodeClass *klass; PinosNodeClass *klass;
gboolean res = FALSE; gboolean res = FALSE;
g_return_val_if_fail (PINOS_IS_NODE (node), FALSE); g_return_val_if_fail (PINOS_IS_NODE (node), FALSE);
priv = node->priv;
klass = PINOS_NODE_GET_CLASS (node); klass = PINOS_NODE_GET_CLASS (node);
@ -740,10 +739,6 @@ pinos_node_remove_port (PinosNode *node, guint id)
return FALSE; return FALSE;
res = klass->remove_port (node, id); res = klass->remove_port (node, id);
if (res) {
g_debug ("node %p: removed port %u", node, id);
g_hash_table_remove (priv->ports, GUINT_TO_POINTER (id));
}
return res; return res;
} }

9
spa/include/spa/control.h
View file

@ -96,6 +96,10 @@ typedef struct {
const SpaProps *props; const SpaProps *props;
} SpaControlCmdNodeUpdate; } SpaControlCmdNodeUpdate;
#define SPA_CONTROL_CMD_NODE_UPDATE_MAX_INPUTS (1 << 0)
#define SPA_CONTROL_CMD_NODE_UPDATE_MAX_OUTPUTS (1 << 1)
#define SPA_CONTROL_CMD_NODE_UPDATE_PROPS (1 << 2)
/* SPA_CONTROL_CMD_PORT_UPDATE */ /* SPA_CONTROL_CMD_PORT_UPDATE */
typedef struct { typedef struct {
uint32_t port_id; uint32_t port_id;
@ -107,6 +111,11 @@ typedef struct {
const SpaPortInfo *info; const SpaPortInfo *info;
} SpaControlCmdPortUpdate; } SpaControlCmdPortUpdate;
#define SPA_CONTROL_CMD_PORT_UPDATE_DIRECTION (1 << 0)
#define SPA_CONTROL_CMD_PORT_UPDATE_POSSIBLE_FORMATS (1 << 1)
#define SPA_CONTROL_CMD_PORT_UPDATE_PROPS (1 << 2)
#define SPA_CONTROL_CMD_PORT_UPDATE_INFO (1 << 3)
/* SPA_CONTROL_CMD_PORT_REMOVED */ /* SPA_CONTROL_CMD_PORT_REMOVED */
typedef struct { typedef struct {
uint32_t port_id; uint32_t port_id;

6
spa/include/spa/event.h
View file

@ -49,6 +49,8 @@ typedef struct _SpaEvent SpaEvent;
*/ */
typedef enum { typedef enum {
SPA_EVENT_TYPE_INVALID = 0, SPA_EVENT_TYPE_INVALID = 0,
SPA_EVENT_TYPE_PORT_ADDED,
SPA_EVENT_TYPE_PORT_REMOVED,
SPA_EVENT_TYPE_STATE_CHANGE, SPA_EVENT_TYPE_STATE_CHANGE,
SPA_EVENT_TYPE_CAN_PULL_OUTPUT, SPA_EVENT_TYPE_CAN_PULL_OUTPUT,
SPA_EVENT_TYPE_CAN_PUSH_INPUT, SPA_EVENT_TYPE_CAN_PUSH_INPUT,
@ -70,6 +72,10 @@ struct _SpaEvent {
size_t size; size_t size;
}; };
typedef struct {
SpaDirection direction;
} SpaEventPortAdded;
typedef struct { typedef struct {
uint32_t buffer_id; uint32_t buffer_id;
off_t offset; off_t offset;

14
spa/include/spa/port.h
View file

@ -74,18 +74,18 @@ typedef struct {
* @SPA_PORT_INFO_FLAG_NONE: no flags * @SPA_PORT_INFO_FLAG_NONE: no flags
* @SPA_PORT_INFO_FLAG_REMOVABLE: port can be removed * @SPA_PORT_INFO_FLAG_REMOVABLE: port can be removed
* @SPA_PORT_INFO_FLAG_OPTIONAL: processing on port is optional * @SPA_PORT_INFO_FLAG_OPTIONAL: processing on port is optional
* @SPA_PORT_INFO_FLAG_CAN_GIVE_BUFFER: the port can give a buffer * @SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS: the port can give a buffer
* @SPA_PORT_INFO_FLAG_CAN_USE_BUFFER: the port can use a provided buffer * @SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS: the port can use a provided buffer
* @SPA_PORT_INFO_FLAG_IN_PLACE: the port can process data in-place and will need * @SPA_PORT_INFO_FLAG_IN_PLACE: the port can process data in-place and will need
* a writable input buffer when no output buffer is specified. * a writable input buffer
* @SPA_PORT_INFO_FLAG_NO_REF: the port does not keep a ref on the buffer * @SPA_PORT_INFO_FLAG_NO_REF: the port does not keep a ref on the buffer
*/ */
typedef enum { typedef enum {
SPA_PORT_INFO_FLAG_NONE = 0, SPA_PORT_INFO_FLAG_NONE = 0,
SPA_PORT_INFO_FLAG_REMOVABLE = 1 << 0, SPA_PORT_INFO_FLAG_REMOVABLE = 1 << 0,
SPA_PORT_INFO_FLAG_OPTIONAL = 1 << 1, SPA_PORT_INFO_FLAG_OPTIONAL = 1 << 1,
SPA_PORT_INFO_FLAG_CAN_GIVE_BUFFER = 1 << 2, SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS = 1 << 2,
SPA_PORT_INFO_FLAG_CAN_USE_BUFFER = 1 << 3, SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS = 1 << 3,
SPA_PORT_INFO_FLAG_IN_PLACE = 1 << 4, SPA_PORT_INFO_FLAG_IN_PLACE = 1 << 4,
SPA_PORT_INFO_FLAG_NO_REF = 1 << 5, SPA_PORT_INFO_FLAG_NO_REF = 1 << 5,
} SpaPortInfoFlags; } SpaPortInfoFlags;
@ -103,10 +103,10 @@ typedef enum {
*/ */
typedef struct { typedef struct {
SpaPortInfoFlags flags; SpaPortInfoFlags flags;
unsigned int maxbuffering; uint64_t maxbuffering;
uint64_t latency; uint64_t latency;
SpaAllocParam **params; SpaAllocParam **params;
uint32_t n_params; unsigned int n_params;
const char **features; const char **features;
} SpaPortInfo; } SpaPortInfo;

80
spa/lib/control.c
View file

@ -398,10 +398,15 @@ spa_control_iter_parse_cmd (SpaControlIter *iter,
switch (si->cmd) { switch (si->cmd) {
/* C -> S */ /* C -> S */
case SPA_CONTROL_CMD_NODE_UPDATE: case SPA_CONTROL_CMD_NODE_UPDATE:
case SPA_CONTROL_CMD_PORT_UPDATE:
fprintf (stderr, "implement iter of %d\n", si->cmd); fprintf (stderr, "implement iter of %d\n", si->cmd);
break; break;
case SPA_CONTROL_CMD_PORT_UPDATE:
if (si->size < sizeof (SpaControlCmdPortUpdate))
return SPA_RESULT_ERROR;
memcpy (command, si->data, sizeof (SpaControlCmdPortUpdate));
break;
case SPA_CONTROL_CMD_PORT_REMOVED: case SPA_CONTROL_CMD_PORT_REMOVED:
if (si->size < sizeof (SpaControlCmdPortRemoved)) if (si->size < sizeof (SpaControlCmdPortRemoved))
return SPA_RESULT_ERROR; return SPA_RESULT_ERROR;
@ -722,6 +727,55 @@ builder_add_cmd (struct stack_builder *sb, SpaControlCmd cmd, size_t size)
return p; return p;
} }
static size_t
calc_props_len (const SpaProps *props)
{
size_t len;
unsigned int i, j;
SpaPropInfo *pi;
SpaPropRangeInfo *ri;
/* props and unset mask */
len = sizeof (SpaProps) + sizeof (uint32_t);
for (i = 0; i < props->n_prop_info; i++) {
pi = (SpaPropInfo *) &props->prop_info[i];
len += sizeof (SpaPropInfo);
len += pi->name ? strlen (pi->name) + 1 : 0;
len += pi->description ? strlen (pi->description) + 1 : 0;
/* for the value and the default value */
len += pi->maxsize + pi->default_size;
for (j = 0; j < pi->n_range_values; j++) {
ri = (SpaPropRangeInfo *)&pi->range_values[j];
len += sizeof (SpaPropRangeInfo);
len += ri->name ? strlen (ri->name) + 1 : 0;
len += ri->description ? strlen (ri->description) + 1 : 0;
/* the size of the range value */
len += ri->size;
}
}
return len;
}
static size_t
calc_format_len (const SpaFormat *format)
{
return calc_props_len (&format->props) - sizeof (SpaProps) + sizeof (SpaFormat);
}
static void
builder_add_port_update (struct stack_builder *sb, SpaControlCmdPortUpdate *pu)
{
size_t len;
void *base;
/* calc len */
len = sizeof (SpaControlCmdPortUpdate);
base = builder_add_cmd (sb, SPA_CONTROL_CMD_PORT_UPDATE, len);
memcpy (base, pu, sizeof (SpaControlCmdPortUpdate));
/* FIXME add more things */
}
static void static void
builder_add_set_format (struct stack_builder *sb, SpaControlCmdSetFormat *sf) builder_add_set_format (struct stack_builder *sb, SpaControlCmdSetFormat *sf)
{ {
@ -738,24 +792,7 @@ builder_add_set_format (struct stack_builder *sb, SpaControlCmdSetFormat *sf)
/* calculate length */ /* calculate length */
/* port_id + format + mask */ /* port_id + format + mask */
len = sizeof (uint32_t) + sizeof (SpaFormat) + sizeof (uint32_t); len = sizeof (uint32_t) + calc_format_len (sf->format);
for (i = 0; i < sp->n_prop_info; i++) {
pi = (SpaPropInfo *) &sp->prop_info[i];
len += sizeof (SpaPropInfo);
len += pi->name ? strlen (pi->name) + 1 : 0;
len += pi->description ? strlen (pi->description) + 1 : 0;
/* for the value and the default value */
len += pi->maxsize + pi->default_size;
for (j = 0; j < pi->n_range_values; j++) {
ri = (SpaPropRangeInfo *)&pi->range_values[j];
len += sizeof (SpaPropRangeInfo);
len += ri->name ? strlen (ri->name) + 1 : 0;
len += ri->description ? strlen (ri->description) + 1 : 0;
/* the size of the range value */
len += ri->size;
}
}
base = builder_add_cmd (sb, SPA_CONTROL_CMD_SET_FORMAT, len); base = builder_add_cmd (sb, SPA_CONTROL_CMD_SET_FORMAT, len);
memcpy (base, &sf->port_id, sizeof (uint32_t)); memcpy (base, &sf->port_id, sizeof (uint32_t));
@ -881,10 +918,13 @@ spa_control_builder_add_cmd (SpaControlBuilder *builder,
switch (cmd) { switch (cmd) {
/* C -> S */ /* C -> S */
case SPA_CONTROL_CMD_NODE_UPDATE: case SPA_CONTROL_CMD_NODE_UPDATE:
case SPA_CONTROL_CMD_PORT_UPDATE:
fprintf (stderr, "implement builder of %d\n", cmd); fprintf (stderr, "implement builder of %d\n", cmd);
break; break;
case SPA_CONTROL_CMD_PORT_UPDATE:
builder_add_port_update (sb, command);
break;
case SPA_CONTROL_CMD_PORT_REMOVED: case SPA_CONTROL_CMD_PORT_REMOVED:
p = builder_add_cmd (sb, cmd, sizeof (SpaControlCmdPortRemoved)); p = builder_add_cmd (sb, cmd, sizeof (SpaControlCmdPortRemoved));
memcpy (p, command, sizeof (SpaControlCmdPortRemoved)); memcpy (p, command, sizeof (SpaControlCmdPortRemoved));

2
spa/lib/debug.c
View file

@ -34,7 +34,7 @@ spa_debug_port_info (const SpaPortInfo *info)
fprintf (stderr, "SpaPortInfo %p:\n", info); fprintf (stderr, "SpaPortInfo %p:\n", info);
fprintf (stderr, " flags: \t%08x\n", info->flags); fprintf (stderr, " flags: \t%08x\n", info->flags);
fprintf (stderr, " maxbuffering: \t%u\n", info->maxbuffering); fprintf (stderr, " maxbuffering: \t%"PRIu64"\n", info->maxbuffering);
fprintf (stderr, " latency: \t%" PRIu64 "\n", info->latency); fprintf (stderr, " latency: \t%" PRIu64 "\n", info->latency);
fprintf (stderr, " n_params: \t%d\n", info->n_params); fprintf (stderr, " n_params: \t%d\n", info->n_params);
for (i = 0; i < info->n_params; i++) { for (i = 0; i < info->n_params; i++) {

6
spa/plugins/audiomixer/audiomixer.c
View file

@ -275,7 +275,7 @@ spa_audiomixer_node_add_port (SpaNode *node,
this->ports[port_id].valid = true; this->ports[port_id].valid = true;
this->port_count++; this->port_count++;
this->ports[port_id].info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFER | this->ports[port_id].info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS |
SPA_PORT_INFO_FLAG_REMOVABLE | SPA_PORT_INFO_FLAG_REMOVABLE |
SPA_PORT_INFO_FLAG_OPTIONAL | SPA_PORT_INFO_FLAG_OPTIONAL |
SPA_PORT_INFO_FLAG_IN_PLACE; SPA_PORT_INFO_FLAG_IN_PLACE;
@ -798,8 +798,8 @@ spa_audiomixer_init (const SpaHandleFactory *factory,
reset_audiomixer_props (&this->props[1]); reset_audiomixer_props (&this->props[1]);
this->ports[0].valid = true; this->ports[0].valid = true;
this->ports[0].info.flags = SPA_PORT_INFO_FLAG_CAN_GIVE_BUFFER | this->ports[0].info.flags = SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS |
SPA_PORT_INFO_FLAG_CAN_USE_BUFFER | SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS |
SPA_PORT_INFO_FLAG_NO_REF; SPA_PORT_INFO_FLAG_NO_REF;
return SPA_RESULT_OK; return SPA_RESULT_OK;
} }

2
spa/plugins/audiotestsrc/audiotestsrc.c
View file

@ -585,7 +585,7 @@ audiotestsrc_init (const SpaHandleFactory *factory,
this->props[1].props.prop_info = prop_info; this->props[1].props.prop_info = prop_info;
reset_audiotestsrc_props (&this->props[1]); reset_audiotestsrc_props (&this->props[1]);
this->info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFER | this->info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS |
SPA_PORT_INFO_FLAG_NO_REF; SPA_PORT_INFO_FLAG_NO_REF;
this->status.flags = SPA_PORT_STATUS_FLAG_HAVE_OUTPUT; this->status.flags = SPA_PORT_STATUS_FLAG_HAVE_OUTPUT;

88
spa/plugins/remote/proxy.c
View file

@ -57,7 +57,6 @@ typedef struct {
SpaFormat formats[2]; SpaFormat formats[2];
unsigned int n_buffers; unsigned int n_buffers;
SpaBuffer **buffers; SpaBuffer **buffers;
SpaBuffer *remote_buffers[128];
} SpaProxyPort; } SpaProxyPort;
struct _SpaProxy { struct _SpaProxy {
@ -334,6 +333,59 @@ spa_proxy_node_get_port_ids (SpaNode *node,
return SPA_RESULT_OK; return SPA_RESULT_OK;
} }
static void
do_init_port (SpaProxy *this,
uint32_t port_id,
SpaDirection direction)
{
SpaEvent event;
SpaProxyPort *port;
SpaEventPortAdded pa;
fprintf (stderr, "%p: adding port %d, %d\n", this, port_id, direction);
port = &this->ports[port_id];
port->direction = direction;
port->valid = true;
port->have_format = false;
if (direction == SPA_DIRECTION_INPUT)
this->n_inputs++;
else
this->n_outputs++;
event.type = SPA_EVENT_TYPE_PORT_ADDED;
event.port_id = port_id;
event.data = &pa;
event.size = sizeof (pa);
pa.direction = direction;
this->event_cb (&this->node, &event, this->user_data);
}
static void
do_uninit_port (SpaProxy *this,
uint32_t port_id)
{
SpaEvent event;
SpaProxyPort *port;
fprintf (stderr, "%p: removing port %d\n", this, port_id);
port = &this->ports[port_id];
if (port->direction == SPA_DIRECTION_INPUT)
this->n_inputs--;
else
this->n_outputs--;
port->direction = SPA_DIRECTION_INVALID;
port->valid = false;
port->have_format = false;
event.type = SPA_EVENT_TYPE_PORT_REMOVED;
event.port_id = port_id;
event.data = NULL;
event.size = 0;
this->event_cb (&this->node, &event, this->user_data);
}
static SpaResult static SpaResult
spa_proxy_node_add_port (SpaNode *node, spa_proxy_node_add_port (SpaNode *node,
@ -341,7 +393,6 @@ spa_proxy_node_add_port (SpaNode *node,
uint32_t port_id) uint32_t port_id)
{ {
SpaProxy *this; SpaProxy *this;
SpaProxyPort *port;
if (node == NULL || node->handle == NULL) if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS; return SPA_RESULT_INVALID_ARGUMENTS;
@ -351,12 +402,7 @@ spa_proxy_node_add_port (SpaNode *node,
if (!CHECK_FREE_PORT_ID (this, port_id)) if (!CHECK_FREE_PORT_ID (this, port_id))
return SPA_RESULT_INVALID_PORT; return SPA_RESULT_INVALID_PORT;
fprintf (stderr, "%p: adding port %d, %d\n", node, port_id, direction); do_init_port (this, port_id, direction);
port = &this->ports[port_id];
port->direction = direction;
port->valid = true;
port->have_format = false;
return SPA_RESULT_OK; return SPA_RESULT_OK;
} }
@ -366,7 +412,6 @@ spa_proxy_node_remove_port (SpaNode *node,
uint32_t port_id) uint32_t port_id)
{ {
SpaProxy *this; SpaProxy *this;
SpaProxyPort *port;
if (node == NULL || node->handle == NULL) if (node == NULL || node->handle == NULL)
return SPA_RESULT_INVALID_ARGUMENTS; return SPA_RESULT_INVALID_ARGUMENTS;
@ -376,8 +421,7 @@ spa_proxy_node_remove_port (SpaNode *node,
if (!CHECK_PORT_ID (this, port_id)) if (!CHECK_PORT_ID (this, port_id))
return SPA_RESULT_INVALID_PORT; return SPA_RESULT_INVALID_PORT;
port = &this->ports[port_id]; do_uninit_port (this, port_id);
port->valid = false;
return SPA_RESULT_OK; return SPA_RESULT_OK;
} }
@ -868,11 +912,31 @@ parse_control (SpaProxy *this,
break; break;
case SPA_CONTROL_CMD_NODE_UPDATE: case SPA_CONTROL_CMD_NODE_UPDATE:
case SPA_CONTROL_CMD_PORT_UPDATE:
case SPA_CONTROL_CMD_PORT_REMOVED: case SPA_CONTROL_CMD_PORT_REMOVED:
fprintf (stderr, "proxy %p: command not implemented %d\n", this, cmd); fprintf (stderr, "proxy %p: command not implemented %d\n", this, cmd);
break; break;
case SPA_CONTROL_CMD_PORT_UPDATE:
{
SpaControlCmdPortUpdate pu;
SpaProxyPort *port;
fprintf (stderr, "proxy %p: got port update %d\n", this, cmd);
if (spa_control_iter_parse_cmd (&it, &pu) < 0)
break;
if (pu.port_id >= MAX_PORTS)
break;
port = &this->ports[pu.port_id];
if (!port->valid && pu.direction != SPA_DIRECTION_INVALID) {
do_init_port (this, pu.port_id, pu.direction);
} else {
do_uninit_port (this, pu.port_id);
}
break;
}
case SPA_CONTROL_CMD_STATE_CHANGE: case SPA_CONTROL_CMD_STATE_CHANGE:
{ {
SpaControlCmdStateChange sc; SpaControlCmdStateChange sc;

3
spa/plugins/v4l2/v4l2-utils.c
View file

@ -398,7 +398,8 @@ spa_v4l2_set_format (SpaV4l2Source *this, V4l2Format *f, bool try_only)
return 0; return 0;
state->fmt = fmt; state->fmt = fmt;
state->info.flags = SPA_PORT_INFO_FLAG_CAN_GIVE_BUFFER; state->info.flags = SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS |
SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
state->info.maxbuffering = -1; state->info.maxbuffering = -1;
state->info.latency = -1; state->info.latency = -1;

6
spa/plugins/volume/volume.c
View file

@ -695,10 +695,10 @@ volume_instantiate (const SpaHandleFactory *factory,
this->props[1].props.prop_info = prop_info; this->props[1].props.prop_info = prop_info;
reset_volume_props (&this->props[1]); reset_volume_props (&this->props[1]);
this->ports[0].info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFER | this->ports[0].info.flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS |
SPA_PORT_INFO_FLAG_IN_PLACE; SPA_PORT_INFO_FLAG_IN_PLACE;
this->ports[1].info.flags = SPA_PORT_INFO_FLAG_CAN_GIVE_BUFFER | this->ports[1].info.flags = SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS |
SPA_PORT_INFO_FLAG_CAN_USE_BUFFER | SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS |
SPA_PORT_INFO_FLAG_NO_REF; SPA_PORT_INFO_FLAG_NO_REF;
this->ports[0].status.flags = SPA_PORT_STATUS_FLAG_NEED_INPUT; this->ports[0].status.flags = SPA_PORT_STATUS_FLAG_NEED_INPUT;