Merge branch 'master' into 0.11
[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., 59 Temple Place - Suite 330,
19  * Boston, MA 02111-1307, USA.
20  */
21
22 /**
23  * SECTION:element-gstrtpssrcdemux
24  *
25  * gstrtpssrcdemux 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 udpsrc caps="application/x-rtp" ! gstrtpssrcdemux ! 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  * Last reviewed on 2007-05-28 (0.10.5)
41  */
42
43 #ifdef HAVE_CONFIG_H
44 #include "config.h"
45 #endif
46
47 #include <string.h>
48 #include <gst/rtp/gstrtpbuffer.h>
49 #include <gst/rtp/gstrtcpbuffer.h>
50
51 #include "gstrtpbin-marshal.h"
52 #include "gstrtpssrcdemux.h"
53
54 GST_DEBUG_CATEGORY_STATIC (gst_rtp_ssrc_demux_debug);
55 #define GST_CAT_DEFAULT gst_rtp_ssrc_demux_debug
56
57 /* generic templates */
58 static GstStaticPadTemplate rtp_ssrc_demux_sink_template =
59 GST_STATIC_PAD_TEMPLATE ("sink",
60     GST_PAD_SINK,
61     GST_PAD_ALWAYS,
62     GST_STATIC_CAPS ("application/x-rtp")
63     );
64
65 static GstStaticPadTemplate rtp_ssrc_demux_rtcp_sink_template =
66 GST_STATIC_PAD_TEMPLATE ("rtcp_sink",
67     GST_PAD_SINK,
68     GST_PAD_ALWAYS,
69     GST_STATIC_CAPS ("application/x-rtcp")
70     );
71
72 static GstStaticPadTemplate rtp_ssrc_demux_src_template =
73 GST_STATIC_PAD_TEMPLATE ("src_%d",
74     GST_PAD_SRC,
75     GST_PAD_SOMETIMES,
76     GST_STATIC_CAPS ("application/x-rtp")
77     );
78
79 static GstStaticPadTemplate rtp_ssrc_demux_rtcp_src_template =
80 GST_STATIC_PAD_TEMPLATE ("rtcp_src_%d",
81     GST_PAD_SRC,
82     GST_PAD_SOMETIMES,
83     GST_STATIC_CAPS ("application/x-rtcp")
84     );
85
86 #define GST_PAD_LOCK(obj)   (g_mutex_lock ((obj)->padlock))
87 #define GST_PAD_UNLOCK(obj) (g_mutex_unlock ((obj)->padlock))
88
89 /* signals */
90 enum
91 {
92   SIGNAL_NEW_SSRC_PAD,
93   SIGNAL_REMOVED_SSRC_PAD,
94   SIGNAL_CLEAR_SSRC,
95   LAST_SIGNAL
96 };
97
98 GST_BOILERPLATE (GstRtpSsrcDemux, gst_rtp_ssrc_demux, GstElement,
99     GST_TYPE_ELEMENT);
100
101
102 /* GObject vmethods */
103 static void gst_rtp_ssrc_demux_dispose (GObject * object);
104 static void gst_rtp_ssrc_demux_finalize (GObject * object);
105
106 /* GstElement vmethods */
107 static GstStateChangeReturn gst_rtp_ssrc_demux_change_state (GstElement *
108     element, GstStateChange transition);
109
110 static void gst_rtp_ssrc_demux_clear_ssrc (GstRtpSsrcDemux * demux,
111     guint32 ssrc);
112
113 /* sinkpad stuff */
114 static GstFlowReturn gst_rtp_ssrc_demux_chain (GstPad * pad, GstBuffer * buf);
115 static gboolean gst_rtp_ssrc_demux_sink_event (GstPad * pad, GstEvent * event);
116
117 static GstFlowReturn gst_rtp_ssrc_demux_rtcp_chain (GstPad * pad,
118     GstBuffer * buf);
119 static gboolean gst_rtp_ssrc_demux_rtcp_sink_event (GstPad * pad,
120     GstEvent * event);
121
122 /* srcpad stuff */
123 static gboolean gst_rtp_ssrc_demux_src_event (GstPad * pad, GstEvent * event);
124 static GstIterator *gst_rtp_ssrc_demux_iterate_internal_links (GstPad * pad);
125 static gboolean gst_rtp_ssrc_demux_src_query (GstPad * pad, GstQuery * query);
126
127 static guint gst_rtp_ssrc_demux_signals[LAST_SIGNAL] = { 0 };
128
129 /*
130  * Item for storing GstPad <-> SSRC pairs.
131  */
132 struct _GstRtpSsrcDemuxPad
133 {
134   guint32 ssrc;
135   GstPad *rtp_pad;
136   GstCaps *caps;
137   GstPad *rtcp_pad;
138 };
139
140 /* find a src pad for a given SSRC, returns NULL if the SSRC was not found
141  */
142 static GstRtpSsrcDemuxPad *
143 find_demux_pad_for_ssrc (GstRtpSsrcDemux * demux, guint32 ssrc)
144 {
145   GSList *walk;
146
147   for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
148     GstRtpSsrcDemuxPad *pad = (GstRtpSsrcDemuxPad *) walk->data;
149
150     if (pad->ssrc == ssrc)
151       return pad;
152   }
153   return NULL;
154 }
155
156 /* with PAD_LOCK */
157 static GstRtpSsrcDemuxPad *
158 create_demux_pad_for_ssrc (GstRtpSsrcDemux * demux, guint32 ssrc,
159     GstClockTime timestamp)
160 {
161   GstPad *rtp_pad, *rtcp_pad;
162   GstElementClass *klass;
163   GstPadTemplate *templ;
164   gchar *padname;
165   GstRtpSsrcDemuxPad *demuxpad;
166
167   GST_DEBUG_OBJECT (demux, "creating pad for SSRC %08x", ssrc);
168
169   klass = GST_ELEMENT_GET_CLASS (demux);
170   templ = gst_element_class_get_pad_template (klass, "src_%d");
171   padname = g_strdup_printf ("src_%d", ssrc);
172   rtp_pad = gst_pad_new_from_template (templ, padname);
173   g_free (padname);
174
175   templ = gst_element_class_get_pad_template (klass, "rtcp_src_%d");
176   padname = g_strdup_printf ("rtcp_src_%d", ssrc);
177   rtcp_pad = gst_pad_new_from_template (templ, padname);
178   g_free (padname);
179
180   /* we use the first timestamp received to calculate the difference between
181    * timestamps on all streams */
182   GST_DEBUG_OBJECT (demux, "SSRC %08x, first timestamp %" GST_TIME_FORMAT,
183       ssrc, GST_TIME_ARGS (timestamp));
184
185   /* wrap in structure and add to list */
186   demuxpad = g_new0 (GstRtpSsrcDemuxPad, 1);
187   demuxpad->ssrc = ssrc;
188   demuxpad->rtp_pad = rtp_pad;
189   demuxpad->rtcp_pad = rtcp_pad;
190
191   GST_DEBUG_OBJECT (demux, "first timestamp %" GST_TIME_FORMAT,
192       GST_TIME_ARGS (timestamp));
193
194   gst_pad_set_element_private (rtp_pad, demuxpad);
195   gst_pad_set_element_private (rtcp_pad, demuxpad);
196
197   demux->srcpads = g_slist_prepend (demux->srcpads, demuxpad);
198
199   /* copy caps from input */
200   gst_pad_set_caps (rtp_pad, GST_PAD_CAPS (demux->rtp_sink));
201   gst_pad_use_fixed_caps (rtp_pad);
202   gst_pad_set_caps (rtcp_pad, GST_PAD_CAPS (demux->rtcp_sink));
203   gst_pad_use_fixed_caps (rtcp_pad);
204
205   gst_pad_set_event_function (rtp_pad, gst_rtp_ssrc_demux_src_event);
206   gst_pad_set_query_function (rtp_pad, gst_rtp_ssrc_demux_src_query);
207   gst_pad_set_iterate_internal_links_function (rtp_pad,
208       gst_rtp_ssrc_demux_iterate_internal_links);
209   gst_pad_set_active (rtp_pad, TRUE);
210
211   gst_pad_set_event_function (rtcp_pad, gst_rtp_ssrc_demux_src_event);
212   gst_pad_set_iterate_internal_links_function (rtcp_pad,
213       gst_rtp_ssrc_demux_iterate_internal_links);
214   gst_pad_set_active (rtcp_pad, TRUE);
215
216   gst_element_add_pad (GST_ELEMENT_CAST (demux), rtp_pad);
217   gst_element_add_pad (GST_ELEMENT_CAST (demux), rtcp_pad);
218
219   g_signal_emit (G_OBJECT (demux),
220       gst_rtp_ssrc_demux_signals[SIGNAL_NEW_SSRC_PAD], 0, ssrc, rtp_pad);
221
222   return demuxpad;
223 }
224
225 static void
226 gst_rtp_ssrc_demux_base_init (gpointer g_class)
227 {
228   GstElementClass *gstelement_klass = GST_ELEMENT_CLASS (g_class);
229
230   gst_element_class_add_pad_template (gstelement_klass,
231       gst_static_pad_template_get (&rtp_ssrc_demux_sink_template));
232   gst_element_class_add_pad_template (gstelement_klass,
233       gst_static_pad_template_get (&rtp_ssrc_demux_rtcp_sink_template));
234   gst_element_class_add_pad_template (gstelement_klass,
235       gst_static_pad_template_get (&rtp_ssrc_demux_src_template));
236   gst_element_class_add_pad_template (gstelement_klass,
237       gst_static_pad_template_get (&rtp_ssrc_demux_rtcp_src_template));
238
239   gst_element_class_set_details_simple (gstelement_klass, "RTP SSRC Demux",
240       "Demux/Network/RTP",
241       "Splits RTP streams based on the SSRC",
242       "Wim Taymans <wim.taymans@gmail.com>");
243 }
244
245 static void
246 gst_rtp_ssrc_demux_class_init (GstRtpSsrcDemuxClass * klass)
247 {
248   GObjectClass *gobject_klass;
249   GstElementClass *gstelement_klass;
250   GstRtpSsrcDemuxClass *gstrtpssrcdemux_klass;
251
252   gobject_klass = (GObjectClass *) klass;
253   gstelement_klass = (GstElementClass *) klass;
254   gstrtpssrcdemux_klass = (GstRtpSsrcDemuxClass *) klass;
255
256   gobject_klass->dispose = gst_rtp_ssrc_demux_dispose;
257   gobject_klass->finalize = gst_rtp_ssrc_demux_finalize;
258
259   /**
260    * GstRtpSsrcDemux::new-ssrc-pad:
261    * @demux: the object which received the signal
262    * @ssrc: the SSRC of the pad
263    * @pad: the new pad.
264    *
265    * Emited when a new SSRC pad has been created.
266    */
267   gst_rtp_ssrc_demux_signals[SIGNAL_NEW_SSRC_PAD] =
268       g_signal_new ("new-ssrc-pad",
269       G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
270       G_STRUCT_OFFSET (GstRtpSsrcDemuxClass, new_ssrc_pad),
271       NULL, NULL, gst_rtp_bin_marshal_VOID__UINT_OBJECT,
272       G_TYPE_NONE, 2, G_TYPE_UINT, GST_TYPE_PAD);
273
274   /**
275    * GstRtpSsrcDemux::removed-ssrc-pad:
276    * @demux: the object which received the signal
277    * @ssrc: the SSRC of the pad
278    * @pad: the removed pad.
279    *
280    * Emited when a SSRC pad has been removed.
281    */
282   gst_rtp_ssrc_demux_signals[SIGNAL_REMOVED_SSRC_PAD] =
283       g_signal_new ("removed-ssrc-pad",
284       G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
285       G_STRUCT_OFFSET (GstRtpSsrcDemuxClass, removed_ssrc_pad),
286       NULL, NULL, gst_rtp_bin_marshal_VOID__UINT_OBJECT,
287       G_TYPE_NONE, 2, G_TYPE_UINT, GST_TYPE_PAD);
288
289   /**
290    * GstRtpSsrcDemux::clear-ssrc:
291    * @demux: the object which received the signal
292    * @ssrc: the SSRC of the pad
293    *
294    * Action signal to remove the pad for SSRC.
295    */
296   gst_rtp_ssrc_demux_signals[SIGNAL_CLEAR_SSRC] =
297       g_signal_new ("clear-ssrc",
298       G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
299       G_STRUCT_OFFSET (GstRtpSsrcDemuxClass, clear_ssrc),
300       NULL, NULL, gst_rtp_bin_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
301
302   gstelement_klass->change_state =
303       GST_DEBUG_FUNCPTR (gst_rtp_ssrc_demux_change_state);
304   gstrtpssrcdemux_klass->clear_ssrc =
305       GST_DEBUG_FUNCPTR (gst_rtp_ssrc_demux_clear_ssrc);
306
307   GST_DEBUG_CATEGORY_INIT (gst_rtp_ssrc_demux_debug,
308       "rtpssrcdemux", 0, "RTP SSRC demuxer");
309 }
310
311 static void
312 gst_rtp_ssrc_demux_init (GstRtpSsrcDemux * demux,
313     GstRtpSsrcDemuxClass * g_class)
314 {
315   GstElementClass *klass = GST_ELEMENT_GET_CLASS (demux);
316
317   demux->rtp_sink =
318       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
319           "sink"), "sink");
320   gst_pad_set_chain_function (demux->rtp_sink, gst_rtp_ssrc_demux_chain);
321   gst_pad_set_event_function (demux->rtp_sink, gst_rtp_ssrc_demux_sink_event);
322   gst_element_add_pad (GST_ELEMENT_CAST (demux), demux->rtp_sink);
323
324   demux->rtcp_sink =
325       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
326           "rtcp_sink"), "rtcp_sink");
327   gst_pad_set_chain_function (demux->rtcp_sink, gst_rtp_ssrc_demux_rtcp_chain);
328   gst_pad_set_event_function (demux->rtcp_sink,
329       gst_rtp_ssrc_demux_rtcp_sink_event);
330   gst_element_add_pad (GST_ELEMENT_CAST (demux), demux->rtcp_sink);
331
332   demux->padlock = g_mutex_new ();
333
334   gst_segment_init (&demux->segment, GST_FORMAT_UNDEFINED);
335 }
336
337 static void
338 gst_rtp_ssrc_demux_reset (GstRtpSsrcDemux * demux)
339 {
340   GSList *walk;
341
342   for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
343     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) walk->data;
344
345     gst_pad_set_active (dpad->rtp_pad, FALSE);
346     gst_pad_set_active (dpad->rtcp_pad, FALSE);
347
348     gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtp_pad);
349     gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtcp_pad);
350     g_free (dpad);
351   }
352   g_slist_free (demux->srcpads);
353   demux->srcpads = NULL;
354 }
355
356 static void
357 gst_rtp_ssrc_demux_dispose (GObject * object)
358 {
359   GstRtpSsrcDemux *demux;
360
361   demux = GST_RTP_SSRC_DEMUX (object);
362
363   gst_rtp_ssrc_demux_reset (demux);
364
365   G_OBJECT_CLASS (parent_class)->dispose (object);
366 }
367
368 static void
369 gst_rtp_ssrc_demux_finalize (GObject * object)
370 {
371   GstRtpSsrcDemux *demux;
372
373   demux = GST_RTP_SSRC_DEMUX (object);
374   g_mutex_free (demux->padlock);
375
376   G_OBJECT_CLASS (parent_class)->finalize (object);
377 }
378
379 static void
380 gst_rtp_ssrc_demux_clear_ssrc (GstRtpSsrcDemux * demux, guint32 ssrc)
381 {
382   GstRtpSsrcDemuxPad *dpad;
383
384   GST_PAD_LOCK (demux);
385   dpad = find_demux_pad_for_ssrc (demux, ssrc);
386   if (dpad == NULL)
387     goto unknown_pad;
388
389   GST_DEBUG_OBJECT (demux, "clearing pad for SSRC %08x", ssrc);
390
391   demux->srcpads = g_slist_remove (demux->srcpads, dpad);
392   GST_PAD_UNLOCK (demux);
393
394   gst_pad_set_active (dpad->rtp_pad, FALSE);
395   gst_pad_set_active (dpad->rtcp_pad, FALSE);
396
397   g_signal_emit (G_OBJECT (demux),
398       gst_rtp_ssrc_demux_signals[SIGNAL_REMOVED_SSRC_PAD], 0, ssrc,
399       dpad->rtp_pad);
400
401   gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtp_pad);
402   gst_element_remove_pad (GST_ELEMENT_CAST (demux), dpad->rtcp_pad);
403
404   g_free (dpad);
405
406   return;
407
408   /* ERRORS */
409 unknown_pad:
410   {
411     g_warning ("unknown SSRC %08x", ssrc);
412     return;
413   }
414 }
415
416 static gboolean
417 gst_rtp_ssrc_demux_sink_event (GstPad * pad, GstEvent * event)
418 {
419   GstRtpSsrcDemux *demux;
420   gboolean res = FALSE;
421
422   demux = GST_RTP_SSRC_DEMUX (gst_pad_get_parent (pad));
423
424   switch (GST_EVENT_TYPE (event)) {
425     case GST_EVENT_FLUSH_STOP:
426       gst_segment_init (&demux->segment, GST_FORMAT_UNDEFINED);
427     case GST_EVENT_NEWSEGMENT:
428     default:
429     {
430       GSList *walk;
431       GSList *pads = NULL;
432
433       res = TRUE;
434       /* need local snapshot of pads;
435        * should not push downstream while holding lock as that might deadlock
436        * with stuff traveling upstream tyring to get this lock while holding
437        * other (stream)lock */
438       GST_PAD_LOCK (demux);
439       for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
440         GstRtpSsrcDemuxPad *pad = (GstRtpSsrcDemuxPad *) walk->data;
441
442         pads = g_slist_prepend (pads, gst_object_ref (pad->rtp_pad));
443       }
444       GST_PAD_UNLOCK (demux);
445       for (walk = pads; walk; walk = g_slist_next (walk)) {
446         GstPad *pad = (GstPad *) walk->data;
447
448         gst_event_ref (event);
449         res &= gst_pad_push_event (pad, event);
450         gst_object_unref (pad);
451       }
452       g_slist_free (pads);
453       gst_event_unref (event);
454       break;
455     }
456   }
457
458   gst_object_unref (demux);
459   return res;
460 }
461
462 static gboolean
463 gst_rtp_ssrc_demux_rtcp_sink_event (GstPad * pad, GstEvent * event)
464 {
465   GstRtpSsrcDemux *demux;
466   gboolean res = FALSE;
467
468   demux = GST_RTP_SSRC_DEMUX (gst_pad_get_parent (pad));
469
470   switch (GST_EVENT_TYPE (event)) {
471     case GST_EVENT_NEWSEGMENT:
472     default:
473     {
474       GSList *walk;
475       GSList *pads = NULL;
476
477       res = TRUE;
478       GST_PAD_LOCK (demux);
479       for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
480         GstRtpSsrcDemuxPad *pad = (GstRtpSsrcDemuxPad *) walk->data;
481
482         pads = g_slist_prepend (pads, gst_object_ref (pad->rtcp_pad));
483       }
484       GST_PAD_UNLOCK (demux);
485       for (walk = pads; walk; walk = g_slist_next (walk)) {
486         GstPad *pad = (GstPad *) walk->data;
487
488         gst_event_ref (event);
489         res &= gst_pad_push_event (pad, event);
490         gst_object_unref (pad);
491       }
492       g_slist_free (pads);
493       gst_event_unref (event);
494       break;
495     }
496   }
497   gst_object_unref (demux);
498   return res;
499 }
500
501 static GstFlowReturn
502 gst_rtp_ssrc_demux_chain (GstPad * pad, GstBuffer * buf)
503 {
504   GstFlowReturn ret;
505   GstRtpSsrcDemux *demux;
506   guint32 ssrc;
507   GstRtpSsrcDemuxPad *dpad;
508
509   demux = GST_RTP_SSRC_DEMUX (GST_OBJECT_PARENT (pad));
510
511   if (!gst_rtp_buffer_validate (buf))
512     goto invalid_payload;
513
514   ssrc = gst_rtp_buffer_get_ssrc (buf);
515
516   GST_DEBUG_OBJECT (demux, "received buffer of SSRC %08x", ssrc);
517
518   GST_PAD_LOCK (demux);
519   dpad = find_demux_pad_for_ssrc (demux, ssrc);
520   if (dpad == NULL) {
521     if (!(dpad =
522             create_demux_pad_for_ssrc (demux, ssrc,
523                 GST_BUFFER_TIMESTAMP (buf))))
524       goto create_failed;
525   }
526   GST_PAD_UNLOCK (demux);
527
528   /* push to srcpad */
529   ret = gst_pad_push (dpad->rtp_pad, buf);
530
531   return ret;
532
533   /* ERRORS */
534 invalid_payload:
535   {
536     /* this is fatal and should be filtered earlier */
537     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
538         ("Dropping invalid RTP payload"));
539     gst_buffer_unref (buf);
540     return GST_FLOW_ERROR;
541   }
542 create_failed:
543   {
544     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
545         ("Could not create new pad"));
546     GST_PAD_UNLOCK (demux);
547     gst_buffer_unref (buf);
548     return GST_FLOW_ERROR;
549   }
550 }
551
552 static GstFlowReturn
553 gst_rtp_ssrc_demux_rtcp_chain (GstPad * pad, GstBuffer * buf)
554 {
555   GstFlowReturn ret;
556   GstRtpSsrcDemux *demux;
557   guint32 ssrc;
558   GstRtpSsrcDemuxPad *dpad;
559   GstRTCPPacket packet;
560
561   demux = GST_RTP_SSRC_DEMUX (GST_OBJECT_PARENT (pad));
562
563   if (!gst_rtcp_buffer_validate (buf))
564     goto invalid_rtcp;
565
566   if (!gst_rtcp_buffer_get_first_packet (buf, &packet))
567     goto invalid_rtcp;
568
569   /* first packet must be SR or RR or else the validate would have failed */
570   switch (gst_rtcp_packet_get_type (&packet)) {
571     case GST_RTCP_TYPE_SR:
572       /* get the ssrc so that we can route it to the right source pad */
573       gst_rtcp_packet_sr_get_sender_info (&packet, &ssrc, NULL, NULL, NULL,
574           NULL);
575       break;
576     default:
577       goto unexpected_rtcp;
578   }
579
580   GST_DEBUG_OBJECT (demux, "received RTCP of SSRC %08x", ssrc);
581
582   GST_PAD_LOCK (demux);
583   dpad = find_demux_pad_for_ssrc (demux, ssrc);
584   if (dpad == NULL) {
585     GST_DEBUG_OBJECT (demux, "creating pad for SSRC %08x", ssrc);
586     if (!(dpad = create_demux_pad_for_ssrc (demux, ssrc, -1)))
587       goto create_failed;
588   }
589   GST_PAD_UNLOCK (demux);
590
591   /* push to srcpad */
592   ret = gst_pad_push (dpad->rtcp_pad, buf);
593
594   return ret;
595
596   /* ERRORS */
597 invalid_rtcp:
598   {
599     /* this is fatal and should be filtered earlier */
600     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
601         ("Dropping invalid RTCP packet"));
602     gst_buffer_unref (buf);
603     return GST_FLOW_ERROR;
604   }
605 unexpected_rtcp:
606   {
607     GST_DEBUG_OBJECT (demux, "dropping unexpected RTCP packet");
608     gst_buffer_unref (buf);
609     return GST_FLOW_OK;
610   }
611 create_failed:
612   {
613     GST_ELEMENT_ERROR (demux, STREAM, DECODE, (NULL),
614         ("Could not create new pad"));
615     GST_PAD_UNLOCK (demux);
616     gst_buffer_unref (buf);
617     return GST_FLOW_ERROR;
618   }
619 }
620
621 static gboolean
622 gst_rtp_ssrc_demux_src_event (GstPad * pad, GstEvent * event)
623 {
624   GstRtpSsrcDemux *demux;
625   const GstStructure *s;
626
627   demux = GST_RTP_SSRC_DEMUX (gst_pad_get_parent (pad));
628
629   switch (GST_EVENT_TYPE (event)) {
630     case GST_EVENT_CUSTOM_UPSTREAM:
631     case GST_EVENT_CUSTOM_BOTH:
632     case GST_EVENT_CUSTOM_BOTH_OOB:
633       s = gst_event_get_structure (event);
634       if (s && !gst_structure_has_field (s, "ssrc")) {
635         GSList *walk;
636
637         for (walk = demux->srcpads; walk; walk = g_slist_next (walk)) {
638           GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) walk->data;
639
640           if (dpad->rtp_pad == pad || dpad->rtcp_pad == pad) {
641             event =
642                 GST_EVENT_CAST (gst_mini_object_make_writable
643                 (GST_MINI_OBJECT_CAST (event)));
644             gst_structure_set (event->structure, "ssrc", G_TYPE_UINT,
645                 dpad->ssrc, NULL);
646             break;
647           }
648         }
649       }
650       break;
651     default:
652       break;
653   }
654
655   gst_object_unref (demux);
656
657   return gst_pad_event_default (pad, event);
658 }
659
660 static GstIterator *
661 gst_rtp_ssrc_demux_iterate_internal_links (GstPad * pad)
662 {
663   GstRtpSsrcDemux *demux;
664   GstPad *otherpad = NULL;
665   GstIterator *it;
666   GSList *current;
667
668   demux = GST_RTP_SSRC_DEMUX (gst_pad_get_parent (pad));
669
670   GST_PAD_LOCK (demux);
671   for (current = demux->srcpads; current; current = g_slist_next (current)) {
672     GstRtpSsrcDemuxPad *dpad = (GstRtpSsrcDemuxPad *) current->data;
673
674     if (pad == demux->rtp_sink) {
675       otherpad = dpad->rtp_pad;
676       break;
677     } else if (pad == demux->rtcp_sink) {
678       otherpad = dpad->rtcp_pad;
679     } else if (pad == dpad->rtp_pad) {
680       otherpad = demux->rtp_sink;
681       break;
682     } else if (pad == dpad->rtcp_pad) {
683       otherpad = demux->rtcp_sink;
684       break;
685     }
686   }
687   it = gst_iterator_new_single (GST_TYPE_PAD, otherpad,
688       (GstCopyFunction) gst_object_ref, (GFreeFunc) gst_object_unref);
689   GST_PAD_UNLOCK (demux);
690
691   gst_object_unref (demux);
692   return it;
693 }
694
695 static gboolean
696 gst_rtp_ssrc_demux_src_query (GstPad * pad, GstQuery * query)
697 {
698   GstRtpSsrcDemux *demux;
699   gboolean res = FALSE;
700
701   demux = GST_RTP_SSRC_DEMUX (gst_pad_get_parent (pad));
702
703   switch (GST_QUERY_TYPE (query)) {
704     case GST_QUERY_LATENCY:
705     {
706
707       if ((res = gst_pad_peer_query (demux->rtp_sink, query))) {
708         gboolean live;
709         GstClockTime min_latency, max_latency;
710         GstRtpSsrcDemuxPad *demuxpad;
711
712         demuxpad = gst_pad_get_element_private (pad);
713
714         gst_query_parse_latency (query, &live, &min_latency, &max_latency);
715
716         GST_DEBUG_OBJECT (demux, "peer min latency %" GST_TIME_FORMAT,
717             GST_TIME_ARGS (min_latency));
718
719         GST_DEBUG_OBJECT (demux, "latency for SSRC %08x", demuxpad->ssrc);
720
721         gst_query_set_latency (query, live, min_latency, max_latency);
722       }
723       break;
724     }
725     default:
726       res = gst_pad_query_default (pad, query);
727       break;
728   }
729   gst_object_unref (demux);
730
731   return res;
732 }
733
734 static GstStateChangeReturn
735 gst_rtp_ssrc_demux_change_state (GstElement * element,
736     GstStateChange transition)
737 {
738   GstStateChangeReturn ret;
739   GstRtpSsrcDemux *demux;
740
741   demux = GST_RTP_SSRC_DEMUX (element);
742
743   switch (transition) {
744     case GST_STATE_CHANGE_NULL_TO_READY:
745     case GST_STATE_CHANGE_READY_TO_PAUSED:
746     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
747     default:
748       break;
749   }
750
751   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
752
753   switch (transition) {
754     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
755       break;
756     case GST_STATE_CHANGE_PAUSED_TO_READY:
757       gst_rtp_ssrc_demux_reset (demux);
758       break;
759     case GST_STATE_CHANGE_READY_TO_NULL:
760     default:
761       break;
762   }
763   return ret;
764 }