rtpsession: Call on-new-ssrc earlier
[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 INTERNAL_STREAM_LOCK(obj)   (g_rec_mutex_lock (&(obj)->padlock))
84 #define INTERNAL_STREAM_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   INTERNAL_STREAM_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     INTERNAL_STREAM_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   INTERNAL_STREAM_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   GST_DEBUG_REGISTER_FUNCPTR (gst_rtp_ssrc_demux_chain);
413   GST_DEBUG_REGISTER_FUNCPTR (gst_rtp_ssrc_demux_rtcp_chain);
414 }
415
416 static void
417 gst_rtp_ssrc_demux_init (GstRtpSsrcDemux * demux)
418 {
419   GstElementClass *klass = GST_ELEMENT_GET_CLASS (demux);
420
421   demux->rtp_sink =
422       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
423           "sink"), "sink");
424   gst_pad_set_chain_function (demux->rtp_sink, gst_rtp_ssrc_demux_chain);
425   gst_pad_set_event_function (demux->rtp_sink, gst_rtp_ssrc_demux_sink_event);
426   gst_pad_set_iterate_internal_links_function (demux->rtp_sink,
427       gst_rtp_ssrc_demux_iterate_internal_links_sink);
428   gst_element_add_pad (GST_ELEMENT_CAST (demux), demux->rtp_sink);
429
430   demux->rtcp_sink =
431       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
432           "rtcp_sink"), "rtcp_sink");
433   gst_pad_set_chain_function (demux->rtcp_sink, gst_rtp_ssrc_demux_rtcp_chain);
434   gst_pad_set_event_function (demux->rtcp_sink, gst_rtp_ssrc_demux_sink_event);
435   gst_pad_set_iterate_internal_links_function (demux->rtcp_sink,
436       gst_rtp_ssrc_demux_iterate_internal_links_sink);
437   gst_element_add_pad (GST_ELEMENT_CAST (demux), demux->rtcp_sink);
438
439   g_rec_mutex_init (&demux->padlock);
440 }
441
442 static void
443 gst_rtp_ssrc_demux_reset (GstRtpSsrcDemux * demux)
444 {
445   GSList *walk;
446
447   for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
448     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) walk->data;
449
450     gst_pad_set_active (dpad->rtp_pad, FALSE);
451     gst_pad_set_active (dpad->rtcp_pad, FALSE);
452
453     gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtp_pad);
454     gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtcp_pad);
455     g_free (dpad);
456   }
457   g_slist_free (demux->srcpads);
458   demux->srcpads = NULL;
459 }
460
461 static void
462 gst_rtp_ssrc_demux_dispose (GObject * object)
463 {
464   GstRtpSsrcDemux *demux;
465
466   demux = GST_RTP_SSRC_DEMUX (object);
467
468   gst_rtp_ssrc_demux_reset (demux);
469
470   G_OBJECT_CLASS (parent_class)->dispose (object);
471 }
472
473 static void
474 gst_rtp_ssrc_demux_finalize (GObject * object)
475 {
476   GstRtpSsrcDemux *demux;
477
478   demux = GST_RTP_SSRC_DEMUX (object);
479   g_rec_mutex_clear (&demux->padlock);
480
481   G_OBJECT_CLASS (parent_class)->finalize (object);
482 }
483
484 static void
485 gst_rtp_ssrc_demux_clear_ssrc (GstRtpSsrcDemux * demux, guint32 ssrc)
486 {
487   GstRtpSsrcDemuxPad *dpad;
488
489   INTERNAL_STREAM_LOCK (demux);
490   dpad = find_demux_pad_for_ssrc (demux, ssrc);
491   if (dpad == NULL) {
492     INTERNAL_STREAM_UNLOCK (demux);
493     goto unknown_pad;
494   }
495
496   GST_DEBUG_OBJECT (demux, "clearing pad for SSRC %08x", ssrc);
497
498   demux->srcpads = g_slist_remove (demux->srcpads, dpad);
499   INTERNAL_STREAM_UNLOCK (demux);
500
501   gst_pad_set_active (dpad->rtp_pad, FALSE);
502   gst_pad_set_active (dpad->rtcp_pad, FALSE);
503
504   g_signal_emit (G_OBJECT (demux),
505       gst_rtp_ssrc_demux_signals[SIGNAL_REMOVED_SSRC_PAD], 0, ssrc,
506       dpad->rtp_pad);
507
508   gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtp_pad);
509   gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtcp_pad);
510
511   g_free (dpad);
512
513   return;
514
515   /* ERRORS */
516 unknown_pad:
517   {
518     GST_WARNING_OBJECT (demux, "unknown SSRC %08x", ssrc);
519     return;
520   }
521 }
522
523 struct ForwardEventData
524 {
525   GstRtpSsrcDemux *demux;
526   GstEvent *event;
527   gboolean res;
528   GstPad *pad;
529 };
530
531 static gboolean
532 forward_event (GstPad * pad, gpointer user_data)
533 {
534   struct ForwardEventData *fdata = user_data;
535   GSList *walk = NULL;
536   GstEvent *newevent = NULL;
537
538   INTERNAL_STREAM_LOCK (fdata->demux);
539   for (walk = fdata->demux->srcpads; walk; walk = walk->next) {
540     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) walk->data;
541
542     if (pad == dpad->rtp_pad || pad == dpad->rtcp_pad) {
543       newevent = add_ssrc_and_ref (fdata->event, dpad->ssrc);
544       break;
545     }
546   }
547   INTERNAL_STREAM_UNLOCK (fdata->demux);
548
549   if (newevent)
550     fdata->res &= gst_pad_push_event (pad, newevent);
551
552   return FALSE;
553 }
554
555
556 static gboolean
557 gst_rtp_ssrc_demux_sink_event (GstPad * pad, GstObject * parent,
558     GstEvent * event)
559 {
560   GstRtpSsrcDemux *demux;
561   struct ForwardEventData fdata;
562
563   demux = GST_RTP_SSRC_DEMUX (parent);
564
565   fdata.demux = demux;
566   fdata.pad = pad;
567   fdata.event = event;
568   fdata.res = TRUE;
569
570   gst_pad_forward (pad, forward_event, &fdata);
571
572   gst_event_unref (event);
573
574   return fdata.res;
575 }
576
577 static GstFlowReturn
578 gst_rtp_ssrc_demux_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
579 {
580   GstFlowReturn ret;
581   GstRtpSsrcDemux *demux;
582   guint32 ssrc;
583   GstRTPBuffer rtp = { NULL };
584   GstPad *srcpad;
585   GstRtpSsrcDemuxPad *dpad;
586
587   demux = GST_RTP_SSRC_DEMUX (parent);
588
589   if (!gst_rtp_buffer_map (buf, GST_MAP_READ, &rtp))
590     goto invalid_payload;
591
592   ssrc = gst_rtp_buffer_get_ssrc (&rtp);
593   gst_rtp_buffer_unmap (&rtp);
594
595   GST_DEBUG_OBJECT (demux, "received buffer of SSRC %08x", ssrc);
596
597   srcpad = find_or_create_demux_pad_for_ssrc (demux, ssrc, RTP_PAD);
598   if (srcpad == NULL)
599     goto create_failed;
600
601   /* push to srcpad */
602   ret = gst_pad_push (srcpad, buf);
603
604   if (ret != GST_FLOW_OK) {
605     /* check if the ssrc still there, may have been removed */
606     INTERNAL_STREAM_LOCK (demux);
607     dpad = find_demux_pad_for_ssrc (demux, ssrc);
608     if (dpad == NULL || dpad->rtp_pad != srcpad) {
609       /* SSRC was removed during the push ... ignore the error */
610       ret = GST_FLOW_OK;
611     }
612     INTERNAL_STREAM_UNLOCK (demux);
613   }
614
615   gst_object_unref (srcpad);
616
617   return ret;
618
619   /* ERRORS */
620 invalid_payload:
621   {
622     /* this is fatal and should be filtered earlier */
623     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
624         ("Dropping invalid RTP payload"));
625     gst_buffer_unref (buf);
626     return GST_FLOW_ERROR;
627   }
628 create_failed:
629   {
630     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
631         ("Could not create new pad"));
632     gst_buffer_unref (buf);
633     return GST_FLOW_ERROR;
634   }
635 }
636
637 static GstFlowReturn
638 gst_rtp_ssrc_demux_rtcp_chain (GstPad * pad, GstObject * parent,
639     GstBuffer * buf)
640 {
641   GstFlowReturn ret;
642   GstRtpSsrcDemux *demux;
643   guint32 ssrc;
644   GstRTCPPacket packet;
645   GstRTCPBuffer rtcp = { NULL, };
646   GstPad *srcpad;
647   GstRtpSsrcDemuxPad *dpad;
648
649   demux = GST_RTP_SSRC_DEMUX (parent);
650
651   if (!gst_rtcp_buffer_validate_reduced (buf))
652     goto invalid_rtcp;
653
654   gst_rtcp_buffer_map (buf, GST_MAP_READ, &rtcp);
655   if (!gst_rtcp_buffer_get_first_packet (&rtcp, &packet)) {
656     gst_rtcp_buffer_unmap (&rtcp);
657     goto invalid_rtcp;
658   }
659
660   /* first packet must be SR or RR, or in case of a reduced size RTCP packet
661    * it must be APP, RTPFB or PSFB feeadback, or else the validate would
662    * have failed */
663   switch (gst_rtcp_packet_get_type (&packet)) {
664     case GST_RTCP_TYPE_SR:
665       /* get the ssrc so that we can route it to the right source pad */
666       gst_rtcp_packet_sr_get_sender_info (&packet, &ssrc, NULL, NULL, NULL,
667           NULL);
668       break;
669     case GST_RTCP_TYPE_RR:
670       ssrc = gst_rtcp_packet_rr_get_ssrc (&packet);
671       break;
672     case GST_RTCP_TYPE_APP:
673     case GST_RTCP_TYPE_RTPFB:
674     case GST_RTCP_TYPE_PSFB:
675       ssrc = gst_rtcp_packet_fb_get_sender_ssrc (&packet);
676       break;
677     default:
678       goto unexpected_rtcp;
679   }
680   gst_rtcp_buffer_unmap (&rtcp);
681
682   GST_DEBUG_OBJECT (demux, "received RTCP of SSRC %08x", ssrc);
683
684   srcpad = find_or_create_demux_pad_for_ssrc (demux, ssrc, RTCP_PAD);
685   if (srcpad == NULL)
686     goto create_failed;
687
688   /* push to srcpad */
689   ret = gst_pad_push (srcpad, buf);
690
691   if (ret != GST_FLOW_OK) {
692     /* check if the ssrc still there, may have been removed */
693     INTERNAL_STREAM_LOCK (demux);
694     dpad = find_demux_pad_for_ssrc (demux, ssrc);
695     if (dpad == NULL || dpad->rtcp_pad != srcpad) {
696       /* SSRC was removed during the push ... ignore the error */
697       ret = GST_FLOW_OK;
698     }
699     INTERNAL_STREAM_UNLOCK (demux);
700   }
701
702   gst_object_unref (srcpad);
703
704   return ret;
705
706   /* ERRORS */
707 invalid_rtcp:
708   {
709     /* this is fatal and should be filtered earlier */
710     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
711         ("Dropping invalid RTCP packet"));
712     gst_buffer_unref (buf);
713     return GST_FLOW_ERROR;
714   }
715 unexpected_rtcp:
716   {
717     GST_DEBUG_OBJECT (demux, "dropping unexpected RTCP packet");
718     gst_buffer_unref (buf);
719     return GST_FLOW_OK;
720   }
721 create_failed:
722   {
723     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
724         ("Could not create new pad"));
725     gst_buffer_unref (buf);
726     return GST_FLOW_ERROR;
727   }
728 }
729
730 static GstRtpSsrcDemuxPad *
731 find_demux_pad_for_pad (GstRtpSsrcDemux * demux, GstPad * pad)
732 {
733   GSList *walk;
734
735   for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
736     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) walk->data;
737     if (dpad->rtp_pad == pad || dpad->rtcp_pad == pad) {
738       return dpad;
739     }
740   }
741
742   return NULL;
743 }
744
745
746 static gboolean
747 gst_rtp_ssrc_demux_src_event (GstPad * pad, GstObject * parent,
748     GstEvent * event)
749 {
750   GstRtpSsrcDemux *demux;
751   const GstStructure *s;
752
753   demux = GST_RTP_SSRC_DEMUX (parent);
754
755   switch (GST_EVENT_TYPE (event)) {
756     case GST_EVENT_CUSTOM_UPSTREAM:
757     case GST_EVENT_CUSTOM_BOTH:
758     case GST_EVENT_CUSTOM_BOTH_OOB:
759       s = gst_event_get_structure (event);
760       if (s && !gst_structure_has_field (s, "ssrc")) {
761         GstRtpSsrcDemuxPad *dpad = find_demux_pad_for_pad (demux, pad);
762
763         if (dpad) {
764           GstStructure *ws;
765
766           event = gst_event_make_writable (event);
767           ws = gst_event_writable_structure (event);
768           gst_structure_set (ws, "ssrc", G_TYPE_UINT, dpad->ssrc, NULL);
769         }
770       }
771       break;
772     default:
773       break;
774   }
775
776   return gst_pad_event_default (pad, parent, event);
777 }
778
779 static GstIterator *
780 gst_rtp_ssrc_demux_iterate_internal_links_src (GstPad * pad, GstObject * parent)
781 {
782   GstRtpSsrcDemux *demux;
783   GstPad *otherpad = NULL;
784   GstIterator *it = NULL;
785   GSList *current;
786
787   demux = GST_RTP_SSRC_DEMUX (parent);
788
789   INTERNAL_STREAM_LOCK (demux);
790   for (current = demux->srcpads; current; current = g_slist_next (current)) {
791     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) current->data;
792
793     if (pad == dpad->rtp_pad) {
794       otherpad = demux->rtp_sink;
795       break;
796     } else if (pad == dpad->rtcp_pad) {
797       otherpad = demux->rtcp_sink;
798       break;
799     }
800   }
801   if (otherpad) {
802     GValue val = { 0, };
803
804     g_value_init (&val, GST_TYPE_PAD);
805     g_value_set_object (&val, otherpad);
806     it = gst_iterator_new_single (GST_TYPE_PAD, &val);
807     g_value_unset (&val);
808
809   }
810   INTERNAL_STREAM_UNLOCK (demux);
811
812   return it;
813 }
814
815 /* Should return 0 for elements to be included */
816 static gint
817 src_pad_compare_func (gconstpointer a, gconstpointer b)
818 {
819   GstPad *pad = GST_PAD (g_value_get_object (a));
820   const gchar *prefix = g_value_get_string (b);
821   gint res;
822
823   /* 0 means equal means we accept the pad, accepted if there is a name
824    * and it starts with the prefix */
825   GST_OBJECT_LOCK (pad);
826   res = !GST_PAD_NAME (pad) || !g_str_has_prefix (GST_PAD_NAME (pad), prefix);
827   GST_OBJECT_UNLOCK (pad);
828
829   return res;
830 }
831
832 static GstIterator *
833 gst_rtp_ssrc_demux_iterate_internal_links_sink (GstPad * pad,
834     GstObject * parent)
835 {
836   GstRtpSsrcDemux *demux;
837   GstIterator *it = NULL;
838   GValue gval = { 0, };
839
840   demux = GST_RTP_SSRC_DEMUX (parent);
841
842   g_value_init (&gval, G_TYPE_STRING);
843   if (pad == demux->rtp_sink)
844     g_value_set_static_string (&gval, "src_");
845   else if (pad == demux->rtcp_sink)
846     g_value_set_static_string (&gval, "rtcp_src_");
847   else
848     g_assert_not_reached ();
849
850   it = gst_element_iterate_src_pads (GST_ELEMENT_CAST (demux));
851   it = gst_iterator_filter (it, src_pad_compare_func, &gval);
852
853   return it;
854 }
855
856
857 static gboolean
858 gst_rtp_ssrc_demux_src_query (GstPad * pad, GstObject * parent,
859     GstQuery * query)
860 {
861   GstRtpSsrcDemux *demux;
862   gboolean res = FALSE;
863
864   demux = GST_RTP_SSRC_DEMUX (parent);
865
866   switch (GST_QUERY_TYPE (query)) {
867     case GST_QUERY_LATENCY:
868     {
869
870       if ((res = gst_pad_peer_query (demux->rtp_sink, query))) {
871         gboolean live;
872         GstClockTime min_latency, max_latency;
873         GstRtpSsrcDemuxPad *demuxpad;
874
875         demuxpad = gst_pad_get_element_private (pad);
876
877         gst_query_parse_latency (query, &live, &min_latency, &max_latency);
878
879         GST_DEBUG_OBJECT (demux, "peer min latency %" GST_TIME_FORMAT,
880             GST_TIME_ARGS (min_latency));
881
882         GST_DEBUG_OBJECT (demux, "latency for SSRC %08x", demuxpad->ssrc);
883
884         gst_query_set_latency (query, live, min_latency, max_latency);
885       }
886       break;
887     }
888     default:
889       res = gst_pad_query_default (pad, parent, query);
890       break;
891   }
892
893   return res;
894 }
895
896 static GstStateChangeReturn
897 gst_rtp_ssrc_demux_change_state (GstElement * element,
898     GstStateChange transition)
899 {
900   GstStateChangeReturn ret;
901   GstRtpSsrcDemux *demux;
902
903   demux = GST_RTP_SSRC_DEMUX (element);
904
905   switch (transition) {
906     case GST_STATE_CHANGE_NULL_TO_READY:
907     case GST_STATE_CHANGE_READY_TO_PAUSED:
908     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
909     default:
910       break;
911   }
912
913   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
914
915   switch (transition) {
916     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
917       break;
918     case GST_STATE_CHANGE_PAUSED_TO_READY:
919       gst_rtp_ssrc_demux_reset (demux);
920       break;
921     case GST_STATE_CHANGE_READY_TO_NULL:
922     default:
923       break;
924   }
925   return ret;
926 }