gst/base/gstbasesink.c

   1 /* GStreamer
   2  * Copyright (C) 2005 Wim Taymans <wim@fluendo.com>
   3  *
   4  * gstbasesink.c:
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  */
  21
  22 /**
  23  * SECTION:gstbasesink
  24  * @short_description: Base class for sink elements
  25  * @see_also: #GstBaseTransformc, #GstBaseSource
  26  *
  27  * This class is for elements that do output operations.
  28  *
  29  * <itemizedlist>
  30  *   <listitem><para>one sinkpad</para></listitem>
  31  *   <listitem><para>handles state changes</para></listitem>
  32  *   <listitem><para>pull/push mode</para></listitem>
  33  *   <listitem><para>handles seeking/query</para></listitem>
  34  *   <listitem><para>handles preroll</para></listitem>
  35  *   <listitem><para>EOS handling</para></listitem>
  36  * </itemizedlist>
  37  */
  38
  39 #ifdef HAVE_CONFIG_H
  40 #  include "config.h"
  41 #endif
  42
  43 #include "gstbasesink.h"
  44 #include <gst/gstmarshal.h>
  45
  46 GST_DEBUG_CATEGORY_STATIC (gst_base_sink_debug);
  47 #define GST_CAT_DEFAULT gst_base_sink_debug
  48
  49 /* BaseSink signals and properties */
  50 enum
  51 {
  52   /* FILL ME */
  53   SIGNAL_HANDOFF,
  54   LAST_SIGNAL
  55 };
  56
  57 /* FIXME, need to figure out a better way to handle the pull mode */
  58 #define DEFAULT_SIZE 1024
  59 #define DEFAULT_HAS_LOOP FALSE
  60 #define DEFAULT_HAS_CHAIN TRUE
  61
  62 enum
  63 {
  64   PROP_0,
  65   PROP_HAS_LOOP,
  66   PROP_HAS_CHAIN,
  67   PROP_PREROLL_QUEUE_LEN
  68 };
  69
  70 static GstElementClass *parent_class = NULL;
  71
  72 static void gst_base_sink_base_init (gpointer g_class);
  73 static void gst_base_sink_class_init (GstBaseSinkClass * klass);
  74 static void gst_base_sink_init (GstBaseSink * trans, gpointer g_class);
  75 static void gst_base_sink_finalize (GObject * object);
  76
  77 GType
  78 gst_base_sink_get_type (void)
  79 {
  80   static GType base_sink_type = 0;
  81
  82   if (!base_sink_type) {
  83     static const GTypeInfo base_sink_info = {
  84       sizeof (GstBaseSinkClass),
  85       (GBaseInitFunc) gst_base_sink_base_init,
  86       NULL,
  87       (GClassInitFunc) gst_base_sink_class_init,
  88       NULL,
  89       NULL,
  90       sizeof (GstBaseSink),
  91       0,
  92       (GInstanceInitFunc) gst_base_sink_init,
  93     };
  94
  95     base_sink_type = g_type_register_static (GST_TYPE_ELEMENT,
  96         "GstBaseSink", &base_sink_info, G_TYPE_FLAG_ABSTRACT);
  97   }
  98   return base_sink_type;
  99 }
 100
 101 static void gst_base_sink_set_clock (GstElement * element, GstClock * clock);
 102
 103 static void gst_base_sink_set_property (GObject * object, guint prop_id,
 104     const GValue * value, GParamSpec * pspec);
 105 static void gst_base_sink_get_property (GObject * object, guint prop_id,
 106     GValue * value, GParamSpec * pspec);
 107
 108 static GstCaps *gst_base_sink_get_caps (GstBaseSink * sink);
 109 static gboolean gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps);
 110 static GstFlowReturn gst_base_sink_buffer_alloc (GstBaseSink * sink,
 111     guint64 offset, guint size, GstCaps * caps, GstBuffer ** buf);
 112 static void gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
 113     GstClockTime * start, GstClockTime * end);
 114
 115 static GstElementStateReturn gst_base_sink_change_state (GstElement * element);
 116
 117 static GstFlowReturn gst_base_sink_chain (GstPad * pad, GstBuffer * buffer);
 118 static void gst_base_sink_loop (GstPad * pad);
 119 static GstFlowReturn gst_base_sink_chain (GstPad * pad, GstBuffer * buffer);
 120 static gboolean gst_base_sink_activate_push (GstPad * pad, gboolean active);
 121 static gboolean gst_base_sink_activate_pull (GstPad * pad, gboolean active);
 122 static gboolean gst_base_sink_event (GstPad * pad, GstEvent * event);
 123 static inline GstFlowReturn gst_base_sink_handle_buffer (GstBaseSink * basesink,
 124     GstBuffer * buf);
 125 static inline gboolean gst_base_sink_handle_event (GstBaseSink * basesink,
 126     GstEvent * event);
 127
 128 static void
 129 gst_base_sink_base_init (gpointer g_class)
 130 {
 131   GST_DEBUG_CATEGORY_INIT (gst_base_sink_debug, "basesink", 0,
 132       "basesink element");
 133 }
 134
 135 static void
 136 gst_base_sink_class_init (GstBaseSinkClass * klass)
 137 {
 138   GObjectClass *gobject_class;
 139   GstElementClass *gstelement_class;
 140
 141   gobject_class = (GObjectClass *) klass;
 142   gstelement_class = (GstElementClass *) klass;
 143
 144   parent_class = g_type_class_ref (GST_TYPE_ELEMENT);
 145
 146   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_base_sink_finalize);
 147   gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_base_sink_set_property);
 148   gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_base_sink_get_property);
 149
 150   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_HAS_LOOP,
 151       g_param_spec_boolean ("has-loop", "has-loop",
 152           "Enable loop-based operation", DEFAULT_HAS_LOOP,
 153           G_PARAM_READWRITE | G_PARAM_CONSTRUCT));
 154   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_HAS_CHAIN,
 155       g_param_spec_boolean ("has-chain", "has-chain",
 156           "Enable chain-based operation", DEFAULT_HAS_CHAIN,
 157           G_PARAM_READWRITE | G_PARAM_CONSTRUCT));
 158   /* FIXME, this next value should be configured using an event from the
 159    * upstream element */
 160   g_object_class_install_property (G_OBJECT_CLASS (klass),
 161       PROP_PREROLL_QUEUE_LEN,
 162       g_param_spec_uint ("preroll-queue-len", "preroll-queue-len",
 163           "Number of buffers to queue during preroll", 0, G_MAXUINT, 0,
 164           G_PARAM_READWRITE | G_PARAM_CONSTRUCT));
 165
 166   gstelement_class->set_clock = GST_DEBUG_FUNCPTR (gst_base_sink_set_clock);
 167   gstelement_class->change_state =
 168       GST_DEBUG_FUNCPTR (gst_base_sink_change_state);
 169
 170   klass->get_caps = GST_DEBUG_FUNCPTR (gst_base_sink_get_caps);
 171   klass->set_caps = GST_DEBUG_FUNCPTR (gst_base_sink_set_caps);
 172   klass->buffer_alloc = GST_DEBUG_FUNCPTR (gst_base_sink_buffer_alloc);
 173   klass->get_times = GST_DEBUG_FUNCPTR (gst_base_sink_get_times);
 174 }
 175
 176 static GstCaps *
 177 gst_base_sink_pad_getcaps (GstPad * pad)
 178 {
 179   GstBaseSinkClass *bclass;
 180   GstBaseSink *bsink;
 181   GstCaps *caps = NULL;
 182
 183   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
 184   bclass = GST_BASE_SINK_GET_CLASS (bsink);
 185   if (bclass->get_caps)
 186     caps = bclass->get_caps (bsink);
 187
 188   if (caps == NULL) {
 189     GstPadTemplate *pad_template;
 190
 191     pad_template =
 192         gst_element_class_get_pad_template (GST_ELEMENT_CLASS (bclass), "sink");
 193     if (pad_template != NULL) {
 194       caps = gst_caps_ref (gst_pad_template_get_caps (pad_template));
 195     }
 196   }
 197   gst_object_unref (bsink);
 198
 199   return caps;
 200 }
 201
 202 static gboolean
 203 gst_base_sink_pad_setcaps (GstPad * pad, GstCaps * caps)
 204 {
 205   GstBaseSinkClass *bclass;
 206   GstBaseSink *bsink;
 207   gboolean res = FALSE;
 208
 209   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
 210   bclass = GST_BASE_SINK_GET_CLASS (bsink);
 211
 212   if (bclass->set_caps)
 213     res = bclass->set_caps (bsink, caps);
 214
 215   gst_object_unref (bsink);
 216
 217   return res;
 218 }
 219
 220 static GstFlowReturn
 221 gst_base_sink_pad_buffer_alloc (GstPad * pad, guint64 offset, guint size,
 222     GstCaps * caps, GstBuffer ** buf)
 223 {
 224   GstBaseSinkClass *bclass;
 225   GstBaseSink *bsink;
 226   GstFlowReturn result = GST_FLOW_OK;
 227
 228   bsink = GST_BASE_SINK (gst_pad_get_parent (pad));
 229   bclass = GST_BASE_SINK_GET_CLASS (bsink);
 230
 231   if (bclass->buffer_alloc)
 232     result = bclass->buffer_alloc (bsink, offset, size, caps, buf);
 233   else
 234     *buf = NULL;
 235
 236   gst_object_unref (bsink);
 237
 238   return result;
 239 }
 240
 241 static void
 242 gst_base_sink_init (GstBaseSink * basesink, gpointer g_class)
 243 {
 244   GstPadTemplate *pad_template;
 245
 246   pad_template =
 247       gst_element_class_get_pad_template (GST_ELEMENT_CLASS (g_class), "sink");
 248   g_return_if_fail (pad_template != NULL);
 249
 250   basesink->sinkpad = gst_pad_new_from_template (pad_template, "sink");
 251
 252   gst_pad_set_getcaps_function (basesink->sinkpad,
 253       GST_DEBUG_FUNCPTR (gst_base_sink_pad_getcaps));
 254   gst_pad_set_setcaps_function (basesink->sinkpad,
 255       GST_DEBUG_FUNCPTR (gst_base_sink_pad_setcaps));
 256   gst_pad_set_bufferalloc_function (basesink->sinkpad,
 257       GST_DEBUG_FUNCPTR (gst_base_sink_pad_buffer_alloc));
 258   gst_element_add_pad (GST_ELEMENT (basesink), basesink->sinkpad);
 259
 260   basesink->pad_mode = GST_ACTIVATE_NONE;
 261   GST_PAD_TASK (basesink->sinkpad) = NULL;
 262   basesink->preroll_queue = g_queue_new ();
 263
 264   GST_FLAG_SET (basesink, GST_ELEMENT_IS_SINK);
 265 }
 266
 267 static void
 268 gst_base_sink_finalize (GObject * object)
 269 {
 270   GstBaseSink *basesink;
 271
 272   basesink = GST_BASE_SINK (object);
 273
 274   g_queue_free (basesink->preroll_queue);
 275
 276   G_OBJECT_CLASS (parent_class)->finalize (object);
 277 }
 278
 279 static void
 280 gst_base_sink_set_pad_functions (GstBaseSink * this, GstPad * pad)
 281 {
 282   gst_pad_set_activatepush_function (pad,
 283       GST_DEBUG_FUNCPTR (gst_base_sink_activate_push));
 284   gst_pad_set_activatepull_function (pad,
 285       GST_DEBUG_FUNCPTR (gst_base_sink_activate_pull));
 286   gst_pad_set_event_function (pad, GST_DEBUG_FUNCPTR (gst_base_sink_event));
 287
 288   if (this->has_chain)
 289     gst_pad_set_chain_function (pad, GST_DEBUG_FUNCPTR (gst_base_sink_chain));
 290   else
 291     gst_pad_set_chain_function (pad, NULL);
 292 }
 293
 294 static void
 295 gst_base_sink_set_all_pad_functions (GstBaseSink * this)
 296 {
 297   GList *l;
 298
 299   for (l = GST_ELEMENT_PADS (this); l; l = l->next)
 300     gst_base_sink_set_pad_functions (this, (GstPad *) l->data);
 301 }
 302
 303 static void
 304 gst_base_sink_set_clock (GstElement * element, GstClock * clock)
 305 {
 306   GstBaseSink *sink;
 307
 308   sink = GST_BASE_SINK (element);
 309
 310   sink->clock = clock;
 311 }
 312
 313 static void
 314 gst_base_sink_set_property (GObject * object, guint prop_id,
 315     const GValue * value, GParamSpec * pspec)
 316 {
 317   GstBaseSink *sink;
 318
 319   sink = GST_BASE_SINK (object);
 320
 321   switch (prop_id) {
 322     case PROP_HAS_LOOP:
 323       GST_LOCK (sink);
 324       sink->has_loop = g_value_get_boolean (value);
 325       gst_base_sink_set_all_pad_functions (sink);
 326       GST_UNLOCK (sink);
 327       break;
 328     case PROP_HAS_CHAIN:
 329       GST_LOCK (sink);
 330       sink->has_chain = g_value_get_boolean (value);
 331       gst_base_sink_set_all_pad_functions (sink);
 332       GST_UNLOCK (sink);
 333       break;
 334     case PROP_PREROLL_QUEUE_LEN:
 335       /* preroll lock necessary to serialize with finish_preroll */
 336       GST_PREROLL_LOCK (sink->sinkpad);
 337       sink->preroll_queue_max_len = g_value_get_uint (value);
 338       GST_PREROLL_UNLOCK (sink->sinkpad);
 339       break;
 340     default:
 341       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 342       break;
 343   }
 344 }
 345
 346 static void
 347 gst_base_sink_get_property (GObject * object, guint prop_id, GValue * value,
 348     GParamSpec * pspec)
 349 {
 350   GstBaseSink *sink;
 351
 352   sink = GST_BASE_SINK (object);
 353
 354   GST_LOCK (sink);
 355   switch (prop_id) {
 356     case PROP_HAS_LOOP:
 357       g_value_set_boolean (value, sink->has_loop);
 358       break;
 359     case PROP_HAS_CHAIN:
 360       g_value_set_boolean (value, sink->has_chain);
 361       break;
 362     case PROP_PREROLL_QUEUE_LEN:
 363       g_value_set_uint (value, sink->preroll_queue_max_len);
 364       break;
 365     default:
 366       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 367       break;
 368   }
 369   GST_UNLOCK (sink);
 370 }
 371
 372 static GstCaps *
 373 gst_base_sink_get_caps (GstBaseSink * sink)
 374 {
 375   return NULL;
 376 }
 377
 378 static gboolean
 379 gst_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
 380 {
 381   return TRUE;
 382 }
 383
 384 static GstFlowReturn
 385 gst_base_sink_buffer_alloc (GstBaseSink * sink, guint64 offset, guint size,
 386     GstCaps * caps, GstBuffer ** buf)
 387 {
 388   *buf = NULL;
 389   return GST_FLOW_OK;
 390 }
 391
 392 /* with PREROLL_LOCK */
 393 static GstFlowReturn
 394 gst_base_sink_preroll_queue_empty (GstBaseSink * basesink, GstPad * pad)
 395 {
 396   GstMiniObject *obj;
 397   GQueue *q = basesink->preroll_queue;
 398   GstFlowReturn ret;
 399
 400   ret = GST_FLOW_OK;
 401
 402   if (q) {
 403     GST_DEBUG ("emptying queue");
 404     while ((obj = g_queue_pop_head (q))) {
 405       gboolean is_buffer;
 406
 407       is_buffer = GST_IS_BUFFER (obj);
 408       if (is_buffer) {
 409         basesink->preroll_queued--;
 410         basesink->buffers_queued--;
 411       } else {
 412         switch (GST_EVENT_TYPE (obj)) {
 413           case GST_EVENT_EOS:
 414             basesink->preroll_queued--;
 415             break;
 416           default:
 417             break;
 418         }
 419         basesink->events_queued--;
 420       }
 421       /* we release the preroll lock while pushing so that we
 422        * can still flush it while blocking on the clock or
 423        * inside the element. */
 424       GST_PREROLL_UNLOCK (pad);
 425
 426       if (is_buffer) {
 427         GST_DEBUG ("popped buffer %p", obj);
 428         ret = gst_base_sink_handle_buffer (basesink, GST_BUFFER (obj));
 429       } else {
 430         GST_DEBUG ("popped event %p", obj);
 431         gst_base_sink_handle_event (basesink, GST_EVENT (obj));
 432         ret = GST_FLOW_OK;
 433       }
 434
 435       GST_PREROLL_LOCK (pad);
 436     }
 437     GST_DEBUG ("queue empty");
 438   }
 439   return ret;
 440 }
 441
 442 /* with PREROLL_LOCK */
 443 static void
 444 gst_base_sink_preroll_queue_flush (GstBaseSink * basesink, GstPad * pad)
 445 {
 446   GstMiniObject *obj;
 447   GQueue *q = basesink->preroll_queue;
 448
 449   GST_DEBUG ("flushing queue %p", basesink);
 450   if (q) {
 451     while ((obj = g_queue_pop_head (q))) {
 452       GST_DEBUG ("popped %p", obj);
 453       gst_mini_object_unref (obj);
 454     }
 455   }
 456   /* we can't have EOS anymore now */
 457   basesink->eos = FALSE;
 458   basesink->preroll_queued = 0;
 459   basesink->buffers_queued = 0;
 460   basesink->events_queued = 0;
 461   basesink->have_preroll = FALSE;
 462   /* and signal any waiters now */
 463   GST_PREROLL_SIGNAL (pad);
 464 }
 465
 466 /* with STREAM_LOCK */
 467 static GstFlowReturn
 468 gst_base_sink_handle_object (GstBaseSink * basesink, GstPad * pad,
 469     GstMiniObject * obj)
 470 {
 471   gint length;
 472   gboolean have_event;
 473
 474   GST_PREROLL_LOCK (pad);
 475   /* push object on the queue */
 476   GST_DEBUG ("push on queue %p", basesink, obj);
 477   g_queue_push_tail (basesink->preroll_queue, obj);
 478
 479   have_event = GST_IS_EVENT (obj);
 480   if (have_event) {
 481     GstEvent *event = GST_EVENT (obj);
 482
 483     switch (GST_EVENT_TYPE (obj)) {
 484       case GST_EVENT_EOS:
 485         basesink->preroll_queued++;
 486         basesink->eos = TRUE;
 487         break;
 488       case GST_EVENT_NEWSEGMENT:
 489       {
 490         GstFormat format;
 491         gdouble rate;
 492
 493         /* the newsegment event is needed to bring the buffer timestamps to the
 494          * stream time */
 495         gst_event_parse_newsegment (event, &rate, &format,
 496             &basesink->discont_start, &basesink->discont_stop, NULL);
 497
 498         if (format != GST_FORMAT_TIME) {
 499           /* this means this sink will not be able to sync to the clock */
 500           basesink->discont_start = 0;
 501           basesink->discont_stop = 0;
 502         }
 503         basesink->have_discont = TRUE;
 504
 505         GST_DEBUG ("received DISCONT %" GST_TIME_FORMAT "-%" GST_TIME_FORMAT,
 506             GST_TIME_ARGS (basesink->discont_start),
 507             GST_TIME_ARGS (basesink->discont_stop));
 508         break;
 509       }
 510       default:
 511         break;
 512     }
 513     basesink->events_queued++;
 514   } else {
 515     if (!basesink->have_discont) {
 516       GST_ELEMENT_ERROR (basesink, STREAM, STOPPED,
 517           ("received buffer without a discont"),
 518           ("received buffer without a discont"));
 519     }
 520     basesink->preroll_queued++;
 521     basesink->buffers_queued++;
 522   }
 523   GST_DEBUG ("now %d preroll, %d buffers, %d events on queue",
 524       basesink->preroll_queued,
 525       basesink->buffers_queued, basesink->events_queued);
 526
 527   /* check if we are prerolling */
 528   if (!basesink->need_preroll)
 529     goto no_preroll;
 530
 531   /* there is a buffer queued */
 532   if (basesink->buffers_queued == 1) {
 533     GST_DEBUG ("do preroll %p", obj);
 534
 535     /* if it's a buffer, we need to call the preroll method */
 536     if (GST_IS_BUFFER (obj)) {
 537       GstBaseSinkClass *bclass;
 538       GstFlowReturn pres;
 539
 540       bclass = GST_BASE_SINK_GET_CLASS (basesink);
 541       if (bclass->preroll)
 542         if ((pres =
 543                 bclass->preroll (basesink, GST_BUFFER (obj))) != GST_FLOW_OK)
 544           goto preroll_failed;
 545     }
 546   }
 547   length = basesink->preroll_queued;
 548   GST_DEBUG ("prerolled length %d", length);
 549
 550   if (length == 1) {
 551     gint t;
 552
 553     basesink->have_preroll = TRUE;
 554     /* we are prerolling */
 555     GST_PREROLL_UNLOCK (pad);
 556
 557     /* have to release STREAM_LOCK as we cannot take the STATE_LOCK
 558      * inside the STREAM_LOCK */
 559     t = GST_STREAM_UNLOCK_FULL (pad);
 560     GST_DEBUG ("released stream lock %d times", t);
 561     if (t <= 0) {
 562       GST_WARNING ("STREAM_LOCK should have been locked !!");
 563       g_warning ("STREAM_LOCK should have been locked !!");
 564     }
 565
 566     /* now we commit our state */
 567     GST_STATE_LOCK (basesink);
 568     GST_DEBUG ("commit state %p >", basesink);
 569     gst_element_commit_state (GST_ELEMENT (basesink));
 570     GST_STATE_UNLOCK (basesink);
 571
 572     /* reacquire stream lock, pad could be flushing now */
 573     /* FIXME in glib, if t==0, the lock is still taken... hmmm */
 574     if (t > 0)
 575       GST_STREAM_LOCK_FULL (pad, t);
 576
 577     /* and wait if needed */
 578     GST_PREROLL_LOCK (pad);
 579
 580     GST_LOCK (pad);
 581     if (G_UNLIKELY (GST_PAD_IS_FLUSHING (pad)))
 582       goto flushing;
 583     GST_UNLOCK (pad);
 584
 585     /* it is possible that the application set the state to PLAYING
 586      * now in which case we don't need to block anymore. */
 587     if (!basesink->need_preroll)
 588       goto no_preroll;
 589
 590     length = basesink->preroll_queued;
 591
 592     g_assert (length == 1);
 593   }
 594
 595   /* see if we need to block now. We cannot block on events, only
 596    * on buffers, the reason is that events can be sent from the
 597    * application thread and we don't want to block there. */
 598   if (length > basesink->preroll_queue_max_len && !have_event) {
 599     /* block until the state changes, or we get a flush, or something */
 600     GST_DEBUG ("element %s waiting to finish preroll",
 601         GST_ELEMENT_NAME (basesink));
 602     GST_PREROLL_WAIT (pad);
 603     GST_DEBUG ("done preroll");
 604   }
 605   GST_LOCK (pad);
 606   if (G_UNLIKELY (GST_PAD_IS_FLUSHING (pad)))
 607     goto flushing;
 608   GST_UNLOCK (pad);
 609
 610   GST_PREROLL_UNLOCK (pad);
 611
 612   return GST_FLOW_OK;
 613
 614 no_preroll:
 615   {
 616     GstFlowReturn ret;
 617
 618     GST_DEBUG ("no preroll needed");
 619     /* maybe it was another sink that blocked in preroll, need to check for
 620        buffers to drain */
 621     basesink->have_preroll = FALSE;
 622     ret = gst_base_sink_preroll_queue_empty (basesink, pad);
 623     GST_PREROLL_UNLOCK (pad);
 624
 625     return ret;
 626   }
 627 flushing:
 628   {
 629     GST_UNLOCK (pad);
 630     gst_base_sink_preroll_queue_flush (basesink, pad);
 631     GST_PREROLL_UNLOCK (pad);
 632     GST_DEBUG ("pad is flushing");
 633     return GST_FLOW_WRONG_STATE;
 634   }
 635 preroll_failed:
 636   {
 637     gint t;
 638
 639     GST_DEBUG ("preroll failed");
 640     gst_base_sink_preroll_queue_flush (basesink, pad);
 641     GST_PREROLL_UNLOCK (pad);
 642
 643     /* have to release STREAM_LOCK as we cannot take the STATE_LOCK
 644      * inside the STREAM_LOCK */
 645     t = GST_STREAM_UNLOCK_FULL (pad);
 646     GST_DEBUG ("released stream lock %d times", t);
 647     if (t <= 0) {
 648       GST_WARNING ("STREAM_LOCK should have been locked !!");
 649       g_warning ("STREAM_LOCK should have been locked !!");
 650     }
 651
 652     /* now we abort our state */
 653     GST_STATE_LOCK (basesink);
 654     GST_DEBUG ("abort state %p >", basesink);
 655     gst_element_abort_state (GST_ELEMENT (basesink));
 656     GST_STATE_UNLOCK (basesink);
 657
 658     /* reacquire stream lock, pad could be flushing now */
 659     if (t > 0)
 660       GST_STREAM_LOCK_FULL (pad, t);
 661
 662     return GST_FLOW_ERROR;
 663   }
 664 }
 665
 666 static gboolean
 667 gst_base_sink_event (GstPad * pad, GstEvent * event)
 668 {
 669   GstBaseSink *basesink;
 670   gboolean result = TRUE;
 671   GstBaseSinkClass *bclass;
 672
 673   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
 674
 675   bclass = GST_BASE_SINK_GET_CLASS (basesink);
 676
 677   GST_DEBUG ("event %p", event);
 678
 679   switch (GST_EVENT_TYPE (event)) {
 680     case GST_EVENT_EOS:
 681     {
 682       GstFlowReturn ret;
 683
 684       GST_STREAM_LOCK (pad);
 685       /* EOS also finishes the preroll */
 686       ret =
 687           gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (event));
 688       GST_STREAM_UNLOCK (pad);
 689       break;
 690     }
 691     case GST_EVENT_NEWSEGMENT:
 692     {
 693       GstFlowReturn ret;
 694
 695       GST_STREAM_LOCK (pad);
 696       ret =
 697           gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (event));
 698       GST_STREAM_UNLOCK (pad);
 699       break;
 700     }
 701     case GST_EVENT_FLUSH_START:
 702       /* make sure we are not blocked on the clock also clear any pending
 703        * eos state. */
 704       if (bclass->event)
 705         bclass->event (basesink, event);
 706
 707       GST_PREROLL_LOCK (pad);
 708       /* we need preroll after the flush */
 709       basesink->need_preroll = TRUE;
 710       /* unlock from a possible state change/preroll */
 711       gst_base_sink_preroll_queue_flush (basesink, pad);
 712
 713       GST_LOCK (basesink);
 714       if (basesink->clock_id) {
 715         gst_clock_id_unschedule (basesink->clock_id);
 716       }
 717       GST_UNLOCK (basesink);
 718       GST_PREROLL_UNLOCK (pad);
 719
 720       /* and we need to commit our state again on the next
 721        * prerolled buffer */
 722       GST_STATE_LOCK (basesink);
 723       GST_STREAM_LOCK (pad);
 724       gst_element_lost_state (GST_ELEMENT (basesink));
 725       GST_STREAM_UNLOCK (pad);
 726       GST_STATE_UNLOCK (basesink);
 727       GST_DEBUG ("event unref %p %p", basesink, event);
 728       gst_event_unref (event);
 729       break;
 730     case GST_EVENT_FLUSH_STOP:
 731       if (bclass->event)
 732         bclass->event (basesink, event);
 733
 734       /* now we are completely unblocked and the _chain method
 735        * will return */
 736       GST_STREAM_LOCK (pad);
 737       GST_STREAM_UNLOCK (pad);
 738
 739       GST_DEBUG ("event unref %p %p", basesink, event);
 740       gst_event_unref (event);
 741       break;
 742     default:
 743       gst_event_unref (event);
 744       break;
 745   }
 746   gst_object_unref (basesink);
 747
 748   return result;
 749 }
 750
 751 /* default implementation to calculate the start and end
 752  * timestamps on a buffer, subclasses can override
 753  */
 754 static void
 755 gst_base_sink_get_times (GstBaseSink * basesink, GstBuffer * buffer,
 756     GstClockTime * start, GstClockTime * end)
 757 {
 758   GstClockTime timestamp, duration;
 759
 760   timestamp = GST_BUFFER_TIMESTAMP (buffer);
 761   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
 762     GstClockTimeDiff diff;
 763
 764     /* bring timestamp to stream time using last
 765      * discont offset. */
 766     if ((diff = timestamp - basesink->discont_start) < 0)
 767       goto too_late;
 768
 769     /* get duration to calculate end time */
 770     duration = GST_BUFFER_DURATION (buffer);
 771     if (GST_CLOCK_TIME_IS_VALID (duration)) {
 772       *end = diff + duration;
 773     }
 774     *start = diff;
 775   }
 776   return;
 777
 778 too_late:
 779   {
 780     *start = GST_CLOCK_TIME_NONE;
 781     *end = GST_CLOCK_TIME_NONE;
 782   }
 783 }
 784
 785 /* perform synchronisation on a buffer
 786  *
 787  * 1) check if we have a clock, if not, do nothing
 788  * 2) calculate the start and end time of the buffer
 789  * 3) create a single shot notification to wait on
 790  *    the clock, save the entry so we can unlock it
 791  * 4) wait on the clock, this blocks
 792  * 5) unref the clockid again
 793  */
 794 static gboolean
 795 gst_base_sink_do_sync (GstBaseSink * basesink, GstBuffer * buffer)
 796 {
 797   gboolean result = TRUE;
 798
 799   if (basesink->clock) {
 800     GstClockTime start, end;
 801     GstBaseSinkClass *bclass;
 802
 803     bclass = GST_BASE_SINK_GET_CLASS (basesink);
 804     start = end = -1;
 805     if (bclass->get_times)
 806       bclass->get_times (basesink, buffer, &start, &end);
 807
 808     GST_DEBUG_OBJECT (basesink, "got times start: %" GST_TIME_FORMAT
 809         ", end: %" GST_TIME_FORMAT, GST_TIME_ARGS (start), GST_TIME_ARGS (end));
 810
 811     if (GST_CLOCK_TIME_IS_VALID (start)) {
 812       GstClockReturn ret;
 813       GstClockTime base_time;
 814
 815       GST_LOCK (basesink);
 816       base_time = GST_ELEMENT (basesink)->base_time;
 817
 818       GST_LOG_OBJECT (basesink,
 819           "waiting for clock, base time %" GST_TIME_FORMAT,
 820           GST_TIME_ARGS (base_time));
 821       /* save clock id so that we can unlock it if needed */
 822       basesink->clock_id = gst_clock_new_single_shot_id (basesink->clock,
 823           start + base_time);
 824       basesink->end_time = end;
 825       GST_UNLOCK (basesink);
 826
 827       ret = gst_clock_id_wait (basesink->clock_id, NULL);
 828
 829       GST_LOCK (basesink);
 830       if (basesink->clock_id) {
 831         gst_clock_id_unref (basesink->clock_id);
 832         basesink->clock_id = NULL;
 833       }
 834       GST_UNLOCK (basesink);
 835
 836       GST_LOG_OBJECT (basesink, "clock entry done: %d", ret);
 837       if (ret == GST_CLOCK_UNSCHEDULED)
 838         result = FALSE;
 839     }
 840   }
 841   return result;
 842 }
 843
 844
 845 /* handle an event
 846  *
 847  * 2) render the event
 848  * 3) unref the event
 849  */
 850 static inline gboolean
 851 gst_base_sink_handle_event (GstBaseSink * basesink, GstEvent * event)
 852 {
 853   GstBaseSinkClass *bclass;
 854   gboolean ret;
 855
 856   switch (GST_EVENT_TYPE (event)) {
 857     case GST_EVENT_EOS:
 858       GST_LOCK (basesink);
 859       if (basesink->clock) {
 860         /* wait for last buffer to finish if we have a valid end time */
 861         if (GST_CLOCK_TIME_IS_VALID (basesink->end_time)) {
 862           basesink->clock_id = gst_clock_new_single_shot_id (basesink->clock,
 863               basesink->end_time + GST_ELEMENT (basesink)->base_time);
 864           GST_UNLOCK (basesink);
 865
 866           gst_clock_id_wait (basesink->clock_id, NULL);
 867
 868           GST_LOCK (basesink);
 869           if (basesink->clock_id) {
 870             gst_clock_id_unref (basesink->clock_id);
 871             basesink->clock_id = NULL;
 872           }
 873           basesink->end_time = GST_CLOCK_TIME_NONE;
 874         }
 875       }
 876       GST_UNLOCK (basesink);
 877       break;
 878     default:
 879       break;
 880   }
 881
 882   bclass = GST_BASE_SINK_GET_CLASS (basesink);
 883   if (bclass->event)
 884     ret = bclass->event (basesink, event);
 885   else
 886     ret = TRUE;
 887
 888   switch (GST_EVENT_TYPE (event)) {
 889     case GST_EVENT_EOS:
 890       GST_PREROLL_LOCK (basesink->sinkpad);
 891       /* if we are still EOS, we can post the EOS message */
 892       if (basesink->eos) {
 893         /* ok, now we can post the message */
 894         GST_DEBUG_OBJECT (basesink, "Now posting EOS");
 895         gst_element_post_message (GST_ELEMENT (basesink),
 896             gst_message_new_eos (GST_OBJECT (basesink)));
 897       }
 898       GST_PREROLL_UNLOCK (basesink->sinkpad);
 899       break;
 900     default:
 901       break;
 902   }
 903
 904   GST_DEBUG ("event unref %p %p", basesink, event);
 905   gst_event_unref (event);
 906
 907   return ret;
 908 }
 909
 910 /* handle a buffer
 911  *
 912  * 1) first sync on the buffer
 913  * 2) render the buffer
 914  * 3) unref the buffer
 915  */
 916 static inline GstFlowReturn
 917 gst_base_sink_handle_buffer (GstBaseSink * basesink, GstBuffer * buf)
 918 {
 919   GstBaseSinkClass *bclass;
 920   GstFlowReturn ret;
 921
 922   gst_base_sink_do_sync (basesink, buf);
 923
 924   bclass = GST_BASE_SINK_GET_CLASS (basesink);
 925   if (bclass->render)
 926     ret = bclass->render (basesink, buf);
 927   else
 928     ret = GST_FLOW_OK;
 929
 930   GST_DEBUG ("buffer unref after render %p", basesink, buf);
 931   gst_buffer_unref (buf);
 932
 933   return ret;
 934 }
 935
 936 static GstFlowReturn
 937 gst_base_sink_chain (GstPad * pad, GstBuffer * buf)
 938 {
 939   GstBaseSink *basesink;
 940   GstFlowReturn result;
 941
 942   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
 943
 944   result = gst_base_sink_handle_object (basesink, pad, GST_MINI_OBJECT (buf));
 945
 946   gst_object_unref (basesink);
 947
 948   return result;
 949 }
 950
 951 /* FIXME, not all sinks can operate in pull mode
 952  */
 953 static void
 954 gst_base_sink_loop (GstPad * pad)
 955 {
 956   GstBaseSink *basesink;
 957   GstBuffer *buf = NULL;
 958   GstFlowReturn result;
 959
 960   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
 961
 962   g_assert (basesink->pad_mode == GST_ACTIVATE_PULL);
 963
 964   result = gst_pad_pull_range (pad, basesink->offset, DEFAULT_SIZE, &buf);
 965   if (result != GST_FLOW_OK)
 966     goto paused;
 967
 968   result = gst_base_sink_chain (pad, buf);
 969   if (result != GST_FLOW_OK)
 970     goto paused;
 971
 972   gst_object_unref (basesink);
 973
 974   /* default */
 975   return;
 976
 977 paused:
 978   {
 979     gst_object_unref (basesink);
 980     gst_pad_pause_task (pad);
 981     return;
 982   }
 983 }
 984
 985 static gboolean
 986 gst_base_sink_deactivate (GstBaseSink * basesink, GstPad * pad)
 987 {
 988   gboolean result = FALSE;
 989   GstBaseSinkClass *bclass;
 990
 991   bclass = GST_BASE_SINK_GET_CLASS (basesink);
 992
 993   /* step 1, unblock clock sync (if any) or any other blocking thing */
 994   GST_PREROLL_LOCK (pad);
 995   GST_LOCK (basesink);
 996   if (basesink->clock_id) {
 997     gst_clock_id_unschedule (basesink->clock_id);
 998   }
 999   GST_UNLOCK (basesink);
1000
1001   /* unlock any subclasses */
1002   if (bclass->unlock)
1003     bclass->unlock (basesink);
1004
1005   /* flush out the data thread if it's locked in finish_preroll */
1006   basesink->need_preroll = FALSE;
1007   gst_base_sink_preroll_queue_flush (basesink, pad);
1008   GST_PREROLL_SIGNAL (pad);
1009   GST_PREROLL_UNLOCK (pad);
1010
1011   /* step 2, make sure streaming finishes */
1012   result = gst_pad_stop_task (pad);
1013
1014   return result;
1015 }
1016
1017 static gboolean
1018 gst_base_sink_activate_push (GstPad * pad, gboolean active)
1019 {
1020   gboolean result = FALSE;
1021   GstBaseSink *basesink;
1022
1023   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1024
1025   if (active) {
1026     if (!basesink->has_chain)
1027       goto done;
1028     result = TRUE;
1029   } else {
1030     result = gst_base_sink_deactivate (basesink, pad);
1031   }
1032   basesink->pad_mode = GST_ACTIVATE_PUSH;
1033
1034 done:
1035   gst_object_unref (basesink);
1036
1037   return result;
1038 }
1039
1040 /* this won't get called until we implement an activate function */
1041 static gboolean
1042 gst_base_sink_activate_pull (GstPad * pad, gboolean active)
1043 {
1044   gboolean result = FALSE;
1045   GstBaseSink *basesink;
1046
1047   basesink = GST_BASE_SINK (gst_pad_get_parent (pad));
1048
1049   if (active) {
1050     /* if we have a scheduler we can start the task */
1051     if (!basesink->has_loop)
1052       goto done;
1053     result =
1054         gst_pad_start_task (pad, (GstTaskFunction) gst_base_sink_loop, pad);
1055   } else {
1056     result = gst_base_sink_deactivate (basesink, pad);
1057   }
1058 done:
1059   gst_object_unref (basesink);
1060
1061   return result;
1062 }
1063
1064 static GstElementStateReturn
1065 gst_base_sink_change_state (GstElement * element)
1066 {
1067   GstElementStateReturn ret = GST_STATE_SUCCESS;
1068   GstBaseSink *basesink = GST_BASE_SINK (element);
1069   GstElementState transition = GST_STATE_TRANSITION (element);
1070   GstBaseSinkClass *bclass;
1071
1072   bclass = GST_BASE_SINK_GET_CLASS (basesink);
1073
1074   switch (transition) {
1075     case GST_STATE_NULL_TO_READY:
1076       if (bclass->start)
1077         if (!bclass->start (basesink))
1078           goto start_failed;
1079       break;
1080     case GST_STATE_READY_TO_PAUSED:
1081       /* need to complete preroll before this state change completes, there
1082        * is no data flow in READY so we can safely assume we need to preroll. */
1083       basesink->offset = 0;
1084       GST_PREROLL_LOCK (basesink->sinkpad);
1085       basesink->have_preroll = FALSE;
1086       basesink->need_preroll = TRUE;
1087       GST_PREROLL_UNLOCK (basesink->sinkpad);
1088       basesink->have_discont = FALSE;
1089       basesink->discont_start = 0;
1090       basesink->discont_stop = 0;
1091       ret = GST_STATE_ASYNC;
1092       break;
1093     case GST_STATE_PAUSED_TO_PLAYING:
1094     {
1095       GST_PREROLL_LOCK (basesink->sinkpad);
1096       /* if we have EOS, we should empty the queue now as there will
1097        * be no more data received in the chain function.
1098        * FIXME, this could block the state change function too long when
1099        * we are pushing and syncing the buffers, better start a new
1100        * thread to do this. */
1101       if (basesink->eos) {
1102         gst_base_sink_preroll_queue_empty (basesink, basesink->sinkpad);
1103       }
1104       /* don't need the preroll anymore */
1105       basesink->need_preroll = FALSE;
1106       /* now let it play */
1107       GST_PREROLL_SIGNAL (basesink->sinkpad);
1108       GST_PREROLL_UNLOCK (basesink->sinkpad);
1109       break;
1110     }
1111     default:
1112       break;
1113   }
1114
1115   GST_ELEMENT_CLASS (parent_class)->change_state (element);
1116
1117   switch (transition) {
1118     case GST_STATE_PLAYING_TO_PAUSED:
1119     {
1120       GstBaseSinkClass *bclass;
1121
1122       bclass = GST_BASE_SINK_GET_CLASS (basesink);
1123
1124       GST_PREROLL_LOCK (basesink->sinkpad);
1125       GST_LOCK (basesink);
1126       /* unlock clock wait if any */
1127       if (basesink->clock_id) {
1128         gst_clock_id_unschedule (basesink->clock_id);
1129       }
1130       GST_UNLOCK (basesink);
1131
1132       /* unlock any subclasses */
1133       if (bclass->unlock)
1134         bclass->unlock (basesink);
1135
1136       /* if we don't have a preroll buffer and we have not received EOS,
1137        * we need to wait for a preroll */
1138       GST_DEBUG ("have_preroll: %d, EOS: %d", basesink->have_preroll,
1139           basesink->eos);
1140       if (!basesink->have_preroll && !basesink->eos) {
1141         basesink->need_preroll = TRUE;
1142         ret = GST_STATE_ASYNC;
1143       }
1144       GST_PREROLL_UNLOCK (basesink->sinkpad);
1145       break;
1146     }
1147     case GST_STATE_PAUSED_TO_READY:
1148       break;
1149     case GST_STATE_READY_TO_NULL:
1150       if (bclass->stop)
1151         if (!bclass->stop (basesink)) {
1152           GST_WARNING ("failed to stop");
1153         }
1154       break;
1155     default:
1156       break;
1157   }
1158
1159   return ret;
1160
1161   /* ERRORS */
1162 start_failed:
1163   {
1164     GST_DEBUG ("failed to start");
1165     return GST_STATE_FAILURE;
1166   }
1167 }