From: Ziqing Hui zhui@codeweavers.com
--- dlls/winegstreamer/gst_private.h | 1 + dlls/winegstreamer/main.c | 13 +++ dlls/winegstreamer/unix_private.h | 1 + dlls/winegstreamer/unixlib.h | 8 ++ dlls/winegstreamer/wg_muxer.c | 148 ++++++++++++++++++++++++++++++ dlls/winegstreamer/wg_parser.c | 17 ++++ 6 files changed, 188 insertions(+)
diff --git a/dlls/winegstreamer/gst_private.h b/dlls/winegstreamer/gst_private.h index 23b59766d0c..ac573515444 100644 --- a/dlls/winegstreamer/gst_private.h +++ b/dlls/winegstreamer/gst_private.h @@ -113,6 +113,7 @@ HRESULT wg_transform_flush(wg_transform_t transform); HRESULT wg_muxer_create(const char *format, wg_muxer_t *muxer); void wg_muxer_destroy(wg_muxer_t muxer); HRESULT wg_muxer_add_stream(wg_muxer_t muxer, UINT32 stream_id, const struct wg_format *format); +HRESULT wg_muxer_push_sample(wg_muxer_t muxer, struct wg_sample *sample);
unsigned int wg_format_get_max_size(const struct wg_format *format);
diff --git a/dlls/winegstreamer/main.c b/dlls/winegstreamer/main.c index 9dca66100c9..d984ed05bb0 100644 --- a/dlls/winegstreamer/main.c +++ b/dlls/winegstreamer/main.c @@ -496,6 +496,19 @@ HRESULT wg_muxer_add_stream(wg_muxer_t muxer, UINT32 stream_id, const struct wg_ return WINE_UNIX_CALL(unix_wg_muxer_add_stream, ¶ms); }
+HRESULT wg_muxer_push_sample(wg_muxer_t muxer, struct wg_sample *sample) +{ + struct wg_muxer_push_sample_params params = + { + .muxer = muxer, + .sample = sample, + }; + + TRACE("muxer %#I64x, sample %p.\n", muxer, sample); + + return WINE_UNIX_CALL(unix_wg_muxer_push_sample, ¶ms); +} + #define ALIGN(n, alignment) (((n) + (alignment) - 1) & ~((alignment) - 1))
unsigned int wg_format_get_stride(const struct wg_format *format) diff --git a/dlls/winegstreamer/unix_private.h b/dlls/winegstreamer/unix_private.h index 9627845b402..0e9a67471c9 100644 --- a/dlls/winegstreamer/unix_private.h +++ b/dlls/winegstreamer/unix_private.h @@ -63,6 +63,7 @@ extern NTSTATUS wg_transform_flush(void *args) DECLSPEC_HIDDEN; extern NTSTATUS wg_muxer_create(void *args) DECLSPEC_HIDDEN; extern NTSTATUS wg_muxer_destroy(void *args) DECLSPEC_HIDDEN; extern NTSTATUS wg_muxer_add_stream(void *args) DECLSPEC_HIDDEN; +extern NTSTATUS wg_muxer_push_sample(void *args) DECLSPEC_HIDDEN;
/* wg_allocator.c */
diff --git a/dlls/winegstreamer/unixlib.h b/dlls/winegstreamer/unixlib.h index 3b320448510..6cbcab81c40 100644 --- a/dlls/winegstreamer/unixlib.h +++ b/dlls/winegstreamer/unixlib.h @@ -185,6 +185,7 @@ struct wg_sample UINT32 max_size; UINT32 size; UINT64 data; /* pointer to user memory */ + UINT32 stream_id; };
struct wg_parser_buffer @@ -379,6 +380,12 @@ struct wg_muxer_add_stream_params const struct wg_format *format; };
+struct wg_muxer_push_sample_params +{ + wg_muxer_t muxer; + struct wg_sample *sample; +}; + enum unix_funcs { unix_wg_init_gstreamer, @@ -422,6 +429,7 @@ enum unix_funcs unix_wg_muxer_create, unix_wg_muxer_destroy, unix_wg_muxer_add_stream, + unix_wg_muxer_push_sample, };
#endif /* __WINE_WINEGSTREAMER_UNIXLIB_H */ diff --git a/dlls/winegstreamer/wg_muxer.c b/dlls/winegstreamer/wg_muxer.c index d33c978fe97..b1d8978767b 100644 --- a/dlls/winegstreamer/wg_muxer.c +++ b/dlls/winegstreamer/wg_muxer.c @@ -27,6 +27,7 @@ #include "ntstatus.h" #define WIN32_NO_STATUS #include "winternl.h" +#include "mferror.h"
#include "unix_private.h"
@@ -35,7 +36,13 @@ struct wg_muxer { GstElement *container, *muxer; + GstPad *my_sink; + GstAtomicQueue *output_queue; + + pthread_mutex_t mutex; + guint64 offset; + struct list streams; };
@@ -56,6 +63,26 @@ static struct wg_muxer *get_muxer(wg_muxer_t muxer) return (struct wg_muxer *)(ULONG_PTR)muxer; }
+static void sample_free_notify(void *arg) +{ + struct wg_sample *sample = arg; + GST_DEBUG("Releasing wg_sample %p", sample); + InterlockedDecrement(&sample->refcount); +} + +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 gboolean muxer_src_query_cb(GstPad *pad, GstObject *parent, GstQuery *query) { struct wg_muxer_stream *stream = gst_pad_get_element_private(pad); @@ -91,6 +118,66 @@ 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) +{ + struct wg_muxer *muxer = gst_pad_get_element_private(pad); + + GST_DEBUG("pad %p, parent %p, buffer %p, muxer %p.", pad, parent, buffer, muxer); + + pthread_mutex_lock(&muxer->mutex); + + GST_BUFFER_OFFSET(buffer) = GST_BUFFER_OFFSET_NONE; + if (muxer->offset != GST_BUFFER_OFFSET_NONE) + { + GST_BUFFER_OFFSET(buffer) = muxer->offset; + muxer->offset = GST_BUFFER_OFFSET_NONE; + } + gst_atomic_queue_push(muxer->output_queue, buffer); + + GST_LOG("Pushed buffer %p to output queue %p, buffer size %" G_GSIZE_FORMAT ", " + "offset %" G_GUINT64_FORMAT ", %u buffers in queue now.", + buffer, muxer->output_queue, gst_buffer_get_size(buffer), GST_BUFFER_OFFSET(buffer), + gst_atomic_queue_length(muxer->output_queue)); + + pthread_mutex_unlock(&muxer->mutex); + + return GST_FLOW_OK; +} + NTSTATUS wg_muxer_create(void *args) { struct wg_muxer_create_params *params = args; @@ -104,8 +191,12 @@ NTSTATUS wg_muxer_create(void *args) if (!(muxer = calloc(1, sizeof(*muxer)))) return E_OUTOFMEMORY; 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; + if (!(muxer->output_queue = gst_atomic_queue_new(8))) + goto out;
/* Create sink pad. */ if (!(sink_caps = gst_caps_from_string(params->format))) @@ -120,6 +211,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);
/* Create gstreamer muxer element. */ if (!(muxer->muxer = find_element(GST_ELEMENT_FACTORY_TYPE_MUXER | GST_ELEMENT_FACTORY_TYPE_FORMATTER, @@ -154,11 +247,13 @@ out: gst_object_unref(template); if (sink_caps) gst_caps_unref(sink_caps); + gst_atomic_queue_unref(muxer->output_queue); if (muxer->container) { gst_element_set_state(muxer->container, GST_STATE_NULL); gst_object_unref(muxer->container); } + pthread_mutex_destroy(&muxer->mutex); free(muxer);
return status; @@ -168,6 +263,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) { @@ -176,9 +272,17 @@ NTSTATUS wg_muxer_destroy(void *args) gst_object_unref(stream->my_src); free(stream); } + + 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_element_set_state(muxer->container, GST_STATE_NULL); gst_object_unref(muxer->container); + + pthread_mutex_destroy(&muxer->mutex); + free(muxer);
return S_OK; @@ -283,3 +387,47 @@ out:
return status; } + +NTSTATUS wg_muxer_push_sample(void *args) +{ + 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, sample->stream_id))) + return MF_E_INVALIDSTREAMNUMBER; + + /* Create sample data buffer by wrapping wg_sample. */ + if (!(buffer = gst_buffer_new_wrapped_full(GST_MEMORY_FLAG_READONLY, wg_sample_data(sample), sample->max_size, + 0, sample->size, sample, sample_free_notify))) + { + GST_ERROR("Failed to allocate input buffer."); + return E_OUTOFMEMORY; + } + + InterlockedIncrement(&sample->refcount); + GST_INFO("Wrapped %u/%u bytes from sample %p to buffer %p.", sample->size, sample->max_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 E_FAIL; + } + + return S_OK; +} diff --git a/dlls/winegstreamer/wg_parser.c b/dlls/winegstreamer/wg_parser.c index 1880b052680..30b30b405b2 100644 --- a/dlls/winegstreamer/wg_parser.c +++ b/dlls/winegstreamer/wg_parser.c @@ -1949,6 +1949,7 @@ const unixlib_entry_t __wine_unix_call_funcs[] = X(wg_muxer_create), X(wg_muxer_destroy), X(wg_muxer_add_stream), + X(wg_muxer_push_sample), };
#ifdef _WIN64 @@ -2190,6 +2191,21 @@ NTSTATUS wow64_wg_muxer_add_stream(void *args) return wg_muxer_add_stream(¶ms); }
+NTSTATUS wow64_wg_muxer_push_sample(void *args) +{ + struct + { + wg_muxer_t muxer; + PTR32 sample; + } *params32 = args; + struct wg_muxer_push_sample_params params = + { + .muxer = params32->muxer, + .sample = ULongToPtr(params32->sample), + }; + return wg_muxer_push_sample(¶ms); +} + const unixlib_entry_t __wine_unix_call_wow64_funcs[] = { #define X64(name) [unix_ ## name] = wow64_ ## name @@ -2234,6 +2250,7 @@ const unixlib_entry_t __wine_unix_call_wow64_funcs[] = X64(wg_muxer_create), X(wg_muxer_destroy), X64(wg_muxer_add_stream), + X64(wg_muxer_push_sample), };
#endif /* _WIN64 */