/**
* SECTION:element-queue
+ * @title: queue
*
* Data is queued until one of the limits specified by the
* #GstQueue:max-size-buffers, #GstQueue:max-size-bytes and/or
static GstFlowReturn gst_queue_push_one (GstQueue * queue);
static void gst_queue_loop (GstPad * pad);
-static gboolean gst_queue_handle_sink_event (GstPad * pad, GstObject * parent,
- GstEvent * event);
+static GstFlowReturn gst_queue_handle_sink_event (GstPad * pad,
+ GstObject * parent, GstEvent * event);
static gboolean gst_queue_handle_sink_query (GstPad * pad, GstObject * parent,
GstQuery * query);
typedef struct
{
- gboolean is_query;
GstMiniObject *item;
gsize size;
+ gboolean is_query;
} GstQueueItem;
#define GST_TYPE_QUEUE_LEAKY (queue_leaky_get_type ())
gst_element_class_set_static_metadata (gstelement_class,
"Queue",
"Generic", "Simple data queue", "Erik Walthinsen <omega@cse.ogi.edu>");
- gst_element_class_add_pad_template (gstelement_class,
- gst_static_pad_template_get (&srctemplate));
- gst_element_class_add_pad_template (gstelement_class,
- gst_static_pad_template_get (&sinktemplate));
+ gst_element_class_add_static_pad_template (gstelement_class, &srctemplate);
+ gst_element_class_add_static_pad_template (gstelement_class, &sinktemplate);
/* Registering debug symbols for function pointers */
GST_DEBUG_REGISTER_FUNCPTR (gst_queue_src_activate_mode);
gst_pad_set_chain_list_function (queue->sinkpad, gst_queue_chain_list);
gst_pad_set_activatemode_function (queue->sinkpad,
gst_queue_sink_activate_mode);
- gst_pad_set_event_function (queue->sinkpad, gst_queue_handle_sink_event);
+ gst_pad_set_event_full_function (queue->sinkpad, gst_queue_handle_sink_event);
gst_pad_set_query_function (queue->sinkpad, gst_queue_handle_sink_query);
GST_PAD_SET_PROXY_CAPS (queue->sinkpad);
gst_element_add_pad (GST_ELEMENT (queue), queue->sinkpad);
g_cond_init (&queue->item_del);
g_cond_init (&queue->query_handled);
- queue->queue = gst_queue_array_new (DEFAULT_MAX_SIZE_BUFFERS * 3 / 2);
+ queue->queue =
+ gst_queue_array_new_for_struct (sizeof (GstQueueItem),
+ DEFAULT_MAX_SIZE_BUFFERS * 3 / 2);
- queue->sinktime = GST_CLOCK_TIME_NONE;
- queue->srctime = GST_CLOCK_TIME_NONE;
+ queue->sinktime = GST_CLOCK_STIME_NONE;
+ queue->srctime = GST_CLOCK_STIME_NONE;
queue->sink_tainted = TRUE;
queue->src_tainted = TRUE;
gst_queue_finalize (GObject * object)
{
GstQueue *queue = GST_QUEUE (object);
+ GstQueueItem *qitem;
GST_DEBUG_OBJECT (queue, "finalizing queue");
- while (!gst_queue_array_is_empty (queue->queue)) {
- GstQueueItem *qitem = gst_queue_array_pop_head (queue->queue);
+ while ((qitem = gst_queue_array_pop_head_struct (queue->queue))) {
/* FIXME: if it's a query, shouldn't we unref that too? */
if (!qitem->is_query)
gst_mini_object_unref (qitem->item);
- g_slice_free (GstQueueItem, qitem);
}
gst_queue_array_free (queue->queue);
G_OBJECT_CLASS (parent_class)->finalize (object);
}
+/* Convenience function */
+static inline GstClockTimeDiff
+my_segment_to_running_time (GstSegment * segment, GstClockTime val)
+{
+ GstClockTimeDiff res = GST_CLOCK_STIME_NONE;
+
+ if (GST_CLOCK_TIME_IS_VALID (val)) {
+ gboolean sign =
+ gst_segment_to_running_time_full (segment, GST_FORMAT_TIME, val, &val);
+ if (sign > 0)
+ res = val;
+ else if (sign < 0)
+ res = -val;
+ }
+ return res;
+}
+
/* calculate the diff between running time on the sink and src of the queue.
* This is the total amount of time in the queue. */
static void
if (queue->sink_tainted) {
GST_LOG_OBJECT (queue, "update sink time");
queue->sinktime =
- gst_segment_to_running_time (&queue->sink_segment, GST_FORMAT_TIME,
+ my_segment_to_running_time (&queue->sink_segment,
queue->sink_segment.position);
queue->sink_tainted = FALSE;
}
if (queue->src_tainted) {
GST_LOG_OBJECT (queue, "update src time");
queue->srctime =
- gst_segment_to_running_time (&queue->src_segment, GST_FORMAT_TIME,
+ my_segment_to_running_time (&queue->src_segment,
queue->src_segment.position);
queue->src_tainted = FALSE;
}
src_time = queue->srctime;
- GST_LOG_OBJECT (queue, "sink %" GST_TIME_FORMAT ", src %" GST_TIME_FORMAT,
- GST_TIME_ARGS (sink_time), GST_TIME_ARGS (src_time));
+ GST_LOG_OBJECT (queue, "sink %" GST_STIME_FORMAT ", src %" GST_STIME_FORMAT,
+ GST_STIME_ARGS (sink_time), GST_STIME_ARGS (src_time));
- if (sink_time >= src_time)
+ if (GST_CLOCK_STIME_IS_VALID (src_time)
+ && GST_CLOCK_STIME_IS_VALID (sink_time) && sink_time >= src_time)
queue->cur_level.time = sink_time - src_time;
else
queue->cur_level.time = 0;
/* take a buffer and update segment, updating the time level of the queue. */
static void
apply_buffer (GstQueue * queue, GstBuffer * buffer, GstSegment * segment,
- gboolean with_duration, gboolean sink)
+ gboolean sink)
{
GstClockTime duration, timestamp;
- timestamp = GST_BUFFER_TIMESTAMP (buffer);
+ timestamp = GST_BUFFER_DTS_OR_PTS (buffer);
duration = GST_BUFFER_DURATION (buffer);
/* if no timestamp is set, assume it's continuous with the previous
timestamp = segment->position;
/* add duration */
- if (with_duration && duration != GST_CLOCK_TIME_NONE)
+ if (duration != GST_CLOCK_TIME_NONE)
timestamp += duration;
- GST_LOG_OBJECT (queue, "position updated to %" GST_TIME_FORMAT,
+ GST_LOG_OBJECT (queue, "%s position updated to %" GST_TIME_FORMAT,
+ segment == &queue->sink_segment ? "sink" : "src",
GST_TIME_ARGS (timestamp));
segment->position = timestamp;
update_time_level (queue);
}
-typedef struct
-{
- GstClockTime timestamp;
- gboolean with_duration;
-} BufferListApplyTimeData;
-
static gboolean
buffer_list_apply_time (GstBuffer ** buf, guint idx, gpointer user_data)
{
- BufferListApplyTimeData *data = user_data;
+ GstClockTime *timestamp = user_data;
+ GstClockTime btime;
- GST_TRACE ("buffer %u has ts %" GST_TIME_FORMAT
- " duration %" GST_TIME_FORMAT, idx,
- GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (*buf)),
+ GST_TRACE ("buffer %u has pts %" GST_TIME_FORMAT " dts %" GST_TIME_FORMAT
+ " duration %" GST_TIME_FORMAT, idx, GST_TIME_ARGS (GST_BUFFER_DTS (*buf)),
+ GST_TIME_ARGS (GST_BUFFER_PTS (*buf)),
GST_TIME_ARGS (GST_BUFFER_DURATION (*buf)));
- if (GST_BUFFER_TIMESTAMP_IS_VALID (*buf))
- data->timestamp = GST_BUFFER_TIMESTAMP (*buf);
+ btime = GST_BUFFER_DTS_OR_PTS (*buf);
+ if (GST_CLOCK_TIME_IS_VALID (btime))
+ *timestamp = btime;
- if (data->with_duration && GST_BUFFER_DURATION_IS_VALID (*buf))
- data->timestamp += GST_BUFFER_DURATION (*buf);
+ if (GST_BUFFER_DURATION_IS_VALID (*buf))
+ *timestamp += GST_BUFFER_DURATION (*buf);
- GST_TRACE ("ts now %" GST_TIME_FORMAT, GST_TIME_ARGS (data->timestamp));
+ GST_TRACE ("ts now %" GST_TIME_FORMAT, GST_TIME_ARGS (*timestamp));
return TRUE;
}
/* take a buffer list and update segment, updating the time level of the queue */
static void
apply_buffer_list (GstQueue * queue, GstBufferList * buffer_list,
- GstSegment * segment, gboolean with_duration, gboolean sink)
+ GstSegment * segment, gboolean sink)
{
- BufferListApplyTimeData data;
+ GstClockTime timestamp;
/* if no timestamp is set, assume it's continuous with the previous time */
- data.timestamp = segment->position;
- data.with_duration = with_duration;
+ timestamp = segment->position;
- gst_buffer_list_foreach (buffer_list, buffer_list_apply_time, &data);
+ gst_buffer_list_foreach (buffer_list, buffer_list_apply_time, ×tamp);
GST_DEBUG_OBJECT (queue, "position updated to %" GST_TIME_FORMAT,
- GST_TIME_ARGS (data.timestamp));
+ GST_TIME_ARGS (timestamp));
- segment->position = data.timestamp;
+ segment->position = timestamp;
if (sink)
queue->sink_tainted = TRUE;
static void
gst_queue_locked_flush (GstQueue * queue, gboolean full)
{
- while (!gst_queue_array_is_empty (queue->queue)) {
- GstQueueItem *qitem = gst_queue_array_pop_head (queue->queue);
+ GstQueueItem *qitem;
+ while ((qitem = gst_queue_array_pop_head_struct (queue->queue))) {
/* Then lose another reference because we are supposed to destroy that
data when flushing */
if (!full && !qitem->is_query && GST_IS_EVENT (qitem->item)
}
if (!qitem->is_query)
gst_mini_object_unref (qitem->item);
- g_slice_free (GstQueueItem, qitem);
+ memset (qitem, 0, sizeof (GstQueueItem));
}
queue->last_query = FALSE;
g_cond_signal (&queue->query_handled);
gst_segment_init (&queue->src_segment, GST_FORMAT_TIME);
queue->head_needs_discont = queue->tail_needs_discont = FALSE;
- queue->sinktime = queue->srctime = GST_CLOCK_TIME_NONE;
+ queue->sinktime = queue->srctime = GST_CLOCK_STIME_NONE;
queue->sink_tainted = queue->src_tainted = TRUE;
/* we deleted a lot of something */
static inline void
gst_queue_locked_enqueue_buffer (GstQueue * queue, gpointer item)
{
- GstQueueItem *qitem;
+ GstQueueItem qitem;
GstBuffer *buffer = GST_BUFFER_CAST (item);
gsize bsize = gst_buffer_get_size (buffer);
/* add buffer to the statistics */
queue->cur_level.buffers++;
queue->cur_level.bytes += bsize;
- apply_buffer (queue, buffer, &queue->sink_segment, TRUE, TRUE);
+ apply_buffer (queue, buffer, &queue->sink_segment, TRUE);
- qitem = g_slice_new (GstQueueItem);
- qitem->item = item;
- qitem->is_query = FALSE;
- qitem->size = bsize;
- gst_queue_array_push_tail (queue->queue, qitem);
+ qitem.item = item;
+ qitem.is_query = FALSE;
+ qitem.size = bsize;
+ gst_queue_array_push_tail_struct (queue->queue, &qitem);
GST_QUEUE_SIGNAL_ADD (queue);
}
-static gboolean
-buffer_list_calc_size (GstBuffer ** buf, guint idx, gpointer data)
-{
- guint *p_size = data;
- gsize buf_size;
-
- buf_size = gst_buffer_get_size (*buf);
- GST_TRACE ("buffer %u in has size %" G_GSIZE_FORMAT, idx, buf_size);
- *p_size += buf_size;
- return TRUE;
-}
-
static inline void
gst_queue_locked_enqueue_buffer_list (GstQueue * queue, gpointer item)
{
- GstQueueItem *qitem;
+ GstQueueItem qitem;
GstBufferList *buffer_list = GST_BUFFER_LIST_CAST (item);
- gsize bsize = 0;
+ gsize bsize;
- gst_buffer_list_foreach (buffer_list, buffer_list_calc_size, &bsize);
+ bsize = gst_buffer_list_calculate_size (buffer_list);
/* add buffer to the statistics */
queue->cur_level.buffers += gst_buffer_list_length (buffer_list);
queue->cur_level.bytes += bsize;
- apply_buffer_list (queue, buffer_list, &queue->sink_segment, TRUE, TRUE);
+ apply_buffer_list (queue, buffer_list, &queue->sink_segment, TRUE);
- qitem = g_slice_new (GstQueueItem);
- qitem->item = item;
- qitem->is_query = FALSE;
- qitem->size = bsize;
- gst_queue_array_push_tail (queue->queue, qitem);
+ qitem.item = item;
+ qitem.is_query = FALSE;
+ qitem.size = bsize;
+ gst_queue_array_push_tail_struct (queue->queue, &qitem);
GST_QUEUE_SIGNAL_ADD (queue);
}
static inline void
gst_queue_locked_enqueue_event (GstQueue * queue, gpointer item)
{
- GstQueueItem *qitem;
+ GstQueueItem qitem;
GstEvent *event = GST_EVENT_CAST (item);
switch (GST_EVENT_TYPE (event)) {
break;
}
- qitem = g_slice_new (GstQueueItem);
- qitem->item = item;
- qitem->is_query = FALSE;
- gst_queue_array_push_tail (queue->queue, qitem);
+ qitem.item = item;
+ qitem.is_query = FALSE;
+ qitem.size = 0;
+ gst_queue_array_push_tail_struct (queue->queue, &qitem);
GST_QUEUE_SIGNAL_ADD (queue);
}
GstMiniObject *item;
gsize bufsize;
- qitem = gst_queue_array_pop_head (queue->queue);
+ qitem = gst_queue_array_pop_head_struct (queue->queue);
if (qitem == NULL)
goto no_item;
item = qitem->item;
bufsize = qitem->size;
- g_slice_free (GstQueueItem, qitem);
if (GST_IS_BUFFER (item)) {
GstBuffer *buffer = GST_BUFFER_CAST (item);
queue->cur_level.buffers--;
queue->cur_level.bytes -= bufsize;
- apply_buffer (queue, buffer, &queue->src_segment, TRUE, FALSE);
+ apply_buffer (queue, buffer, &queue->src_segment, FALSE);
/* if the queue is empty now, update the other side */
if (queue->cur_level.buffers == 0)
queue->cur_level.buffers -= gst_buffer_list_length (buffer_list);
queue->cur_level.bytes -= bufsize;
- apply_buffer_list (queue, buffer_list, &queue->src_segment, TRUE, FALSE);
+ apply_buffer_list (queue, buffer_list, &queue->src_segment, FALSE);
/* if the queue is empty now, update the other side */
if (queue->cur_level.buffers == 0)
}
}
-static gboolean
+static GstFlowReturn
gst_queue_handle_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
{
gboolean ret = TRUE;
queue = GST_QUEUE (parent);
+ GST_CAT_LOG_OBJECT (queue_dataflow, queue, "Received event '%s'",
+ GST_EVENT_TYPE_NAME (event));
+
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_FLUSH_START:
- STATUS (queue, pad, "received flush start event");
/* forward event */
ret = gst_pad_push_event (queue->srcpad, event);
/* unblock the loop and chain functions */
GST_QUEUE_SIGNAL_ADD (queue);
GST_QUEUE_SIGNAL_DEL (queue);
- queue->last_query = FALSE;
- g_cond_signal (&queue->query_handled);
GST_QUEUE_MUTEX_UNLOCK (queue);
/* make sure it pauses, this should happen since we sent
* flush_start downstream. */
gst_pad_pause_task (queue->srcpad);
GST_CAT_LOG_OBJECT (queue_dataflow, queue, "loop stopped");
+
+ /* unblock query handler after the streaming thread is shut down.
+ * Otherwise downstream might have a query that is already unreffed
+ * upstream */
+ GST_QUEUE_MUTEX_LOCK (queue);
+ queue->last_query = FALSE;
+ g_cond_signal (&queue->query_handled);
+ GST_QUEUE_MUTEX_UNLOCK (queue);
break;
case GST_EVENT_FLUSH_STOP:
- STATUS (queue, pad, "received flush stop event");
/* forward event */
ret = gst_pad_push_event (queue->srcpad, event);
if (GST_EVENT_IS_SERIALIZED (event)) {
/* serialized events go in the queue */
GST_QUEUE_MUTEX_LOCK (queue);
+
+ /* STREAM_START and SEGMENT reset the EOS status of a
+ * pad. Change the cached sinkpad flow result accordingly */
+ if (queue->srcresult == GST_FLOW_EOS
+ && (GST_EVENT_TYPE (event) == GST_EVENT_STREAM_START
+ || GST_EVENT_TYPE (event) == GST_EVENT_SEGMENT))
+ queue->srcresult = GST_FLOW_OK;
+
if (queue->srcresult != GST_FLOW_OK) {
/* Errors in sticky event pushing are no problem and ignored here
* as they will cause more meaningful errors during data flow.
if (queue->srcresult == GST_FLOW_NOT_LINKED
|| queue->srcresult < GST_FLOW_EOS) {
GST_QUEUE_MUTEX_UNLOCK (queue);
- GST_ELEMENT_ERROR (queue, STREAM, FAILED,
- (_("Internal data flow error.")),
- ("streaming task paused, reason %s (%d)",
- gst_flow_get_name (queue->srcresult), queue->srcresult));
+ GST_ELEMENT_FLOW_ERROR (queue, queue->srcresult);
} else {
GST_QUEUE_MUTEX_UNLOCK (queue);
}
goto out_flow_error;
}
}
- /* refuse more events on EOS */
- if (queue->eos)
- goto out_eos;
+
+ /* refuse more events on EOS unless they unset the EOS status */
+ if (queue->eos) {
+ switch (GST_EVENT_TYPE (event)) {
+ case GST_EVENT_STREAM_START:
+ case GST_EVENT_SEGMENT:
+ /* Restart the loop */
+ if (GST_PAD_MODE (queue->srcpad) == GST_PAD_MODE_PUSH) {
+ queue->srcresult = GST_FLOW_OK;
+ queue->eos = FALSE;
+ queue->unexpected = FALSE;
+ gst_pad_start_task (queue->srcpad,
+ (GstTaskFunction) gst_queue_loop, queue->srcpad, NULL);
+ } else {
+ queue->eos = FALSE;
+ queue->unexpected = FALSE;
+ }
+
+ break;
+ default:
+ goto out_eos;
+ }
+ }
+
gst_queue_locked_enqueue_event (queue, event);
GST_QUEUE_MUTEX_UNLOCK (queue);
} else {
}
break;
}
- return ret;
+ if (ret == FALSE) {
+ GST_ERROR_OBJECT (queue, "Failed to push event");
+ return GST_FLOW_ERROR;
+ }
+ return GST_FLOW_OK;
/* ERRORS */
out_eos:
GST_CAT_LOG_OBJECT (queue_dataflow, queue, "refusing event, we are EOS");
GST_QUEUE_MUTEX_UNLOCK (queue);
gst_event_unref (event);
- return FALSE;
+ return GST_FLOW_EOS;
}
out_flow_error:
{
"refusing event, we have a downstream flow error: %s",
gst_flow_get_name (queue->srcresult));
gst_event_unref (event);
- return FALSE;
+ return queue->srcresult;
}
}
switch (GST_QUERY_TYPE (query)) {
default:
if (G_UNLIKELY (GST_QUERY_IS_SERIALIZED (query))) {
- GstQueueItem *qitem;
+ GstQueueItem qitem;
GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
GST_LOG_OBJECT (queue, "queuing query %p (%s)", query,
GST_QUERY_TYPE_NAME (query));
- qitem = g_slice_new (GstQueueItem);
- qitem->item = GST_MINI_OBJECT_CAST (query);
- qitem->is_query = TRUE;
- gst_queue_array_push_tail (queue->queue, qitem);
+ qitem.item = GST_MINI_OBJECT_CAST (query);
+ qitem.is_query = TRUE;
+ qitem.size = 0;
+ gst_queue_array_push_tail_struct (queue->queue, &qitem);
GST_QUEUE_SIGNAL_ADD (queue);
- g_cond_wait (&queue->query_handled, &queue->qlock);
+ while (queue->srcresult == GST_FLOW_OK &&
+ queue->last_handled_query != query)
+ g_cond_wait (&queue->query_handled, &queue->qlock);
+ queue->last_handled_query = NULL;
if (queue->srcresult != GST_FLOW_OK)
goto out_flushing;
res = queue->last_query;
static gboolean
gst_queue_is_empty (GstQueue * queue)
{
- GstQueueItem *head;
+ GstQueueItem *tail;
- if (gst_queue_array_is_empty (queue->queue))
+ tail = gst_queue_array_peek_tail_struct (queue->queue);
+
+ if (tail == NULL)
return TRUE;
/* Only consider the queue empty if the minimum thresholds
- * are not reached and data is at the queue head. Otherwise
+ * are not reached and data is at the queue tail. Otherwise
* we would block forever on serialized queries.
*/
- head = gst_queue_array_peek_head (queue->queue);
- if (!GST_IS_BUFFER (head->item) && !GST_IS_BUFFER_LIST (head->item))
+ if (!GST_IS_BUFFER (tail->item) && !GST_IS_BUFFER_LIST (tail->item))
return FALSE;
/* It is possible that a max size is reached before all min thresholds are.
GstClockTime duration, timestamp;
GstBuffer *buffer = GST_BUFFER_CAST (obj);
- timestamp = GST_BUFFER_TIMESTAMP (buffer);
+ timestamp = GST_BUFFER_DTS_OR_PTS (buffer);
duration = GST_BUFFER_DURATION (buffer);
GST_CAT_LOG_OBJECT (queue_dataflow, queue, "received buffer %p of size %"
GstEvent *event = GST_EVENT_CAST (data);
GstEventType type = GST_EVENT_TYPE (event);
- if (type == GST_EVENT_EOS || type == GST_EVENT_SEGMENT) {
+ if (type == GST_EVENT_EOS || type == GST_EVENT_SEGMENT
+ || type == GST_EVENT_STREAM_START) {
/* we found a pushable item in the queue, push it out */
GST_CAT_LOG_OBJECT (queue_dataflow, queue,
"pushing pushable event %s after EOS",
ret = gst_pad_peer_query (queue->srcpad, query);
GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing_query);
queue->last_query = ret;
+ queue->last_handled_query = query;
g_cond_signal (&queue->query_handled);
GST_CAT_LOG_OBJECT (queue_dataflow, queue,
"did query %p, return %d", query, queue->last_query);
/* ERRORS */
no_item:
{
- GST_CAT_LOG_OBJECT (queue_dataflow, queue,
+ GST_CAT_ERROR_OBJECT (queue_dataflow, queue,
"exit because we have no item in the queue");
return GST_FLOW_ERROR;
}
out_flushing:
{
- GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we are flushing");
- return GST_FLOW_FLUSHING;
+ GstFlowReturn ret = queue->srcresult;
+ GST_CAT_LOG_OBJECT (queue_dataflow, queue,
+ "exit because task paused, reason: %s", gst_flow_get_name (ret));
+ return ret;
}
out_flushing_query:
{
+ GstFlowReturn ret = queue->srcresult;
queue->last_query = FALSE;
g_cond_signal (&queue->query_handled);
- GST_CAT_LOG_OBJECT (queue_dataflow, queue, "exit because we are flushing");
- return GST_FLOW_FLUSHING;
+ GST_CAT_LOG_OBJECT (queue_dataflow, queue,
+ "exit because task paused, reason: %s", gst_flow_get_name (ret));
+ return ret;
}
}
gst_pad_pause_task (queue->srcpad);
GST_CAT_LOG_OBJECT (queue_dataflow, queue,
"pause task, reason: %s", gst_flow_get_name (ret));
- if (ret == GST_FLOW_FLUSHING)
+ if (ret == GST_FLOW_FLUSHING) {
gst_queue_locked_flush (queue, FALSE);
- else
+ } else {
GST_QUEUE_SIGNAL_DEL (queue);
+ queue->last_query = FALSE;
+ g_cond_signal (&queue->query_handled);
+ }
GST_QUEUE_MUTEX_UNLOCK (queue);
/* let app know about us giving up if upstream is not expected to do so */
/* EOS is already taken care of elsewhere */
if (eos && (ret == GST_FLOW_NOT_LINKED || ret < GST_FLOW_EOS)) {
- GST_ELEMENT_ERROR (queue, STREAM, FAILED,
- (_("Internal data flow error.")),
- ("streaming task paused, reason %s (%d)",
- gst_flow_get_name (ret), ret));
+ GST_ELEMENT_FLOW_ERROR (queue, ret);
gst_pad_push_event (queue->srcpad, gst_event_new_eos ());
}
return;
switch (format) {
case GST_FORMAT_BYTES:
peer_pos -= queue->cur_level.bytes;
+ if (peer_pos < 0) /* Clamp result to 0 */
+ peer_pos = 0;
break;
case GST_FORMAT_TIME:
peer_pos -= queue->cur_level.time;
+ if (peer_pos < 0) /* Clamp result to 0 */
+ peer_pos = 0;
break;
default:
GST_DEBUG_OBJECT (queue, "Can't adjust query in %s format, don't "
GST_QUEUE_MUTEX_LOCK (queue);
queue->srcresult = GST_FLOW_FLUSHING;
/* the item del signal will unblock */
- g_cond_signal (&queue->item_del);
- /* unblock query handler */
- queue->last_query = FALSE;
- g_cond_signal (&queue->query_handled);
+ GST_QUEUE_SIGNAL_DEL (queue);
GST_QUEUE_MUTEX_UNLOCK (queue);
/* step 2, wait until streaming thread stopped and flush queue */
/* step 2, make sure streaming finishes */
result = gst_pad_stop_task (pad);
+
+ GST_QUEUE_MUTEX_LOCK (queue);
+ gst_queue_locked_flush (queue, FALSE);
+ GST_QUEUE_MUTEX_UNLOCK (queue);
}
break;
default: