winegstreamer: Implement wg_muxer_finalize.

dlls/winegstreamer/wg_muxer.c

@@ -63,6 +63,9 @@ struct wg_muxer
     GstBuffer *buffer;
 
     pthread_mutex_t mutex;
+    pthread_cond_t cond;
+
+    bool eos;
     guint64 offset; /* Write offset of the output buffer generated by muxer. */
 
     struct list streams;
@@ -175,6 +178,13 @@ static gboolean muxer_sink_event_cb(GstPad *pad, GstObject *parent, GstEvent *ev
 
     switch (event->type)
     {
+        case GST_EVENT_EOS:
+            pthread_mutex_lock(&muxer->mutex);
+            muxer->eos = true;
+            pthread_mutex_unlock(&muxer->mutex);
+            pthread_cond_signal(&muxer->cond);
+            break;
+
         case GST_EVENT_SEGMENT:
             pthread_mutex_lock(&muxer->mutex);
 
@@ -248,6 +258,7 @@ NTSTATUS wg_muxer_create(void *args)
     list_init(&muxer->streams);
     muxer->offset = GST_BUFFER_OFFSET_NONE;
     pthread_mutex_init(&muxer->mutex, NULL);
+    pthread_cond_init(&muxer->cond, NULL);
     if (!(muxer->container = gst_bin_new("wg_muxer")))
         goto out;
     if (!(muxer->output_queue = gst_atomic_queue_new(8)))
@@ -287,6 +298,7 @@ out:
         gst_atomic_queue_unref(muxer->output_queue);
     if (muxer->container)
         gst_object_unref(muxer->container);
+    pthread_cond_destroy(&muxer->cond);
     pthread_mutex_destroy(&muxer->mutex);
     free(muxer);
 
@@ -317,6 +329,7 @@ NTSTATUS wg_muxer_destroy(void *args)
     gst_element_set_state(muxer->container, GST_STATE_NULL);
     gst_object_unref(muxer->container);
 
+    pthread_cond_destroy(&muxer->cond);
     pthread_mutex_destroy(&muxer->mutex);
 
     free(muxer);
@@ -526,6 +539,22 @@ NTSTATUS wg_muxer_read_data(void *args)
 
 NTSTATUS wg_muxer_finalize(void *args)
 {
-    GST_FIXME("Not implemented.");
-    return STATUS_NOT_IMPLEMENTED;
+    struct wg_muxer *muxer = get_muxer(*(wg_muxer_t *)args);
+    struct wg_muxer_stream *stream;
+
+    /* Notify each stream of EOS. */
+    LIST_FOR_EACH_ENTRY(stream, &muxer->streams, struct wg_muxer_stream, entry)
+    {
+        if (!push_event(stream->my_src, gst_event_new_segment_done(GST_FORMAT_BYTES, -1))
+                || !push_event(stream->my_src, gst_event_new_eos()))
+            return STATUS_UNSUCCESSFUL;
+    }
+
+    /* Wait for muxer EOS. */
+    pthread_mutex_lock(&muxer->mutex);
+    while (!muxer->eos)
+        pthread_cond_wait(&muxer->cond, &muxer->mutex);
+    pthread_mutex_unlock(&muxer->mutex);
+
+    return STATUS_SUCCESS;
 }