[PATCH 2/2] winegstreamer: Implement wg_muxer_read_data.

[Ziqing Hui]

From: Ziqing Hui

---
 dlls/winegstreamer/wg_muxer.c | 88 +++++++++++++++++++++++++++++++++--
 1 file changed, 83 insertions(+), 5 deletions(-)

diff --git a/dlls/winegstreamer/wg_muxer.c b/dlls/winegstreamer/wg_muxer.c
index 0d25687a7d1..bf4e8090dcf 100644
--- a/dlls/winegstreamer/wg_muxer.c
+++ b/dlls/winegstreamer/wg_muxer.c
@@ -59,8 +59,13 @@ struct wg_muxer
     GstPad *my_sink;
     GstCaps *my_sink_caps;

+    GstAtomicQueue *output_queue;
+    GstBuffer *buffer;
+    GstMapInfo buffer_map;
+    size_t buffer_read;
+
     pthread_mutex_t mutex;
-    guint64 offset;
+    guint64 offset; /* Write offset of the output buffer generated by muxer. */

struct list streams;
 };
@@ -198,8 +203,31 @@ static gboolean muxer_sink_event_cb(GstPad *pad, GstObject *parent, GstEvent *ev

static GstFlowReturn muxer_sink_chain_cb(GstPad *pad, GstObject *parent, GstBuffer *buffer)
 {
-    GST_FIXME("Stub.");
-    return GST_FLOW_ERROR;
+    GstBuffer *buffer_writable= gst_buffer_make_writable(buffer);
+    struct wg_muxer *muxer = gst_pad_get_element_private(pad);
+
+    GST_DEBUG("muxer %p, pad %"GST_PTR_FORMAT", parent %"GST_PTR_FORMAT", buffer %"GST_PTR_FORMAT".",
+            muxer, pad, parent, buffer);
+
+    pthread_mutex_lock(&muxer->mutex);
+
+    GST_BUFFER_OFFSET(buffer_writable) = GST_BUFFER_OFFSET_NONE;
+    if (muxer->offset != GST_BUFFER_OFFSET_NONE)
+    {
+        GST_BUFFER_OFFSET(buffer_writable) = muxer->offset;
+        muxer->offset = GST_BUFFER_OFFSET_NONE;
+    }
+
+    gst_atomic_queue_push(muxer->output_queue, buffer_writable);
+
+    GST_DEBUG("Pushed buffer %"GST_PTR_FORMAT" to output queue %p,"
+            " buffer size %" G_GSIZE_FORMAT ", offset %" G_GUINT64_FORMAT ", %u buffers in queue now.",
+            buffer_writable, muxer->output_queue, gst_buffer_get_size(buffer_writable),
+            GST_BUFFER_OFFSET(buffer_writable), gst_atomic_queue_length(muxer->output_queue));
+
+    pthread_mutex_unlock(&muxer->mutex);
+
+    return GST_FLOW_OK;
 }

static void stream_free(struct wg_muxer_stream *stream)
@@ -269,6 +297,7 @@ NTSTATUS wg_muxer_destroy(void *args)
 {
     struct wg_muxer *muxer = get_muxer(*(wg_muxer_t *)args);
     struct wg_muxer_stream *stream, *next;
+    GstBuffer *buffer;

LIST_FOR_EACH_ENTRY_SAFE(stream, next, &muxer->streams, struct wg_muxer_stream, entry)
     {
@@ -276,6 +305,16 @@ NTSTATUS wg_muxer_destroy(void *args)
         stream_free(stream);
     }

+    if (muxer->buffer)
+    {
+        gst_buffer_unmap(muxer->buffer, &muxer->buffer_map);
+        gst_buffer_unref(muxer->buffer);
+    }
+
+    while ((buffer = gst_atomic_queue_pop(muxer->output_queue)))
+        gst_buffer_unref(buffer);
+    gst_atomic_queue_unref(muxer->output_queue);
+
     gst_object_unref(muxer->my_sink);
     gst_caps_unref(muxer->my_sink_caps);
     gst_element_set_state(muxer->container, GST_STATE_NULL);
@@ -454,6 +493,45 @@ NTSTATUS wg_muxer_push_sample(void *args)

NTSTATUS wg_muxer_read_data(void *args)
 {
-    GST_FIXME("Not implemented.");
-    return STATUS_NOT_IMPLEMENTED;
+    struct wg_muxer_read_data_params *params = args;
+    struct wg_muxer *muxer = get_muxer(params->muxer);
+    size_t size;
+
+    /* Pop buffer from output queue, and map it. */
+    if (!muxer->buffer)
+    {
+        if (!(muxer->buffer = gst_atomic_queue_pop(muxer->output_queue)))
+            return STATUS_NO_MEMORY;
+
+        if (!gst_buffer_map(muxer->buffer, &muxer->buffer_map, GST_MAP_READ))
+        {
+            GST_ERROR("Failed to map buffer %p, dropped.", muxer->buffer);
+            gst_buffer_unref(muxer->buffer);
+            muxer->buffer = NULL;
+            return STATUS_UNSUCCESSFUL;
+        }
+        muxer->buffer_read = 0;
+    }
+
+    /* Copy data. */
+    size = min(muxer->buffer_map.size - muxer->buffer_read, params->size);
+    memcpy(params->buffer, muxer->buffer_map.data + muxer->buffer_read, size);
+
+    /* Set offset and size.
+     * We may continuously read data from a same buffer multiple times.
+     * But we only need to set the offset at the first reading. */
+    params->size = size;
+    if (GST_BUFFER_OFFSET_IS_VALID(muxer->buffer) && !muxer->buffer_read)
+        params->offset = GST_BUFFER_OFFSET(muxer->buffer);
+
+    /* Unmap buffer if all data is read. */
+    muxer->buffer_read += size;
+    if (muxer->buffer_read == muxer->buffer_map.size)
+    {
+        gst_buffer_unmap(muxer->buffer, &muxer->buffer_map);
+        gst_buffer_unref(muxer->buffer);
+        muxer->buffer = NULL;
+    }
+
+    return STATUS_SUCCESS;
 }