static gboolean
gst_base_sink_get_sync_times (GstBaseSink * basesink, GstMiniObject * obj,
GstClockTime * rsstart, GstClockTime * rsstop,
- GstClockTime * rrstart, GstClockTime * rrstop, gboolean * do_sync)
+ GstClockTime * rrstart, GstClockTime * rrstop, gboolean * do_sync,
+ GstSegment * segment)
{
GstBaseSinkClass *bclass;
GstBuffer *buffer;
gint64 rstart, rstop; /* clipped timestamps converted to running time */
GstClockTime sstart, sstop; /* clipped timestamps converted to stream time */
GstFormat format;
- GstSegment *segment;
+ GstBaseSinkPrivate *priv;
+
+ priv = basesink->priv;
/* start with nothing */
start = stop = sstart = sstop = rstart = rstop = -1;
switch (GST_EVENT_TYPE (event)) {
/* EOS event needs syncing */
case GST_EVENT_EOS:
- sstart = sstop = basesink->priv->current_sstop;
- rstart = rstop = basesink->priv->eos_rtime;
+ sstart = sstop = priv->current_sstop;
+ rstart = rstop = priv->eos_rtime;
*do_sync = rstart != -1;
GST_DEBUG_OBJECT (basesink, "sync times for EOS %" GST_TIME_FORMAT,
GST_TIME_ARGS (rstart));
GST_TIME_ARGS (stop), *do_sync);
/* collect segment and format for code clarity */
- segment = &basesink->segment;
format = segment->format;
/* no timestamp clipping if we did not * get a TIME segment format */
GstClockTimeDiff jitter;
gboolean syncable;
GstClockReturn status = GST_CLOCK_OK;
- GstClockTime sstart, sstop, rstart, rstop;
+ GstClockTime rstart, rstop, sstart, sstop;
gboolean do_sync;
+ GstBaseSinkPrivate *priv;
+
+ priv = basesink->priv;
sstart = sstop = rstart = rstop = -1;
do_sync = TRUE;
- basesink->priv->current_rstart = -1;
+ priv->current_rstart = -1;
- /* update timing information for this object */
+ /* get timing information for this object against the render segment */
syncable = gst_base_sink_get_sync_times (basesink, obj,
- &sstart, &sstop, &rstart, &rstop, &do_sync);
+ &sstart, &sstop, &rstart, &rstop, &do_sync, &basesink->segment);
/* a syncable object needs to participate in preroll and
* clocking. All buffers and EOS are syncable. */
goto not_syncable;
/* store timing info for current object */
- basesink->priv->current_rstart = rstart;
- basesink->priv->current_rstop = (rstop != -1 ? rstop : rstart);
+ priv->current_rstart = rstart;
+ priv->current_rstop = (rstop != -1 ? rstop : rstart);
/* save sync time for eos when the previous object needed sync */
- basesink->priv->eos_rtime = (do_sync ? basesink->priv->current_rstop : -1);
-
- /* lock because we read this when answering the POSITION
- * query. */
- GST_OBJECT_LOCK (basesink);
- basesink->priv->current_sstart = sstart;
- basesink->priv->current_sstop = (sstop != -1 ? sstop : sstart);
- GST_OBJECT_UNLOCK (basesink);
+ priv->eos_rtime = (do_sync ? priv->current_rstop : -1);
again:
/* first do preroll, this makes sure we commit our state
}
}
+ /* After rendering we store the position of the last buffer so that we can use
+ * it to report the position. We need to take the lock here. */
+ GST_OBJECT_LOCK (basesink);
+ priv->current_sstart = sstart;
+ priv->current_sstop = (sstop != -1 ? sstop : sstart);
+ GST_OBJECT_UNLOCK (basesink);
+
if (!do_sync)
goto done;
}
/* successful syncing done, record observation */
- basesink->priv->current_jitter = jitter;
+ priv->current_jitter = jitter;
/* check if the object should be dropped */
*late = gst_base_sink_is_too_late (basesink, obj, rstart, rstop,
if (G_LIKELY (GST_IS_BUFFER (obj))) {
GstBaseSinkClass *bclass;
GstBuffer *buf;
+ GstClockTime timestamp;
buf = GST_BUFFER_CAST (obj);
+ timestamp = GST_BUFFER_TIMESTAMP (buf);
GST_DEBUG_OBJECT (basesink, "preroll buffer %" GST_TIME_FORMAT,
- GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
+ GST_TIME_ARGS (timestamp));
gst_base_sink_set_last_buffer (basesink, buf);
gst_segment_init (basesink->abidata.ABI.clip_segment,
GST_FORMAT_UNDEFINED);
basesink->have_newsegment = FALSE;
+
/* for position reporting */
GST_OBJECT_LOCK (basesink);
basesink->priv->current_sstart = -1;
{
/* return last observed stream time */
*cur = basesink->priv->current_sstop;
+ GST_DEBUG_OBJECT (basesink, "POSITION: %" GST_TIME_FORMAT,
+ GST_TIME_ARGS (*cur));
return TRUE;
}
-/* get the position when we are PAUSED */
-/* FIXME, not entirely correct if we have preroll_queue_len > 1 and
- * there are multiple segments in the queue since we calculate on the
- * total segments, not the first one. */
+/* get the position when we are PAUSED, this is the stream time of the buffer
+ * that prerolled. If no buffer is prerolled (we are still flushing), this
+ * value will be -1. */
static gboolean
gst_base_sink_get_position_paused (GstBaseSink * basesink, gint64 * cur)
{
+ gboolean res;
+
*cur = basesink->priv->current_sstart;
if (*cur != -1)
else
*cur = basesink->abidata.ABI.clip_segment->time;
- return TRUE;
+ res = (*cur != -1);
+ GST_DEBUG_OBJECT (basesink, "POSITION: %" GST_TIME_FORMAT,
+ GST_TIME_ARGS (*cur));
+
+ return res;
}
static gboolean
/* special cases */
in_eos:
{
+ GST_DEBUG_OBJECT (basesink, "position in EOS");
res = gst_base_sink_get_position_last (basesink, cur);
GST_OBJECT_UNLOCK (basesink);
goto done;
}
in_pause:
{
+ GST_DEBUG_OBJECT (basesink, "position in PAUSED");
res = gst_base_sink_get_position_paused (basesink, cur);
GST_OBJECT_UNLOCK (basesink);
goto done;
wrong_state:
{
/* in NULL or READY we always return 0 */
- res = TRUE;
- *cur = 0;
+ GST_DEBUG_OBJECT (basesink, "position in wrong state, return -1");
+ res = FALSE;
+ *cur = -1;
GST_OBJECT_UNLOCK (basesink);
goto done;
}
* that upstream can answer */
if ((*cur = basesink->priv->current_sstart) != -1)
res = TRUE;
+ GST_DEBUG_OBJECT (basesink, "no sync, res %d, POSITION %" GST_TIME_FORMAT,
+ res, GST_TIME_ARGS (*cur));
GST_OBJECT_UNLOCK (basesink);
return res;
}
basesink->have_preroll = FALSE;
basesink->need_preroll = TRUE;
basesink->playing_async = TRUE;
- priv->current_sstart = 0;
- priv->current_sstop = 0;
+ priv->current_sstart = -1;
+ priv->current_sstop = -1;
priv->eos_rtime = -1;
priv->latency = 0;
basesink->eos = FALSE;
} else {
GST_DEBUG_OBJECT (basesink, "PAUSED to READY, don't need_preroll");
}
- priv->current_sstart = 0;
- priv->current_sstop = 0;
+ priv->current_sstart = -1;
+ priv->current_sstop = -1;
priv->have_latency = FALSE;
gst_base_sink_set_last_buffer (basesink, NULL);
GST_PAD_PREROLL_UNLOCK (basesink->sinkpad);
GST_END_TEST;
+/* this variable is updated in the same thread, first it is set by the
+ * handoff-preroll signal, then it is checked when the ASYNC_DONE is posted on
+ * the bus */
+static gboolean have_preroll = FALSE;
+
+static void
+async_done_handoff (GstElement * element, GstBuffer * buf, GstPad * pad,
+ GstElement * sink)
+{
+ GST_DEBUG ("we have the preroll buffer");
+ have_preroll = TRUE;
+}
+
+/* when we get the ASYNC_DONE, query the position */
+static GstBusSyncReply
+async_done_func (GstBus * bus, GstMessage * msg, GstElement * sink)
+{
+ if (GST_MESSAGE_TYPE (msg) == GST_MESSAGE_ASYNC_DONE) {
+ GstFormat format;
+ gint64 position;
+
+ GST_DEBUG ("we have ASYNC_DONE now");
+ fail_unless (have_preroll == TRUE, "no preroll buffer received");
+
+ /* get the position now */
+ format = GST_FORMAT_TIME;
+ gst_element_query_position (sink, &format, &position);
+
+ GST_DEBUG ("we have position %" GST_TIME_FORMAT, GST_TIME_ARGS (position));
+
+ fail_unless (position == 10 * GST_SECOND, "position is wrong");
+ }
+
+ /* we can drop the message, nothing is listening for it. */
+ return GST_BUS_DROP;
+}
+
+static void
+send_buffer (GstPad * sinkpad)
+{
+ GstBuffer *buffer;
+ GstStateChangeReturn ret;
+
+ /* push a second buffer */
+ GST_DEBUG ("pushing last buffer");
+ buffer = gst_buffer_new_and_alloc (10);
+ GST_BUFFER_TIMESTAMP (buffer) = 200 * GST_SECOND;
+ GST_BUFFER_DURATION (buffer) = 100 * GST_SECOND;
+
+ /* this function will initially block */
+ ret = gst_pad_chain (sinkpad, buffer);
+ fail_unless (ret == GST_FLOW_OK, "no OK flow return");
+}
+
+/* when we get the ASYNC_DONE message from a sink, we want the sink to be able
+ * to report the duration and position. The sink should also have called the
+ * render method. */
+GST_START_TEST (test_async_done)
+{
+ GstElement *sink;
+ GstBuffer *buffer;
+ GstEvent *event;
+ GstStateChangeReturn ret;
+ GstPad *sinkpad;
+ GstFlowReturn res;
+ GstBus *bus;
+ GThread *thread;
+ GstFormat format;
+ gint64 position;
+ gboolean qret;
+
+ sink = gst_element_factory_make ("fakesink", "sink");
+ g_object_set (G_OBJECT (sink), "sync", TRUE, NULL);
+ g_object_set (G_OBJECT (sink), "preroll-queue-len", 2, NULL);
+ g_object_set (G_OBJECT (sink), "signal-handoffs", TRUE, NULL);
+
+ g_signal_connect (sink, "preroll-handoff", (GCallback) async_done_handoff,
+ sink);
+
+ sinkpad = gst_element_get_pad (sink, "sink");
+
+ ret = gst_element_set_state (sink, GST_STATE_PAUSED);
+ fail_unless (ret == GST_STATE_CHANGE_ASYNC, "no ASYNC state return");
+
+ /* set bus on element synchronously listen for ASYNC_DONE */
+ bus = gst_bus_new ();
+ gst_element_set_bus (sink, bus);
+ gst_bus_set_sync_handler (bus, (GstBusSyncHandler) async_done_func, sink);
+
+ /* make newsegment, this sets the position to 10sec when the buffer prerolls */
+ GST_DEBUG ("sending segment");
+ event =
+ gst_event_new_new_segment (FALSE, 1.0, GST_FORMAT_TIME, 0, -1,
+ 10 * GST_SECOND);
+ res = gst_pad_send_event (sinkpad, event);
+
+ /* We have not yet received any buffers so we are still in the READY state,
+ * the position is therefore still not queryable. */
+ format = GST_FORMAT_TIME;
+ position = -1;
+ qret = gst_element_query_position (sink, &format, &position);
+ fail_unless (qret == FALSE, "position wrong");
+ fail_unless (position == -1, "position is wrong");
+
+ /* Since we are paused and the preroll queue has a length of 2, this function
+ * will return immediatly, the preroll handoff will be called and the stream
+ * position should now be 10 seconds. */
+ GST_DEBUG ("pushing first buffer");
+ buffer = gst_buffer_new_and_alloc (10);
+ GST_BUFFER_TIMESTAMP (buffer) = 1 * GST_SECOND;
+ GST_BUFFER_DURATION (buffer) = 100 * GST_SECOND;
+ res = gst_pad_chain (sinkpad, buffer);
+ fail_unless (res == GST_FLOW_OK, "no OK flow return");
+
+ /* scond buffer, will not block either but position should still be 10
+ * seconds */
+ GST_DEBUG ("pushing second buffer");
+ buffer = gst_buffer_new_and_alloc (10);
+ GST_BUFFER_TIMESTAMP (buffer) = 100 * GST_SECOND;
+ GST_BUFFER_DURATION (buffer) = 100 * GST_SECOND;
+ res = gst_pad_chain (sinkpad, buffer);
+ fail_unless (res == GST_FLOW_OK, "no OK flow return");
+
+ /* check if position is still 10 seconds */
+ format = GST_FORMAT_TIME;
+ gst_element_query_position (sink, &format, &position);
+ GST_DEBUG ("first buffer position %" GST_TIME_FORMAT,
+ GST_TIME_ARGS (position));
+ fail_unless (position == 10 * GST_SECOND, "position is wrong");
+
+ /* last buffer, blocks because preroll queue is filled. Start the push in a
+ * new thread so that we can check the position */
+ GST_DEBUG ("starting thread");
+ thread = g_thread_create ((GThreadFunc) send_buffer, sinkpad, TRUE, NULL);
+ fail_if (thread == NULL, "no thread");
+
+ GST_DEBUG ("waiting 1 second");
+ g_usleep (G_USEC_PER_SEC);
+ GST_DEBUG ("waiting done");
+
+ /* check if position is still 10 seconds. This is racy because the above
+ * thread might not yet have started the push, because of the above sleep,
+ * this is very unlikely, though. */
+ format = GST_FORMAT_TIME;
+ gst_element_query_position (sink, &format, &position);
+ GST_DEBUG ("second buffer position %" GST_TIME_FORMAT,
+ GST_TIME_ARGS (position));
+ fail_unless (position == 10 * GST_SECOND, "position is wrong");
+
+ /* now we go to playing. This should unlock and stop the above thread. */
+ GST_DEBUG ("going to PLAYING");
+ ret = gst_element_set_state (sink, GST_STATE_PLAYING);
+
+ /* join the thread. At this point we know the sink processed the last buffer
+ * and the position should now be 210 seconds; the time of the last buffer we
+ * pushed */
+ GST_DEBUG ("joining thread");
+ g_thread_join (thread);
+
+ format = GST_FORMAT_TIME;
+ gst_element_query_position (sink, &format, &position);
+ GST_DEBUG ("last buffer position %" GST_TIME_FORMAT,
+ GST_TIME_ARGS (position));
+ fail_unless (position == 210 * GST_SECOND, "position is wrong");
+
+ gst_object_unref (sinkpad);
+
+ gst_element_set_state (sink, GST_STATE_NULL);
+ gst_object_unref (sink);
+ gst_object_unref (bus);
+}
+
+GST_END_TEST;
+
+/* when we get the ASYNC_DONE, query the position */
+static GstBusSyncReply
+async_done_eos_func (GstBus * bus, GstMessage * msg, GstElement * sink)
+{
+ if (GST_MESSAGE_TYPE (msg) == GST_MESSAGE_ASYNC_DONE) {
+ GstFormat format;
+ gint64 position;
+
+ GST_DEBUG ("we have ASYNC_DONE now");
+
+ /* get the position now */
+ format = GST_FORMAT_TIME;
+ gst_element_query_position (sink, &format, &position);
+
+ GST_DEBUG ("we have position %" GST_TIME_FORMAT, GST_TIME_ARGS (position));
+
+ fail_unless (position == 10 * GST_SECOND, "position is wrong");
+ }
+ /* we can drop the message, nothing is listening for it. */
+ return GST_BUS_DROP;
+}
+
+/* when we get the ASYNC_DONE message from a sink, we want the sink to be able
+ * to report the duration and position. The sink should also have called the
+ * render method. */
+GST_START_TEST (test_async_done_eos)
+{
+ GstElement *sink;
+ GstEvent *event;
+ GstStateChangeReturn ret;
+ GstPad *sinkpad;
+ gboolean res;
+ GstBus *bus;
+ GstFormat format;
+ gint64 position;
+ gboolean qret;
+
+ sink = gst_element_factory_make ("fakesink", "sink");
+ g_object_set (G_OBJECT (sink), "sync", TRUE, NULL);
+ g_object_set (G_OBJECT (sink), "preroll-queue-len", 1, NULL);
+
+ sinkpad = gst_element_get_pad (sink, "sink");
+
+ ret = gst_element_set_state (sink, GST_STATE_PAUSED);
+ fail_unless (ret == GST_STATE_CHANGE_ASYNC, "no ASYNC state return");
+
+ /* set bus on element synchronously listen for ASYNC_DONE */
+ bus = gst_bus_new ();
+ gst_element_set_bus (sink, bus);
+ gst_bus_set_sync_handler (bus, (GstBusSyncHandler) async_done_eos_func, sink);
+
+ /* make newsegment, this sets the position to 10sec when the buffer prerolls */
+ GST_DEBUG ("sending segment");
+ event =
+ gst_event_new_new_segment (FALSE, 1.0, GST_FORMAT_TIME, 0, -1,
+ 10 * GST_SECOND);
+ res = gst_pad_send_event (sinkpad, event);
+
+ /* We have not yet received any buffers so we are still in the READY state,
+ * the position is therefore still not queryable. */
+ format = GST_FORMAT_TIME;
+ position = -1;
+ qret = gst_element_query_position (sink, &format, &position);
+ fail_unless (qret == FALSE, "position wrong");
+ fail_unless (position == -1, "position is wrong");
+
+ /* Since we are paused and the preroll queue has a length of 1, this function
+ * will return immediatly. The EOS will complete the preroll and the
+ * position should now be 10 seconds. */
+ GST_DEBUG ("pushing EOS");
+ event = gst_event_new_eos ();
+ res = gst_pad_send_event (sinkpad, event);
+ fail_unless (res == TRUE, "no TRUE return");
+
+ /* check if position is still 10 seconds */
+ format = GST_FORMAT_TIME;
+ gst_element_query_position (sink, &format, &position);
+ GST_DEBUG ("EOS position %" GST_TIME_FORMAT, GST_TIME_ARGS (position));
+ fail_unless (position == 10 * GST_SECOND, "position is wrong");
+
+ gst_object_unref (sinkpad);
+
+ gst_element_set_state (sink, GST_STATE_NULL);
+ gst_object_unref (sink);
+ gst_object_unref (bus);
+}
+
+GST_END_TEST;
+
/* test: try changing state of sinks */
Suite *
gst_sinks_suite (void)
tcase_add_test (tc_chain, test_add_live2);
tcase_add_test (tc_chain, test_bin_live);
tcase_add_test (tc_chain, test_fake_eos);
+ tcase_add_test (tc_chain, test_async_done);
+ tcase_add_test (tc_chain, test_async_done_eos);
return s;
}
projects / platform / upstream / gstreamer.git / commitdiff