mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-10 13:30:05 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

Make structure private

Browse source 
Make structs private. Expose methods for things we need.
Signals only work on exposed structures so make a new callback helper to
signal events.
This commit is contained in:
Wim Taymans 2017-08-04 10:18:54 +02:00
parent e24c934a1b
commit b898eb46cd
72 changed files with 2980 additions and 2120 deletions
Download patch file Download diff file Expand all files Collapse all files

168
src/gst/gstpipewiredeviceprovider.c
View file

@ -201,13 +201,14 @@ new_node (GstPipeWireDeviceProvider *self, const struct pw_node_info *info, uint
struct spa_dict_item *item;
GstPipeWireDeviceType type;
int i;
struct pw_type *t = self->type;
caps = gst_caps_new_empty ();
if (info->max_input_ports > 0 && info->max_output_ports == 0) {
type = GST_PIPEWIRE_DEVICE_TYPE_SINK;
for (i = 0; i < info->n_input_formats; i++) {
GstCaps *c1 = gst_caps_from_format (info->input_formats[i], self->core->type.map);
GstCaps *c1 = gst_caps_from_format (info->input_formats[i], t->map);
if (c1)
gst_caps_append (caps, c1);
}
@ -215,7 +216,7 @@ new_node (GstPipeWireDeviceProvider *self, const struct pw_node_info *info, uint
else if (info->max_output_ports > 0 && info->max_input_ports == 0) {
type = GST_PIPEWIRE_DEVICE_TYPE_SOURCE;
for (i = 0; i < info->n_output_formats; i++) {
GstCaps *c1 = gst_caps_from_format (info->output_formats[i], self->core->type.map);
GstCaps *c1 = gst_caps_from_format (info->output_formats[i], t->map);
if (c1)
gst_caps_append (caps, c1);
}
@ -267,7 +268,7 @@ get_core_info (struct pw_remote *remote,
void *user_data)
{
GstDeviceProvider *provider = user_data;
struct pw_core_info *info = remote->info;
const struct pw_core_info *info = pw_remote_get_core_info(remote);
const gchar *value;
if (info == NULL || info->props == NULL)
@ -291,25 +292,53 @@ get_core_info (struct pw_remote *remote,
}
static void
on_sync_reply (struct pw_listener *listener, struct pw_remote *remote, uint32_t seq)
on_sync_reply (void *data, uint32_t seq)
{
GstPipeWireDeviceProvider *self = SPA_CONTAINER_OF (listener, GstPipeWireDeviceProvider, on_sync_reply);
GstPipeWireDeviceProvider *self = data;
if (seq == 1)
pw_core_proxy_sync(self->remote->core_proxy, 2);
pw_core_proxy_sync(self->core_proxy, 2);
else if (seq == 2)
self->end = true;
}
static void
on_state_changed (void *data, enum pw_remote_state old, enum pw_remote_state state, const char *error)
{
GstPipeWireDeviceProvider *self = data;
GST_DEBUG ("got remote state %d", state);
switch (state) {
case PW_REMOTE_STATE_CONNECTING:
break;
case PW_REMOTE_STATE_UNCONNECTED:
case PW_REMOTE_STATE_CONNECTED:
break;
case PW_REMOTE_STATE_ERROR:
GST_ERROR_OBJECT (self, "remote error: %s", error);
break;
}
pw_thread_loop_signal (self->main_loop, FALSE);
}
struct node_data {
GstPipeWireDeviceProvider *self;
struct pw_node_proxy *node;
uint32_t id;
uint32_t parent_id;
struct pw_callback_info node_listener;
};
static void node_event_info(void *object, struct pw_node_info *info)
struct registry_data {
GstPipeWireDeviceProvider *self;
struct pw_registry_proxy *registry;
struct pw_callback_info registry_listener;
};
static void node_event_info(void *data, struct pw_node_info *info)
{
struct pw_proxy *proxy = object;
struct node_data *node_data = proxy->user_data;
struct node_data *node_data = data;
GstPipeWireDeviceProvider *self = node_data->self;
GstDevice *dev;
@ -324,33 +353,32 @@ static void node_event_info(void *object, struct pw_node_info *info)
static const struct pw_node_events node_events = {
PW_VERSION_NODE_EVENTS,
&node_event_info
.info = node_event_info
};
static void registry_event_global(void *object, uint32_t id, uint32_t parent_id, uint32_t permissions,
static void registry_event_global(void *data, uint32_t id, uint32_t parent_id, uint32_t permissions,
uint32_t type, uint32_t version)
{
struct pw_registry_proxy *registry = object;
GstPipeWireDeviceProvider *self = registry->proxy.user_data;
struct pw_remote *remote = self->remote;
struct pw_core *core = remote->core;
struct registry_data *rd = data;
GstPipeWireDeviceProvider *self = rd->self;
struct pw_node_proxy *node;
struct node_data *data;
struct node_data *nd;
if (type != core->type.node)
if (type != self->type->node)
return;
node = pw_registry_proxy_bind(registry, id, core->type.node, PW_VERSION_NODE,
sizeof(struct node_data), NULL);
node = pw_registry_proxy_bind(rd->registry, id, self->type->node, PW_VERSION_NODE, sizeof(*nd));
if (node == NULL)
goto no_mem;
data = node->proxy.user_data;
data->id = id;
data->parent_id = parent_id;
data->self = self;
pw_node_proxy_add_listener(node, node, &node_events);
nd = pw_proxy_get_user_data((struct pw_proxy*)node);
nd->self = self;
nd->node = node;
nd->id = id;
nd->parent_id = parent_id;
pw_node_proxy_add_listener(node, &nd->node_listener, &node_events, nd);
return;
no_mem:
@ -358,10 +386,9 @@ no_mem:
return;
}
static void registry_event_global_remove(void *object, uint32_t id)
static void registry_event_global_remove(void *data, uint32_t id)
{
struct pw_registry_proxy *registry = object;
GstPipeWireDeviceProvider *self = registry->proxy.user_data;
GstPipeWireDeviceProvider *self = data;
GstDeviceProvider *provider = GST_DEVICE_PROVIDER (self);
GstPipeWireDevice *dev;
@ -374,8 +401,14 @@ static void registry_event_global_remove(void *object, uint32_t id)
static const struct pw_registry_events registry_events = {
PW_VERSION_REGISTRY_EVENTS,
registry_event_global,
registry_event_global_remove,
.global = registry_event_global,
.global_remove = registry_event_global_remove,
};
static const struct pw_remote_callbacks remote_callbacks = {
PW_VERSION_REMOTE_CALLBACKS,
.state_changed = on_state_changed,
.sync_reply = on_sync_reply,
};
static GList *
@ -384,8 +417,11 @@ gst_pipewire_device_provider_probe (GstDeviceProvider * provider)
GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (provider);
struct pw_loop *l = NULL;
struct pw_core *c = NULL;
struct pw_type *t = NULL;
struct pw_remote *r = NULL;
struct pw_registry_proxy *reg = NULL;
struct registry_data *data;
struct pw_callback_info callbacks;
GST_DEBUG_OBJECT (self, "starting probe");
@ -395,20 +431,23 @@ gst_pipewire_device_provider_probe (GstDeviceProvider * provider)
if (!(c = pw_core_new (l, NULL)))
return NULL;
t = pw_core_get_type(c);
if (!(r = pw_remote_new (c, NULL)))
goto failed;
pw_signal_add(&r->sync_reply, &self->on_sync_reply, on_sync_reply);
pw_remote_add_callbacks(r, &callbacks, &remote_callbacks, self);
pw_remote_connect (r);
for (;;) {
enum pw_remote_state state;
const char *error = NULL;
state = r->state;
state = pw_remote_get_state(r, &error);
if (state <= 0) {
GST_ERROR_OBJECT (self, "Failed to connect: %s", r->error);
GST_ERROR_OBJECT (self, "Failed to connect: %s", error);
goto failed;
}
@ -426,13 +465,17 @@ gst_pipewire_device_provider_probe (GstDeviceProvider * provider)
self->list_only = TRUE;
self->devices = NULL;
reg = pw_core_proxy_get_registry(r->core_proxy, PW_VERSION_REGISTRY, 0, NULL);
reg->proxy.user_data = self;
pw_registry_proxy_add_listener(reg, reg, &registry_events);
pw_core_proxy_sync(r->core_proxy, 1);
self->core_proxy = pw_remote_get_core_proxy(r);
reg = pw_core_proxy_get_registry(self->core_proxy, t->registry, PW_VERSION_REGISTRY, sizeof(*data));
data = pw_proxy_get_user_data((struct pw_proxy*)reg);
data->self = self;
data->registry = reg;
pw_registry_proxy_add_listener(reg, &data->registry_listener, &registry_events, data);
pw_core_proxy_sync(self->core_proxy, 1);
for (;;) {
if (r->state <= 0)
if (pw_remote_get_state(r, NULL) <= 0)
break;
if (self->end)
break;
@ -451,34 +494,11 @@ failed:
return NULL;
}
static void
on_remote_state_changed (struct pw_listener *listener,
struct pw_remote *remote)
{
GstPipeWireDeviceProvider *self = SPA_CONTAINER_OF (listener, GstPipeWireDeviceProvider, remote_state_changed);
enum pw_remote_state state;
state= remote->state;
GST_DEBUG ("got remote state %d", state);
switch (state) {
case PW_REMOTE_STATE_CONNECTING:
break;
case PW_REMOTE_STATE_UNCONNECTED:
case PW_REMOTE_STATE_CONNECTED:
break;
case PW_REMOTE_STATE_ERROR:
GST_ERROR_OBJECT (self, "remote error: %s", remote->error);
break;
}
pw_thread_loop_signal (self->main_loop, FALSE);
}
static gboolean
gst_pipewire_device_provider_start (GstDeviceProvider * provider)
{
GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (provider);
struct registry_data *data;
GST_DEBUG_OBJECT (self, "starting provider");
@ -494,6 +514,7 @@ gst_pipewire_device_provider_start (GstDeviceProvider * provider)
GST_ERROR_OBJECT (self, "Could not create PipeWire core");
goto failed_core;
}
self->type = pw_core_get_type (self->core);
if (pw_thread_loop_start (self->main_loop) != SPA_RESULT_OK) {
GST_ERROR_OBJECT (self, "Could not start PipeWire mainloop");
@ -507,18 +528,17 @@ gst_pipewire_device_provider_start (GstDeviceProvider * provider)
goto failed_remote;
}
pw_signal_add (&self->remote->state_changed,
&self->remote_state_changed,
on_remote_state_changed);
pw_remote_add_callbacks (self->remote, &self->remote_callbacks, &remote_callbacks, self);
pw_remote_connect (self->remote);
for (;;) {
enum pw_remote_state state;
const char *error = NULL;
state = self->remote->state;
state = pw_remote_get_state(self->remote, &error);
if (state <= 0) {
GST_WARNING_OBJECT (self, "Failed to connect: %s", self->remote->error);
GST_WARNING_OBJECT (self, "Failed to connect: %s", error);
goto not_running;
}
@ -531,10 +551,16 @@ gst_pipewire_device_provider_start (GstDeviceProvider * provider)
GST_DEBUG_OBJECT (self, "connected");
get_core_info (self->remote, self);
self->registry = pw_core_proxy_get_registry(self->remote->core_proxy, PW_VERSION_REGISTRY, 0, NULL);
self->registry->proxy.user_data = self;
pw_registry_proxy_add_listener(self->registry, self->registry, &registry_events);
pw_core_proxy_sync(self->remote->core_proxy, 1);
self->core_proxy = pw_remote_get_core_proxy(self->remote);
self->registry = pw_core_proxy_get_registry(self->core_proxy, self->type->registry,
PW_VERSION_REGISTRY, sizeof(*data));
data = pw_proxy_get_user_data((struct pw_proxy*)self->registry);
data->self = self;
data->registry = self->registry;
pw_registry_proxy_add_listener(self->registry, &data->registry_listener, &registry_events, data);
pw_core_proxy_sync(self->core_proxy, 1);
pw_thread_loop_unlock (self->main_loop);
@ -548,6 +574,7 @@ failed_remote:
failed_start:
pw_core_destroy (self->core);
self->core = NULL;
self->type = NULL;
failed_core:
pw_thread_loop_destroy (self->main_loop);
self->main_loop = NULL;
@ -570,6 +597,7 @@ gst_pipewire_device_provider_stop (GstDeviceProvider * provider)
if (self->core) {
pw_core_destroy (self->core);
self->core = NULL;
self->type = NULL;
}
if (self->main_loop) {
pw_thread_loop_destroy (self->main_loop);

6
src/gst/gstpipewiredeviceprovider.h
View file

@ -85,13 +85,15 @@ struct _GstPipeWireDeviceProvider {
struct pw_thread_loop *main_loop;
struct pw_core *core;
struct pw_type *type;
struct pw_remote *remote;
struct pw_core_proxy *core_proxy;
struct pw_registry_proxy *registry;
gboolean end;
gboolean list_only;
GList **devices;
struct pw_listener remote_state_changed;
struct pw_listener on_sync_reply;
struct pw_callback_info remote_callbacks;
};
struct _GstPipeWireDeviceProviderClass {

151
src/gst/gstpipewiresink.c
View file

@ -235,8 +235,7 @@ gst_pipewire_sink_class_init (GstPipeWireSinkClass * klass)
static void
pool_activated (GstPipeWirePool *pool, GstPipeWireSink *sink)
{
struct pw_remote *remote = sink->stream->remote;
struct pw_core *core = remote->core;
struct pw_type *t = sink->type;
GstStructure *config;
GstCaps *caps;
guint size;
@ -251,36 +250,36 @@ pool_activated (GstPipeWirePool *pool, GstPipeWireSink *sink)
gst_buffer_pool_config_get_params (config, &caps, &size, &min_buffers, &max_buffers);
spa_pod_builder_init (&b, buffer, sizeof (buffer));
spa_pod_builder_push_object (&b, &f[0], 0, core->type.param_alloc_buffers.Buffers);
spa_pod_builder_push_object (&b, &f[0], 0, t->param_alloc_buffers.Buffers);
if (size == 0)
spa_pod_builder_add (&b,
PROP_U_MM (&f[1], core->type.param_alloc_buffers.size, SPA_POD_TYPE_INT, 0, 0, INT32_MAX), 0);
PROP_U_MM (&f[1], t->param_alloc_buffers.size, SPA_POD_TYPE_INT, 0, 0, INT32_MAX), 0);
else
spa_pod_builder_add (&b,
PROP_MM (&f[1], core->type.param_alloc_buffers.size, SPA_POD_TYPE_INT, size, size, INT32_MAX), 0);
PROP_MM (&f[1], t->param_alloc_buffers.size, SPA_POD_TYPE_INT, size, size, INT32_MAX), 0);
spa_pod_builder_add (&b,
PROP_MM (&f[1], core->type.param_alloc_buffers.stride, SPA_POD_TYPE_INT, 0, 0, INT32_MAX),
PROP_U_MM (&f[1], core->type.param_alloc_buffers.buffers, SPA_POD_TYPE_INT, min_buffers, min_buffers, max_buffers ? max_buffers : INT32_MAX),
PROP (&f[1], core->type.param_alloc_buffers.align, SPA_POD_TYPE_INT, 16),
PROP_MM (&f[1], t->param_alloc_buffers.stride, SPA_POD_TYPE_INT, 0, 0, INT32_MAX),
PROP_U_MM (&f[1], t->param_alloc_buffers.buffers, SPA_POD_TYPE_INT, min_buffers, min_buffers, max_buffers ? max_buffers : INT32_MAX),
PROP (&f[1], t->param_alloc_buffers.align, SPA_POD_TYPE_INT, 16),
0);
spa_pod_builder_pop (&b, &f[0]);
port_params[0] = SPA_POD_BUILDER_DEREF (&b, f[0].ref, struct spa_param);
spa_pod_builder_object (&b, &f[0], 0, core->type.param_alloc_meta_enable.MetaEnable,
PROP (&f[1], core->type.param_alloc_meta_enable.type, SPA_POD_TYPE_ID, core->type.meta.Header),
PROP (&f[1], core->type.param_alloc_meta_enable.size, SPA_POD_TYPE_INT, sizeof (struct spa_meta_header)));
spa_pod_builder_object (&b, &f[0], 0, t->param_alloc_meta_enable.MetaEnable,
PROP (&f[1], t->param_alloc_meta_enable.type, SPA_POD_TYPE_ID, t->meta.Header),
PROP (&f[1], t->param_alloc_meta_enable.size, SPA_POD_TYPE_INT, sizeof (struct spa_meta_header)));
port_params[1] = SPA_POD_BUILDER_DEREF (&b, f[0].ref, struct spa_param);
spa_pod_builder_object (&b, &f[0], 0, core->type.param_alloc_meta_enable.MetaEnable,
PROP (&f[1], core->type.param_alloc_meta_enable.type, SPA_POD_TYPE_ID, core->type.meta.Ringbuffer),
PROP (&f[1], core->type.param_alloc_meta_enable.size, SPA_POD_TYPE_INT, sizeof (struct spa_meta_ringbuffer)),
PROP (&f[1], core->type.param_alloc_meta_enable.ringbufferSize, SPA_POD_TYPE_INT,
size * SPA_MAX (4,
SPA_MAX (min_buffers, max_buffers))),
PROP (&f[1], core->type.param_alloc_meta_enable.ringbufferStride, SPA_POD_TYPE_INT, 0),
PROP (&f[1], core->type.param_alloc_meta_enable.ringbufferBlocks, SPA_POD_TYPE_INT, 1),
PROP (&f[1], core->type.param_alloc_meta_enable.ringbufferAlign, SPA_POD_TYPE_INT, 16));
spa_pod_builder_object (&b, &f[0], 0, t->param_alloc_meta_enable.MetaEnable,
PROP (&f[1], t->param_alloc_meta_enable.type, SPA_POD_TYPE_ID, t->meta.Ringbuffer),
PROP (&f[1], t->param_alloc_meta_enable.size, SPA_POD_TYPE_INT, sizeof (struct spa_meta_ringbuffer)),
PROP (&f[1], t->param_alloc_meta_enable.ringbufferSize, SPA_POD_TYPE_INT,
size * SPA_MAX (4,
SPA_MAX (min_buffers, max_buffers))),
PROP (&f[1], t->param_alloc_meta_enable.ringbufferStride, SPA_POD_TYPE_INT, 0),
PROP (&f[1], t->param_alloc_meta_enable.ringbufferBlocks, SPA_POD_TYPE_INT, 1),
PROP (&f[1], t->param_alloc_meta_enable.ringbufferAlign, SPA_POD_TYPE_INT, 16));
port_params[2] = SPA_POD_BUILDER_DEREF (&b, f[0].ref, struct spa_param);
pw_thread_loop_lock (sink->main_loop);
@ -306,6 +305,7 @@ gst_pipewire_sink_init (GstPipeWireSink * sink)
sink->loop = pw_loop_new ();
sink->main_loop = pw_thread_loop_new (sink->loop, "pipewire-sink-loop");
sink->core = pw_core_new (sink->loop, NULL);
sink->type = pw_core_get_type (sink->core);
GST_DEBUG ("loop %p %p", sink->loop, sink->main_loop);
}
@ -434,16 +434,15 @@ process_mem_data_destroy (gpointer user_data)
}
static void
on_add_buffer (struct pw_listener *listener,
struct pw_stream *stream,
uint32_t id)
on_add_buffer (void *_data,
uint32_t id)
{
GstPipeWireSink *pwsink = SPA_CONTAINER_OF (listener, GstPipeWireSink, stream_add_buffer);
GstPipeWireSink *pwsink = _data;
struct spa_buffer *b;
GstBuffer *buf;
uint32_t i;
ProcessMemData data;
struct pw_core *core = pwsink->remote->core;
struct pw_type *t = pwsink->type;
GST_LOG_OBJECT (pwsink, "add buffer");
@ -457,20 +456,20 @@ on_add_buffer (struct pw_listener *listener,
data.sink = gst_object_ref (pwsink);
data.id = id;
data.buf = b;
data.header = spa_buffer_find_meta (b, core->type.meta.Header);
data.header = spa_buffer_find_meta (b, t->meta.Header);
for (i = 0; i < b->n_datas; i++) {
struct spa_data *d = &b->datas[i];
GstMemory *gmem = NULL;
if (d->type == core->type.data.MemFd ||
d->type == core->type.data.DmaBuf) {
if (d->type == t->data.MemFd ||
d->type == t->data.DmaBuf) {
gmem = gst_fd_allocator_alloc (pwsink->allocator, dup (d->fd),
d->mapoffset + d->maxsize, GST_FD_MEMORY_FLAG_NONE);
gst_memory_resize (gmem, d->chunk->offset + d->mapoffset, d->chunk->size);
data.offset = d->mapoffset;
}
else if (d->type == core->type.data.MemPtr) {
else if (d->type == t->data.MemPtr) {
gmem = gst_memory_new_wrapped (0, d->data, d->maxsize, d->chunk->offset,
d->chunk->size, NULL, NULL);
data.offset = 0;
@ -491,11 +490,10 @@ on_add_buffer (struct pw_listener *listener,
}
static void
on_remove_buffer (struct pw_listener *listener,
struct pw_stream *stream,
uint32_t id)
on_remove_buffer (void *data,
uint32_t id)
{
GstPipeWireSink *pwsink = SPA_CONTAINER_OF (listener, GstPipeWireSink, stream_remove_buffer);
GstPipeWireSink *pwsink = data;
GstBuffer *buf;
GST_LOG_OBJECT (pwsink, "remove buffer");
@ -511,11 +509,10 @@ on_remove_buffer (struct pw_listener *listener,
}
static void
on_new_buffer (struct pw_listener *listener,
struct pw_stream *stream,
uint32_t id)
on_new_buffer (void *data,
uint32_t id)
{
GstPipeWireSink *pwsink = SPA_CONTAINER_OF (listener, GstPipeWireSink, stream_new_buffer);
GstPipeWireSink *pwsink = data;
GstBuffer *buf;
GST_LOG_OBJECT (pwsink, "got new buffer %u", id);
@ -569,24 +566,19 @@ do_send_buffer (GstPipeWireSink *pwsink)
static void
on_need_buffer (struct pw_listener *listener,
struct pw_stream *stream)
on_need_buffer (void *data)
{
GstPipeWireSink *pwsink = SPA_CONTAINER_OF (listener, GstPipeWireSink, stream_need_buffer);
GstPipeWireSink *pwsink = data;
pwsink->need_ready++;
GST_DEBUG ("need buffer %u", pwsink->need_ready);
do_send_buffer (pwsink);
}
static void
on_state_changed (struct pw_listener *listener,
struct pw_stream *stream)
on_state_changed (void *data, enum pw_stream_state old, enum pw_stream_state state, const char *error)
{
GstPipeWireSink *pwsink = SPA_CONTAINER_OF (listener, GstPipeWireSink, stream_state_changed);
enum pw_stream_state state;
GstPipeWireSink *pwsink = data;
state = stream->state;
GST_DEBUG ("got stream state %d", state);
switch (state) {
@ -599,18 +591,16 @@ on_state_changed (struct pw_listener *listener,
break;
case PW_STREAM_STATE_ERROR:
GST_ELEMENT_ERROR (pwsink, RESOURCE, FAILED,
("stream error: %s", stream->error), (NULL));
("stream error: %s", error), (NULL));
break;
}
pw_thread_loop_signal (pwsink->main_loop, FALSE);
}
static void
on_format_changed (struct pw_listener *listener,
struct pw_stream *stream,
struct spa_format *format)
on_format_changed (void *data, struct spa_format *format)
{
GstPipeWireSink *pwsink = SPA_CONTAINER_OF (listener, GstPipeWireSink, stream_format_changed);
GstPipeWireSink *pwsink = data;
if (gst_buffer_pool_is_active (GST_BUFFER_POOL_CAST (pwsink->pool)))
pool_activated (pwsink->pool, pwsink);
@ -622,14 +612,15 @@ gst_pipewire_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
GstPipeWireSink *pwsink;
GPtrArray *possible;
enum pw_stream_state state;
const char *error = NULL;
gboolean res = FALSE;
pwsink = GST_PIPEWIRE_SINK (bsink);
possible = gst_caps_to_format_all (caps, pwsink->remote->core->type.map);
possible = gst_caps_to_format_all (caps, pwsink->type->map);
pw_thread_loop_lock (pwsink->main_loop);
state = pwsink->stream->state;
state = pw_stream_get_state (pwsink->stream, &error);
if (state == PW_STREAM_STATE_ERROR)
goto start_error;
@ -649,7 +640,7 @@ gst_pipewire_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
(const struct spa_format **) possible->pdata);
while (TRUE) {
state = pwsink->stream->state;
state = pw_stream_get_state (pwsink->stream, &error);
if (state == PW_STREAM_STATE_READY)
break;
@ -670,7 +661,7 @@ gst_pipewire_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
start_error:
{
GST_ERROR ("could not start stream");
GST_ERROR ("could not start stream: %s", error);
pw_thread_loop_unlock (pwsink->main_loop);
g_ptr_array_unref (possible);
return FALSE;
@ -682,6 +673,7 @@ gst_pipewire_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
{
GstPipeWireSink *pwsink;
GstFlowReturn res = GST_FLOW_OK;
const char *error = NULL;
pwsink = GST_PIPEWIRE_SINK (bsink);
@ -689,7 +681,7 @@ gst_pipewire_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
goto not_negotiated;
pw_thread_loop_lock (pwsink->main_loop);
if (pwsink->stream->state != PW_STREAM_STATE_STREAMING)
if (pw_stream_get_state (pwsink->stream, &error) != PW_STREAM_STATE_STREAMING)
goto done;
if (buffer->pool != GST_BUFFER_POOL_CAST (pwsink->pool)) {
@ -742,6 +734,16 @@ copy_properties (GQuark field_id,
return TRUE;
}
static const struct pw_stream_callbacks stream_callbacks = {
PW_VERSION_STREAM_CALLBACKS,
.state_changed = on_state_changed,
.format_changed = on_format_changed,
.add_buffer = on_add_buffer,
.remove_buffer = on_remove_buffer,
.new_buffer = on_new_buffer,
.need_buffer = on_need_buffer,
};
static gboolean
gst_pipewire_sink_start (GstBaseSink * basesink)
{
@ -761,12 +763,11 @@ gst_pipewire_sink_start (GstBaseSink * basesink)
pwsink->stream = pw_stream_new (pwsink->remote, pwsink->client_name, props);
pwsink->pool->stream = pwsink->stream;
pw_signal_add (&pwsink->stream->state_changed, &pwsink->stream_state_changed, on_state_changed);
pw_signal_add (&pwsink->stream->format_changed, &pwsink->stream_format_changed, on_format_changed);
pw_signal_add (&pwsink->stream->add_buffer, &pwsink->stream_add_buffer, on_add_buffer);
pw_signal_add (&pwsink->stream->remove_buffer, &pwsink->stream_remove_buffer, on_remove_buffer);
pw_signal_add (&pwsink->stream->new_buffer, &pwsink->stream_new_buffer, on_new_buffer);
pw_signal_add (&pwsink->stream->need_buffer, &pwsink->stream_need_buffer, on_need_buffer);
pw_stream_add_callbacks(pwsink->stream,
&pwsink->stream_callbacks,
&stream_callbacks,
pwsink);
pw_thread_loop_unlock (pwsink->main_loop);
return TRUE;
@ -792,13 +793,10 @@ gst_pipewire_sink_stop (GstBaseSink * basesink)
}
static void
on_remote_state_changed (struct pw_listener *listener,
struct pw_remote *remote)
on_remote_state_changed (void *data, enum pw_remote_state old, enum pw_remote_state state, const char *error)
{
GstPipeWireSink *pwsink = SPA_CONTAINER_OF (listener, GstPipeWireSink, remote_state_changed);
enum pw_remote_state state;
GstPipeWireSink *pwsink = data;
state = remote->state;
GST_DEBUG ("got remote state %d", state);
switch (state) {
@ -808,27 +806,36 @@ on_remote_state_changed (struct pw_listener *listener,
break;
case PW_REMOTE_STATE_ERROR:
GST_ELEMENT_ERROR (pwsink, RESOURCE, FAILED,
("remote error: %s", remote->error), (NULL));
("remote error: %s", error), (NULL));
break;
}
pw_thread_loop_signal (pwsink->main_loop, FALSE);
}
static const struct pw_remote_callbacks remote_callbacks = {
PW_VERSION_REMOTE_CALLBACKS,
.state_changed = on_remote_state_changed,
};
static gboolean
gst_pipewire_sink_open (GstPipeWireSink * pwsink)
{
const char *error = NULL;
if (pw_thread_loop_start (pwsink->main_loop) != SPA_RESULT_OK)
goto mainloop_error;
pw_thread_loop_lock (pwsink->main_loop);
pwsink->remote = pw_remote_new (pwsink->core, NULL);
pw_signal_add (&pwsink->remote->state_changed, &pwsink->remote_state_changed, on_remote_state_changed);
pw_remote_add_callbacks (pwsink->remote,
&pwsink->remote_callbacks,
&remote_callbacks, pwsink);
pw_remote_connect (pwsink->remote);
while (TRUE) {
enum pw_remote_state state = pwsink->remote->state;
enum pw_remote_state state = pw_remote_get_state (pwsink->remote, &error);
if (state == PW_REMOTE_STATE_CONNECTED)
break;
@ -859,6 +866,8 @@ connect_error:
static gboolean
gst_pipewire_sink_close (GstPipeWireSink * pwsink)
{
const char *error = NULL;
pw_thread_loop_lock (pwsink->main_loop);
if (pwsink->stream) {
pw_stream_disconnect (pwsink->stream);
@ -867,7 +876,7 @@ gst_pipewire_sink_close (GstPipeWireSink * pwsink)
pw_remote_disconnect (pwsink->remote);
while (TRUE) {
enum pw_remote_state state = pwsink->remote->state;
enum pw_remote_state state = pw_remote_get_state (pwsink->remote, &error);
if (state == PW_REMOTE_STATE_UNCONNECTED)
break;

10
src/gst/gstpipewiresink.h
View file

@ -81,16 +81,12 @@ struct _GstPipeWireSink {
struct pw_thread_loop *main_loop;
struct pw_core *core;
struct pw_type *type;
struct pw_remote *remote;
struct pw_listener remote_state_changed;
struct pw_callback_info remote_callbacks;
struct pw_stream *stream;
struct pw_listener stream_state_changed;
struct pw_listener stream_format_changed;
struct pw_listener stream_add_buffer;
struct pw_listener stream_remove_buffer;
struct pw_listener stream_new_buffer;
struct pw_listener stream_need_buffer;
struct pw_callback_info stream_callbacks;
GstAllocator *allocator;
GstStructure *properties;

139
src/gst/gstpipewiresrc.c
View file

@ -196,6 +196,7 @@ gst_pipewire_src_finalize (GObject * object)
pw_core_destroy (pwsrc->core);
pwsrc->core = NULL;
pwsrc->type = NULL;
pw_thread_loop_destroy (pwsrc->main_loop);
pwsrc->main_loop = NULL;
pw_loop_destroy (pwsrc->loop);
@ -310,6 +311,7 @@ gst_pipewire_src_init (GstPipeWireSrc * src)
src->loop = pw_loop_new ();
src->main_loop = pw_thread_loop_new (src->loop, "pipewire-main-loop");
src->core = pw_core_new (src->loop, NULL);
src->type = pw_core_get_type (src->core);
GST_DEBUG ("loop %p, mainloop %p", src->loop, src->main_loop);
}
@ -353,17 +355,15 @@ buffer_recycle (GstMiniObject *obj)
}
static void
on_add_buffer (struct pw_listener *listener,
struct pw_stream *stream,
guint id)
on_add_buffer (void *_data, guint id)
{
GstPipeWireSrc *pwsrc = SPA_CONTAINER_OF (listener, GstPipeWireSrc, stream_add_buffer);
GstPipeWireSrc *pwsrc = _data;
struct spa_buffer *b;
GstBuffer *buf;
uint32_t i;
ProcessMemData data;
struct pw_remote *remote = pwsrc->stream->remote;
struct pw_core *core = remote->core;
struct pw_core *core = pwsrc->core;
struct pw_type *t = pw_core_get_type(core);
GST_LOG_OBJECT (pwsrc, "add buffer");
@ -378,19 +378,19 @@ on_add_buffer (struct pw_listener *listener,
data.src = gst_object_ref (pwsrc);
data.id = id;
data.buf = b;
data.header = spa_buffer_find_meta (b, core->type.meta.Header);
data.header = spa_buffer_find_meta (b, t->meta.Header);
for (i = 0; i < b->n_datas; i++) {
struct spa_data *d = &b->datas[i];
GstMemory *gmem = NULL;
if (d->type == core->type.data.MemFd || d->type == core->type.data.DmaBuf) {
if (d->type == t->data.MemFd || d->type == t->data.DmaBuf) {
gmem = gst_fd_allocator_alloc (pwsrc->fd_allocator, dup (d->fd),
d->mapoffset + d->maxsize, GST_FD_MEMORY_FLAG_NONE);
gst_memory_resize (gmem, d->chunk->offset + d->mapoffset, d->chunk->size);
data.offset = d->mapoffset;
}
else if (d->type == core->type.data.MemPtr) {
else if (d->type == t->data.MemPtr) {
gmem = gst_memory_new_wrapped (0, d->data, d->maxsize, d->chunk->offset + d->mapoffset,
d->chunk->size, NULL, NULL);
data.offset = 0;
@ -408,11 +408,10 @@ on_add_buffer (struct pw_listener *listener,
}
static void
on_remove_buffer (struct pw_listener *listener,
struct pw_stream *stream,
guint id)
on_remove_buffer (void *data,
guint id)
{
GstPipeWireSrc *pwsrc = SPA_CONTAINER_OF (listener, GstPipeWireSrc, stream_remove_buffer);
GstPipeWireSrc *pwsrc = data;
GstBuffer *buf;
GST_LOG_OBJECT (pwsrc, "remove buffer");
@ -437,11 +436,10 @@ on_remove_buffer (struct pw_listener *listener,
}
static void
on_new_buffer (struct pw_listener *listener,
struct pw_stream *stream,
guint id)
on_new_buffer (void *_data,
guint id)
{
GstPipeWireSrc *pwsrc = SPA_CONTAINER_OF (listener, GstPipeWireSrc, stream_new_buffer);
GstPipeWireSrc *pwsrc = _data;
GstBuffer *buf;
ProcessMemData *data;
struct spa_meta_header *h;
@ -486,11 +484,11 @@ on_new_buffer (struct pw_listener *listener,
}
static void
on_state_changed (struct pw_listener *listener,
struct pw_stream *stream)
on_state_changed (void *data,
enum pw_stream_state old,
enum pw_stream_state state, const char *error)
{
GstPipeWireSrc *pwsrc = SPA_CONTAINER_OF (listener, GstPipeWireSrc, stream_state_changed);
enum pw_stream_state state = stream->state;
GstPipeWireSrc *pwsrc = data;
GST_DEBUG ("got stream state %s", pw_stream_state_as_string (state));
@ -504,7 +502,7 @@ on_state_changed (struct pw_listener *listener,
break;
case PW_STREAM_STATE_ERROR:
GST_ELEMENT_ERROR (pwsrc, RESOURCE, FAILED,
("stream error: %s", stream->error), (NULL));
("stream error: %s", error), (NULL));
break;
}
pw_thread_loop_signal (pwsrc->main_loop, FALSE);
@ -535,10 +533,11 @@ parse_stream_properties (GstPipeWireSrc *pwsrc, struct pw_properties *props)
static gboolean
gst_pipewire_src_stream_start (GstPipeWireSrc *pwsrc)
{
const char *error = NULL;
pw_thread_loop_lock (pwsrc->main_loop);
GST_DEBUG_OBJECT (pwsrc, "doing stream start");
while (TRUE) {
enum pw_stream_state state = pwsrc->stream->state;
enum pw_stream_state state = pw_stream_get_state (pwsrc->stream, &error);
GST_DEBUG_OBJECT (pwsrc, "waiting for STREAMING, now %s", pw_stream_state_as_string (state));
if (state == PW_STREAM_STATE_STREAMING)
@ -547,13 +546,13 @@ gst_pipewire_src_stream_start (GstPipeWireSrc *pwsrc)
if (state == PW_STREAM_STATE_ERROR)
goto start_error;
if (pwsrc->remote->state == PW_REMOTE_STATE_ERROR)
if (pw_remote_get_state(pwsrc->remote, &error) == PW_REMOTE_STATE_ERROR)
goto start_error;
pw_thread_loop_wait (pwsrc->main_loop);
}
parse_stream_properties (pwsrc, pwsrc->stream->properties);
parse_stream_properties (pwsrc, pw_stream_get_properties (pwsrc->stream));
GST_DEBUG_OBJECT (pwsrc, "signal started");
pwsrc->started = TRUE;
pw_thread_loop_signal (pwsrc->main_loop, FALSE);
@ -563,7 +562,7 @@ gst_pipewire_src_stream_start (GstPipeWireSrc *pwsrc)
start_error:
{
GST_DEBUG_OBJECT (pwsrc, "error starting stream");
GST_DEBUG_OBJECT (pwsrc, "error starting stream: %s", error);
pw_thread_loop_unlock (pwsrc->main_loop);
return FALSE;
}
@ -573,10 +572,11 @@ static enum pw_stream_state
wait_negotiated (GstPipeWireSrc *this)
{
enum pw_stream_state state;
const char *error = NULL;
pw_thread_loop_lock (this->main_loop);
while (TRUE) {
state = this->stream->state;
state = pw_stream_get_state (this->stream, &error);
GST_DEBUG_OBJECT (this, "waiting for started signal, state now %s",
pw_stream_state_as_string (state));
@ -584,7 +584,7 @@ wait_negotiated (GstPipeWireSrc *this)
if (state == PW_STREAM_STATE_ERROR)
break;
if (this->remote->state == PW_REMOTE_STATE_ERROR)
if (pw_remote_get_state(this->remote, &error) == PW_REMOTE_STATE_ERROR)
break;
if (this->started)
@ -607,6 +607,7 @@ gst_pipewire_src_negotiate (GstBaseSrc * basesrc)
GstCaps *peercaps = NULL;
gboolean result = FALSE;
GPtrArray *possible;
const char *error = NULL;
/* first see what is possible on our source pad */
thiscaps = gst_pad_query_caps (GST_BASE_SRC_PAD (basesrc), NULL);
@ -635,16 +636,16 @@ gst_pipewire_src_negotiate (GstBaseSrc * basesrc)
GST_DEBUG_OBJECT (basesrc, "have common caps: %" GST_PTR_FORMAT, caps);
/* open a connection with these caps */
possible = gst_caps_to_format_all (caps, pwsrc->remote->core->type.map);
possible = gst_caps_to_format_all (caps, pwsrc->type->map);
gst_caps_unref (caps);
/* first disconnect */
pw_thread_loop_lock (pwsrc->main_loop);
if (pwsrc->stream->state != PW_STREAM_STATE_UNCONNECTED) {
if (pw_stream_get_state(pwsrc->stream, &error) != PW_STREAM_STATE_UNCONNECTED) {
GST_DEBUG_OBJECT (basesrc, "disconnect capture");
pw_stream_disconnect (pwsrc->stream);
while (TRUE) {
enum pw_stream_state state = pwsrc->stream->state;
enum pw_stream_state state = pw_stream_get_state (pwsrc->stream, &error);
GST_DEBUG_OBJECT (basesrc, "waiting for UNCONNECTED, now %s", pw_stream_state_as_string (state));
if (state == PW_STREAM_STATE_UNCONNECTED)
@ -670,7 +671,7 @@ gst_pipewire_src_negotiate (GstBaseSrc * basesrc)
g_ptr_array_free (possible, TRUE);
while (TRUE) {
enum pw_stream_state state = pwsrc->stream->state;
enum pw_stream_state state = pw_stream_get_state (pwsrc->stream, &error);
GST_DEBUG_OBJECT (basesrc, "waiting for PAUSED, now %s", pw_stream_state_as_string (state));
if (state == PW_STREAM_STATE_PAUSED ||
@ -680,7 +681,7 @@ gst_pipewire_src_negotiate (GstBaseSrc * basesrc)
if (state == PW_STREAM_STATE_ERROR)
goto connect_error;
if (pwsrc->remote->state == PW_REMOTE_STATE_ERROR)
if (pw_remote_get_state(pwsrc->remote, &error) == PW_REMOTE_STATE_ERROR)
goto connect_error;
pw_thread_loop_wait (pwsrc->main_loop);
@ -732,15 +733,14 @@ connect_error:
SPA_POD_PROP_RANGE_MIN_MAX,type,3,__VA_ARGS__)
static void
on_format_changed (struct pw_listener *listener,
struct pw_stream *stream,
struct spa_format *format)
on_format_changed (void *data,
struct spa_format *format)
{
GstPipeWireSrc *pwsrc = SPA_CONTAINER_OF (listener, GstPipeWireSrc, stream_format_changed);
GstPipeWireSrc *pwsrc = data;
GstCaps *caps;
gboolean res;
struct pw_remote *remote = stream->remote;
struct pw_core *core = remote->core;
struct pw_core *core = pwsrc->core;
struct pw_type *t = pw_core_get_type(core);
if (format == NULL) {
GST_DEBUG_OBJECT (pwsrc, "clear format");
@ -748,7 +748,7 @@ on_format_changed (struct pw_listener *listener,
return;
}
caps = gst_caps_from_format (format, core->type.map);
caps = gst_caps_from_format (format, t->map);
GST_DEBUG_OBJECT (pwsrc, "we got format %" GST_PTR_FORMAT, caps);
res = gst_base_src_set_caps (GST_BASE_SRC (pwsrc), caps);
gst_caps_unref (caps);
@ -760,16 +760,16 @@ on_format_changed (struct pw_listener *listener,
struct spa_pod_frame f[2];
spa_pod_builder_init (&b, buffer, sizeof (buffer));
spa_pod_builder_object (&b, &f[0], 0, core->type.param_alloc_buffers.Buffers,
PROP_U_MM (&f[1], core->type.param_alloc_buffers.size, SPA_POD_TYPE_INT, 0, 0, INT32_MAX),
PROP_U_MM (&f[1], core->type.param_alloc_buffers.stride, SPA_POD_TYPE_INT, 0, 0, INT32_MAX),
PROP_U_MM (&f[1], core->type.param_alloc_buffers.buffers, SPA_POD_TYPE_INT, 16, 0, INT32_MAX),
PROP (&f[1], core->type.param_alloc_buffers.align, SPA_POD_TYPE_INT, 16));
spa_pod_builder_object (&b, &f[0], 0, t->param_alloc_buffers.Buffers,
PROP_U_MM (&f[1], t->param_alloc_buffers.size, SPA_POD_TYPE_INT, 0, 0, INT32_MAX),
PROP_U_MM (&f[1], t->param_alloc_buffers.stride, SPA_POD_TYPE_INT, 0, 0, INT32_MAX),
PROP_U_MM (&f[1], t->param_alloc_buffers.buffers, SPA_POD_TYPE_INT, 16, 0, INT32_MAX),
PROP (&f[1], t->param_alloc_buffers.align, SPA_POD_TYPE_INT, 16));
params[0] = SPA_POD_BUILDER_DEREF (&b, f[0].ref, struct spa_param);
spa_pod_builder_object (&b, &f[0], 0, core->type.param_alloc_meta_enable.MetaEnable,
PROP (&f[1], core->type.param_alloc_meta_enable.type, SPA_POD_TYPE_ID, core->type.meta.Header),
PROP (&f[1], core->type.param_alloc_meta_enable.size, SPA_POD_TYPE_INT, sizeof (struct spa_meta_header)));
spa_pod_builder_object (&b, &f[0], 0, t->param_alloc_meta_enable.MetaEnable,
PROP (&f[1], t->param_alloc_meta_enable.type, SPA_POD_TYPE_ID, t->meta.Header),
PROP (&f[1], t->param_alloc_meta_enable.size, SPA_POD_TYPE_INT, sizeof (struct spa_meta_header)));
params[1] = SPA_POD_BUILDER_DEREF (&b, f[0].ref, struct spa_param);
GST_DEBUG_OBJECT (pwsrc, "doing finish format");
@ -882,6 +882,7 @@ gst_pipewire_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
{
GstPipeWireSrc *pwsrc;
GstClockTime pts, dts, base_time;
const char *error = NULL;
pwsrc = GST_PIPEWIRE_SRC (psrc);
@ -898,7 +899,7 @@ gst_pipewire_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
if (pwsrc->stream == NULL)
goto streaming_error;
state = pwsrc->stream->state;
state = pw_stream_get_state (pwsrc->stream, &error);
if (state == PW_STREAM_STATE_ERROR)
goto streaming_error;
@ -974,11 +975,9 @@ gst_pipewire_src_stop (GstBaseSrc * basesrc)
}
static void
on_remote_state_changed (struct pw_listener *listener,
struct pw_remote *remote)
on_remote_state_changed (void *data, enum pw_remote_state old, enum pw_remote_state state, const char *error)
{
GstPipeWireSrc *pwsrc = SPA_CONTAINER_OF (listener, GstPipeWireSrc, remote_state_changed);
enum pw_remote_state state = remote->state;
GstPipeWireSrc *pwsrc = data;
GST_DEBUG ("got remote state %s", pw_remote_state_as_string (state));
@ -989,7 +988,7 @@ on_remote_state_changed (struct pw_listener *listener,
break;
case PW_REMOTE_STATE_ERROR:
GST_ELEMENT_ERROR (pwsrc, RESOURCE, FAILED,
("remote error: %s", remote->error), (NULL));
("remote error: %s", error), (NULL));
break;
}
pw_thread_loop_signal (pwsrc->main_loop, FALSE);
@ -1009,10 +1008,25 @@ copy_properties (GQuark field_id,
return TRUE;
}
static const struct pw_remote_callbacks remote_callbacks = {
PW_VERSION_REMOTE_CALLBACKS,
.state_changed = on_remote_state_changed,
};
static const struct pw_stream_callbacks stream_callbacks = {
PW_VERSION_STREAM_CALLBACKS,
.state_changed = on_state_changed,
.format_changed = on_format_changed,
.add_buffer = on_add_buffer,
.remove_buffer = on_remove_buffer,
.new_buffer = on_new_buffer,
};
static gboolean
gst_pipewire_src_open (GstPipeWireSrc * pwsrc)
{
struct pw_properties *props;
const char *error = NULL;
if (pw_thread_loop_start (pwsrc->main_loop) != SPA_RESULT_OK)
goto mainloop_failed;
@ -1021,12 +1035,14 @@ gst_pipewire_src_open (GstPipeWireSrc * pwsrc)
if ((pwsrc->remote = pw_remote_new (pwsrc->core, NULL)) == NULL)
goto no_remote;
pw_signal_add (&pwsrc->remote->state_changed, &pwsrc->remote_state_changed, on_remote_state_changed);
pw_remote_add_callbacks (pwsrc->remote,
&pwsrc->remote_callbacks,
&remote_callbacks, pwsrc);
pw_remote_connect (pwsrc->remote);
while (TRUE) {
enum pw_remote_state state = pwsrc->remote->state;
enum pw_remote_state state = pw_remote_get_state(pwsrc->remote, &error);
GST_DEBUG ("waiting for CONNECTED, now %s", pw_remote_state_as_string (state));
if (state == PW_REMOTE_STATE_CONNECTED)
@ -1048,11 +1064,12 @@ gst_pipewire_src_open (GstPipeWireSrc * pwsrc)
if ((pwsrc->stream = pw_stream_new (pwsrc->remote, pwsrc->client_name, props)) == NULL)
goto no_stream;
pw_signal_add (&pwsrc->stream->state_changed, &pwsrc->stream_state_changed, on_state_changed);
pw_signal_add (&pwsrc->stream->format_changed, &pwsrc->stream_format_changed, on_format_changed);
pw_signal_add (&pwsrc->stream->add_buffer, &pwsrc->stream_add_buffer, on_add_buffer);
pw_signal_add (&pwsrc->stream->remove_buffer, &pwsrc->stream_remove_buffer, on_remove_buffer);
pw_signal_add (&pwsrc->stream->new_buffer, &pwsrc->stream_new_buffer, on_new_buffer);
pw_stream_add_callbacks(pwsrc->stream,
&pwsrc->stream_callbacks,
&stream_callbacks,
pwsrc);
pwsrc->clock = gst_pipewire_clock_new (pwsrc->stream);
pw_thread_loop_unlock (pwsrc->main_loop);

9
src/gst/gstpipewiresrc.h
View file

@ -68,15 +68,12 @@ struct _GstPipeWireSrc {
struct pw_thread_loop *main_loop;
struct pw_core *core;
struct pw_type *type;
struct pw_remote *remote;
struct pw_listener remote_state_changed;
struct pw_callback_info remote_callbacks;
struct pw_stream *stream;
struct pw_listener stream_state_changed;
struct pw_listener stream_format_changed;
struct pw_listener stream_add_buffer;
struct pw_listener stream_remove_buffer;
struct pw_listener stream_new_buffer;
struct pw_callback_info stream_callbacks;
GstAllocator *fd_allocator;
GstStructure *properties;