diff options
Diffstat (limited to '')
-rw-r--r-- | src/gst/gstpipewiresink.c | 924 |
1 files changed, 924 insertions, 0 deletions
diff --git a/src/gst/gstpipewiresink.c b/src/gst/gstpipewiresink.c new file mode 100644 index 0000000..91bdfcc --- /dev/null +++ b/src/gst/gstpipewiresink.c @@ -0,0 +1,924 @@ +/* GStreamer + * + * Copyright © 2018 Wim Taymans + * + * Permission is hereby granted, free of charge, to any person obtaining a + * copy of this software and associated documentation files (the "Software"), + * to deal in the Software without restriction, including without limitation + * the rights to use, copy, modify, merge, publish, distribute, sublicense, + * and/or sell copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice (including the next + * paragraph) shall be included in all copies or substantial portions of the + * Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER + * DEALINGS IN THE SOFTWARE. + */ + +/** + * SECTION:element-pipewiresink + * + * <refsect2> + * <title>Example launch line</title> + * |[ + * gst-launch -v videotestsrc ! pipewiresink + * ]| Sends a test video source to PipeWire + * </refsect2> + */ + +#define PW_ENABLE_DEPRECATED + +#include "config.h" +#include "gstpipewiresink.h" + +#include <string.h> +#include <stdlib.h> +#include <fcntl.h> +#include <unistd.h> +#include <sys/socket.h> + +#include <spa/pod/builder.h> +#include <spa/utils/result.h> + +#include <gst/video/video.h> + +#include "gstpipewireformat.h" + +GST_DEBUG_CATEGORY_STATIC (pipewire_sink_debug); +#define GST_CAT_DEFAULT pipewire_sink_debug + +#define DEFAULT_PROP_MODE GST_PIPEWIRE_SINK_MODE_DEFAULT + +#define MIN_BUFFERS 8u + +enum +{ + PROP_0, + PROP_PATH, + PROP_TARGET_OBJECT, + PROP_CLIENT_NAME, + PROP_CLIENT_PROPERTIES, + PROP_STREAM_PROPERTIES, + PROP_MODE, + PROP_FD +}; + +GType +gst_pipewire_sink_mode_get_type (void) +{ + static gsize mode_type = 0; + static const GEnumValue mode[] = { + {GST_PIPEWIRE_SINK_MODE_DEFAULT, "GST_PIPEWIRE_SINK_MODE_DEFAULT", "default"}, + {GST_PIPEWIRE_SINK_MODE_RENDER, "GST_PIPEWIRE_SINK_MODE_RENDER", "render"}, + {GST_PIPEWIRE_SINK_MODE_PROVIDE, "GST_PIPEWIRE_SINK_MODE_PROVIDE", "provide"}, + {0, NULL, NULL}, + }; + + if (g_once_init_enter (&mode_type)) { + GType tmp = + g_enum_register_static ("GstPipeWireSinkMode", mode); + g_once_init_leave (&mode_type, tmp); + } + + return (GType) mode_type; +} + + +static GstStaticPadTemplate gst_pipewire_sink_template = +GST_STATIC_PAD_TEMPLATE ("sink", + GST_PAD_SINK, + GST_PAD_ALWAYS, + GST_STATIC_CAPS_ANY + ); + +#define gst_pipewire_sink_parent_class parent_class +G_DEFINE_TYPE (GstPipeWireSink, gst_pipewire_sink, GST_TYPE_BASE_SINK); + +static void gst_pipewire_sink_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec); +static void gst_pipewire_sink_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec); + +static GstStateChangeReturn +gst_pipewire_sink_change_state (GstElement * element, GstStateChange transition); + +static gboolean gst_pipewire_sink_setcaps (GstBaseSink * bsink, GstCaps * caps); +static GstCaps *gst_pipewire_sink_sink_fixate (GstBaseSink * bsink, + GstCaps * caps); + +static GstFlowReturn gst_pipewire_sink_render (GstBaseSink * psink, + GstBuffer * buffer); +static gboolean gst_pipewire_sink_start (GstBaseSink * basesink); +static gboolean gst_pipewire_sink_stop (GstBaseSink * basesink); + +static void +gst_pipewire_sink_finalize (GObject * object) +{ + GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (object); + + g_object_unref (pwsink->pool); + + if (pwsink->stream_properties) + gst_structure_free (pwsink->stream_properties); + if (pwsink->client_properties) + gst_structure_free (pwsink->client_properties); + g_free (pwsink->path); + g_free (pwsink->target_object); + g_free (pwsink->client_name); + + G_OBJECT_CLASS (parent_class)->finalize (object); +} + +static gboolean +gst_pipewire_sink_propose_allocation (GstBaseSink * bsink, GstQuery * query) +{ + GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (bsink); + + gst_query_add_allocation_pool (query, GST_BUFFER_POOL_CAST (pwsink->pool), 0, 0, 0); + return TRUE; +} + +static void +gst_pipewire_sink_class_init (GstPipeWireSinkClass * klass) +{ + GObjectClass *gobject_class; + GstElementClass *gstelement_class; + GstBaseSinkClass *gstbasesink_class; + + gobject_class = (GObjectClass *) klass; + gstelement_class = (GstElementClass *) klass; + gstbasesink_class = (GstBaseSinkClass *) klass; + + gobject_class->finalize = gst_pipewire_sink_finalize; + gobject_class->set_property = gst_pipewire_sink_set_property; + gobject_class->get_property = gst_pipewire_sink_get_property; + + g_object_class_install_property (gobject_class, + PROP_PATH, + g_param_spec_string ("path", + "Path", + "The sink path to connect to (NULL = default)", + NULL, + G_PARAM_READWRITE | + G_PARAM_STATIC_STRINGS | + G_PARAM_DEPRECATED)); + + g_object_class_install_property (gobject_class, + PROP_TARGET_OBJECT, + g_param_spec_string ("target-object", + "Target object", + "The sink name/serial to connect to (NULL = default)", + NULL, + G_PARAM_READWRITE | + G_PARAM_STATIC_STRINGS)); + + g_object_class_install_property (gobject_class, + PROP_CLIENT_NAME, + g_param_spec_string ("client-name", + "Client Name", + "The client name to use (NULL = default)", + NULL, + G_PARAM_READWRITE | + G_PARAM_STATIC_STRINGS)); + + g_object_class_install_property (gobject_class, + PROP_CLIENT_PROPERTIES, + g_param_spec_boxed ("client-properties", + "Client properties", + "List of PipeWire client properties", + GST_TYPE_STRUCTURE, + G_PARAM_READWRITE | + G_PARAM_STATIC_STRINGS)); + + g_object_class_install_property (gobject_class, + PROP_STREAM_PROPERTIES, + g_param_spec_boxed ("stream-properties", + "Stream properties", + "List of PipeWire stream properties", + GST_TYPE_STRUCTURE, + G_PARAM_READWRITE | + G_PARAM_STATIC_STRINGS)); + + g_object_class_install_property (gobject_class, + PROP_MODE, + g_param_spec_enum ("mode", + "Mode", + "The mode to operate in", + GST_TYPE_PIPEWIRE_SINK_MODE, + DEFAULT_PROP_MODE, + G_PARAM_READWRITE | + G_PARAM_STATIC_STRINGS)); + + g_object_class_install_property (gobject_class, + PROP_FD, + g_param_spec_int ("fd", + "Fd", + "The fd to connect with", + -1, G_MAXINT, -1, + G_PARAM_READWRITE | + G_PARAM_STATIC_STRINGS)); + + gstelement_class->change_state = gst_pipewire_sink_change_state; + + gst_element_class_set_static_metadata (gstelement_class, + "PipeWire sink", "Sink/Video", + "Send video to PipeWire", "Wim Taymans <wim.taymans@gmail.com>"); + + gst_element_class_add_pad_template (gstelement_class, + gst_static_pad_template_get (&gst_pipewire_sink_template)); + + gstbasesink_class->set_caps = gst_pipewire_sink_setcaps; + gstbasesink_class->fixate = gst_pipewire_sink_sink_fixate; + gstbasesink_class->propose_allocation = gst_pipewire_sink_propose_allocation; + gstbasesink_class->start = gst_pipewire_sink_start; + gstbasesink_class->stop = gst_pipewire_sink_stop; + gstbasesink_class->render = gst_pipewire_sink_render; + + GST_DEBUG_CATEGORY_INIT (pipewire_sink_debug, "pipewiresink", 0, + "PipeWire Sink"); +} + +static void +pool_activated (GstPipeWirePool *pool, GstPipeWireSink *sink) +{ + GstStructure *config; + GstCaps *caps; + guint size; + guint min_buffers; + guint max_buffers; + const struct spa_pod *port_params[3]; + struct spa_pod_builder b = { NULL }; + uint8_t buffer[1024]; + struct spa_pod_frame f; + + config = gst_buffer_pool_get_config (GST_BUFFER_POOL (pool)); + 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, SPA_TYPE_OBJECT_ParamBuffers, SPA_PARAM_Buffers); + if (size == 0) + spa_pod_builder_add (&b, + SPA_PARAM_BUFFERS_size, SPA_POD_CHOICE_RANGE_Int(0, 0, INT32_MAX), + 0); + else + spa_pod_builder_add (&b, + SPA_PARAM_BUFFERS_size, SPA_POD_CHOICE_RANGE_Int(size, size, INT32_MAX), + 0); + + spa_pod_builder_add (&b, + SPA_PARAM_BUFFERS_stride, SPA_POD_CHOICE_RANGE_Int(0, 0, INT32_MAX), + SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int( + SPA_MAX(MIN_BUFFERS, min_buffers), + SPA_MAX(MIN_BUFFERS, min_buffers), + max_buffers ? max_buffers : INT32_MAX), + SPA_PARAM_BUFFERS_dataType, SPA_POD_CHOICE_FLAGS_Int( + (1<<SPA_DATA_MemFd) | + (1<<SPA_DATA_MemPtr)), + 0); + port_params[0] = spa_pod_builder_pop (&b, &f); + + port_params[1] = spa_pod_builder_add_object (&b, + SPA_TYPE_OBJECT_ParamMeta, SPA_PARAM_Meta, + SPA_PARAM_META_type, SPA_POD_Int(SPA_META_Header), + SPA_PARAM_META_size, SPA_POD_Int(sizeof (struct spa_meta_header))); + + port_params[2] = spa_pod_builder_add_object (&b, + SPA_TYPE_OBJECT_ParamMeta, SPA_PARAM_Meta, + SPA_PARAM_META_type, SPA_POD_Int(SPA_META_VideoCrop), + SPA_PARAM_META_size, SPA_POD_Int(sizeof (struct spa_meta_region))); + + pw_thread_loop_lock (sink->core->loop); + pw_stream_update_params (sink->stream, port_params, 3); + pw_thread_loop_unlock (sink->core->loop); +} + +static void +gst_pipewire_sink_init (GstPipeWireSink * sink) +{ + sink->pool = gst_pipewire_pool_new (); + sink->client_name = g_strdup(pw_get_client_name()); + sink->mode = DEFAULT_PROP_MODE; + sink->fd = -1; + + g_signal_connect (sink->pool, "activated", G_CALLBACK (pool_activated), sink); +} + +static GstCaps * +gst_pipewire_sink_sink_fixate (GstBaseSink * bsink, GstCaps * caps) +{ + GstStructure *structure; + + caps = gst_caps_make_writable (caps); + + structure = gst_caps_get_structure (caps, 0); + + if (gst_structure_has_name (structure, "video/x-raw")) { + gst_structure_fixate_field_nearest_int (structure, "width", 320); + gst_structure_fixate_field_nearest_int (structure, "height", 240); + gst_structure_fixate_field_nearest_fraction (structure, "framerate", 30, 1); + + if (gst_structure_has_field (structure, "pixel-aspect-ratio")) + gst_structure_fixate_field_nearest_fraction (structure, + "pixel-aspect-ratio", 1, 1); + else + gst_structure_set (structure, "pixel-aspect-ratio", GST_TYPE_FRACTION, 1, 1, + NULL); + + if (gst_structure_has_field (structure, "colorimetry")) + gst_structure_fixate_field_string (structure, "colorimetry", "bt601"); + if (gst_structure_has_field (structure, "chroma-site")) + gst_structure_fixate_field_string (structure, "chroma-site", "mpeg2"); + + if (gst_structure_has_field (structure, "interlace-mode")) + gst_structure_fixate_field_string (structure, "interlace-mode", + "progressive"); + else + gst_structure_set (structure, "interlace-mode", G_TYPE_STRING, + "progressive", NULL); + } else if (gst_structure_has_name (structure, "audio/x-raw")) { + gst_structure_fixate_field_string (structure, "format", "S16LE"); + gst_structure_fixate_field_nearest_int (structure, "channels", 2); + gst_structure_fixate_field_nearest_int (structure, "rate", 44100); + } else if (gst_structure_has_name (structure, "audio/mpeg")) { + gst_structure_fixate_field_string (structure, "format", "Encoded"); + gst_structure_fixate_field_nearest_int (structure, "channels", 2); + gst_structure_fixate_field_nearest_int (structure, "rate", 44100); + } else if (gst_structure_has_name (structure, "audio/x-flac")) { + gst_structure_fixate_field_string (structure, "format", "Encoded"); + gst_structure_fixate_field_nearest_int (structure, "channels", 2); + gst_structure_fixate_field_nearest_int (structure, "rate", 44100); + } + + caps = GST_BASE_SINK_CLASS (parent_class)->fixate (bsink, caps); + + return caps; +} + +static void +gst_pipewire_sink_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec) +{ + GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (object); + + switch (prop_id) { + case PROP_PATH: + g_free (pwsink->path); + pwsink->path = g_value_dup_string (value); + break; + + case PROP_TARGET_OBJECT: + g_free (pwsink->target_object); + pwsink->target_object = g_value_dup_string (value); + break; + + case PROP_CLIENT_NAME: + g_free (pwsink->client_name); + pwsink->client_name = g_value_dup_string (value); + break; + + case PROP_CLIENT_PROPERTIES: + if (pwsink->client_properties) + gst_structure_free (pwsink->client_properties); + pwsink->client_properties = + gst_structure_copy (gst_value_get_structure (value)); + break; + + case PROP_STREAM_PROPERTIES: + if (pwsink->stream_properties) + gst_structure_free (pwsink->stream_properties); + pwsink->stream_properties = + gst_structure_copy (gst_value_get_structure (value)); + break; + + case PROP_MODE: + pwsink->mode = g_value_get_enum (value); + break; + + case PROP_FD: + pwsink->fd = g_value_get_int (value); + break; + + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static void +gst_pipewire_sink_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec) +{ + GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (object); + + switch (prop_id) { + case PROP_PATH: + g_value_set_string (value, pwsink->path); + break; + + case PROP_TARGET_OBJECT: + g_value_set_string (value, pwsink->target_object); + break; + + case PROP_CLIENT_NAME: + g_value_set_string (value, pwsink->client_name); + break; + + case PROP_CLIENT_PROPERTIES: + gst_value_set_structure (value, pwsink->client_properties); + break; + + case PROP_STREAM_PROPERTIES: + gst_value_set_structure (value, pwsink->stream_properties); + break; + + case PROP_MODE: + g_value_set_enum (value, pwsink->mode); + break; + + case PROP_FD: + g_value_set_int (value, pwsink->fd); + break; + + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static void +on_add_buffer (void *_data, struct pw_buffer *b) +{ + GstPipeWireSink *pwsink = _data; + gst_pipewire_pool_wrap_buffer (pwsink->pool, b); +} + +static void +on_remove_buffer (void *_data, struct pw_buffer *b) +{ + GstPipeWireSink *pwsink = _data; + GstPipeWirePoolData *data = b->user_data; + + GST_LOG_OBJECT (pwsink, "remove buffer"); + + gst_buffer_unref (data->buf); +} + +static void +do_send_buffer (GstPipeWireSink *pwsink, GstBuffer *buffer) +{ + GstPipeWirePoolData *data; + gboolean res; + guint i; + struct spa_buffer *b; + + data = gst_pipewire_pool_get_data(buffer); + + b = data->b->buffer; + + if (data->header) { + data->header->seq = GST_BUFFER_OFFSET (buffer); + data->header->pts = GST_BUFFER_PTS (buffer); + data->header->dts_offset = GST_BUFFER_DTS (buffer); + } + if (data->crop) { + GstVideoCropMeta *meta = gst_buffer_get_video_crop_meta (buffer); + if (meta) { + data->crop->region.position.x = meta->x; + data->crop->region.position.y = meta->y; + data->crop->region.size.width = meta->width; + data->crop->region.size.height = meta->width; + } + } + for (i = 0; i < b->n_datas; i++) { + struct spa_data *d = &b->datas[i]; + GstMemory *mem = gst_buffer_peek_memory (buffer, i); + d->chunk->offset = mem->offset; + d->chunk->size = mem->size; + d->chunk->stride = 0; + } + + if ((res = pw_stream_queue_buffer (pwsink->stream, data->b)) < 0) { + g_warning ("can't send buffer %s", spa_strerror(res)); + } +} + + +static void +on_process (void *data) +{ + GstPipeWireSink *pwsink = data; + GST_DEBUG ("signal"); + g_cond_signal (&pwsink->pool->cond); +} + +static void +on_state_changed (void *data, enum pw_stream_state old, enum pw_stream_state state, const char *error) +{ + GstPipeWireSink *pwsink = data; + + GST_DEBUG ("got stream state %d", state); + + switch (state) { + case PW_STREAM_STATE_UNCONNECTED: + case PW_STREAM_STATE_CONNECTING: + case PW_STREAM_STATE_PAUSED: + break; + case PW_STREAM_STATE_STREAMING: + if (pw_stream_is_driving (pwsink->stream)) + pw_stream_trigger_process (pwsink->stream); + break; + case PW_STREAM_STATE_ERROR: + GST_ELEMENT_ERROR (pwsink, RESOURCE, FAILED, + ("stream error: %s", error), (NULL)); + break; + } + pw_thread_loop_signal (pwsink->core->loop, FALSE); +} + +static void +on_param_changed (void *data, uint32_t id, const struct spa_pod *param) +{ + GstPipeWireSink *pwsink = data; + + if (param == NULL || id != SPA_PARAM_Format) + return; + + if (gst_buffer_pool_is_active (GST_BUFFER_POOL_CAST (pwsink->pool))) + pool_activated (pwsink->pool, pwsink); +} + +static gboolean +gst_pipewire_sink_setcaps (GstBaseSink * bsink, GstCaps * caps) +{ + GstPipeWireSink *pwsink; + GPtrArray *possible; + enum pw_stream_state state; + const char *error = NULL; + gboolean res = FALSE; + GstStructure *config; + guint size; + guint min_buffers; + guint max_buffers; + struct timespec abstime; + + pwsink = GST_PIPEWIRE_SINK (bsink); + + possible = gst_caps_to_format_all (caps, SPA_PARAM_EnumFormat); + + pw_thread_loop_lock (pwsink->core->loop); + state = pw_stream_get_state (pwsink->stream, &error); + + if (state == PW_STREAM_STATE_ERROR) + goto start_error; + + if (state == PW_STREAM_STATE_UNCONNECTED) { + enum pw_stream_flags flags = 0; + uint32_t target_id; + + if (pwsink->mode != GST_PIPEWIRE_SINK_MODE_PROVIDE) + flags |= PW_STREAM_FLAG_AUTOCONNECT; + else + flags |= PW_STREAM_FLAG_DRIVER; + + target_id = pwsink->path ? (uint32_t)atoi(pwsink->path) : PW_ID_ANY; + + if (pwsink->target_object) { + struct spa_dict_item items[2] = { + SPA_DICT_ITEM_INIT(PW_KEY_TARGET_OBJECT, pwsink->target_object), + /* XXX deprecated but the portal and some example apps only + * provide the object id */ + SPA_DICT_ITEM_INIT(PW_KEY_NODE_TARGET, NULL), + }; + struct spa_dict dict = SPA_DICT_INIT_ARRAY(items); + uint64_t serial; + + /* If target.object is a name, set it also to node.target */ + if (spa_atou64(pwsink->target_object, &serial, 0)) { + dict.n_items = 1; + } else { + target_id = PW_ID_ANY; + items[1].value = pwsink->target_object; + } + + pw_stream_update_properties (pwsink->stream, &dict); + } + + pw_stream_connect (pwsink->stream, + PW_DIRECTION_OUTPUT, + target_id, + flags, + (const struct spa_pod **) possible->pdata, + possible->len); + + pw_thread_loop_get_time (pwsink->core->loop, &abstime, + GST_PIPEWIRE_DEFAULT_TIMEOUT * SPA_NSEC_PER_SEC); + + while (TRUE) { + state = pw_stream_get_state (pwsink->stream, &error); + + if (state >= PW_STREAM_STATE_PAUSED) + break; + + if (state == PW_STREAM_STATE_ERROR) + goto start_error; + + if (pw_thread_loop_timed_wait_full (pwsink->core->loop, &abstime) < 0) { + error = "timeout"; + goto start_error; + } + } + } + res = TRUE; + + config = gst_buffer_pool_get_config (GST_BUFFER_POOL_CAST (pwsink->pool)); + gst_buffer_pool_config_get_params (config, NULL, &size, &min_buffers, &max_buffers); + gst_buffer_pool_config_set_params (config, caps, size, min_buffers, max_buffers); + gst_buffer_pool_set_config (GST_BUFFER_POOL_CAST (pwsink->pool), config); + + pw_thread_loop_unlock (pwsink->core->loop); + + pwsink->negotiated = res; + + return res; + +start_error: + { + GST_ERROR ("could not start stream: %s", error); + pw_thread_loop_unlock (pwsink->core->loop); + g_ptr_array_unref (possible); + return FALSE; + } +} + +static GstFlowReturn +gst_pipewire_sink_render (GstBaseSink * bsink, GstBuffer * buffer) +{ + GstPipeWireSink *pwsink; + GstFlowReturn res = GST_FLOW_OK; + const char *error = NULL; + gboolean unref_buffer = FALSE; + + pwsink = GST_PIPEWIRE_SINK (bsink); + + if (!pwsink->negotiated) + goto not_negotiated; + + if (buffer->pool != GST_BUFFER_POOL_CAST (pwsink->pool) && + !gst_buffer_pool_is_active (GST_BUFFER_POOL_CAST (pwsink->pool))) { + GstStructure *config; + GstCaps *caps; + guint size, min_buffers, max_buffers; + + config = gst_buffer_pool_get_config (GST_BUFFER_POOL_CAST (pwsink->pool)); + gst_buffer_pool_config_get_params (config, &caps, &size, &min_buffers, &max_buffers); + + size = (size == 0) ? gst_buffer_get_size (buffer) : size; + + gst_buffer_pool_config_set_params (config, caps, size, min_buffers, max_buffers); + gst_buffer_pool_set_config (GST_BUFFER_POOL_CAST (pwsink->pool), config); + + gst_buffer_pool_set_active (GST_BUFFER_POOL_CAST (pwsink->pool), TRUE); + } + + pw_thread_loop_lock (pwsink->core->loop); + if (pw_stream_get_state (pwsink->stream, &error) != PW_STREAM_STATE_STREAMING) + goto done_unlock; + + if (buffer->pool != GST_BUFFER_POOL_CAST (pwsink->pool)) { + GstBuffer *b = NULL; + GstMapInfo info = { 0, }; + GstBufferPoolAcquireParams params = { 0, }; + + pw_thread_loop_unlock (pwsink->core->loop); + + if ((res = gst_buffer_pool_acquire_buffer (GST_BUFFER_POOL_CAST (pwsink->pool), &b, ¶ms)) != GST_FLOW_OK) + goto done; + + gst_buffer_map (b, &info, GST_MAP_WRITE); + gst_buffer_extract (buffer, 0, info.data, info.maxsize); + gst_buffer_unmap (b, &info); + gst_buffer_resize (b, 0, gst_buffer_get_size (buffer)); + buffer = b; + unref_buffer = TRUE; + + pw_thread_loop_lock (pwsink->core->loop); + if (pw_stream_get_state (pwsink->stream, &error) != PW_STREAM_STATE_STREAMING) + goto done_unlock; + } + + GST_DEBUG ("push buffer"); + do_send_buffer (pwsink, buffer); + if (unref_buffer) + gst_buffer_unref (buffer); + + if (pw_stream_is_driving (pwsink->stream)) + pw_stream_trigger_process (pwsink->stream); + +done_unlock: + pw_thread_loop_unlock (pwsink->core->loop); +done: + return res; + +not_negotiated: + { + return GST_FLOW_NOT_NEGOTIATED; + } +} + +static gboolean +copy_properties (GQuark field_id, + const GValue *value, + gpointer user_data) +{ + struct pw_properties *properties = user_data; + GValue dst = { 0 }; + + if (g_value_type_transformable (G_VALUE_TYPE(value), G_TYPE_STRING)) { + g_value_init(&dst, G_TYPE_STRING); + if (g_value_transform(value, &dst)) { + pw_properties_set (properties, + g_quark_to_string (field_id), + g_value_get_string (&dst)); + } + g_value_unset(&dst); + } + return TRUE; +} + +static const struct pw_stream_events stream_events = { + PW_VERSION_STREAM_EVENTS, + .state_changed = on_state_changed, + .param_changed = on_param_changed, + .add_buffer = on_add_buffer, + .remove_buffer = on_remove_buffer, + .process = on_process, +}; + +static gboolean +gst_pipewire_sink_start (GstBaseSink * basesink) +{ + GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (basesink); + struct pw_properties *props; + + pwsink->negotiated = FALSE; + + pw_thread_loop_lock (pwsink->core->loop); + + props = pw_properties_new (NULL, NULL); + if (pwsink->client_name) { + pw_properties_set (props, PW_KEY_NODE_NAME, pwsink->client_name); + pw_properties_set (props, PW_KEY_NODE_DESCRIPTION, pwsink->client_name); + } + if (pwsink->stream_properties) { + gst_structure_foreach (pwsink->stream_properties, copy_properties, props); + } + + if ((pwsink->stream = pw_stream_new (pwsink->core->core, pwsink->client_name, props)) == NULL) + goto no_stream; + + pwsink->pool->stream = pwsink->stream; + + pw_stream_add_listener(pwsink->stream, + &pwsink->stream_listener, + &stream_events, + pwsink); + + pw_thread_loop_unlock (pwsink->core->loop); + + return TRUE; + +no_stream: + { + GST_ELEMENT_ERROR (pwsink, RESOURCE, FAILED, ("can't create stream"), (NULL)); + pw_thread_loop_unlock (pwsink->core->loop); + return FALSE; + } +} + +static gboolean +gst_pipewire_sink_stop (GstBaseSink * basesink) +{ + GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (basesink); + + pw_thread_loop_lock (pwsink->core->loop); + if (pwsink->stream) { + pw_stream_destroy (pwsink->stream); + pwsink->stream = NULL; + pwsink->pool->stream = NULL; + } + pw_thread_loop_unlock (pwsink->core->loop); + + pwsink->negotiated = FALSE; + + return TRUE; +} + +static gboolean +gst_pipewire_sink_open (GstPipeWireSink * pwsink) +{ + struct pw_properties *props; + + GST_DEBUG_OBJECT (pwsink, "open"); + + pwsink->core = gst_pipewire_core_get(pwsink->fd); + if (pwsink->core == NULL) + goto connect_error; + + pw_thread_loop_lock (pwsink->core->loop); + + props = pw_properties_new (NULL, NULL); + if (pwsink->client_properties) { + gst_structure_foreach (pwsink->client_properties, copy_properties, props); + pw_core_update_properties (pwsink->core->core, &props->dict); + } + pw_properties_free(props); + pw_thread_loop_unlock (pwsink->core->loop); + + return TRUE; + + /* ERRORS */ +connect_error: + { + GST_ELEMENT_ERROR (pwsink, RESOURCE, FAILED, + ("Failed to connect"), (NULL)); + return FALSE; + } +} + +static gboolean +gst_pipewire_sink_close (GstPipeWireSink * pwsink) +{ + pw_thread_loop_lock (pwsink->core->loop); + if (pwsink->stream) { + pw_stream_destroy (pwsink->stream); + pwsink->stream = NULL; + } + pw_thread_loop_unlock (pwsink->core->loop); + + if (pwsink->core) { + gst_pipewire_core_release (pwsink->core); + pwsink->core = NULL; + } + return TRUE; +} + +static GstStateChangeReturn +gst_pipewire_sink_change_state (GstElement * element, GstStateChange transition) +{ + GstStateChangeReturn ret; + GstPipeWireSink *this = GST_PIPEWIRE_SINK_CAST (element); + + switch (transition) { + case GST_STATE_CHANGE_NULL_TO_READY: + if (!gst_pipewire_sink_open (this)) + goto open_failed; + break; + case GST_STATE_CHANGE_READY_TO_PAUSED: + break; + case GST_STATE_CHANGE_PAUSED_TO_PLAYING: + /* uncork and start play */ + pw_thread_loop_lock (this->core->loop); + pw_stream_set_active(this->stream, true); + pw_thread_loop_unlock (this->core->loop); + gst_buffer_pool_set_flushing(GST_BUFFER_POOL_CAST(this->pool), FALSE); + break; + case GST_STATE_CHANGE_PLAYING_TO_PAUSED: + /* stop play ASAP by corking */ + pw_thread_loop_lock (this->core->loop); + pw_stream_set_active(this->stream, false); + pw_thread_loop_unlock (this->core->loop); + gst_buffer_pool_set_flushing(GST_BUFFER_POOL_CAST(this->pool), TRUE); + break; + default: + break; + } + + ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition); + + switch (transition) { + case GST_STATE_CHANGE_PLAYING_TO_PAUSED: + break; + case GST_STATE_CHANGE_PAUSED_TO_READY: + gst_buffer_pool_set_active(GST_BUFFER_POOL_CAST(this->pool), FALSE); + break; + case GST_STATE_CHANGE_READY_TO_NULL: + gst_pipewire_sink_close (this); + break; + default: + break; + } + return ret; + + /* ERRORS */ +open_failed: + { + return GST_STATE_CHANGE_FAILURE; + } +} |