gst/encoding/gststreamsplitter.c

   1 /* GStreamer Stream Splitter
   2  * Copyright (C) 2010 Edward Hervey <edward.hervey@collabora.co.uk>
   3  *           (C) 2009 Nokia Corporation
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  18  * Boston, MA 02110-1301, USA.
  19  */
  20
  21 #ifdef HAVE_CONFIG_H
  22 #include "config.h"
  23 #endif
  24
  25 #include "gststreamsplitter.h"
  26
  27 static GstStaticPadTemplate src_template =
  28 GST_STATIC_PAD_TEMPLATE ("src_%u", GST_PAD_SRC, GST_PAD_REQUEST,
  29     GST_STATIC_CAPS_ANY);
  30
  31 static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
  32     GST_PAD_SINK,
  33     GST_PAD_ALWAYS,
  34     GST_STATIC_CAPS_ANY);
  35
  36 GST_DEBUG_CATEGORY_STATIC (gst_stream_splitter_debug);
  37 #define GST_CAT_DEFAULT gst_stream_splitter_debug
  38
  39 G_DEFINE_TYPE (GstStreamSplitter, gst_stream_splitter, GST_TYPE_ELEMENT);
  40
  41 #define STREAMS_LOCK(obj) (g_mutex_lock(&obj->lock))
  42 #define STREAMS_UNLOCK(obj) (g_mutex_unlock(&obj->lock))
  43
  44 static void gst_stream_splitter_dispose (GObject * object);
  45 static void gst_stream_splitter_finalize (GObject * object);
  46
  47 static gboolean gst_stream_splitter_sink_setcaps (GstPad * pad, GstCaps * caps);
  48
  49 static GstPad *gst_stream_splitter_request_new_pad (GstElement * element,
  50     GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
  51 static void gst_stream_splitter_release_pad (GstElement * element,
  52     GstPad * pad);
  53
  54 static void
  55 gst_stream_splitter_class_init (GstStreamSplitterClass * klass)
  56 {
  57   GObjectClass *gobject_klass;
  58   GstElementClass *gstelement_klass;
  59
  60   gobject_klass = (GObjectClass *) klass;
  61   gstelement_klass = (GstElementClass *) klass;
  62
  63   gobject_klass->dispose = gst_stream_splitter_dispose;
  64   gobject_klass->finalize = gst_stream_splitter_finalize;
  65
  66   GST_DEBUG_CATEGORY_INIT (gst_stream_splitter_debug, "streamsplitter", 0,
  67       "Stream Splitter");
  68
  69   gst_element_class_add_pad_template (gstelement_klass,
  70       gst_static_pad_template_get (&src_template));
  71   gst_element_class_add_pad_template (gstelement_klass,
  72       gst_static_pad_template_get (&sink_template));
  73
  74   gstelement_klass->request_new_pad =
  75       GST_DEBUG_FUNCPTR (gst_stream_splitter_request_new_pad);
  76   gstelement_klass->release_pad =
  77       GST_DEBUG_FUNCPTR (gst_stream_splitter_release_pad);
  78
  79   gst_element_class_set_static_metadata (gstelement_klass,
  80       "streamsplitter", "Generic",
  81       "Splits streams based on their media type",
  82       "Edward Hervey <edward.hervey@collabora.co.uk>");
  83 }
  84
  85 static void
  86 gst_stream_splitter_dispose (GObject * object)
  87 {
  88   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) object;
  89
  90   g_list_foreach (stream_splitter->pending_events, (GFunc) gst_event_unref,
  91       NULL);
  92   g_list_free (stream_splitter->pending_events);
  93   stream_splitter->pending_events = NULL;
  94
  95   G_OBJECT_CLASS (gst_stream_splitter_parent_class)->dispose (object);
  96 }
  97
  98 static void
  99 gst_stream_splitter_finalize (GObject * object)
 100 {
 101   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) object;
 102
 103   g_mutex_clear (&stream_splitter->lock);
 104
 105   G_OBJECT_CLASS (gst_stream_splitter_parent_class)->finalize (object);
 106 }
 107
 108 static GstFlowReturn
 109 gst_stream_splitter_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
 110 {
 111   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) parent;
 112   GstFlowReturn res;
 113   GstPad *srcpad = NULL;
 114
 115   STREAMS_LOCK (stream_splitter);
 116   if (stream_splitter->current)
 117     srcpad = gst_object_ref (stream_splitter->current);
 118   STREAMS_UNLOCK (stream_splitter);
 119
 120   if (G_UNLIKELY (srcpad == NULL))
 121     goto nopad;
 122
 123   if (G_UNLIKELY (stream_splitter->pending_events)) {
 124     GList *tmp;
 125     GST_DEBUG_OBJECT (srcpad, "Pushing out pending events");
 126
 127     for (tmp = stream_splitter->pending_events; tmp; tmp = tmp->next) {
 128       GstEvent *event = (GstEvent *) tmp->data;
 129       gst_pad_push_event (srcpad, event);
 130     }
 131     g_list_free (stream_splitter->pending_events);
 132     stream_splitter->pending_events = NULL;
 133   }
 134
 135   /* Forward to currently activated stream */
 136   res = gst_pad_push (srcpad, buf);
 137   gst_object_unref (srcpad);
 138
 139   return res;
 140
 141 nopad:
 142   GST_WARNING_OBJECT (stream_splitter, "No output pad was configured");
 143   return GST_FLOW_ERROR;
 144 }
 145
 146 static gboolean
 147 gst_stream_splitter_sink_event (GstPad * pad, GstObject * parent,
 148     GstEvent * event)
 149 {
 150   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) parent;
 151   gboolean res = TRUE;
 152   gboolean toall = FALSE;
 153   gboolean store = FALSE;
 154   gboolean eos = FALSE;
 155   gboolean flushpending = FALSE;
 156
 157   /* FLUSH_START/STOP : forward to all
 158    * EOS : transform to CUSTOM_REAL_EOS and forward to all
 159    * INBAND events : store to send in chain function to selected chain
 160    * OUT_OF_BAND events : send to all
 161    */
 162
 163   GST_DEBUG_OBJECT (stream_splitter, "Got event %s",
 164       GST_EVENT_TYPE_NAME (event));
 165
 166   switch (GST_EVENT_TYPE (event)) {
 167     case GST_EVENT_CAPS:
 168     {
 169       GstCaps *caps;
 170
 171       gst_event_parse_caps (event, &caps);
 172       res = gst_stream_splitter_sink_setcaps (pad, caps);
 173
 174       store = TRUE;
 175       break;
 176     }
 177     case GST_EVENT_FLUSH_STOP:
 178       flushpending = TRUE;
 179       toall = TRUE;
 180       break;
 181     case GST_EVENT_FLUSH_START:
 182       toall = TRUE;
 183       break;
 184     case GST_EVENT_EOS:
 185       /* Replace with our custom eos event */
 186       gst_event_unref (event);
 187       event =
 188           gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM,
 189           gst_structure_new_empty ("stream-switching-eos"));
 190       toall = TRUE;
 191       eos = TRUE;
 192       break;
 193     default:
 194       if (GST_EVENT_TYPE (event) & GST_EVENT_TYPE_SERIALIZED)
 195         store = TRUE;
 196   }
 197
 198   if (flushpending) {
 199     g_list_foreach (stream_splitter->pending_events, (GFunc) gst_event_unref,
 200         NULL);
 201     g_list_free (stream_splitter->pending_events);
 202     stream_splitter->pending_events = NULL;
 203   }
 204
 205   if (store) {
 206     stream_splitter->pending_events =
 207         g_list_append (stream_splitter->pending_events, event);
 208   } else if (toall || eos) {
 209     GList *tmp;
 210     guint32 cookie;
 211
 212     /* Send to all pads */
 213     STREAMS_LOCK (stream_splitter);
 214   resync:
 215     if (G_UNLIKELY (stream_splitter->srcpads == NULL)) {
 216       STREAMS_UNLOCK (stream_splitter);
 217       /* No source pads */
 218       gst_event_unref (event);
 219       res = FALSE;
 220       goto beach;
 221     }
 222     tmp = stream_splitter->srcpads;
 223     cookie = stream_splitter->cookie;
 224     while (tmp) {
 225       GstPad *srcpad = (GstPad *) tmp->data;
 226       STREAMS_UNLOCK (stream_splitter);
 227       /* In case of EOS, we first push out the real one to flush out
 228        * each streams (but which will be discarded in the streamcombiner)
 229        * before our custom one (which will be converted back to and EOS
 230        * in the streamcombiner) */
 231       if (eos)
 232         gst_pad_push_event (srcpad, gst_event_new_eos ());
 233       gst_event_ref (event);
 234       res = gst_pad_push_event (srcpad, event);
 235       STREAMS_LOCK (stream_splitter);
 236       if (G_UNLIKELY (cookie != stream_splitter->cookie))
 237         goto resync;
 238       tmp = tmp->next;
 239     }
 240     STREAMS_UNLOCK (stream_splitter);
 241     gst_event_unref (event);
 242   } else {
 243     GstPad *pad;
 244
 245     /* Only send to current pad */
 246
 247     STREAMS_LOCK (stream_splitter);
 248     pad = stream_splitter->current;
 249     STREAMS_UNLOCK (stream_splitter);
 250     if (pad)
 251       res = gst_pad_push_event (pad, event);
 252     else {
 253       gst_event_unref (event);
 254       res = FALSE;
 255     }
 256   }
 257
 258 beach:
 259   return res;
 260 }
 261
 262 static GstCaps *
 263 gst_stream_splitter_sink_getcaps (GstPad * pad, GstCaps * filter)
 264 {
 265   GstStreamSplitter *stream_splitter =
 266       (GstStreamSplitter *) GST_PAD_PARENT (pad);
 267   guint32 cookie;
 268   GList *tmp;
 269   GstCaps *res = NULL;
 270
 271   /* Return the combination of all downstream caps */
 272
 273   STREAMS_LOCK (stream_splitter);
 274
 275 resync:
 276   if (G_UNLIKELY (stream_splitter->srcpads == NULL)) {
 277     res = (filter ? gst_caps_ref (filter) : gst_caps_new_any ());
 278     goto beach;
 279   }
 280
 281   res = NULL;
 282   cookie = stream_splitter->cookie;
 283   tmp = stream_splitter->srcpads;
 284
 285   while (tmp) {
 286     GstPad *srcpad = (GstPad *) tmp->data;
 287
 288     STREAMS_UNLOCK (stream_splitter);
 289     if (res) {
 290       GstCaps *peercaps = gst_pad_peer_query_caps (srcpad, filter);
 291       if (peercaps)
 292         res = gst_caps_merge (res, peercaps);
 293     } else {
 294       res = gst_pad_peer_query_caps (srcpad, filter);
 295     }
 296     STREAMS_LOCK (stream_splitter);
 297
 298     if (G_UNLIKELY (cookie != stream_splitter->cookie)) {
 299       if (res)
 300         gst_caps_unref (res);
 301       goto resync;
 302     }
 303     tmp = tmp->next;
 304   }
 305
 306 beach:
 307   STREAMS_UNLOCK (stream_splitter);
 308   return res;
 309 }
 310
 311 static gboolean
 312 gst_stream_splitter_sink_query (GstPad * pad, GstObject * parent,
 313     GstQuery * query)
 314 {
 315   gboolean res;
 316
 317   switch (GST_QUERY_TYPE (query)) {
 318     case GST_QUERY_CAPS:
 319     {
 320       GstCaps *filter, *caps;
 321
 322       gst_query_parse_caps (query, &filter);
 323       caps = gst_stream_splitter_sink_getcaps (pad, filter);
 324       gst_query_set_caps_result (query, caps);
 325       gst_caps_unref (caps);
 326       res = TRUE;
 327       break;
 328     }
 329     default:
 330       res = gst_pad_query_default (pad, parent, query);
 331       break;
 332   }
 333   return res;
 334 }
 335
 336 static gboolean
 337 gst_stream_splitter_sink_setcaps (GstPad * pad, GstCaps * caps)
 338 {
 339   GstStreamSplitter *stream_splitter =
 340       (GstStreamSplitter *) GST_PAD_PARENT (pad);
 341   guint32 cookie;
 342   GList *tmp;
 343   gboolean res;
 344
 345   GST_DEBUG_OBJECT (stream_splitter, "caps %" GST_PTR_FORMAT, caps);
 346
 347   /* Try on all pads, choose the one that succeeds as the current stream */
 348   STREAMS_LOCK (stream_splitter);
 349
 350 resync:
 351   if (G_UNLIKELY (stream_splitter->srcpads == NULL)) {
 352     res = FALSE;
 353     goto beach;
 354   }
 355
 356   res = FALSE;
 357   tmp = stream_splitter->srcpads;
 358   cookie = stream_splitter->cookie;
 359
 360   while (tmp) {
 361     GstPad *srcpad = (GstPad *) tmp->data;
 362     GstCaps *peercaps;
 363
 364     STREAMS_UNLOCK (stream_splitter);
 365     peercaps = gst_pad_peer_query_caps (srcpad, NULL);
 366     if (peercaps) {
 367       res = gst_caps_can_intersect (caps, peercaps);
 368       gst_caps_unref (peercaps);
 369     }
 370     STREAMS_LOCK (stream_splitter);
 371
 372     if (G_UNLIKELY (cookie != stream_splitter->cookie))
 373       goto resync;
 374
 375     if (res) {
 376       /* FIXME : we need to switch properly */
 377       GST_DEBUG_OBJECT (srcpad, "Setting caps on this pad was successful");
 378       stream_splitter->current = srcpad;
 379       goto beach;
 380     }
 381     tmp = tmp->next;
 382   }
 383
 384 beach:
 385   STREAMS_UNLOCK (stream_splitter);
 386   return res;
 387 }
 388
 389 static gboolean
 390 gst_stream_splitter_src_event (GstPad * pad, GstObject * parent,
 391     GstEvent * event)
 392 {
 393   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) parent;
 394
 395   GST_DEBUG_OBJECT (pad, "%s", GST_EVENT_TYPE_NAME (event));
 396
 397   /* Forward upstream as is */
 398   return gst_pad_push_event (stream_splitter->sinkpad, event);
 399 }
 400
 401 static gboolean
 402 gst_stream_splitter_src_query (GstPad * pad, GstObject * parent,
 403     GstQuery * query)
 404 {
 405   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) parent;
 406
 407   GST_DEBUG_OBJECT (pad, "%s", GST_QUERY_TYPE_NAME (query));
 408
 409   /* Forward upstream as is */
 410   return gst_pad_peer_query (stream_splitter->sinkpad, query);
 411 }
 412
 413 static void
 414 gst_stream_splitter_init (GstStreamSplitter * stream_splitter)
 415 {
 416   stream_splitter->sinkpad =
 417       gst_pad_new_from_static_template (&sink_template, "sink");
 418   gst_pad_set_chain_function (stream_splitter->sinkpad,
 419       gst_stream_splitter_chain);
 420   gst_pad_set_event_function (stream_splitter->sinkpad,
 421       gst_stream_splitter_sink_event);
 422   gst_pad_set_query_function (stream_splitter->sinkpad,
 423       gst_stream_splitter_sink_query);
 424   gst_element_add_pad (GST_ELEMENT (stream_splitter), stream_splitter->sinkpad);
 425
 426   g_mutex_init (&stream_splitter->lock);
 427 }
 428
 429 static GstPad *
 430 gst_stream_splitter_request_new_pad (GstElement * element,
 431     GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
 432 {
 433   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) element;
 434   GstPad *srcpad;
 435
 436   srcpad = gst_pad_new_from_static_template (&src_template, name);
 437   gst_pad_set_event_function (srcpad, gst_stream_splitter_src_event);
 438   gst_pad_set_query_function (srcpad, gst_stream_splitter_src_query);
 439
 440   STREAMS_LOCK (stream_splitter);
 441   stream_splitter->srcpads = g_list_append (stream_splitter->srcpads, srcpad);
 442   gst_pad_set_active (srcpad, TRUE);
 443   gst_element_add_pad (element, srcpad);
 444   stream_splitter->cookie++;
 445   STREAMS_UNLOCK (stream_splitter);
 446
 447   return srcpad;
 448 }
 449
 450 static void
 451 gst_stream_splitter_release_pad (GstElement * element, GstPad * pad)
 452 {
 453   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) element;
 454   GList *tmp;
 455
 456   STREAMS_LOCK (stream_splitter);
 457   tmp = g_list_find (stream_splitter->srcpads, pad);
 458   if (tmp) {
 459     GstPad *pad = (GstPad *) tmp->data;
 460
 461     stream_splitter->srcpads =
 462         g_list_delete_link (stream_splitter->srcpads, tmp);
 463     stream_splitter->cookie++;
 464
 465     if (pad == stream_splitter->current) {
 466       /* Deactivate current flow */
 467       GST_DEBUG_OBJECT (element, "Removed pad was the current one");
 468       stream_splitter->current = NULL;
 469     }
 470
 471     gst_element_remove_pad (element, pad);
 472   }
 473   STREAMS_UNLOCK (stream_splitter);
 474
 475   return;
 476 }