From e9e6b0797dd6d970c2e0abd94d31992f958d717a Mon Sep 17 00:00:00 2001 From: Steve Baker Date: Sat, 8 Jun 2002 16:13:18 +0000 Subject: [PATCH] put upstream events in an async queue and resend them at the start of the chain. not working as expected yet but a start Original commit message from CVS: put upstream events in an async queue and resend them at the start of the chain. not working as expected yet but a start --- gst/gstqueue.c | 50 ++++++++++++++++++++++++++------------------- gst/gstqueue.h | 1 + plugins/elements/gstqueue.c | 50 ++++++++++++++++++++++++++------------------- plugins/elements/gstqueue.h | 1 + 4 files changed, 60 insertions(+), 42 deletions(-) diff --git a/gst/gstqueue.c b/gst/gstqueue.c index f2ca083..c54037f 100644 --- a/gst/gstqueue.c +++ b/gst/gstqueue.c @@ -236,6 +236,7 @@ gst_queue_init (GstQueue *queue) queue->writer = FALSE; queue->not_empty = g_cond_new (); queue->not_full = g_cond_new (); + queue->events = g_async_queue_new(); GST_DEBUG_ELEMENT (GST_CAT_THREAD, queue, "initialized queue's not_empty & not_full conditions"); } @@ -248,6 +249,8 @@ gst_queue_dispose (GObject *object) g_cond_free (queue->not_empty); g_cond_free (queue->not_full); + g_async_queue_unref(queue->events); + G_OBJECT_CLASS (parent_class)->dispose (object); } @@ -297,6 +300,16 @@ gst_queue_chain (GstPad *pad, GstBuffer *buf) g_return_if_fail (buf != NULL); queue = GST_QUEUE (GST_OBJECT_PARENT (pad)); + + /* check for events to send upstream */ + g_async_queue_lock(queue->events); + while (g_async_queue_length_unlocked(queue->events) > 0){ + GstEvent *event = (GstEvent*)g_async_queue_pop_unlocked(queue->events); + g_print("sending event upstream\n"); + gst_pad_event_default (pad, event); + g_print("event sent\n"); + } + g_async_queue_unlock(queue->events); restart: /* we have to lock the queue since we span threads */ @@ -441,6 +454,8 @@ gst_queue_get (GstPad *pad) queue = GST_QUEUE (GST_OBJECT_PARENT (pad)); + + restart: /* have to lock for thread-safety */ GST_DEBUG_ELEMENT (GST_CAT_DATAFLOW, queue, "locking t:%ld", pthread_self ()); @@ -542,35 +557,28 @@ static gboolean gst_queue_handle_src_event (GstPad *pad, GstEvent *event) { GstQueue *queue; - gboolean res = TRUE; queue = GST_QUEUE (GST_OBJECT_PARENT (pad)); - g_mutex_lock (queue->qlock); - - if (gst_element_get_state (GST_ELEMENT (queue)) == GST_STATE_PLAYING) { - g_mutex_unlock (queue->qlock); - g_warning ("queue event in playing state"); - return FALSE; - } + /* push the event to the queue for upstream consumption */ + g_async_queue_push(queue->events, event); - res = gst_pad_event_default (pad, event); - if (res) { - switch (GST_EVENT_TYPE (event)) { - case GST_EVENT_FLUSH: - GST_DEBUG_ELEMENT (GST_CAT_DATAFLOW, queue, "FLUSH event, flushing queue\n"); + g_mutex_lock (queue->qlock); + switch (GST_EVENT_TYPE (event)) { + case GST_EVENT_FLUSH: + GST_DEBUG_ELEMENT (GST_CAT_DATAFLOW, queue, "FLUSH event, flushing queue\n"); + gst_queue_locked_flush (queue); + break; + case GST_EVENT_SEEK: + if (GST_EVENT_SEEK_FLAGS (event) & GST_SEEK_FLAG_FLUSH) gst_queue_locked_flush (queue); - break; - case GST_EVENT_SEEK: - if (GST_EVENT_SEEK_FLAGS (event) & GST_SEEK_FLAG_FLUSH) - gst_queue_locked_flush (queue); - default: - break; - } + default: + break; } g_mutex_unlock (queue->qlock); - return res; + /* we have to claim success, but we don't really know */ + return TRUE; } static gboolean diff --git a/gst/gstqueue.h b/gst/gstqueue.h index 1362007..5fabe89 100644 --- a/gst/gstqueue.h +++ b/gst/gstqueue.h @@ -87,6 +87,7 @@ struct _GstQueue { GCond *not_full; /* signals space now available for writing */ GTimeVal *timeval; /* the timeout for the queue locking */ + GAsyncQueue *events; /* upstream events get decoupled here */ }; struct _GstQueueClass { diff --git a/plugins/elements/gstqueue.c b/plugins/elements/gstqueue.c index f2ca083..c54037f 100644 --- a/plugins/elements/gstqueue.c +++ b/plugins/elements/gstqueue.c @@ -236,6 +236,7 @@ gst_queue_init (GstQueue *queue) queue->writer = FALSE; queue->not_empty = g_cond_new (); queue->not_full = g_cond_new (); + queue->events = g_async_queue_new(); GST_DEBUG_ELEMENT (GST_CAT_THREAD, queue, "initialized queue's not_empty & not_full conditions"); } @@ -248,6 +249,8 @@ gst_queue_dispose (GObject *object) g_cond_free (queue->not_empty); g_cond_free (queue->not_full); + g_async_queue_unref(queue->events); + G_OBJECT_CLASS (parent_class)->dispose (object); } @@ -297,6 +300,16 @@ gst_queue_chain (GstPad *pad, GstBuffer *buf) g_return_if_fail (buf != NULL); queue = GST_QUEUE (GST_OBJECT_PARENT (pad)); + + /* check for events to send upstream */ + g_async_queue_lock(queue->events); + while (g_async_queue_length_unlocked(queue->events) > 0){ + GstEvent *event = (GstEvent*)g_async_queue_pop_unlocked(queue->events); + g_print("sending event upstream\n"); + gst_pad_event_default (pad, event); + g_print("event sent\n"); + } + g_async_queue_unlock(queue->events); restart: /* we have to lock the queue since we span threads */ @@ -441,6 +454,8 @@ gst_queue_get (GstPad *pad) queue = GST_QUEUE (GST_OBJECT_PARENT (pad)); + + restart: /* have to lock for thread-safety */ GST_DEBUG_ELEMENT (GST_CAT_DATAFLOW, queue, "locking t:%ld", pthread_self ()); @@ -542,35 +557,28 @@ static gboolean gst_queue_handle_src_event (GstPad *pad, GstEvent *event) { GstQueue *queue; - gboolean res = TRUE; queue = GST_QUEUE (GST_OBJECT_PARENT (pad)); - g_mutex_lock (queue->qlock); - - if (gst_element_get_state (GST_ELEMENT (queue)) == GST_STATE_PLAYING) { - g_mutex_unlock (queue->qlock); - g_warning ("queue event in playing state"); - return FALSE; - } + /* push the event to the queue for upstream consumption */ + g_async_queue_push(queue->events, event); - res = gst_pad_event_default (pad, event); - if (res) { - switch (GST_EVENT_TYPE (event)) { - case GST_EVENT_FLUSH: - GST_DEBUG_ELEMENT (GST_CAT_DATAFLOW, queue, "FLUSH event, flushing queue\n"); + g_mutex_lock (queue->qlock); + switch (GST_EVENT_TYPE (event)) { + case GST_EVENT_FLUSH: + GST_DEBUG_ELEMENT (GST_CAT_DATAFLOW, queue, "FLUSH event, flushing queue\n"); + gst_queue_locked_flush (queue); + break; + case GST_EVENT_SEEK: + if (GST_EVENT_SEEK_FLAGS (event) & GST_SEEK_FLAG_FLUSH) gst_queue_locked_flush (queue); - break; - case GST_EVENT_SEEK: - if (GST_EVENT_SEEK_FLAGS (event) & GST_SEEK_FLAG_FLUSH) - gst_queue_locked_flush (queue); - default: - break; - } + default: + break; } g_mutex_unlock (queue->qlock); - return res; + /* we have to claim success, but we don't really know */ + return TRUE; } static gboolean diff --git a/plugins/elements/gstqueue.h b/plugins/elements/gstqueue.h index 1362007..5fabe89 100644 --- a/plugins/elements/gstqueue.h +++ b/plugins/elements/gstqueue.h @@ -87,6 +87,7 @@ struct _GstQueue { GCond *not_full; /* signals space now available for writing */ GTimeVal *timeval; /* the timeout for the queue locking */ + GAsyncQueue *events; /* upstream events get decoupled here */ }; struct _GstQueueClass { -- 2.7.4