plugins/elements/gsttee.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *               2000,2001,2002,2003,2004,2005 Wim Taymans <wim@fluendo.com>
   4  *               2007 Wim Taymans <wim.taymans@gmail.com>
   5  *
   6  * gsttee.c: Tee element, one in N out
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Library General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Library General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Library General Public
  19  * License along with this library; if not, write to the
  20  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  21  * Boston, MA 02111-1307, USA.
  22  */
  23
  24 /**
  25  * SECTION:element-tee
  26  * @see_also: #GstIdentity
  27  *
  28  * Split data to multiple pads.
  29  */
  30
  31 #ifdef HAVE_CONFIG_H
  32 #  include "config.h"
  33 #endif
  34
  35 #include "gsttee.h"
  36
  37 #include <string.h>
  38
  39 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  40     GST_PAD_SINK,
  41     GST_PAD_ALWAYS,
  42     GST_STATIC_CAPS_ANY);
  43
  44 GST_DEBUG_CATEGORY_STATIC (gst_tee_debug);
  45 #define GST_CAT_DEFAULT gst_tee_debug
  46
  47 #define GST_TYPE_TEE_PULL_MODE (gst_tee_pull_mode_get_type())
  48 static GType
  49 gst_tee_pull_mode_get_type (void)
  50 {
  51   static GType type = 0;
  52   static const GEnumValue data[] = {
  53     {GST_TEE_PULL_MODE_NEVER, "Never activate in pull mode", "never"},
  54     {GST_TEE_PULL_MODE_SINGLE, "Only one src pad can be active in pull mode",
  55         "single"},
  56     {0, NULL, NULL},
  57   };
  58
  59   if (!type) {
  60     type = g_enum_register_static ("GstTeePullMode", data);
  61   }
  62   return type;
  63 }
  64
  65 /* lock to protect request pads from being removed while downstream */
  66 #define GST_TEE_DYN_LOCK(tee) g_mutex_lock ((tee)->dyn_lock)
  67 #define GST_TEE_DYN_UNLOCK(tee) g_mutex_unlock ((tee)->dyn_lock)
  68
  69 #define DEFAULT_PROP_NUM_SRC_PADS       0
  70 #define DEFAULT_PROP_HAS_SINK_LOOP      FALSE
  71 #define DEFAULT_PROP_HAS_CHAIN          TRUE
  72 #define DEFAULT_PROP_SILENT             TRUE
  73 #define DEFAULT_PROP_LAST_MESSAGE       NULL
  74 #define DEFAULT_PULL_MODE               GST_TEE_PULL_MODE_NEVER
  75
  76 enum
  77 {
  78   PROP_0,
  79   PROP_NUM_SRC_PADS,
  80   PROP_HAS_SINK_LOOP,
  81   PROP_HAS_CHAIN,
  82   PROP_SILENT,
  83   PROP_LAST_MESSAGE,
  84   PROP_PULL_MODE,
  85   PROP_ALLOC_PAD,
  86 };
  87
  88 static GstStaticPadTemplate tee_src_template = GST_STATIC_PAD_TEMPLATE ("src%d",
  89     GST_PAD_SRC,
  90     GST_PAD_REQUEST,
  91     GST_STATIC_CAPS_ANY);
  92
  93 #define _do_init(bla) \
  94     GST_DEBUG_CATEGORY_INIT (gst_tee_debug, "tee", 0, "tee element");
  95
  96 GST_BOILERPLATE_FULL (GstTee, gst_tee, GstElement, GST_TYPE_ELEMENT, _do_init);
  97
  98 /* structure and quark to keep track of which pads have been pushed */
  99 static GQuark push_data;
 100
 101 typedef struct
 102 {
 103   gboolean pushed;
 104   GstFlowReturn result;
 105   gboolean removed;
 106 } PushData;
 107
 108 static GstPad *gst_tee_request_new_pad (GstElement * element,
 109     GstPadTemplate * temp, const gchar * unused);
 110 static void gst_tee_release_pad (GstElement * element, GstPad * pad);
 111
 112 static void gst_tee_finalize (GObject * object);
 113 static void gst_tee_set_property (GObject * object, guint prop_id,
 114     const GValue * value, GParamSpec * pspec);
 115 static void gst_tee_get_property (GObject * object, guint prop_id,
 116     GValue * value, GParamSpec * pspec);
 117
 118 static GstFlowReturn gst_tee_chain (GstPad * pad, GstBuffer * buffer);
 119 static GstFlowReturn gst_tee_chain_list (GstPad * pad, GstBufferList * list);
 120 static GstFlowReturn gst_tee_buffer_alloc (GstPad * pad, guint64 offset,
 121     guint size, GstCaps * caps, GstBuffer ** buf);
 122 static gboolean gst_tee_sink_acceptcaps (GstPad * pad, GstCaps * caps);
 123 static gboolean gst_tee_sink_activate_push (GstPad * pad, gboolean active);
 124 static gboolean gst_tee_src_check_get_range (GstPad * pad);
 125 static gboolean gst_tee_src_activate_pull (GstPad * pad, gboolean active);
 126 static GstFlowReturn gst_tee_src_get_range (GstPad * pad, guint64 offset,
 127     guint length, GstBuffer ** buf);
 128
 129
 130 static void
 131 gst_tee_base_init (gpointer g_class)
 132 {
 133   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
 134
 135   gst_element_class_set_details_simple (gstelement_class,
 136       "Tee pipe fitting",
 137       "Generic",
 138       "1-to-N pipe fitting",
 139       "Erik Walthinsen <omega@cse.ogi.edu>, " "Wim Taymans <wim@fluendo.com>");
 140   gst_element_class_add_pad_template (gstelement_class,
 141       gst_static_pad_template_get (&sinktemplate));
 142   gst_element_class_add_pad_template (gstelement_class,
 143       gst_static_pad_template_get (&tee_src_template));
 144
 145   push_data = g_quark_from_static_string ("tee-push-data");
 146 }
 147
 148 static void
 149 gst_tee_finalize (GObject * object)
 150 {
 151   GstTee *tee;
 152
 153   tee = GST_TEE (object);
 154
 155   g_free (tee->last_message);
 156
 157   g_mutex_free (tee->dyn_lock);
 158
 159   G_OBJECT_CLASS (parent_class)->finalize (object);
 160 }
 161
 162 static void
 163 gst_tee_class_init (GstTeeClass * klass)
 164 {
 165   GObjectClass *gobject_class;
 166   GstElementClass *gstelement_class;
 167
 168   gobject_class = G_OBJECT_CLASS (klass);
 169   gstelement_class = GST_ELEMENT_CLASS (klass);
 170
 171   gobject_class->finalize = gst_tee_finalize;
 172   gobject_class->set_property = gst_tee_set_property;
 173   gobject_class->get_property = gst_tee_get_property;
 174
 175   g_object_class_install_property (gobject_class, PROP_NUM_SRC_PADS,
 176       g_param_spec_int ("num-src-pads", "Num Src Pads",
 177           "The number of source pads", 0, G_MAXINT, DEFAULT_PROP_NUM_SRC_PADS,
 178           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 179   g_object_class_install_property (gobject_class, PROP_HAS_SINK_LOOP,
 180       g_param_spec_boolean ("has-sink-loop", "Has Sink Loop",
 181           "If the element should spawn a thread (unimplemented and deprecated)",
 182           DEFAULT_PROP_HAS_SINK_LOOP,
 183           G_PARAM_CONSTRUCT | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 184   g_object_class_install_property (gobject_class, PROP_HAS_CHAIN,
 185       g_param_spec_boolean ("has-chain", "Has Chain",
 186           "If the element can operate in push mode", DEFAULT_PROP_HAS_CHAIN,
 187           G_PARAM_CONSTRUCT | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 188   g_object_class_install_property (gobject_class, PROP_SILENT,
 189       g_param_spec_boolean ("silent", "Silent",
 190           "Don't produce last_message events", DEFAULT_PROP_SILENT,
 191           G_PARAM_CONSTRUCT | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 192   g_object_class_install_property (gobject_class, PROP_LAST_MESSAGE,
 193       g_param_spec_string ("last-message", "Last Message",
 194           "The message describing current status", DEFAULT_PROP_LAST_MESSAGE,
 195           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 196   g_object_class_install_property (gobject_class, PROP_PULL_MODE,
 197       g_param_spec_enum ("pull-mode", "Pull mode",
 198           "Behavior of tee in pull mode", GST_TYPE_TEE_PULL_MODE,
 199           DEFAULT_PULL_MODE,
 200           G_PARAM_CONSTRUCT | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 201   g_object_class_install_property (gobject_class, PROP_ALLOC_PAD,
 202       g_param_spec_object ("alloc-pad", "Allocation Src Pad",
 203           "The pad used for gst_pad_alloc_buffer", GST_TYPE_PAD,
 204           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 205
 206   gstelement_class->request_new_pad =
 207       GST_DEBUG_FUNCPTR (gst_tee_request_new_pad);
 208   gstelement_class->release_pad = GST_DEBUG_FUNCPTR (gst_tee_release_pad);
 209 }
 210
 211 static void
 212 gst_tee_init (GstTee * tee, GstTeeClass * g_class)
 213 {
 214   tee->dyn_lock = g_mutex_new ();
 215
 216   tee->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 217   tee->sink_mode = GST_ACTIVATE_NONE;
 218
 219   gst_pad_set_setcaps_function (tee->sinkpad,
 220       GST_DEBUG_FUNCPTR (gst_pad_proxy_setcaps));
 221   gst_pad_set_getcaps_function (tee->sinkpad,
 222       GST_DEBUG_FUNCPTR (gst_pad_proxy_getcaps));
 223   gst_pad_set_acceptcaps_function (tee->sinkpad,
 224       GST_DEBUG_FUNCPTR (gst_tee_sink_acceptcaps));
 225   gst_pad_set_bufferalloc_function (tee->sinkpad,
 226       GST_DEBUG_FUNCPTR (gst_tee_buffer_alloc));
 227   gst_pad_set_activatepush_function (tee->sinkpad,
 228       GST_DEBUG_FUNCPTR (gst_tee_sink_activate_push));
 229   gst_pad_set_chain_function (tee->sinkpad, GST_DEBUG_FUNCPTR (gst_tee_chain));
 230   gst_pad_set_chain_list_function (tee->sinkpad,
 231       GST_DEBUG_FUNCPTR (gst_tee_chain_list));
 232   gst_element_add_pad (GST_ELEMENT (tee), tee->sinkpad);
 233
 234   tee->last_message = NULL;
 235 }
 236
 237 static GstPad *
 238 gst_tee_request_new_pad (GstElement * element, GstPadTemplate * templ,
 239     const gchar * unused)
 240 {
 241   gchar *name;
 242   GstPad *srcpad;
 243   GstTee *tee;
 244   GstActivateMode mode;
 245   gboolean res;
 246   PushData *data;
 247
 248   tee = GST_TEE (element);
 249
 250   GST_DEBUG_OBJECT (tee, "requesting pad");
 251
 252   GST_OBJECT_LOCK (tee);
 253   name = g_strdup_printf ("src%d", tee->pad_counter++);
 254
 255   srcpad = gst_pad_new_from_template (templ, name);
 256   g_free (name);
 257
 258   mode = tee->sink_mode;
 259
 260   /* install the data, we automatically free it when the pad is disposed because
 261    * of _release_pad or when the element goes away. */
 262   data = g_new0 (PushData, 1);
 263   data->pushed = FALSE;
 264   data->result = GST_FLOW_NOT_LINKED;
 265   data->removed = FALSE;
 266   g_object_set_qdata_full (G_OBJECT (srcpad), push_data, data, g_free);
 267
 268   GST_OBJECT_UNLOCK (tee);
 269
 270   switch (mode) {
 271     case GST_ACTIVATE_PULL:
 272       /* we already have a src pad in pull mode, and our pull mode can only be
 273          SINGLE, so fall through to activate this new pad in push mode */
 274     case GST_ACTIVATE_PUSH:
 275       res = gst_pad_activate_push (srcpad, TRUE);
 276       break;
 277     default:
 278       res = TRUE;
 279       break;
 280   }
 281
 282   if (!res)
 283     goto activate_failed;
 284
 285   gst_pad_set_setcaps_function (srcpad,
 286       GST_DEBUG_FUNCPTR (gst_pad_proxy_setcaps));
 287   gst_pad_set_getcaps_function (srcpad,
 288       GST_DEBUG_FUNCPTR (gst_pad_proxy_getcaps));
 289   gst_pad_set_activatepull_function (srcpad,
 290       GST_DEBUG_FUNCPTR (gst_tee_src_activate_pull));
 291   gst_pad_set_checkgetrange_function (srcpad,
 292       GST_DEBUG_FUNCPTR (gst_tee_src_check_get_range));
 293   gst_pad_set_getrange_function (srcpad,
 294       GST_DEBUG_FUNCPTR (gst_tee_src_get_range));
 295   gst_element_add_pad (GST_ELEMENT_CAST (tee), srcpad);
 296
 297   return srcpad;
 298
 299   /* ERRORS */
 300 activate_failed:
 301   {
 302     gboolean changed = FALSE;
 303
 304     GST_OBJECT_LOCK (tee);
 305     GST_DEBUG_OBJECT (tee, "warning failed to activate request pad");
 306     if (tee->allocpad == srcpad) {
 307       tee->allocpad = NULL;
 308       changed = TRUE;
 309     }
 310     GST_OBJECT_UNLOCK (tee);
 311     gst_object_unref (srcpad);
 312     if (changed) {
 313       g_object_notify (G_OBJECT (tee), "alloc-pad");
 314     }
 315     return NULL;
 316   }
 317 }
 318
 319 static void
 320 gst_tee_release_pad (GstElement * element, GstPad * pad)
 321 {
 322   GstTee *tee;
 323   PushData *data;
 324   gboolean changed = FALSE;
 325
 326   tee = GST_TEE (element);
 327
 328   GST_DEBUG_OBJECT (tee, "releasing pad");
 329
 330   GST_OBJECT_LOCK (tee);
 331   if (tee->allocpad == pad) {
 332     tee->allocpad = NULL;
 333     changed = TRUE;
 334   }
 335   GST_OBJECT_UNLOCK (tee);
 336   if (changed) {
 337     g_object_notify (G_OBJECT (tee), "alloc-pad");
 338   }
 339
 340   /* wait for pending pad_alloc to finish */
 341   GST_TEE_DYN_LOCK (tee);
 342   /* mark the pad as removed so that future pad_alloc fails with NOT_LINKED. */
 343   data = g_object_get_qdata (G_OBJECT (pad), push_data);
 344   data->removed = TRUE;
 345
 346   gst_pad_set_active (pad, FALSE);
 347
 348   gst_element_remove_pad (GST_ELEMENT_CAST (tee), pad);
 349   GST_TEE_DYN_UNLOCK (tee);
 350 }
 351
 352 static void
 353 gst_tee_set_property (GObject * object, guint prop_id, const GValue * value,
 354     GParamSpec * pspec)
 355 {
 356   GstTee *tee = GST_TEE (object);
 357
 358   GST_OBJECT_LOCK (tee);
 359   switch (prop_id) {
 360     case PROP_HAS_SINK_LOOP:
 361       tee->has_sink_loop = g_value_get_boolean (value);
 362       if (tee->has_sink_loop) {
 363         g_warning ("tee will never implement has-sink-loop==TRUE");
 364       }
 365       break;
 366     case PROP_HAS_CHAIN:
 367       tee->has_chain = g_value_get_boolean (value);
 368       break;
 369     case PROP_SILENT:
 370       tee->silent = g_value_get_boolean (value);
 371       break;
 372     case PROP_PULL_MODE:
 373       tee->pull_mode = g_value_get_enum (value);
 374       break;
 375     case PROP_ALLOC_PAD:
 376     {
 377       GstPad *pad = g_value_get_object (value);
 378       GST_OBJECT_LOCK (pad);
 379       if (GST_OBJECT_PARENT (pad) == GST_OBJECT_CAST (object))
 380         tee->allocpad = pad;
 381       else
 382         GST_WARNING_OBJECT (object, "Tried to set alloc pad %s which"
 383             " is not my pad", GST_OBJECT_NAME (pad));
 384       GST_OBJECT_UNLOCK (pad);
 385       break;
 386     }
 387     default:
 388       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 389       break;
 390   }
 391   GST_OBJECT_UNLOCK (tee);
 392 }
 393
 394 static void
 395 gst_tee_get_property (GObject * object, guint prop_id, GValue * value,
 396     GParamSpec * pspec)
 397 {
 398   GstTee *tee = GST_TEE (object);
 399
 400   GST_OBJECT_LOCK (tee);
 401   switch (prop_id) {
 402     case PROP_NUM_SRC_PADS:
 403       g_value_set_int (value, GST_ELEMENT (tee)->numsrcpads);
 404       break;
 405     case PROP_HAS_SINK_LOOP:
 406       g_value_set_boolean (value, tee->has_sink_loop);
 407       break;
 408     case PROP_HAS_CHAIN:
 409       g_value_set_boolean (value, tee->has_chain);
 410       break;
 411     case PROP_SILENT:
 412       g_value_set_boolean (value, tee->silent);
 413       break;
 414     case PROP_LAST_MESSAGE:
 415       g_value_set_string (value, tee->last_message);
 416       break;
 417     case PROP_PULL_MODE:
 418       g_value_set_enum (value, tee->pull_mode);
 419       break;
 420     case PROP_ALLOC_PAD:
 421       g_value_set_object (value, tee->allocpad);
 422       break;
 423     default:
 424       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 425       break;
 426   }
 427   GST_OBJECT_UNLOCK (tee);
 428 }
 429
 430 /* we have no previous source pad we can use to proxy the pad alloc. Loop over
 431  * the source pads, try to alloc a buffer on each one of them. Keep a reference
 432  * to the first pad that succeeds, we will be using it to alloc more buffers
 433  * later.  must be called with the OBJECT_LOCK on tee. */
 434 static GstFlowReturn
 435 gst_tee_find_buffer_alloc (GstTee * tee, guint64 offset, guint size,
 436     GstCaps * caps, GstBuffer ** buf)
 437 {
 438   GstFlowReturn res;
 439   GList *pads;
 440   guint32 cookie;
 441
 442   res = GST_FLOW_NOT_LINKED;
 443
 444 retry:
 445   pads = GST_ELEMENT_CAST (tee)->srcpads;
 446   cookie = GST_ELEMENT_CAST (tee)->pads_cookie;
 447
 448   while (pads) {
 449     GstPad *pad;
 450     PushData *data;
 451
 452     pad = GST_PAD_CAST (pads->data);
 453     gst_object_ref (pad);
 454     GST_DEBUG_OBJECT (tee, "try alloc on pad %s:%s", GST_DEBUG_PAD_NAME (pad));
 455     GST_OBJECT_UNLOCK (tee);
 456
 457     GST_TEE_DYN_LOCK (tee);
 458     data = g_object_get_qdata (G_OBJECT (pad), push_data);
 459     if (!data->removed)
 460       res = gst_pad_alloc_buffer (pad, offset, size, caps, buf);
 461     else
 462       res = GST_FLOW_NOT_LINKED;
 463     GST_TEE_DYN_UNLOCK (tee);
 464
 465     GST_DEBUG_OBJECT (tee, "got return value %d", res);
 466
 467     gst_object_unref (pad);
 468
 469     GST_OBJECT_LOCK (tee);
 470     if (GST_ELEMENT_CAST (tee)->pads_cookie != cookie) {
 471       GST_DEBUG_OBJECT (tee, "pad list changed, restart");
 472       /* pad list changed, restart. If the pad alloc function returned OK we
 473        * need to unref the buffer */
 474       if (res == GST_FLOW_OK)
 475         gst_buffer_unref (*buf);
 476       *buf = NULL;
 477       goto retry;
 478     }
 479     if (res == GST_FLOW_OK) {
 480       GST_DEBUG_OBJECT (tee, "we have a buffer on pad %s:%s",
 481           GST_DEBUG_PAD_NAME (pad));
 482       /* we have a buffer, keep the pad for later and exit the loop. */
 483       tee->allocpad = pad;
 484       GST_OBJECT_UNLOCK (tee);
 485       g_object_notify (G_OBJECT (tee), "alloc-pad");
 486       GST_OBJECT_LOCK (tee);
 487       break;
 488     }
 489     /* no valid buffer, try another pad */
 490     pads = g_list_next (pads);
 491   }
 492
 493   return res;
 494 }
 495
 496 static GstFlowReturn
 497 gst_tee_buffer_alloc (GstPad * pad, guint64 offset, guint size,
 498     GstCaps * caps, GstBuffer ** buf)
 499 {
 500   GstTee *tee;
 501   GstFlowReturn res;
 502   GstPad *allocpad;
 503
 504   tee = GST_TEE (GST_PAD_PARENT (pad));
 505
 506   res = GST_FLOW_NOT_LINKED;
 507
 508   GST_OBJECT_LOCK (tee);
 509   if ((allocpad = tee->allocpad)) {
 510     PushData *data;
 511
 512     /* if we had a previous pad we used for allocating a buffer, continue using
 513      * it. */
 514     GST_DEBUG_OBJECT (tee, "using pad %s:%s for alloc",
 515         GST_DEBUG_PAD_NAME (allocpad));
 516     gst_object_ref (allocpad);
 517     GST_OBJECT_UNLOCK (tee);
 518
 519     GST_TEE_DYN_LOCK (tee);
 520     data = g_object_get_qdata (G_OBJECT (allocpad), push_data);
 521     if (!data->removed)
 522       res = gst_pad_alloc_buffer (allocpad, offset, size, caps, buf);
 523     else
 524       res = GST_FLOW_NOT_LINKED;
 525     GST_TEE_DYN_UNLOCK (tee);
 526
 527     gst_object_unref (allocpad);
 528
 529     GST_OBJECT_LOCK (tee);
 530   }
 531   /* either we failed to alloc on the the previous pad or we did not have a
 532    * previous pad. */
 533   if (res == GST_FLOW_NOT_LINKED) {
 534     /* find a new pad to alloc a buffer on */
 535     GST_DEBUG_OBJECT (tee, "finding pad for alloc");
 536     res = gst_tee_find_buffer_alloc (tee, offset, size, caps, buf);
 537   }
 538   GST_OBJECT_UNLOCK (tee);
 539
 540   return res;
 541 }
 542
 543 /* on the sink we accept caps that are acceptable to all srcpads */
 544 static gboolean
 545 gst_tee_sink_acceptcaps (GstPad * pad, GstCaps * caps)
 546 {
 547   GstTee *tee;
 548   gboolean res, done;
 549   GstIterator *it;
 550
 551   tee = GST_TEE (GST_PAD_PARENT (pad));
 552
 553   it = gst_element_iterate_src_pads (GST_ELEMENT_CAST (tee));
 554
 555   res = TRUE;
 556   done = FALSE;
 557   while (!done && res) {
 558     gpointer item;
 559
 560     switch (gst_iterator_next (it, &item)) {
 561       case GST_ITERATOR_OK:
 562         res &= gst_pad_peer_accept_caps (GST_PAD_CAST (item), caps);
 563         gst_object_unref (item);
 564         break;
 565       case GST_ITERATOR_RESYNC:
 566         res = TRUE;
 567         gst_iterator_resync (it);
 568         break;
 569       case GST_ITERATOR_ERROR:
 570         res = FALSE;
 571         done = TRUE;
 572         break;
 573       case GST_ITERATOR_DONE:
 574         done = TRUE;
 575         break;
 576     }
 577   }
 578   gst_iterator_free (it);
 579
 580   return res;
 581 }
 582
 583 static void
 584 gst_tee_do_message (GstTee * tee, GstPad * pad, gpointer data, gboolean is_list)
 585 {
 586   GST_OBJECT_LOCK (tee);
 587   g_free (tee->last_message);
 588   if (is_list) {
 589     tee->last_message =
 590         g_strdup_printf ("chain-list   ******* (%s:%s)t %p",
 591         GST_DEBUG_PAD_NAME (pad), data);
 592   } else {
 593     tee->last_message =
 594         g_strdup_printf ("chain        ******* (%s:%s)t (%d bytes, %"
 595         G_GUINT64_FORMAT ") %p", GST_DEBUG_PAD_NAME (pad),
 596         GST_BUFFER_SIZE (data), GST_BUFFER_TIMESTAMP (data), data);
 597   }
 598   GST_OBJECT_UNLOCK (tee);
 599   g_object_notify (G_OBJECT (tee), "last_message");
 600 }
 601
 602 static GstFlowReturn
 603 gst_tee_do_push (GstTee * tee, GstPad * pad, gpointer data, gboolean is_list)
 604 {
 605   GstFlowReturn res;
 606
 607   /* Push */
 608   if (pad == tee->pull_pad) {
 609     /* don't push on the pad we're pulling from */
 610     res = GST_FLOW_OK;
 611   } else if (is_list) {
 612     res =
 613         gst_pad_push_list (pad,
 614         gst_buffer_list_ref (GST_BUFFER_LIST_CAST (data)));
 615   } else {
 616     res = gst_pad_push (pad, gst_buffer_ref (GST_BUFFER_CAST (data)));
 617   }
 618   return res;
 619 }
 620
 621 static void
 622 clear_pads (GstPad * pad, GstTee * tee)
 623 {
 624   PushData *data;
 625
 626   data = g_object_get_qdata (G_OBJECT (pad), push_data);
 627
 628   /* the data must be there or we have a screwed up internal state */
 629   g_assert (data != NULL);
 630
 631   data->pushed = FALSE;
 632   data->result = GST_FLOW_NOT_LINKED;
 633 }
 634
 635 static GstFlowReturn
 636 gst_tee_handle_data (GstTee * tee, gpointer data, gboolean is_list)
 637 {
 638   GList *pads;
 639   guint32 cookie;
 640   GstFlowReturn ret, cret;
 641
 642   if (G_UNLIKELY (!tee->silent))
 643     gst_tee_do_message (tee, tee->sinkpad, data, is_list);
 644
 645   GST_OBJECT_LOCK (tee);
 646
 647   /* mark all pads as 'not pushed on yet' */
 648   g_list_foreach (GST_ELEMENT_CAST (tee)->srcpads, (GFunc) clear_pads, tee);
 649
 650 restart:
 651   cret = GST_FLOW_NOT_LINKED;
 652   pads = GST_ELEMENT_CAST (tee)->srcpads;
 653   cookie = GST_ELEMENT_CAST (tee)->pads_cookie;
 654
 655   while (pads) {
 656     GstPad *pad;
 657     PushData *pdata;
 658
 659     pad = GST_PAD_CAST (pads->data);
 660
 661     /* get the private data, something is really wrong with the internal state
 662      * when it is not there */
 663     pdata = g_object_get_qdata (G_OBJECT (pad), push_data);
 664     g_assert (pdata != NULL);
 665
 666     if (!pdata->pushed) {
 667       /* not yet pushed, release lock and start pushing */
 668       gst_object_ref (pad);
 669       GST_OBJECT_UNLOCK (tee);
 670
 671       GST_LOG_OBJECT (tee, "Starting to push %s %p",
 672           is_list ? "list" : "buffer", data);
 673
 674       ret = gst_tee_do_push (tee, pad, data, is_list);
 675
 676       GST_LOG_OBJECT (tee, "Pushing item %p yielded result %s", data,
 677           gst_flow_get_name (ret));
 678
 679       GST_OBJECT_LOCK (tee);
 680       /* keep track of which pad we pushed and the result value. We need to do
 681        * this before we release the refcount on the pad, the PushData is
 682        * destroyed when the last ref of the pad goes away. */
 683       pdata->pushed = TRUE;
 684       pdata->result = ret;
 685       gst_object_unref (pad);
 686     } else {
 687       /* already pushed, use previous return value */
 688       ret = pdata->result;
 689       GST_LOG_OBJECT (tee, "pad already pushed with %s",
 690           gst_flow_get_name (ret));
 691     }
 692
 693     /* before we go combining the return value, check if the pad list is still
 694      * the same. It could be possible that the pad we just pushed was removed
 695      * and the return value it not valid anymore */
 696     if (GST_ELEMENT_CAST (tee)->pads_cookie != cookie) {
 697       GST_LOG_OBJECT (tee, "pad list changed");
 698       /* the list of pads changed, restart iteration. Pads that we already
 699        * pushed on and are still in the new list, will not be pushed on
 700        * again. */
 701       goto restart;
 702     }
 703
 704     /* stop pushing more buffers when we have a fatal error */
 705     if (ret != GST_FLOW_OK && ret != GST_FLOW_NOT_LINKED)
 706       goto error;
 707
 708     /* keep all other return values, overwriting the previous one. */
 709     if (ret != GST_FLOW_NOT_LINKED) {
 710       GST_LOG_OBJECT (tee, "Replacing ret val %d with %d", cret, ret);
 711       cret = ret;
 712     }
 713     pads = g_list_next (pads);
 714   }
 715   GST_OBJECT_UNLOCK (tee);
 716
 717   gst_mini_object_unref (GST_MINI_OBJECT_CAST (data));
 718
 719   /* no need to unset gvalue */
 720   return cret;
 721
 722   /* ERRORS */
 723 error:
 724   {
 725     GST_DEBUG_OBJECT (tee, "received error %s", gst_flow_get_name (ret));
 726     gst_mini_object_unref (GST_MINI_OBJECT_CAST (data));
 727     GST_OBJECT_UNLOCK (tee);
 728     return ret;
 729   }
 730 }
 731
 732 static GstFlowReturn
 733 gst_tee_chain (GstPad * pad, GstBuffer * buffer)
 734 {
 735   GstFlowReturn res;
 736   GstTee *tee;
 737
 738   tee = GST_TEE (gst_pad_get_parent (pad));
 739
 740   GST_DEBUG_OBJECT (tee, "received buffer %p", buffer);
 741
 742   res = gst_tee_handle_data (tee, buffer, FALSE);
 743
 744   GST_DEBUG_OBJECT (tee, "handled buffer %s", gst_flow_get_name (res));
 745
 746   gst_object_unref (tee);
 747
 748   return res;
 749 }
 750
 751 static GstFlowReturn
 752 gst_tee_chain_list (GstPad * pad, GstBufferList * list)
 753 {
 754   GstFlowReturn res;
 755   GstTee *tee;
 756
 757   tee = GST_TEE (gst_pad_get_parent (pad));
 758
 759   GST_DEBUG_OBJECT (tee, "received list %p", list);
 760
 761   res = gst_tee_handle_data (tee, list, TRUE);
 762
 763   GST_DEBUG_OBJECT (tee, "handled list %s", gst_flow_get_name (res));
 764
 765   gst_object_unref (tee);
 766
 767   return res;
 768 }
 769
 770 static gboolean
 771 gst_tee_sink_activate_push (GstPad * pad, gboolean active)
 772 {
 773   GstTee *tee;
 774
 775   tee = GST_TEE (GST_OBJECT_PARENT (pad));
 776
 777   GST_OBJECT_LOCK (tee);
 778   tee->sink_mode = active && GST_ACTIVATE_PUSH;
 779
 780   if (active && !tee->has_chain)
 781     goto no_chain;
 782   GST_OBJECT_UNLOCK (tee);
 783
 784   return TRUE;
 785
 786   /* ERRORS */
 787 no_chain:
 788   {
 789     GST_OBJECT_UNLOCK (tee);
 790     GST_INFO_OBJECT (tee,
 791         "Tee cannot operate in push mode with has-chain==FALSE");
 792     return FALSE;
 793   }
 794 }
 795
 796 static gboolean
 797 gst_tee_src_activate_pull (GstPad * pad, gboolean active)
 798 {
 799   GstTee *tee;
 800   gboolean res;
 801   GstPad *sinkpad;
 802
 803   tee = GST_TEE (gst_pad_get_parent (pad));
 804
 805   GST_OBJECT_LOCK (tee);
 806
 807   if (tee->pull_mode == GST_TEE_PULL_MODE_NEVER)
 808     goto cannot_pull;
 809
 810   if (tee->pull_mode == GST_TEE_PULL_MODE_SINGLE && active && tee->pull_pad)
 811     goto cannot_pull_multiple_srcs;
 812
 813   sinkpad = gst_object_ref (tee->sinkpad);
 814
 815   GST_OBJECT_UNLOCK (tee);
 816
 817   res = gst_pad_activate_pull (sinkpad, active);
 818   gst_object_unref (sinkpad);
 819
 820   if (!res)
 821     goto sink_activate_failed;
 822
 823   GST_OBJECT_LOCK (tee);
 824   if (active) {
 825     if (tee->pull_mode == GST_TEE_PULL_MODE_SINGLE)
 826       tee->pull_pad = pad;
 827   } else {
 828     if (pad == tee->pull_pad)
 829       tee->pull_pad = NULL;
 830   }
 831   tee->sink_mode = active && GST_ACTIVATE_PULL;
 832   GST_OBJECT_UNLOCK (tee);
 833
 834   gst_object_unref (tee);
 835
 836   return res;
 837
 838   /* ERRORS */
 839 cannot_pull:
 840   {
 841     GST_OBJECT_UNLOCK (tee);
 842     GST_INFO_OBJECT (tee, "Cannot activate in pull mode, pull-mode "
 843         "set to NEVER");
 844     gst_object_unref (tee);
 845     return FALSE;
 846   }
 847 cannot_pull_multiple_srcs:
 848   {
 849     GST_OBJECT_UNLOCK (tee);
 850     GST_INFO_OBJECT (tee, "Cannot activate multiple src pads in pull mode, "
 851         "pull-mode set to SINGLE");
 852     gst_object_unref (tee);
 853     return FALSE;
 854   }
 855 sink_activate_failed:
 856   {
 857     GST_INFO_OBJECT (tee, "Failed to %sactivate sink pad in pull mode",
 858         active ? "" : "de");
 859     gst_object_unref (tee);
 860     return FALSE;
 861   }
 862 }
 863
 864 static gboolean
 865 gst_tee_src_check_get_range (GstPad * pad)
 866 {
 867   GstTee *tee;
 868   gboolean res;
 869   GstPad *sinkpad;
 870
 871   tee = GST_TEE (gst_pad_get_parent (pad));
 872
 873   GST_OBJECT_LOCK (tee);
 874
 875   if (tee->pull_mode == GST_TEE_PULL_MODE_NEVER)
 876     goto cannot_pull;
 877
 878   if (tee->pull_mode == GST_TEE_PULL_MODE_SINGLE && tee->pull_pad)
 879     goto cannot_pull_multiple_srcs;
 880
 881   sinkpad = gst_object_ref (tee->sinkpad);
 882
 883   GST_OBJECT_UNLOCK (tee);
 884
 885   res = gst_pad_check_pull_range (sinkpad);
 886   gst_object_unref (sinkpad);
 887
 888   gst_object_unref (tee);
 889
 890   return res;
 891
 892   /* ERRORS */
 893 cannot_pull:
 894   {
 895     GST_OBJECT_UNLOCK (tee);
 896     GST_INFO_OBJECT (tee, "Cannot activate in pull mode, pull-mode "
 897         "set to NEVER");
 898     gst_object_unref (tee);
 899     return FALSE;
 900   }
 901 cannot_pull_multiple_srcs:
 902   {
 903     GST_OBJECT_UNLOCK (tee);
 904     GST_INFO_OBJECT (tee, "Cannot activate multiple src pads in pull mode, "
 905         "pull-mode set to SINGLE");
 906     gst_object_unref (tee);
 907     return FALSE;
 908   }
 909 }
 910
 911 static void
 912 gst_tee_push_eos (GstPad * pad, GstTee * tee)
 913 {
 914   if (pad != tee->pull_pad)
 915     gst_pad_push_event (pad, gst_event_new_eos ());
 916   gst_object_unref (pad);
 917 }
 918
 919 static void
 920 gst_tee_pull_eos (GstTee * tee)
 921 {
 922   GstIterator *iter;
 923
 924   iter = gst_element_iterate_src_pads (GST_ELEMENT (tee));
 925   gst_iterator_foreach (iter, (GFunc) gst_tee_push_eos, tee);
 926   gst_iterator_free (iter);
 927 }
 928
 929 static GstFlowReturn
 930 gst_tee_src_get_range (GstPad * pad, guint64 offset, guint length,
 931     GstBuffer ** buf)
 932 {
 933   GstTee *tee;
 934   GstFlowReturn ret;
 935
 936   tee = GST_TEE (gst_pad_get_parent (pad));
 937
 938   ret = gst_pad_pull_range (tee->sinkpad, offset, length, buf);
 939
 940   if (ret == GST_FLOW_OK)
 941     ret = gst_tee_handle_data (tee, gst_buffer_ref (*buf), FALSE);
 942   else if (ret == GST_FLOW_UNEXPECTED)
 943     gst_tee_pull_eos (tee);
 944
 945   gst_object_unref (tee);
 946
 947   return ret;
 948 }