rtpssrcdemux: Hold on internal stream lock while pushing sticky
[platform/upstream/gst-plugins-good.git] / gst / rtpmanager / gstrtpssrcdemux.c
1 /* GStreamer
2  * Copyright (C) <2007> Wim Taymans <wim.taymans@gmail.com>
3  *
4  * RTP SSRC demuxer
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., 51 Franklin St, Fifth Floor,
19  * Boston, MA 02110-1301, USA.
20  */
21
22 /**
23  * SECTION:element-rtpssrcdemux
24  *
25  * rtpssrcdemux acts as a demuxer for RTP packets based on the SSRC of the
26  * packets. Its main purpose is to allow an application to easily receive and
27  * decode an RTP stream with multiple SSRCs.
28  * 
29  * For each SSRC that is detected, a new pad will be created and the
30  * #GstRtpSsrcDemux::new-ssrc-pad signal will be emitted. 
31  * 
32  * <refsect2>
33  * <title>Example pipelines</title>
34  * |[
35  * gst-launch-1.0 udpsrc caps="application/x-rtp" ! rtpssrcdemux ! fakesink
36  * ]| Takes an RTP stream and send the RTP packets with the first detected SSRC
37  * to fakesink, discarding the other SSRCs.
38  * </refsect2>
39  */
40
41 #ifdef HAVE_CONFIG_H
42 #include "config.h"
43 #endif
44
45 #include <string.h>
46 #include <gst/rtp/gstrtpbuffer.h>
47 #include <gst/rtp/gstrtcpbuffer.h>
48
49 #include "gstrtpssrcdemux.h"
50
51 GST_DEBUG_CATEGORY_STATIC (gst_rtp_ssrc_demux_debug);
52 #define GST_CAT_DEFAULT gst_rtp_ssrc_demux_debug
53
54 /* generic templates */
55 static GstStaticPadTemplate rtp_ssrc_demux_sink_template =
56 GST_STATIC_PAD_TEMPLATE ("sink",
57     GST_PAD_SINK,
58     GST_PAD_ALWAYS,
59     GST_STATIC_CAPS ("application/x-rtp")
60     );
61
62 static GstStaticPadTemplate rtp_ssrc_demux_rtcp_sink_template =
63 GST_STATIC_PAD_TEMPLATE ("rtcp_sink",
64     GST_PAD_SINK,
65     GST_PAD_ALWAYS,
66     GST_STATIC_CAPS ("application/x-rtcp")
67     );
68
69 static GstStaticPadTemplate rtp_ssrc_demux_src_template =
70 GST_STATIC_PAD_TEMPLATE ("src_%u",
71     GST_PAD_SRC,
72     GST_PAD_SOMETIMES,
73     GST_STATIC_CAPS ("application/x-rtp")
74     );
75
76 static GstStaticPadTemplate rtp_ssrc_demux_rtcp_src_template =
77 GST_STATIC_PAD_TEMPLATE ("rtcp_src_%u",
78     GST_PAD_SRC,
79     GST_PAD_SOMETIMES,
80     GST_STATIC_CAPS ("application/x-rtcp")
81     );
82
83 #define GST_PAD_LOCK(obj)   (g_rec_mutex_lock (&(obj)->padlock))
84 #define GST_PAD_UNLOCK(obj) (g_rec_mutex_unlock (&(obj)->padlock))
85
86 typedef enum
87 {
88   RTP_PAD,
89   RTCP_PAD
90 } PadType;
91
92 /* signals */
93 enum
94 {
95   SIGNAL_NEW_SSRC_PAD,
96   SIGNAL_REMOVED_SSRC_PAD,
97   SIGNAL_CLEAR_SSRC,
98   LAST_SIGNAL
99 };
100
101 #define gst_rtp_ssrc_demux_parent_class parent_class
102 G_DEFINE_TYPE (GstRtpSsrcDemux, gst_rtp_ssrc_demux, GST_TYPE_ELEMENT);
103
104 /* GObject vmethods */
105 static void gst_rtp_ssrc_demux_dispose (GObject * object);
106 static void gst_rtp_ssrc_demux_finalize (GObject * object);
107
108 /* GstElement vmethods */
109 static GstStateChangeReturn gst_rtp_ssrc_demux_change_state (GstElement *
110     element, GstStateChange transition);
111
112 static void gst_rtp_ssrc_demux_clear_ssrc (GstRtpSsrcDemux * demux,
113     guint32 ssrc);
114
115 /* sinkpad stuff */
116 static GstFlowReturn gst_rtp_ssrc_demux_chain (GstPad * pad, GstObject * parent,
117     GstBuffer * buf);
118 static gboolean gst_rtp_ssrc_demux_sink_event (GstPad * pad, GstObject * parent,
119     GstEvent * event);
120
121 static GstFlowReturn gst_rtp_ssrc_demux_rtcp_chain (GstPad * pad,
122     GstObject * parent, GstBuffer * buf);
123 static GstIterator *gst_rtp_ssrc_demux_iterate_internal_links_sink (GstPad *
124     pad, GstObject * parent);
125
126 /* srcpad stuff */
127 static gboolean gst_rtp_ssrc_demux_src_event (GstPad * pad, GstObject * parent,
128     GstEvent * event);
129 static GstIterator *gst_rtp_ssrc_demux_iterate_internal_links_src (GstPad * pad,
130     GstObject * parent);
131 static gboolean gst_rtp_ssrc_demux_src_query (GstPad * pad, GstObject * parent,
132     GstQuery * query);
133
134 static guint gst_rtp_ssrc_demux_signals[LAST_SIGNAL] = { 0 };
135
136 /*
137  * Item for storing GstPad <-> SSRC pairs.
138  */
139 struct _GstRtpSsrcDemuxPad
140 {
141   guint32 ssrc;
142   GstPad *rtp_pad;
143   GstCaps *caps;
144   GstPad *rtcp_pad;
145 };
146
147 /* find a src pad for a given SSRC, returns NULL if the SSRC was not found
148  */
149 static GstRtpSsrcDemuxPad *
150 find_demux_pad_for_ssrc (GstRtpSsrcDemux * demux, guint32 ssrc)
151 {
152   GSList *walk;
153
154   for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
155     GstRtpSsrcDemuxPad *pad = (GstRtpSsrcDemuxPad *) walk->data;
156
157     if (pad->ssrc == ssrc)
158       return pad;
159   }
160   return NULL;
161 }
162
163 static GstEvent *
164 add_ssrc_and_ref (GstEvent * event, guint32 ssrc)
165 {
166   /* Set the ssrc on the output caps */
167   switch (GST_EVENT_TYPE (event)) {
168     case GST_EVENT_CAPS:
169     {
170       GstCaps *caps;
171       GstCaps *newcaps;
172       GstStructure *s;
173
174       gst_event_parse_caps (event, &caps);
175       newcaps = gst_caps_copy (caps);
176
177       s = gst_caps_get_structure (newcaps, 0);
178       gst_structure_set (s, "ssrc", G_TYPE_UINT, ssrc, NULL);
179       event = gst_event_new_caps (newcaps);
180       gst_caps_unref (newcaps);
181       break;
182     }
183     default:
184       gst_event_ref (event);
185       break;
186   }
187
188   return event;
189 }
190
191 struct ForwardStickyEventData
192 {
193   GstPad *pad;
194   guint32 ssrc;
195 };
196
197 /* With internal stream lock held */
198 static gboolean
199 forward_sticky_events (GstPad * pad, GstEvent ** event, gpointer user_data)
200 {
201   struct ForwardStickyEventData *data = user_data;
202   GstEvent *newevent;
203
204   newevent = add_ssrc_and_ref (*event, data->ssrc);
205
206   gst_pad_push_event (data->pad, newevent);
207
208   return TRUE;
209 }
210
211 /* With internal stream lock held */
212 static void
213 forward_initial_events (GstRtpSsrcDemux * demux, guint32 ssrc, GstPad * pad,
214     PadType padtype)
215 {
216   struct ForwardStickyEventData fdata;
217   GstPad *sinkpad = NULL;
218
219   if (padtype == RTP_PAD)
220     sinkpad = demux->rtp_sink;
221   else if (padtype == RTCP_PAD)
222     sinkpad = demux->rtcp_sink;
223   else
224     g_assert_not_reached ();
225
226   fdata.ssrc = ssrc;
227   fdata.pad = pad;
228
229   gst_pad_sticky_events_foreach (sinkpad, forward_sticky_events, &fdata);
230 }
231
232 static GstPad *
233 find_or_create_demux_pad_for_ssrc (GstRtpSsrcDemux * demux, guint32 ssrc,
234     PadType padtype)
235 {
236   GstPad *rtp_pad, *rtcp_pad;
237   GstElementClass *klass;
238   GstPadTemplate *templ;
239   gchar *padname;
240   GstRtpSsrcDemuxPad *demuxpad;
241   GstPad *retpad;
242
243   GST_PAD_LOCK (demux);
244
245   demuxpad = find_demux_pad_for_ssrc (demux, ssrc);
246   if (demuxpad != NULL) {
247     switch (padtype) {
248       case RTP_PAD:
249         retpad = gst_object_ref (demuxpad->rtp_pad);
250         break;
251       case RTCP_PAD:
252         retpad = gst_object_ref (demuxpad->rtcp_pad);
253         break;
254       default:
255         retpad = NULL;
256         g_assert_not_reached ();
257     }
258
259     GST_PAD_UNLOCK (demux);
260
261     return retpad;
262   }
263
264   GST_DEBUG_OBJECT (demux, "creating new pad for SSRC %08x", ssrc);
265
266   klass = GST_ELEMENT_GET_CLASS (demux);
267   templ = gst_element_class_get_pad_template (klass, "src_%u");
268   padname = g_strdup_printf ("src_%u", ssrc);
269   rtp_pad = gst_pad_new_from_template (templ, padname);
270   g_free (padname);
271
272   templ = gst_element_class_get_pad_template (klass, "rtcp_src_%u");
273   padname = g_strdup_printf ("rtcp_src_%u", ssrc);
274   rtcp_pad = gst_pad_new_from_template (templ, padname);
275   g_free (padname);
276
277   /* wrap in structure and add to list */
278   demuxpad = g_new0 (GstRtpSsrcDemuxPad, 1);
279   demuxpad->ssrc = ssrc;
280   demuxpad->rtp_pad = rtp_pad;
281   demuxpad->rtcp_pad = rtcp_pad;
282
283   gst_pad_set_element_private (rtp_pad, demuxpad);
284   gst_pad_set_element_private (rtcp_pad, demuxpad);
285
286   demux->srcpads = g_slist_prepend (demux->srcpads, demuxpad);
287
288   gst_pad_set_query_function (rtp_pad, gst_rtp_ssrc_demux_src_query);
289   gst_pad_set_iterate_internal_links_function (rtp_pad,
290       gst_rtp_ssrc_demux_iterate_internal_links_src);
291   gst_pad_set_event_function (rtp_pad, gst_rtp_ssrc_demux_src_event);
292   gst_pad_use_fixed_caps (rtp_pad);
293   gst_pad_set_active (rtp_pad, TRUE);
294
295   gst_pad_set_event_function (rtcp_pad, gst_rtp_ssrc_demux_src_event);
296   gst_pad_set_iterate_internal_links_function (rtcp_pad,
297       gst_rtp_ssrc_demux_iterate_internal_links_src);
298   gst_pad_use_fixed_caps (rtcp_pad);
299   gst_pad_set_active (rtcp_pad, TRUE);
300
301   forward_initial_events (demux, ssrc, rtp_pad, RTP_PAD);
302   forward_initial_events (demux, ssrc, rtcp_pad, RTCP_PAD);
303
304   gst_element_add_pad (GST_ELEMENT_CAST (demux), rtp_pad);
305   gst_element_add_pad (GST_ELEMENT_CAST (demux), rtcp_pad);
306
307   switch (padtype) {
308     case RTP_PAD:
309       retpad = gst_object_ref (demuxpad->rtp_pad);
310       break;
311     case RTCP_PAD:
312       retpad = gst_object_ref (demuxpad->rtcp_pad);
313       break;
314     default:
315       retpad = NULL;
316       g_assert_not_reached ();
317   }
318
319   gst_object_ref (rtp_pad);
320   gst_object_ref (rtcp_pad);
321
322   g_signal_emit (G_OBJECT (demux),
323       gst_rtp_ssrc_demux_signals[SIGNAL_NEW_SSRC_PAD], 0, ssrc, rtp_pad);
324
325   GST_PAD_UNLOCK (demux);
326
327   gst_object_unref (rtp_pad);
328   gst_object_unref (rtcp_pad);
329
330   return retpad;
331 }
332
333 static void
334 gst_rtp_ssrc_demux_class_init (GstRtpSsrcDemuxClass * klass)
335 {
336   GObjectClass *gobject_klass;
337   GstElementClass *gstelement_klass;
338   GstRtpSsrcDemuxClass *gstrtpssrcdemux_klass;
339
340   gobject_klass = (GObjectClass *) klass;
341   gstelement_klass = (GstElementClass *) klass;
342   gstrtpssrcdemux_klass = (GstRtpSsrcDemuxClass *) klass;
343
344   gobject_klass->dispose = gst_rtp_ssrc_demux_dispose;
345   gobject_klass->finalize = gst_rtp_ssrc_demux_finalize;
346
347   /**
348    * GstRtpSsrcDemux::new-ssrc-pad:
349    * @demux: the object which received the signal
350    * @ssrc: the SSRC of the pad
351    * @pad: the new pad.
352    *
353    * Emited when a new SSRC pad has been created.
354    */
355   gst_rtp_ssrc_demux_signals[SIGNAL_NEW_SSRC_PAD] =
356       g_signal_new ("new-ssrc-pad",
357       G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
358       G_STRUCT_OFFSET (GstRtpSsrcDemuxClass, new_ssrc_pad),
359       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
360       GST_TYPE_PAD);
361
362   /**
363    * GstRtpSsrcDemux::removed-ssrc-pad:
364    * @demux: the object which received the signal
365    * @ssrc: the SSRC of the pad
366    * @pad: the removed pad.
367    *
368    * Emited when a SSRC pad has been removed.
369    */
370   gst_rtp_ssrc_demux_signals[SIGNAL_REMOVED_SSRC_PAD] =
371       g_signal_new ("removed-ssrc-pad",
372       G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
373       G_STRUCT_OFFSET (GstRtpSsrcDemuxClass, removed_ssrc_pad),
374       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
375       GST_TYPE_PAD);
376
377   /**
378    * GstRtpSsrcDemux::clear-ssrc:
379    * @demux: the object which received the signal
380    * @ssrc: the SSRC of the pad
381    *
382    * Action signal to remove the pad for SSRC.
383    */
384   gst_rtp_ssrc_demux_signals[SIGNAL_CLEAR_SSRC] =
385       g_signal_new ("clear-ssrc",
386       G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
387       G_STRUCT_OFFSET (GstRtpSsrcDemuxClass, clear_ssrc),
388       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 1, G_TYPE_UINT);
389
390   gstelement_klass->change_state =
391       GST_DEBUG_FUNCPTR (gst_rtp_ssrc_demux_change_state);
392   gstrtpssrcdemux_klass->clear_ssrc =
393       GST_DEBUG_FUNCPTR (gst_rtp_ssrc_demux_clear_ssrc);
394
395   gst_element_class_add_static_pad_template (gstelement_klass,
396       &rtp_ssrc_demux_sink_template);
397   gst_element_class_add_static_pad_template (gstelement_klass,
398       &rtp_ssrc_demux_rtcp_sink_template);
399   gst_element_class_add_static_pad_template (gstelement_klass,
400       &rtp_ssrc_demux_src_template);
401   gst_element_class_add_static_pad_template (gstelement_klass,
402       &rtp_ssrc_demux_rtcp_src_template);
403
404   gst_element_class_set_static_metadata (gstelement_klass, "RTP SSRC Demux",
405       "Demux/Network/RTP",
406       "Splits RTP streams based on the SSRC",
407       "Wim Taymans <wim.taymans@gmail.com>");
408
409   GST_DEBUG_CATEGORY_INIT (gst_rtp_ssrc_demux_debug,
410       "rtpssrcdemux", 0, "RTP SSRC demuxer");
411 }
412
413 static void
414 gst_rtp_ssrc_demux_init (GstRtpSsrcDemux * demux)
415 {
416   GstElementClass *klass = GST_ELEMENT_GET_CLASS (demux);
417
418   demux->rtp_sink =
419       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
420           "sink"), "sink");
421   gst_pad_set_chain_function (demux->rtp_sink, gst_rtp_ssrc_demux_chain);
422   gst_pad_set_event_function (demux->rtp_sink, gst_rtp_ssrc_demux_sink_event);
423   gst_pad_set_iterate_internal_links_function (demux->rtp_sink,
424       gst_rtp_ssrc_demux_iterate_internal_links_sink);
425   gst_element_add_pad (GST_ELEMENT_CAST (demux), demux->rtp_sink);
426
427   demux->rtcp_sink =
428       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
429           "rtcp_sink"), "rtcp_sink");
430   gst_pad_set_chain_function (demux->rtcp_sink, gst_rtp_ssrc_demux_rtcp_chain);
431   gst_pad_set_event_function (demux->rtcp_sink, gst_rtp_ssrc_demux_sink_event);
432   gst_pad_set_iterate_internal_links_function (demux->rtcp_sink,
433       gst_rtp_ssrc_demux_iterate_internal_links_sink);
434   gst_element_add_pad (GST_ELEMENT_CAST (demux), demux->rtcp_sink);
435
436   g_rec_mutex_init (&demux->padlock);
437 }
438
439 static void
440 gst_rtp_ssrc_demux_reset (GstRtpSsrcDemux * demux)
441 {
442   GSList *walk;
443
444   for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
445     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) walk->data;
446
447     gst_pad_set_active (dpad->rtp_pad, FALSE);
448     gst_pad_set_active (dpad->rtcp_pad, FALSE);
449
450     gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtp_pad);
451     gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtcp_pad);
452     g_free (dpad);
453   }
454   g_slist_free (demux->srcpads);
455   demux->srcpads = NULL;
456 }
457
458 static void
459 gst_rtp_ssrc_demux_dispose (GObject * object)
460 {
461   GstRtpSsrcDemux *demux;
462
463   demux = GST_RTP_SSRC_DEMUX (object);
464
465   gst_rtp_ssrc_demux_reset (demux);
466
467   G_OBJECT_CLASS (parent_class)->dispose (object);
468 }
469
470 static void
471 gst_rtp_ssrc_demux_finalize (GObject * object)
472 {
473   GstRtpSsrcDemux *demux;
474
475   demux = GST_RTP_SSRC_DEMUX (object);
476   g_rec_mutex_clear (&demux->padlock);
477
478   G_OBJECT_CLASS (parent_class)->finalize (object);
479 }
480
481 static void
482 gst_rtp_ssrc_demux_clear_ssrc (GstRtpSsrcDemux * demux, guint32 ssrc)
483 {
484   GstRtpSsrcDemuxPad *dpad;
485
486   GST_PAD_LOCK (demux);
487   dpad = find_demux_pad_for_ssrc (demux, ssrc);
488   if (dpad == NULL) {
489     GST_PAD_UNLOCK (demux);
490     goto unknown_pad;
491   }
492
493   GST_DEBUG_OBJECT (demux, "clearing pad for SSRC %08x", ssrc);
494
495   demux->srcpads = g_slist_remove (demux->srcpads, dpad);
496   GST_PAD_UNLOCK (demux);
497
498   gst_pad_set_active (dpad->rtp_pad, FALSE);
499   gst_pad_set_active (dpad->rtcp_pad, FALSE);
500
501   g_signal_emit (G_OBJECT (demux),
502       gst_rtp_ssrc_demux_signals[SIGNAL_REMOVED_SSRC_PAD], 0, ssrc,
503       dpad->rtp_pad);
504
505   gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtp_pad);
506   gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtcp_pad);
507
508   g_free (dpad);
509
510   return;
511
512   /* ERRORS */
513 unknown_pad:
514   {
515     GST_WARNING_OBJECT (demux, "unknown SSRC %08x", ssrc);
516     return;
517   }
518 }
519
520 struct ForwardEventData
521 {
522   GstRtpSsrcDemux *demux;
523   GstEvent *event;
524   gboolean res;
525   GstPad *pad;
526 };
527
528 static gboolean
529 forward_event (GstPad * pad, gpointer user_data)
530 {
531   struct ForwardEventData *fdata = user_data;
532   GSList *walk = NULL;
533   GstEvent *newevent = NULL;
534
535   GST_PAD_LOCK (fdata->demux);
536   for (walk = fdata->demux->srcpads; walk; walk = walk->next) {
537     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) walk->data;
538
539     if (pad == dpad->rtp_pad || pad == dpad->rtcp_pad) {
540       newevent = add_ssrc_and_ref (fdata->event, dpad->ssrc);
541       break;
542     }
543   }
544   GST_PAD_UNLOCK (fdata->demux);
545
546   if (newevent)
547     fdata->res &= gst_pad_push_event (pad, newevent);
548
549   return FALSE;
550 }
551
552
553 static gboolean
554 gst_rtp_ssrc_demux_sink_event (GstPad * pad, GstObject * parent,
555     GstEvent * event)
556 {
557   GstRtpSsrcDemux *demux;
558   struct ForwardEventData fdata;
559
560   demux = GST_RTP_SSRC_DEMUX (parent);
561
562   fdata.demux = demux;
563   fdata.pad = pad;
564   fdata.event = event;
565   fdata.res = TRUE;
566
567   gst_pad_forward (pad, forward_event, &fdata);
568
569   gst_event_unref (event);
570
571   return fdata.res;
572 }
573
574 static GstFlowReturn
575 gst_rtp_ssrc_demux_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
576 {
577   GstFlowReturn ret;
578   GstRtpSsrcDemux *demux;
579   guint32 ssrc;
580   GstRTPBuffer rtp = { NULL };
581   GstPad *srcpad;
582   GstRtpSsrcDemuxPad *dpad;
583
584   demux = GST_RTP_SSRC_DEMUX (parent);
585
586   if (!gst_rtp_buffer_map (buf, GST_MAP_READ, &rtp))
587     goto invalid_payload;
588
589   ssrc = gst_rtp_buffer_get_ssrc (&rtp);
590   gst_rtp_buffer_unmap (&rtp);
591
592   GST_DEBUG_OBJECT (demux, "received buffer of SSRC %08x", ssrc);
593
594   srcpad = find_or_create_demux_pad_for_ssrc (demux, ssrc, RTP_PAD);
595   if (srcpad == NULL)
596     goto create_failed;
597
598   /* push to srcpad */
599   ret = gst_pad_push (srcpad, buf);
600
601   if (ret != GST_FLOW_OK) {
602     /* check if the ssrc still there, may have been removed */
603     GST_PAD_LOCK (demux);
604     dpad = find_demux_pad_for_ssrc (demux, ssrc);
605     if (dpad == NULL || dpad->rtp_pad != srcpad) {
606       /* SSRC was removed during the push ... ignore the error */
607       ret = GST_FLOW_OK;
608     }
609     GST_PAD_UNLOCK (demux);
610   }
611
612   gst_object_unref (srcpad);
613
614   return ret;
615
616   /* ERRORS */
617 invalid_payload:
618   {
619     /* this is fatal and should be filtered earlier */
620     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
621         ("Dropping invalid RTP payload"));
622     gst_buffer_unref (buf);
623     return GST_FLOW_ERROR;
624   }
625 create_failed:
626   {
627     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
628         ("Could not create new pad"));
629     gst_buffer_unref (buf);
630     return GST_FLOW_ERROR;
631   }
632 }
633
634 static GstFlowReturn
635 gst_rtp_ssrc_demux_rtcp_chain (GstPad * pad, GstObject * parent,
636     GstBuffer * buf)
637 {
638   GstFlowReturn ret;
639   GstRtpSsrcDemux *demux;
640   guint32 ssrc;
641   GstRTCPPacket packet;
642   GstRTCPBuffer rtcp = { NULL, };
643   GstPad *srcpad;
644   GstRtpSsrcDemuxPad *dpad;
645
646   demux = GST_RTP_SSRC_DEMUX (parent);
647
648   if (!gst_rtcp_buffer_validate_reduced (buf))
649     goto invalid_rtcp;
650
651   gst_rtcp_buffer_map (buf, GST_MAP_READ, &rtcp);
652   if (!gst_rtcp_buffer_get_first_packet (&rtcp, &packet)) {
653     gst_rtcp_buffer_unmap (&rtcp);
654     goto invalid_rtcp;
655   }
656
657   /* first packet must be SR or RR, or in case of a reduced size RTCP packet
658    * it must be APP, RTPFB or PSFB feeadback, or else the validate would
659    * have failed */
660   switch (gst_rtcp_packet_get_type (&packet)) {
661     case GST_RTCP_TYPE_SR:
662       /* get the ssrc so that we can route it to the right source pad */
663       gst_rtcp_packet_sr_get_sender_info (&packet, &ssrc, NULL, NULL, NULL,
664           NULL);
665       break;
666     case GST_RTCP_TYPE_RR:
667       ssrc = gst_rtcp_packet_rr_get_ssrc (&packet);
668       break;
669     case GST_RTCP_TYPE_APP:
670     case GST_RTCP_TYPE_RTPFB:
671     case GST_RTCP_TYPE_PSFB:
672       ssrc = gst_rtcp_packet_fb_get_sender_ssrc (&packet);
673       break;
674     default:
675       goto unexpected_rtcp;
676   }
677   gst_rtcp_buffer_unmap (&rtcp);
678
679   GST_DEBUG_OBJECT (demux, "received RTCP of SSRC %08x", ssrc);
680
681   srcpad = find_or_create_demux_pad_for_ssrc (demux, ssrc, RTCP_PAD);
682   if (srcpad == NULL)
683     goto create_failed;
684
685   /* push to srcpad */
686   ret = gst_pad_push (srcpad, buf);
687
688   if (ret != GST_FLOW_OK) {
689     /* check if the ssrc still there, may have been removed */
690     GST_PAD_LOCK (demux);
691     dpad = find_demux_pad_for_ssrc (demux, ssrc);
692     if (dpad == NULL || dpad->rtcp_pad != srcpad) {
693       /* SSRC was removed during the push ... ignore the error */
694       ret = GST_FLOW_OK;
695     }
696     GST_PAD_UNLOCK (demux);
697   }
698
699   gst_object_unref (srcpad);
700
701   return ret;
702
703   /* ERRORS */
704 invalid_rtcp:
705   {
706     /* this is fatal and should be filtered earlier */
707     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
708         ("Dropping invalid RTCP packet"));
709     gst_buffer_unref (buf);
710     return GST_FLOW_ERROR;
711   }
712 unexpected_rtcp:
713   {
714     GST_DEBUG_OBJECT (demux, "dropping unexpected RTCP packet");
715     gst_buffer_unref (buf);
716     return GST_FLOW_OK;
717   }
718 create_failed:
719   {
720     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
721         ("Could not create new pad"));
722     gst_buffer_unref (buf);
723     return GST_FLOW_ERROR;
724   }
725 }
726
727 static GstRtpSsrcDemuxPad *
728 find_demux_pad_for_pad (GstRtpSsrcDemux * demux, GstPad * pad)
729 {
730   GSList *walk;
731
732   for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
733     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) walk->data;
734     if (dpad->rtp_pad == pad || dpad->rtcp_pad == pad) {
735       return dpad;
736     }
737   }
738
739   return NULL;
740 }
741
742
743 static gboolean
744 gst_rtp_ssrc_demux_src_event (GstPad * pad, GstObject * parent,
745     GstEvent * event)
746 {
747   GstRtpSsrcDemux *demux;
748   const GstStructure *s;
749
750   demux = GST_RTP_SSRC_DEMUX (parent);
751
752   switch (GST_EVENT_TYPE (event)) {
753     case GST_EVENT_CUSTOM_UPSTREAM:
754     case GST_EVENT_CUSTOM_BOTH:
755     case GST_EVENT_CUSTOM_BOTH_OOB:
756       s = gst_event_get_structure (event);
757       if (s && !gst_structure_has_field (s, "ssrc")) {
758         GstRtpSsrcDemuxPad *dpad = find_demux_pad_for_pad (demux, pad);
759
760         if (dpad) {
761           GstStructure *ws;
762
763           event = gst_event_make_writable (event);
764           ws = gst_event_writable_structure (event);
765           gst_structure_set (ws, "ssrc", G_TYPE_UINT, dpad->ssrc, NULL);
766         }
767       }
768       break;
769     default:
770       break;
771   }
772
773   return gst_pad_event_default (pad, parent, event);
774 }
775
776 static GstIterator *
777 gst_rtp_ssrc_demux_iterate_internal_links_src (GstPad * pad, GstObject * parent)
778 {
779   GstRtpSsrcDemux *demux;
780   GstPad *otherpad = NULL;
781   GstIterator *it = NULL;
782   GSList *current;
783
784   demux = GST_RTP_SSRC_DEMUX (parent);
785
786   GST_PAD_LOCK (demux);
787   for (current = demux->srcpads; current; current = g_slist_next (current)) {
788     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) current->data;
789
790     if (pad == dpad->rtp_pad) {
791       otherpad = demux->rtp_sink;
792       break;
793     } else if (pad == dpad->rtcp_pad) {
794       otherpad = demux->rtcp_sink;
795       break;
796     }
797   }
798   if (otherpad) {
799     GValue val = { 0, };
800
801     g_value_init (&val, GST_TYPE_PAD);
802     g_value_set_object (&val, otherpad);
803     it = gst_iterator_new_single (GST_TYPE_PAD, &val);
804     g_value_unset (&val);
805
806   }
807   GST_PAD_UNLOCK (demux);
808
809   return it;
810 }
811
812 /* Should return 0 for elements to be included */
813 static gint
814 src_pad_compare_func (gconstpointer a, gconstpointer b)
815 {
816   GstPad *pad = GST_PAD (g_value_get_object (a));
817   const gchar *prefix = g_value_get_string (b);
818   gint res;
819
820   /* 0 means equal means we accept the pad, accepted if there is a name
821    * and it starts with the prefix */
822   GST_OBJECT_LOCK (pad);
823   res = !GST_PAD_NAME (pad) || !g_str_has_prefix (GST_PAD_NAME (pad), prefix);
824   GST_OBJECT_UNLOCK (pad);
825
826   return res;
827 }
828
829 static GstIterator *
830 gst_rtp_ssrc_demux_iterate_internal_links_sink (GstPad * pad,
831     GstObject * parent)
832 {
833   GstRtpSsrcDemux *demux;
834   GstIterator *it = NULL;
835   GValue gval = { 0, };
836
837   demux = GST_RTP_SSRC_DEMUX (parent);
838
839   g_value_init (&gval, G_TYPE_STRING);
840   if (pad == demux->rtp_sink)
841     g_value_set_static_string (&gval, "src_");
842   else if (pad == demux->rtcp_sink)
843     g_value_set_static_string (&gval, "rtcp_src_");
844   else
845     g_assert_not_reached ();
846
847   it = gst_element_iterate_src_pads (GST_ELEMENT_CAST (demux));
848   it = gst_iterator_filter (it, src_pad_compare_func, &gval);
849
850   return it;
851 }
852
853
854 static gboolean
855 gst_rtp_ssrc_demux_src_query (GstPad * pad, GstObject * parent,
856     GstQuery * query)
857 {
858   GstRtpSsrcDemux *demux;
859   gboolean res = FALSE;
860
861   demux = GST_RTP_SSRC_DEMUX (parent);
862
863   switch (GST_QUERY_TYPE (query)) {
864     case GST_QUERY_LATENCY:
865     {
866
867       if ((res = gst_pad_peer_query (demux->rtp_sink, query))) {
868         gboolean live;
869         GstClockTime min_latency, max_latency;
870         GstRtpSsrcDemuxPad *demuxpad;
871
872         demuxpad = gst_pad_get_element_private (pad);
873
874         gst_query_parse_latency (query, &live, &min_latency, &max_latency);
875
876         GST_DEBUG_OBJECT (demux, "peer min latency %" GST_TIME_FORMAT,
877             GST_TIME_ARGS (min_latency));
878
879         GST_DEBUG_OBJECT (demux, "latency for SSRC %08x", demuxpad->ssrc);
880
881         gst_query_set_latency (query, live, min_latency, max_latency);
882       }
883       break;
884     }
885     default:
886       res = gst_pad_query_default (pad, parent, query);
887       break;
888   }
889
890   return res;
891 }
892
893 static GstStateChangeReturn
894 gst_rtp_ssrc_demux_change_state (GstElement * element,
895     GstStateChange transition)
896 {
897   GstStateChangeReturn ret;
898   GstRtpSsrcDemux *demux;
899
900   demux = GST_RTP_SSRC_DEMUX (element);
901
902   switch (transition) {
903     case GST_STATE_CHANGE_NULL_TO_READY:
904     case GST_STATE_CHANGE_READY_TO_PAUSED:
905     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
906     default:
907       break;
908   }
909
910   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
911
912   switch (transition) {
913     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
914       break;
915     case GST_STATE_CHANGE_PAUSED_TO_READY:
916       gst_rtp_ssrc_demux_reset (demux);
917       break;
918     case GST_STATE_CHANGE_READY_TO_NULL:
919     default:
920       break;
921   }
922   return ret;
923 }