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  *
   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  * @short_description: 1-to-N pipe fitting
  27  * @see_also: #GstIdentity
  28  *
  29  * Split data to multiple pads.
  30  */
  31
  32 #ifdef HAVE_CONFIG_H
  33 #  include "config.h"
  34 #endif
  35
  36 #include "gsttee.h"
  37
  38 #include <string.h>
  39
  40 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  41     GST_PAD_SINK,
  42     GST_PAD_ALWAYS,
  43     GST_STATIC_CAPS_ANY);
  44
  45 GST_DEBUG_CATEGORY_STATIC (gst_tee_debug);
  46 #define GST_CAT_DEFAULT gst_tee_debug
  47
  48 static GstElementDetails gst_tee_details =
  49 GST_ELEMENT_DETAILS ("Tee pipe fitting",
  50     "Generic",
  51     "1-to-N pipe fitting",
  52     "Erik Walthinsen <omega@cse.ogi.edu>, "
  53     "Wim \"Tim\" Taymans <wim@fluendo.com>");
  54
  55 enum
  56 {
  57   PROP_0,
  58   PROP_NUM_SRC_PADS,
  59   PROP_HAS_SINK_LOOP,
  60   PROP_HAS_CHAIN,
  61   PROP_SILENT,
  62   PROP_LAST_MESSAGE
  63       /* FILL ME */
  64 };
  65
  66 GstStaticPadTemplate tee_src_template = GST_STATIC_PAD_TEMPLATE ("src%d",
  67     GST_PAD_SRC,
  68     GST_PAD_REQUEST,
  69     GST_STATIC_CAPS_ANY);
  70
  71 #define _do_init(bla) \
  72     GST_DEBUG_CATEGORY_INIT (gst_tee_debug, "tee", 0, "tee element");
  73
  74 GST_BOILERPLATE_FULL (GstTee, gst_tee, GstElement, GST_TYPE_ELEMENT, _do_init);
  75
  76 static GstPad *gst_tee_request_new_pad (GstElement * element,
  77     GstPadTemplate * temp, const gchar * unused);
  78
  79 static void gst_tee_finalize (GObject * object);
  80 static void gst_tee_set_property (GObject * object, guint prop_id,
  81     const GValue * value, GParamSpec * pspec);
  82 static void gst_tee_get_property (GObject * object, guint prop_id,
  83     GValue * value, GParamSpec * pspec);
  84
  85 static GstFlowReturn gst_tee_chain (GstPad * pad, GstBuffer * buffer);
  86 static void gst_tee_loop (GstPad * pad);
  87 static gboolean gst_tee_sink_activate_push (GstPad * pad, gboolean active);
  88 static gboolean gst_tee_sink_activate_pull (GstPad * pad, gboolean active);
  89
  90
  91 static void
  92 gst_tee_base_init (gpointer g_class)
  93 {
  94   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
  95
  96   gst_element_class_add_pad_template (gstelement_class,
  97       gst_static_pad_template_get (&sinktemplate));
  98   gst_element_class_set_details (gstelement_class, &gst_tee_details);
  99   gst_element_class_add_pad_template (gstelement_class,
 100       gst_static_pad_template_get (&tee_src_template));
 101 }
 102
 103 static void
 104 gst_tee_finalize (GObject * object)
 105 {
 106   GstTee *tee;
 107
 108   tee = GST_TEE (object);
 109
 110   g_free (tee->last_message);
 111
 112   G_OBJECT_CLASS (parent_class)->finalize (object);
 113 }
 114
 115 static void
 116 gst_tee_class_init (GstTeeClass * klass)
 117 {
 118   GObjectClass *gobject_class;
 119   GstElementClass *gstelement_class;
 120
 121   gobject_class = (GObjectClass *) klass;
 122   gstelement_class = (GstElementClass *) klass;
 123
 124   gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_tee_finalize);
 125   gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_tee_set_property);
 126   gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_tee_get_property);
 127
 128   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_NUM_SRC_PADS,
 129       g_param_spec_int ("num-src-pads", "num-src-pads", "num-src-pads",
 130           0, G_MAXINT, 0, G_PARAM_READABLE));
 131   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_HAS_SINK_LOOP,
 132       g_param_spec_boolean ("has-sink-loop", "has-sink-loop", "has-sink-loop",
 133           FALSE, G_PARAM_CONSTRUCT | G_PARAM_READWRITE));
 134   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_HAS_CHAIN,
 135       g_param_spec_boolean ("has-chain", "has-chain", "has-chain",
 136           TRUE, G_PARAM_CONSTRUCT | G_PARAM_READWRITE));
 137   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SILENT,
 138       g_param_spec_boolean ("silent", "silent", "silent",
 139           TRUE, G_PARAM_CONSTRUCT | G_PARAM_READWRITE));
 140   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_LAST_MESSAGE,
 141       g_param_spec_string ("last_message", "last_message", "last_message",
 142           NULL, G_PARAM_READABLE));
 143
 144   gstelement_class->request_new_pad =
 145       GST_DEBUG_FUNCPTR (gst_tee_request_new_pad);
 146 }
 147
 148 static void
 149 gst_tee_init (GstTee * tee, GstTeeClass * g_class)
 150 {
 151   tee->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 152   gst_pad_set_setcaps_function (tee->sinkpad,
 153       GST_DEBUG_FUNCPTR (gst_pad_proxy_setcaps));
 154   gst_pad_set_getcaps_function (tee->sinkpad,
 155       GST_DEBUG_FUNCPTR (gst_pad_proxy_getcaps));
 156   gst_element_add_pad (GST_ELEMENT (tee), tee->sinkpad);
 157
 158   tee->last_message = NULL;
 159 }
 160
 161 static void
 162 gst_tee_update_pad_functions (GstTee * tee)
 163 {
 164   gst_pad_set_activatepush_function (tee->sinkpad,
 165       GST_DEBUG_FUNCPTR (gst_tee_sink_activate_push));
 166   gst_pad_set_activatepull_function (tee->sinkpad,
 167       GST_DEBUG_FUNCPTR (gst_tee_sink_activate_pull));
 168
 169   if (tee->has_chain)
 170     gst_pad_set_chain_function (tee->sinkpad,
 171         GST_DEBUG_FUNCPTR (gst_tee_chain));
 172   else
 173     gst_pad_set_chain_function (tee->sinkpad, NULL);
 174 }
 175
 176 static GstPad *
 177 gst_tee_request_new_pad (GstElement * element, GstPadTemplate * templ,
 178     const gchar * unused)
 179 {
 180   gchar *name;
 181   GstPad *srcpad;
 182   GstTee *tee;
 183
 184   tee = GST_TEE (element);
 185
 186   GST_OBJECT_LOCK (tee);
 187   name = g_strdup_printf ("src%d", tee->pad_counter++);
 188   GST_OBJECT_UNLOCK (tee);
 189
 190   srcpad = gst_pad_new_from_template (templ, name);
 191   g_free (name);
 192
 193   gst_pad_set_setcaps_function (srcpad,
 194       GST_DEBUG_FUNCPTR (gst_pad_proxy_setcaps));
 195   gst_pad_set_getcaps_function (srcpad,
 196       GST_DEBUG_FUNCPTR (gst_pad_proxy_getcaps));
 197   gst_element_add_pad (GST_ELEMENT (tee), srcpad);
 198
 199   return srcpad;
 200 }
 201
 202 static void
 203 gst_tee_set_property (GObject * object, guint prop_id, const GValue * value,
 204     GParamSpec * pspec)
 205 {
 206   GstTee *tee = GST_TEE (object);
 207
 208   GST_OBJECT_LOCK (tee);
 209   switch (prop_id) {
 210     case PROP_HAS_SINK_LOOP:
 211       tee->has_sink_loop = g_value_get_boolean (value);
 212       gst_tee_update_pad_functions (tee);
 213       break;
 214     case PROP_HAS_CHAIN:
 215       tee->has_chain = g_value_get_boolean (value);
 216       gst_tee_update_pad_functions (tee);
 217       break;
 218     case PROP_SILENT:
 219       tee->silent = g_value_get_boolean (value);
 220       break;
 221     default:
 222       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 223       break;
 224   }
 225   GST_OBJECT_UNLOCK (tee);
 226 }
 227
 228 static void
 229 gst_tee_get_property (GObject * object, guint prop_id, GValue * value,
 230     GParamSpec * pspec)
 231 {
 232   GstTee *tee = GST_TEE (object);
 233
 234   GST_OBJECT_LOCK (tee);
 235   switch (prop_id) {
 236     case PROP_NUM_SRC_PADS:
 237       g_value_set_int (value, GST_ELEMENT (tee)->numsrcpads);
 238       break;
 239     case PROP_HAS_SINK_LOOP:
 240       g_value_set_boolean (value, tee->has_sink_loop);
 241       break;
 242     case PROP_HAS_CHAIN:
 243       g_value_set_boolean (value, tee->has_chain);
 244       break;
 245     case PROP_SILENT:
 246       g_value_set_boolean (value, tee->silent);
 247       break;
 248     case PROP_LAST_MESSAGE:
 249       g_value_set_string (value, tee->last_message);
 250       break;
 251     default:
 252       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 253       break;
 254   }
 255   GST_OBJECT_UNLOCK (tee);
 256 }
 257
 258 typedef struct
 259 {
 260   GstTee *tee;
 261   GstBuffer *buffer;
 262 } PushData;
 263
 264 static gboolean
 265 gst_tee_do_push (GstPad * pad, GValue * ret, PushData * data)
 266 {
 267   GstFlowReturn res;
 268   GstTee *tee = data->tee;
 269
 270   if (G_UNLIKELY (!data->tee->silent)) {
 271     GstBuffer *buf = data->buffer;
 272
 273     GST_OBJECT_LOCK (tee);
 274     g_free (tee->last_message);
 275     tee->last_message =
 276         g_strdup_printf ("chain        ******* (%s:%s)t (%d bytes, %"
 277         G_GUINT64_FORMAT ") %p", GST_DEBUG_PAD_NAME (pad),
 278         GST_BUFFER_SIZE (buf), GST_BUFFER_TIMESTAMP (buf), buf);
 279     GST_OBJECT_UNLOCK (tee);
 280     g_object_notify (G_OBJECT (tee), "last_message");
 281   }
 282
 283   /* Push */
 284   res = gst_pad_push (pad, gst_buffer_ref (data->buffer));
 285   GST_LOG_OBJECT (tee, "Pushing buffer %p to %" GST_PTR_FORMAT
 286       " yielded result=%d", data->buffer, pad, res);
 287
 288   /* If it's fatal or OK, or if ret is currently
 289    * not-linked, we overwrite the previous value */
 290   if (GST_FLOW_IS_FATAL (res) || (res == GST_FLOW_OK) ||
 291       (g_value_get_enum (ret) == GST_FLOW_NOT_LINKED)) {
 292     GST_LOG_OBJECT (tee, "Replacing ret val %d with %d",
 293         g_value_get_enum (ret), res);
 294     g_value_set_enum (ret, res);
 295   }
 296
 297   gst_object_unref (pad);
 298
 299   /* Stop iterating if flow return is fatal */
 300   return (!GST_FLOW_IS_FATAL (res));
 301 }
 302
 303 static GstFlowReturn
 304 gst_tee_handle_buffer (GstTee * tee, GstBuffer * buffer)
 305 {
 306   GstIterator *iter;
 307   PushData data;
 308   GValue ret = { 0, };
 309   GstIteratorResult res;
 310
 311   tee->offset += GST_BUFFER_SIZE (buffer);
 312
 313   g_value_init (&ret, GST_TYPE_FLOW_RETURN);
 314   g_value_set_enum (&ret, GST_FLOW_NOT_LINKED);
 315   iter = gst_element_iterate_src_pads (GST_ELEMENT (tee));
 316   data.tee = tee;
 317   data.buffer = buffer;
 318
 319   GST_LOG_OBJECT (tee, "Starting to push buffer %p", buffer);
 320   /* FIXME: Not sure how tee would handle RESEND buffer from some of the
 321    * pads but not from others. */
 322   res = gst_iterator_fold (iter, (GstIteratorFoldFunction) gst_tee_do_push,
 323       &ret, &data);
 324   gst_iterator_free (iter);
 325
 326   GST_LOG_OBJECT (tee, "Pushing buffer %p yielded result=%d", buffer,
 327       g_value_get_enum (&ret));
 328
 329   gst_buffer_unref (buffer);
 330
 331   /* no need to unset gvalue */
 332   return g_value_get_enum (&ret);
 333 }
 334
 335 static GstFlowReturn
 336 gst_tee_chain (GstPad * pad, GstBuffer * buffer)
 337 {
 338   GstFlowReturn res;
 339   GstTee *tee;
 340
 341   tee = GST_TEE (GST_PAD_PARENT (pad));
 342
 343   res = gst_tee_handle_buffer (tee, buffer);
 344
 345   return res;
 346 }
 347
 348 #define DEFAULT_SIZE 1024
 349
 350 static void
 351 gst_tee_loop (GstPad * pad)
 352 {
 353   GstBuffer *buffer;
 354   GstFlowReturn res;
 355   GstTee *tee;
 356
 357   tee = GST_TEE (GST_PAD_PARENT (pad));
 358
 359   res = gst_pad_pull_range (pad, tee->offset, DEFAULT_SIZE, &buffer);
 360   if (res != GST_FLOW_OK)
 361     goto pause_task;
 362
 363   res = gst_tee_handle_buffer (tee, buffer);
 364   if (res != GST_FLOW_OK)
 365     goto pause_task;
 366
 367   return;
 368
 369 pause_task:
 370   {
 371     gst_pad_pause_task (pad);
 372     return;
 373   }
 374 }
 375
 376 static gboolean
 377 gst_tee_sink_activate_push (GstPad * pad, gboolean active)
 378 {
 379   GstTee *tee;
 380
 381   tee = GST_TEE (GST_OBJECT_PARENT (pad));
 382
 383   tee->sink_mode = active && GST_ACTIVATE_PUSH;
 384
 385   if (active) {
 386     g_return_val_if_fail (tee->has_chain, FALSE);
 387   }
 388
 389   return TRUE;
 390 }
 391
 392 /* won't be called until we implement an activate function */
 393 static gboolean
 394 gst_tee_sink_activate_pull (GstPad * pad, gboolean active)
 395 {
 396   GstTee *tee;
 397
 398   tee = GST_TEE (GST_OBJECT_PARENT (pad));
 399
 400   tee->sink_mode = active && GST_ACTIVATE_PULL;
 401
 402   if (active) {
 403     g_return_val_if_fail (tee->has_sink_loop, FALSE);
 404     return gst_pad_start_task (pad, (GstTaskFunction) gst_tee_loop, pad);
 405   } else {
 406     return gst_pad_stop_task (pad);
 407   }
 408 }