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., 59 Temple Place - Suite 330,
  18  * Boston, MA 02111-1307, USA.
  19  */
  20
  21 #ifdef HAVE_CONFIG_H
  22 #include "config.h"
  23 #endif
  24
  25 #include "gststreamsplitter.h"
  26 #include "gst/glib-compat-private.h"
  27
  28 static GstStaticPadTemplate src_template =
  29 GST_STATIC_PAD_TEMPLATE ("src_%d", GST_PAD_SRC, GST_PAD_REQUEST,
  30     GST_STATIC_CAPS_ANY);
  31
  32 static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
  33     GST_PAD_SINK,
  34     GST_PAD_ALWAYS,
  35     GST_STATIC_CAPS_ANY);
  36
  37 GST_DEBUG_CATEGORY_STATIC (gst_stream_splitter_debug);
  38 #define GST_CAT_DEFAULT gst_stream_splitter_debug
  39
  40 G_DEFINE_TYPE (GstStreamSplitter, gst_stream_splitter, GST_TYPE_ELEMENT);
  41
  42 #define STREAMS_LOCK(obj) (g_mutex_lock(obj->lock))
  43 #define STREAMS_UNLOCK(obj) (g_mutex_unlock(obj->lock))
  44
  45 static void gst_stream_splitter_dispose (GObject * object);
  46
  47 static GstPad *gst_stream_splitter_request_new_pad (GstElement * element,
  48     GstPadTemplate * templ, const gchar * name);
  49 static void gst_stream_splitter_release_pad (GstElement * element,
  50     GstPad * pad);
  51
  52 static void
  53 gst_stream_splitter_class_init (GstStreamSplitterClass * klass)
  54 {
  55   GObjectClass *gobject_klass;
  56   GstElementClass *gstelement_klass;
  57
  58   gobject_klass = (GObjectClass *) klass;
  59   gstelement_klass = (GstElementClass *) klass;
  60
  61   gobject_klass->dispose = gst_stream_splitter_dispose;
  62
  63   GST_DEBUG_CATEGORY_INIT (gst_stream_splitter_debug, "streamsplitter", 0,
  64       "Stream Splitter");
  65
  66   gst_element_class_add_static_pad_template (gstelement_klass, &src_template);
  67   gst_element_class_add_static_pad_template (gstelement_klass, &sink_template);
  68
  69   gstelement_klass->request_new_pad =
  70       GST_DEBUG_FUNCPTR (gst_stream_splitter_request_new_pad);
  71   gstelement_klass->release_pad =
  72       GST_DEBUG_FUNCPTR (gst_stream_splitter_release_pad);
  73
  74   gst_element_class_set_details_simple (gstelement_klass,
  75       "streamsplitter", "Generic",
  76       "Splits streams based on their media type",
  77       "Edward Hervey <edward.hervey@collabora.co.uk>");
  78 }
  79
  80 static void
  81 gst_stream_splitter_dispose (GObject * object)
  82 {
  83   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) object;
  84
  85   if (stream_splitter->lock) {
  86     g_mutex_free (stream_splitter->lock);
  87     stream_splitter->lock = NULL;
  88   }
  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 GstFlowReturn
  99 gst_stream_splitter_chain (GstPad * pad, GstBuffer * buf)
 100 {
 101   GstStreamSplitter *stream_splitter =
 102       (GstStreamSplitter *) GST_PAD_PARENT (pad);
 103   GstFlowReturn res;
 104   GstPad *srcpad = NULL;
 105
 106   STREAMS_LOCK (stream_splitter);
 107   if (stream_splitter->current)
 108     srcpad = gst_object_ref (stream_splitter->current);
 109   STREAMS_UNLOCK (stream_splitter);
 110
 111   if (G_UNLIKELY (srcpad == NULL))
 112     goto nopad;
 113
 114   if (G_UNLIKELY (stream_splitter->pending_events)) {
 115     GList *tmp;
 116     GST_DEBUG_OBJECT (srcpad, "Pushing out pending events");
 117
 118     for (tmp = stream_splitter->pending_events; tmp; tmp = tmp->next) {
 119       GstEvent *event = (GstEvent *) tmp->data;
 120       gst_pad_push_event (srcpad, event);
 121     }
 122     g_list_free (stream_splitter->pending_events);
 123     stream_splitter->pending_events = NULL;
 124   }
 125
 126   /* Forward to currently activated stream */
 127   res = gst_pad_push (srcpad, buf);
 128   gst_object_unref (srcpad);
 129
 130   return res;
 131
 132 nopad:
 133   GST_WARNING_OBJECT (stream_splitter, "No output pad was configured");
 134   return GST_FLOW_ERROR;
 135 }
 136
 137 static gboolean
 138 gst_stream_splitter_sink_event (GstPad * pad, GstEvent * event)
 139 {
 140   GstStreamSplitter *stream_splitter =
 141       (GstStreamSplitter *) GST_PAD_PARENT (pad);
 142   gboolean res = TRUE;
 143   gboolean toall = FALSE;
 144   gboolean store = FALSE;
 145   gboolean eos = FALSE;
 146   gboolean flushpending = FALSE;
 147
 148   /* FLUSH_START/STOP : forward to all
 149    * EOS : transform to CUSTOM_REAL_EOS and forward to all
 150    * INBAND events : store to send in chain function to selected chain
 151    * OUT_OF_BAND events : send to all
 152    */
 153
 154   GST_DEBUG_OBJECT (stream_splitter, "Got event %s",
 155       GST_EVENT_TYPE_NAME (event));
 156
 157   switch (GST_EVENT_TYPE (event)) {
 158     case GST_EVENT_FLUSH_STOP:
 159       flushpending = TRUE;
 160       toall = TRUE;
 161       break;
 162     case GST_EVENT_FLUSH_START:
 163       toall = TRUE;
 164       break;
 165     case GST_EVENT_EOS:
 166       /* Replace with our custom eos event */
 167       gst_event_unref (event);
 168       event =
 169           gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM,
 170           gst_structure_empty_new ("stream-switching-eos"));
 171       toall = TRUE;
 172       eos = TRUE;
 173       break;
 174     default:
 175       if (GST_EVENT_TYPE (event) & GST_EVENT_TYPE_SERIALIZED)
 176         store = TRUE;
 177   }
 178
 179   if (flushpending) {
 180     g_list_foreach (stream_splitter->pending_events, (GFunc) gst_event_unref,
 181         NULL);
 182     g_list_free (stream_splitter->pending_events);
 183     stream_splitter->pending_events = NULL;
 184   }
 185
 186   if (store) {
 187     stream_splitter->pending_events =
 188         g_list_append (stream_splitter->pending_events, event);
 189   } else if (toall || eos) {
 190     GList *tmp;
 191     guint32 cookie;
 192
 193     /* Send to all pads */
 194     STREAMS_LOCK (stream_splitter);
 195   resync:
 196     if (G_UNLIKELY (stream_splitter->srcpads == NULL)) {
 197       STREAMS_UNLOCK (stream_splitter);
 198       /* No source pads */
 199       gst_event_unref (event);
 200       res = FALSE;
 201       goto beach;
 202     }
 203     tmp = stream_splitter->srcpads;
 204     cookie = stream_splitter->cookie;
 205     while (tmp) {
 206       GstPad *srcpad = (GstPad *) tmp->data;
 207       STREAMS_UNLOCK (stream_splitter);
 208       /* In case of EOS, we first push out the real one to flush out
 209        * each streams (but which will be discarded in the streamcombiner)
 210        * before our custom one (which will be converted back to and EOS
 211        * in the streamcombiner) */
 212       if (eos)
 213         gst_pad_push_event (srcpad, gst_event_new_eos ());
 214       gst_event_ref (event);
 215       res = gst_pad_push_event (srcpad, event);
 216       STREAMS_LOCK (stream_splitter);
 217       if (G_UNLIKELY (cookie != stream_splitter->cookie))
 218         goto resync;
 219       tmp = tmp->next;
 220     }
 221     STREAMS_UNLOCK (stream_splitter);
 222     gst_event_unref (event);
 223   } else {
 224     GstPad *pad;
 225
 226     /* Only send to current pad */
 227
 228     STREAMS_LOCK (stream_splitter);
 229     pad = stream_splitter->current;
 230     STREAMS_UNLOCK (stream_splitter);
 231     if (pad)
 232       res = gst_pad_push_event (pad, event);
 233     else {
 234       gst_event_unref (event);
 235       res = FALSE;
 236     }
 237   }
 238
 239 beach:
 240   return res;
 241 }
 242
 243 static GstCaps *
 244 gst_stream_splitter_sink_getcaps (GstPad * pad)
 245 {
 246   GstStreamSplitter *stream_splitter =
 247       (GstStreamSplitter *) GST_PAD_PARENT (pad);
 248   guint32 cookie;
 249   GList *tmp;
 250   GstCaps *res = NULL;
 251
 252   /* Return the combination of all downstream caps */
 253
 254   STREAMS_LOCK (stream_splitter);
 255
 256 resync:
 257   if (G_UNLIKELY (stream_splitter->srcpads == NULL)) {
 258     res = gst_caps_new_any ();
 259     goto beach;
 260   }
 261
 262   res = NULL;
 263   cookie = stream_splitter->cookie;
 264   tmp = stream_splitter->srcpads;
 265
 266   while (tmp) {
 267     GstPad *srcpad = (GstPad *) tmp->data;
 268
 269     STREAMS_UNLOCK (stream_splitter);
 270     if (res) {
 271       GstCaps *peercaps = gst_pad_peer_get_caps_reffed (srcpad);
 272       if (peercaps)
 273         gst_caps_merge (res, gst_caps_make_writable (peercaps));
 274     } else {
 275       res = gst_pad_peer_get_caps (srcpad);
 276     }
 277     STREAMS_LOCK (stream_splitter);
 278
 279     if (G_UNLIKELY (cookie != stream_splitter->cookie)) {
 280       if (res)
 281         gst_caps_unref (res);
 282       goto resync;
 283     }
 284     tmp = tmp->next;
 285   }
 286
 287 beach:
 288   STREAMS_UNLOCK (stream_splitter);
 289   return res;
 290 }
 291
 292 static gboolean
 293 gst_stream_splitter_sink_setcaps (GstPad * pad, GstCaps * caps)
 294 {
 295   GstStreamSplitter *stream_splitter =
 296       (GstStreamSplitter *) GST_PAD_PARENT (pad);
 297   guint32 cookie;
 298   GList *tmp;
 299   gboolean res;
 300
 301   GST_DEBUG_OBJECT (stream_splitter, "caps %" GST_PTR_FORMAT, caps);
 302
 303   /* Try on all pads, choose the one that succeeds as the current stream */
 304   STREAMS_LOCK (stream_splitter);
 305
 306 resync:
 307   if (G_UNLIKELY (stream_splitter->srcpads == NULL)) {
 308     res = FALSE;
 309     goto beach;
 310   }
 311
 312   res = FALSE;
 313   tmp = stream_splitter->srcpads;
 314   cookie = stream_splitter->cookie;
 315
 316   while (tmp) {
 317     GstPad *srcpad = (GstPad *) tmp->data;
 318     GstCaps *peercaps;
 319
 320     STREAMS_UNLOCK (stream_splitter);
 321     peercaps = gst_pad_peer_get_caps_reffed (srcpad);
 322     if (peercaps) {
 323       res = gst_caps_can_intersect (caps, peercaps);
 324       gst_caps_unref (peercaps);
 325     }
 326     STREAMS_LOCK (stream_splitter);
 327
 328     if (G_UNLIKELY (cookie != stream_splitter->cookie))
 329       goto resync;
 330
 331     if (res) {
 332       /* FIXME : we need to switch properly */
 333       GST_DEBUG_OBJECT (srcpad, "Setting caps on this pad was successful");
 334       stream_splitter->current = srcpad;
 335       goto beach;
 336     }
 337     tmp = tmp->next;
 338   }
 339
 340 beach:
 341   STREAMS_UNLOCK (stream_splitter);
 342   return res;
 343 }
 344
 345 static gboolean
 346 gst_stream_splitter_src_event (GstPad * pad, GstEvent * event)
 347 {
 348   GstStreamSplitter *stream_splitter =
 349       (GstStreamSplitter *) GST_PAD_PARENT (pad);
 350
 351   GST_DEBUG_OBJECT (pad, "%s", GST_EVENT_TYPE_NAME (event));
 352
 353   /* Forward upstream as is */
 354   return gst_pad_push_event (stream_splitter->sinkpad, event);
 355 }
 356
 357 static gboolean
 358 gst_stream_splitter_src_query (GstPad * pad, GstQuery * query)
 359 {
 360   GstStreamSplitter *stream_splitter =
 361       (GstStreamSplitter *) GST_PAD_PARENT (pad);
 362
 363   GST_DEBUG_OBJECT (pad, "%s", GST_QUERY_TYPE_NAME (query));
 364
 365   /* Forward upstream as is */
 366   return gst_pad_peer_query (stream_splitter->sinkpad, query);
 367 }
 368
 369 static void
 370 gst_stream_splitter_init (GstStreamSplitter * stream_splitter)
 371 {
 372   stream_splitter->sinkpad =
 373       gst_pad_new_from_static_template (&sink_template, "sink");
 374   /* FIXME : No buffer alloc for the time being, it will resort to the fallback */
 375   /* gst_pad_set_bufferalloc_function (stream_splitter->sinkpad, */
 376   /*     gst_stream_splitter_buffer_alloc); */
 377   gst_pad_set_chain_function (stream_splitter->sinkpad,
 378       gst_stream_splitter_chain);
 379   gst_pad_set_event_function (stream_splitter->sinkpad,
 380       gst_stream_splitter_sink_event);
 381   gst_pad_set_getcaps_function (stream_splitter->sinkpad,
 382       gst_stream_splitter_sink_getcaps);
 383   gst_pad_set_setcaps_function (stream_splitter->sinkpad,
 384       gst_stream_splitter_sink_setcaps);
 385   gst_element_add_pad (GST_ELEMENT (stream_splitter), stream_splitter->sinkpad);
 386
 387   stream_splitter->lock = g_mutex_new ();
 388 }
 389
 390 static GstPad *
 391 gst_stream_splitter_request_new_pad (GstElement * element,
 392     GstPadTemplate * templ, const gchar * name)
 393 {
 394   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) element;
 395   GstPad *srcpad;
 396
 397   srcpad = gst_pad_new_from_static_template (&src_template, name);
 398   gst_pad_set_event_function (srcpad, gst_stream_splitter_src_event);
 399   gst_pad_set_query_function (srcpad, gst_stream_splitter_src_query);
 400
 401   STREAMS_LOCK (stream_splitter);
 402   stream_splitter->srcpads = g_list_append (stream_splitter->srcpads, srcpad);
 403   gst_pad_set_active (srcpad, TRUE);
 404   gst_element_add_pad (element, srcpad);
 405   stream_splitter->cookie++;
 406   STREAMS_UNLOCK (stream_splitter);
 407
 408   return srcpad;
 409 }
 410
 411 static void
 412 gst_stream_splitter_release_pad (GstElement * element, GstPad * pad)
 413 {
 414   GstStreamSplitter *stream_splitter = (GstStreamSplitter *) element;
 415   GList *tmp;
 416
 417   STREAMS_LOCK (stream_splitter);
 418   tmp = g_list_find (stream_splitter->srcpads, pad);
 419   if (tmp) {
 420     GstPad *pad = (GstPad *) tmp->data;
 421
 422     stream_splitter->srcpads =
 423         g_list_delete_link (stream_splitter->srcpads, tmp);
 424     stream_splitter->cookie++;
 425
 426     if (pad == stream_splitter->current) {
 427       /* Deactivate current flow */
 428       GST_DEBUG_OBJECT (element, "Removed pad was the current one");
 429       stream_splitter->current = NULL;
 430     }
 431
 432     gst_element_remove_pad (element, pad);
 433   }
 434   STREAMS_UNLOCK (stream_splitter);
 435
 436   return;
 437 }