[PATCH v3 6/6] winegstreamer: Implement wg_muxer_push_sample.

[Ziqing Hui]

From: Ziqing Hui

---
 dlls/winegstreamer/wg_muxer.c | 105 +++++++++++++++++++++++++++++++++-
 1 file changed, 104 insertions(+), 1 deletion(-)

diff --git a/dlls/winegstreamer/wg_muxer.c b/dlls/winegstreamer/wg_muxer.c
index 1a6c2bd05c8..ef5ce898b82 100644
--- a/dlls/winegstreamer/wg_muxer.c
+++ b/dlls/winegstreamer/wg_muxer.c
@@ -42,6 +42,9 @@ struct wg_muxer
     GstPad *my_sink;
     GstCaps *my_sink_caps;

+    pthread_mutex_t mutex;
+    guint64 offset;
+
     struct list streams;
 };

@@ -183,6 +186,19 @@ static bool stream_link(struct wg_muxer_stream *stream)
     return true;
 }

+static struct wg_muxer_stream *muxer_get_stream_by_id(struct wg_muxer *muxer, DWORD id)
+{
+    struct wg_muxer_stream *stream;
+
+    LIST_FOR_EACH_ENTRY(stream, &muxer->streams, struct wg_muxer_stream, entry)
+    {
+        if (stream->id == id)
+            return stream;
+    }
+
+    return NULL;
+}
+
 static bool muxer_try(struct wg_muxer *muxer, GstElementFactory *muxer_factory, GstRank min_rank)
 {
     char *element_name, *plugin_name = gst_plugin_feature_get_name(GST_PLUGIN_FEATURE(muxer_factory));
@@ -291,6 +307,46 @@ static gboolean muxer_sink_query_cb(GstPad *pad, GstObject *parent, GstQuery *qu
     }
 }

+static gboolean muxer_sink_event_cb(GstPad *pad, GstObject *parent, GstEvent *event)
+{
+    struct wg_muxer *muxer = gst_pad_get_element_private(pad);
+    const GstSegment *segment;
+
+    GST_DEBUG("pad %p, parent %p, event %p, muxer %p, type "%s".",
+            pad, parent, event, muxer, GST_EVENT_TYPE_NAME(event));
+
+    switch (event->type)
+    {
+        case GST_EVENT_SEGMENT:
+            pthread_mutex_lock(&muxer->mutex);
+
+            gst_event_parse_segment(event, &segment);
+            if (segment->format != GST_FORMAT_BYTES)
+            {
+                pthread_mutex_unlock(&muxer->mutex);
+                GST_FIXME("Unhandled segment format "%s".", gst_format_get_name(segment->format));
+                break;
+            }
+            muxer->offset = segment->start;
+
+            pthread_mutex_unlock(&muxer->mutex);
+            break;
+
+        default:
+            GST_WARNING("Ignoring "%s" event.", GST_EVENT_TYPE_NAME(event));
+            break;
+    }
+
+    gst_event_unref(event);
+    return TRUE;
+}
+
+static GstFlowReturn muxer_sink_chain_cb(GstPad *pad, GstObject *parent, GstBuffer *buffer)
+{
+    GST_FIXME("Stub.");
+    return GST_FLOW_ERROR;
+}
+
 NTSTATUS wg_muxer_create(void *args)
 {
     struct wg_muxer_create_params *params = args;
@@ -302,6 +358,8 @@ NTSTATUS wg_muxer_create(void *args)
     if (!(muxer = calloc(1, sizeof(*muxer))))
         return STATUS_NO_MEMORY;
     list_init(&muxer->streams);
+    muxer->offset = GST_BUFFER_OFFSET_NONE;
+    pthread_mutex_init(&muxer->mutex, NULL);
     if (!(muxer->container = gst_bin_new("wg_muxer")))
         goto out;

@@ -318,6 +376,8 @@ NTSTATUS wg_muxer_create(void *args)
         goto out;
     gst_pad_set_element_private(muxer->my_sink, muxer);
     gst_pad_set_query_function(muxer->my_sink, muxer_sink_query_cb);
+    gst_pad_set_event_function(muxer->my_sink, muxer_sink_event_cb);
+    gst_pad_set_chain_function(muxer->my_sink, muxer_sink_chain_cb);

gst_object_unref(template);

@@ -335,6 +395,7 @@ out:
         gst_caps_unref(muxer->my_sink_caps);
     if (muxer->container)
         gst_object_unref(muxer->container);
+    pthread_mutex_destroy(&muxer->mutex);
     free(muxer);

return status;
@@ -352,10 +413,14 @@ NTSTATUS wg_muxer_destroy(void *args)
         gst_caps_unref(stream->my_src_caps);
         free(stream);
     }
+
     gst_object_unref(muxer->my_sink);
     gst_caps_unref(muxer->my_sink_caps);
     gst_element_set_state(muxer->container, GST_STATE_NULL);
     gst_object_unref(muxer->container);
+
+    pthread_mutex_destroy(&muxer->mutex);
+
     free(muxer);

return S_OK;
@@ -463,5 +528,43 @@ NTSTATUS wg_muxer_start(void *args)

NTSTATUS wg_muxer_push_sample(void *args)
 {
-    return STATUS_NOT_IMPLEMENTED;
+    struct wg_muxer_push_sample_params *params = args;
+    struct wg_muxer *muxer = get_muxer(params->muxer);
+    struct wg_sample *sample = params->sample;
+    struct wg_muxer_stream *stream;
+    GstFlowReturn ret;
+    GstBuffer *buffer;
+
+    if (!(stream = muxer_get_stream_by_id(muxer, params->stream_id)))
+        return STATUS_NOT_FOUND;
+
+    /* Create sample data buffer. */
+    if (!(buffer = gst_buffer_new_and_alloc(sample->size))
+            || !gst_buffer_fill(buffer, 0, wg_sample_data(sample), sample->size))
+    {
+        GST_ERROR("Failed to allocate input buffer.");
+        return STATUS_NO_MEMORY;
+    }
+
+    GST_INFO("Copied %u bytes from sample %p to buffer %p.", sample->size, sample, buffer);
+
+    /* Set sample properties. */
+    if (sample->flags & WG_SAMPLE_FLAG_HAS_PTS)
+        GST_BUFFER_PTS(buffer) = sample->pts * 100;
+    if (sample->flags & WG_SAMPLE_FLAG_HAS_DURATION)
+        GST_BUFFER_DURATION(buffer) = sample->duration * 100;
+    if (!(sample->flags & WG_SAMPLE_FLAG_SYNC_POINT))
+        GST_BUFFER_FLAG_SET(buffer, GST_BUFFER_FLAG_DELTA_UNIT);
+    if (sample->flags & WG_SAMPLE_FLAG_DISCONTINUITY)
+        GST_BUFFER_FLAG_SET(buffer, GST_BUFFER_FLAG_DISCONT);
+
+    /* Push sample data buffer to stream src pad. */
+    if ((ret = gst_pad_push(stream->my_src, buffer)) < 0)
+    {
+        GST_ERROR("Failed to push buffer %p to pad %s, reason %s.",
+                buffer, gst_pad_get_name(stream->my_src), gst_flow_get_name(ret));
+        return STATUS_UNSUCCESSFUL;
+    }
+
+    return STATUS_SUCCESS;
 }