gst/rtpmanager/gstrtpbin.c

   1 /* GStreamer
   2  * Copyright (C) <2007> Wim Taymans <wim.taymans@gmail.com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19
  20 /**
  21  * SECTION:element-rtpbin
  22  * @see_also: rtpjitterbuffer, rtpsession, rtpptdemux, rtpssrcdemux
  23  *
  24  * RTP bin combines the functions of #GstRtpSession, #GstRtpSsrcDemux,
  25  * #GstRtpJitterBuffer and #GstRtpPtDemux in one element. It allows for multiple
  26  * RTP sessions that will be synchronized together using RTCP SR packets.
  27  *
  28  * #GstRtpBin is configured with a number of request pads that define the
  29  * functionality that is activated, similar to the #GstRtpSession element.
  30  *
  31  * To use #GstRtpBin as an RTP receiver, request a recv_rtp_sink_\%u pad. The session
  32  * number must be specified in the pad name.
  33  * Data received on the recv_rtp_sink_\%u pad will be processed in the #GstRtpSession
  34  * manager and after being validated forwarded on #GstRtpSsrcDemux element. Each
  35  * RTP stream is demuxed based on the SSRC and send to a #GstRtpJitterBuffer. After
  36  * the packets are released from the jitterbuffer, they will be forwarded to a
  37  * #GstRtpPtDemux element. The #GstRtpPtDemux element will demux the packets based
  38  * on the payload type and will create a unique pad recv_rtp_src_\%u_\%u_\%u on
  39  * rtpbin with the session number, SSRC and payload type respectively as the pad
  40  * name.
  41  *
  42  * To also use #GstRtpBin as an RTCP receiver, request a recv_rtcp_sink_\%u pad. The
  43  * session number must be specified in the pad name.
  44  *
  45  * If you want the session manager to generate and send RTCP packets, request
  46  * the send_rtcp_src_\%u pad with the session number in the pad name. Packet pushed
  47  * on this pad contain SR/RR RTCP reports that should be sent to all participants
  48  * in the session.
  49  *
  50  * To use #GstRtpBin as a sender, request a send_rtp_sink_\%u pad, which will
  51  * automatically create a send_rtp_src_\%u pad. If the session number is not provided,
  52  * the pad from the lowest available session will be returned. The session manager will modify the
  53  * SSRC in the RTP packets to its own SSRC and wil forward the packets on the
  54  * send_rtp_src_\%u pad after updating its internal state.
  55  *
  56  * The session manager needs the clock-rate of the payload types it is handling
  57  * and will signal the #GstRtpSession::request-pt-map signal when it needs such a
  58  * mapping. One can clear the cached values with the #GstRtpSession::clear-pt-map
  59  * signal.
  60  *
  61  * Access to the internal statistics of rtpbin is provided with the
  62  * get-internal-session property. This action signal gives access to the
  63  * RTPSession object which further provides action signals to retrieve the
  64  * internal source and other sources.
  65  *
  66  * #GstRtpBin also has signals (#GstRtpBin::request-rtp-encoder,
  67  * #GstRtpBin::request-rtp-decoder, #GstRtpBin::request-rtcp-encoder and
  68  * #GstRtpBin::request-rtp-decoder) to dynamically request for RTP and RTCP encoders
  69  * and decoders in order to support SRTP. The encoders must provide the pads
  70  * rtp_sink_\%u and rtp_src_\%u for RTP and rtcp_sink_\%u and rtcp_src_\%u for
  71  * RTCP. The session number will be used in the pad name. The decoders must provide
  72  * rtp_sink and rtp_src for RTP and rtcp_sink and rtcp_src for RTCP. The decoders will
  73  * be placed before the #GstRtpSession element, thus they must support SSRC demuxing
  74  * internally.
  75  *
  76  * #GstRtpBin has signals (#GstRtpBin::request-aux-sender and
  77  * #GstRtpBin::request-aux-receiver to dynamically request an element that can be
  78  * used to create or merge additional RTP streams. AUX elements are needed to
  79  * implement FEC or retransmission (such as RFC 4588). An AUX sender must have one
  80  * sink_\%u pad that matches the sessionid in the signal and it should have 1 or
  81  * more src_\%u pads. For each src_%\u pad, a session will be made (if needed)
  82  * and the pad will be linked to the session send_rtp_sink pad. Each session will
  83  * then expose its source pad ad send_rtp_src_\%u on #GstRtpBin.
  84  * An AUX receiver has 1 src_\%u pad that much match the sessionid in the signal
  85  * and 1 or more sink_\%u pads. A session will be made for each sink_\%u pad
  86  * when the corresponding recv_rtp_sink_\%u pad is requested on #GstRtpBin.
  87  *
  88  * <refsect2>
  89  * <title>Example pipelines</title>
  90  * |[
  91  * gst-launch-1.0 udpsrc port=5000 caps="application/x-rtp, ..." ! .recv_rtp_sink_0 \
  92  *     rtpbin ! rtptheoradepay ! theoradec ! xvimagesink
  93  * ]| Receive RTP data from port 5000 and send to the session 0 in rtpbin.
  94  * |[
  95  * gst-launch-1.0 rtpbin name=rtpbin \
  96  *         v4l2src ! videoconvert ! ffenc_h263 ! rtph263ppay ! rtpbin.send_rtp_sink_0 \
  97  *                   rtpbin.send_rtp_src_0 ! udpsink port=5000                            \
  98  *                   rtpbin.send_rtcp_src_0 ! udpsink port=5001 sync=false async=false    \
  99  *                   udpsrc port=5005 ! rtpbin.recv_rtcp_sink_0                           \
 100  *         audiotestsrc ! amrnbenc ! rtpamrpay ! rtpbin.send_rtp_sink_1                   \
 101  *                   rtpbin.send_rtp_src_1 ! udpsink port=5002                            \
 102  *                   rtpbin.send_rtcp_src_1 ! udpsink port=5003 sync=false async=false    \
 103  *                   udpsrc port=5007 ! rtpbin.recv_rtcp_sink_1
 104  * ]| Encode and payload H263 video captured from a v4l2src. Encode and payload AMR
 105  * audio generated from audiotestsrc. The video is sent to session 0 in rtpbin
 106  * and the audio is sent to session 1. Video packets are sent on UDP port 5000
 107  * and audio packets on port 5002. The video RTCP packets for session 0 are sent
 108  * on port 5001 and the audio RTCP packets for session 0 are sent on port 5003.
 109  * RTCP packets for session 0 are received on port 5005 and RTCP for session 1
 110  * is received on port 5007. Since RTCP packets from the sender should be sent
 111  * as soon as possible and do not participate in preroll, sync=false and
 112  * async=false is configured on udpsink
 113  * |[
 114  * gst-launch-1.0 -v rtpbin name=rtpbin                                          \
 115  *     udpsrc caps="application/x-rtp,media=(string)video,clock-rate=(int)90000,encoding-name=(string)H263-1998" \
 116  *             port=5000 ! rtpbin.recv_rtp_sink_0                                \
 117  *         rtpbin. ! rtph263pdepay ! ffdec_h263 ! xvimagesink                    \
 118  *      udpsrc port=5001 ! rtpbin.recv_rtcp_sink_0                               \
 119  *      rtpbin.send_rtcp_src_0 ! udpsink port=5005 sync=false async=false        \
 120  *     udpsrc caps="application/x-rtp,media=(string)audio,clock-rate=(int)8000,encoding-name=(string)AMR,encoding-params=(string)1,octet-align=(string)1" \
 121  *             port=5002 ! rtpbin.recv_rtp_sink_1                                \
 122  *         rtpbin. ! rtpamrdepay ! amrnbdec ! alsasink                           \
 123  *      udpsrc port=5003 ! rtpbin.recv_rtcp_sink_1                               \
 124  *      rtpbin.send_rtcp_src_1 ! udpsink port=5007 sync=false async=false
 125  * ]| Receive H263 on port 5000, send it through rtpbin in session 0, depayload,
 126  * decode and display the video.
 127  * Receive AMR on port 5002, send it through rtpbin in session 1, depayload,
 128  * decode and play the audio.
 129  * Receive server RTCP packets for session 0 on port 5001 and RTCP packets for
 130  * session 1 on port 5003. These packets will be used for session management and
 131  * synchronisation.
 132  * Send RTCP reports for session 0 on port 5005 and RTCP reports for session 1
 133  * on port 5007.
 134  * </refsect2>
 135  */
 136
 137 #ifdef HAVE_CONFIG_H
 138 #include "config.h"
 139 #endif
 140 #include <stdio.h>
 141 #include <string.h>
 142
 143 #include <gst/rtp/gstrtpbuffer.h>
 144 #include <gst/rtp/gstrtcpbuffer.h>
 145
 146 #include "gstrtpbin.h"
 147 #include "rtpsession.h"
 148 #include "gstrtpsession.h"
 149 #include "gstrtpjitterbuffer.h"
 150
 151 #include <gst/glib-compat-private.h>
 152
 153 GST_DEBUG_CATEGORY_STATIC (gst_rtp_bin_debug);
 154 #define GST_CAT_DEFAULT gst_rtp_bin_debug
 155
 156 /* sink pads */
 157 static GstStaticPadTemplate rtpbin_recv_rtp_sink_template =
 158     GST_STATIC_PAD_TEMPLATE ("recv_rtp_sink_%u",
 159     GST_PAD_SINK,
 160     GST_PAD_REQUEST,
 161     GST_STATIC_CAPS ("application/x-rtp;application/x-srtp")
 162     );
 163
 164 static GstStaticPadTemplate rtpbin_recv_rtcp_sink_template =
 165     GST_STATIC_PAD_TEMPLATE ("recv_rtcp_sink_%u",
 166     GST_PAD_SINK,
 167     GST_PAD_REQUEST,
 168     GST_STATIC_CAPS ("application/x-rtcp;application/x-srtcp")
 169     );
 170
 171 static GstStaticPadTemplate rtpbin_send_rtp_sink_template =
 172 GST_STATIC_PAD_TEMPLATE ("send_rtp_sink_%u",
 173     GST_PAD_SINK,
 174     GST_PAD_REQUEST,
 175     GST_STATIC_CAPS ("application/x-rtp")
 176     );
 177
 178 /* src pads */
 179 static GstStaticPadTemplate rtpbin_recv_rtp_src_template =
 180 GST_STATIC_PAD_TEMPLATE ("recv_rtp_src_%u_%u_%u",
 181     GST_PAD_SRC,
 182     GST_PAD_SOMETIMES,
 183     GST_STATIC_CAPS ("application/x-rtp")
 184     );
 185
 186 static GstStaticPadTemplate rtpbin_send_rtcp_src_template =
 187     GST_STATIC_PAD_TEMPLATE ("send_rtcp_src_%u",
 188     GST_PAD_SRC,
 189     GST_PAD_REQUEST,
 190     GST_STATIC_CAPS ("application/x-rtcp;application/x-srtcp")
 191     );
 192
 193 static GstStaticPadTemplate rtpbin_send_rtp_src_template =
 194     GST_STATIC_PAD_TEMPLATE ("send_rtp_src_%u",
 195     GST_PAD_SRC,
 196     GST_PAD_SOMETIMES,
 197     GST_STATIC_CAPS ("application/x-rtp;application/x-srtp")
 198     );
 199
 200 #define GST_RTP_BIN_GET_PRIVATE(obj)  \
 201    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTP_BIN, GstRtpBinPrivate))
 202
 203 #define GST_RTP_BIN_LOCK(bin)   g_mutex_lock (&(bin)->priv->bin_lock)
 204 #define GST_RTP_BIN_UNLOCK(bin) g_mutex_unlock (&(bin)->priv->bin_lock)
 205
 206 /* lock to protect dynamic callbacks, like pad-added and new ssrc. */
 207 #define GST_RTP_BIN_DYN_LOCK(bin)    g_mutex_lock (&(bin)->priv->dyn_lock)
 208 #define GST_RTP_BIN_DYN_UNLOCK(bin)  g_mutex_unlock (&(bin)->priv->dyn_lock)
 209
 210 /* lock for shutdown */
 211 #define GST_RTP_BIN_SHUTDOWN_LOCK(bin,label)     \
 212 G_STMT_START {                                   \
 213   if (g_atomic_int_get (&bin->priv->shutdown))   \
 214     goto label;                                  \
 215   GST_RTP_BIN_DYN_LOCK (bin);                    \
 216   if (g_atomic_int_get (&bin->priv->shutdown)) { \
 217     GST_RTP_BIN_DYN_UNLOCK (bin);                \
 218     goto label;                                  \
 219   }                                              \
 220 } G_STMT_END
 221
 222 /* unlock for shutdown */
 223 #define GST_RTP_BIN_SHUTDOWN_UNLOCK(bin)         \
 224   GST_RTP_BIN_DYN_UNLOCK (bin);                  \
 225
 226 struct _GstRtpBinPrivate
 227 {
 228   GMutex bin_lock;
 229
 230   /* lock protecting dynamic adding/removing */
 231   GMutex dyn_lock;
 232
 233   /* if we are shutting down or not */
 234   gint shutdown;
 235
 236   gboolean autoremove;
 237
 238   /* UNIX (ntp) time of last SR sync used */
 239   guint64 last_unix;
 240
 241   /* list of extra elements */
 242   GList *elements;
 243 };
 244
 245 /* signals and args */
 246 enum
 247 {
 248   SIGNAL_REQUEST_PT_MAP,
 249   SIGNAL_PAYLOAD_TYPE_CHANGE,
 250   SIGNAL_CLEAR_PT_MAP,
 251   SIGNAL_RESET_SYNC,
 252   SIGNAL_GET_INTERNAL_SESSION,
 253
 254   SIGNAL_ON_NEW_SSRC,
 255   SIGNAL_ON_SSRC_COLLISION,
 256   SIGNAL_ON_SSRC_VALIDATED,
 257   SIGNAL_ON_SSRC_ACTIVE,
 258   SIGNAL_ON_SSRC_SDES,
 259   SIGNAL_ON_BYE_SSRC,
 260   SIGNAL_ON_BYE_TIMEOUT,
 261   SIGNAL_ON_TIMEOUT,
 262   SIGNAL_ON_SENDER_TIMEOUT,
 263   SIGNAL_ON_NPT_STOP,
 264
 265   SIGNAL_REQUEST_RTP_ENCODER,
 266   SIGNAL_REQUEST_RTP_DECODER,
 267   SIGNAL_REQUEST_RTCP_ENCODER,
 268   SIGNAL_REQUEST_RTCP_DECODER,
 269
 270   SIGNAL_NEW_JITTERBUFFER,
 271
 272   SIGNAL_REQUEST_AUX_SENDER,
 273   SIGNAL_REQUEST_AUX_RECEIVER,
 274
 275   LAST_SIGNAL
 276 };
 277
 278 #define DEFAULT_LATENCY_MS           200
 279 #define DEFAULT_DROP_ON_LATENCY      FALSE
 280 #define DEFAULT_SDES                 NULL
 281 #define DEFAULT_DO_LOST              FALSE
 282 #define DEFAULT_IGNORE_PT            FALSE
 283 #define DEFAULT_NTP_SYNC             FALSE
 284 #define DEFAULT_AUTOREMOVE           FALSE
 285 #define DEFAULT_BUFFER_MODE          RTP_JITTER_BUFFER_MODE_SLAVE
 286 #define DEFAULT_USE_PIPELINE_CLOCK   FALSE
 287 #define DEFAULT_RTCP_SYNC            GST_RTP_BIN_RTCP_SYNC_ALWAYS
 288 #define DEFAULT_RTCP_SYNC_INTERVAL   0
 289 #define DEFAULT_DO_SYNC_EVENT        FALSE
 290 #define DEFAULT_DO_RETRANSMISSION    FALSE
 291
 292 enum
 293 {
 294   PROP_0,
 295   PROP_LATENCY,
 296   PROP_DROP_ON_LATENCY,
 297   PROP_SDES,
 298   PROP_DO_LOST,
 299   PROP_IGNORE_PT,
 300   PROP_NTP_SYNC,
 301   PROP_RTCP_SYNC,
 302   PROP_RTCP_SYNC_INTERVAL,
 303   PROP_AUTOREMOVE,
 304   PROP_BUFFER_MODE,
 305   PROP_USE_PIPELINE_CLOCK,
 306   PROP_DO_SYNC_EVENT,
 307   PROP_DO_RETRANSMISSION,
 308   PROP_LAST
 309 };
 310
 311 enum
 312 {
 313   GST_RTP_BIN_RTCP_SYNC_ALWAYS,
 314   GST_RTP_BIN_RTCP_SYNC_INITIAL,
 315   GST_RTP_BIN_RTCP_SYNC_RTP
 316 };
 317
 318 #define GST_RTP_BIN_RTCP_SYNC_TYPE (gst_rtp_bin_rtcp_sync_get_type())
 319 static GType
 320 gst_rtp_bin_rtcp_sync_get_type (void)
 321 {
 322   static GType rtcp_sync_type = 0;
 323   static const GEnumValue rtcp_sync_types[] = {
 324     {GST_RTP_BIN_RTCP_SYNC_ALWAYS, "always", "always"},
 325     {GST_RTP_BIN_RTCP_SYNC_INITIAL, "initial", "initial"},
 326     {GST_RTP_BIN_RTCP_SYNC_RTP, "rtp-info", "rtp-info"},
 327     {0, NULL, NULL},
 328   };
 329
 330   if (!rtcp_sync_type) {
 331     rtcp_sync_type = g_enum_register_static ("GstRTCPSync", rtcp_sync_types);
 332   }
 333   return rtcp_sync_type;
 334 }
 335
 336 /* helper objects */
 337 typedef struct _GstRtpBinSession GstRtpBinSession;
 338 typedef struct _GstRtpBinStream GstRtpBinStream;
 339 typedef struct _GstRtpBinClient GstRtpBinClient;
 340
 341 static guint gst_rtp_bin_signals[LAST_SIGNAL] = { 0 };
 342
 343 static GstCaps *pt_map_requested (GstElement * element, guint pt,
 344     GstRtpBinSession * session);
 345 static void payload_type_change (GstElement * element, guint pt,
 346     GstRtpBinSession * session);
 347 static void remove_recv_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session);
 348 static void remove_recv_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session);
 349 static void remove_send_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session);
 350 static void remove_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session);
 351 static void free_client (GstRtpBinClient * client, GstRtpBin * bin);
 352 static void free_stream (GstRtpBinStream * stream, GstRtpBin * bin);
 353
 354 /* Manages the RTP stream for one SSRC.
 355  *
 356  * We pipe the stream (comming from the SSRC demuxer) into a jitterbuffer.
 357  * If we see an SDES RTCP packet that links multiple SSRCs together based on a
 358  * common CNAME, we create a GstRtpBinClient structure to group the SSRCs
 359  * together (see below).
 360  */
 361 struct _GstRtpBinStream
 362 {
 363   /* the SSRC of this stream */
 364   guint32 ssrc;
 365
 366   /* parent bin */
 367   GstRtpBin *bin;
 368
 369   /* the session this SSRC belongs to */
 370   GstRtpBinSession *session;
 371
 372   /* the jitterbuffer of the SSRC */
 373   GstElement *buffer;
 374   gulong buffer_handlesync_sig;
 375   gulong buffer_ptreq_sig;
 376   gulong buffer_ntpstop_sig;
 377   gint percent;
 378
 379   /* the PT demuxer of the SSRC */
 380   GstElement *demux;
 381   gulong demux_newpad_sig;
 382   gulong demux_padremoved_sig;
 383   gulong demux_ptreq_sig;
 384   gulong demux_ptchange_sig;
 385
 386   /* if we have calculated a valid rt_delta for this stream */
 387   gboolean have_sync;
 388   /* mapping to local RTP and NTP time */
 389   gint64 rt_delta;
 390   gint64 rtp_delta;
 391   /* base rtptime in gst time */
 392   gint64 clock_base;
 393 };
 394
 395 #define GST_RTP_SESSION_LOCK(sess)   g_mutex_lock (&(sess)->lock)
 396 #define GST_RTP_SESSION_UNLOCK(sess) g_mutex_unlock (&(sess)->lock)
 397
 398 /* Manages the receiving end of the packets.
 399  *
 400  * There is one such structure for each RTP session (audio/video/...).
 401  * We get the RTP/RTCP packets and stuff them into the session manager. From
 402  * there they are pushed into an SSRC demuxer that splits the stream based on
 403  * SSRC. Each of the SSRC streams go into their own jitterbuffer (managed with
 404  * the GstRtpBinStream above).
 405  */
 406 struct _GstRtpBinSession
 407 {
 408   /* session id */
 409   gint id;
 410   /* the parent bin */
 411   GstRtpBin *bin;
 412   /* the session element */
 413   GstElement *session;
 414   /* the SSRC demuxer */
 415   GstElement *demux;
 416   gulong demux_newpad_sig;
 417   gulong demux_padremoved_sig;
 418
 419   GMutex lock;
 420
 421   /* list of GstRtpBinStream */
 422   GSList *streams;
 423
 424   /* list of elements */
 425   GSList *elements;
 426
 427   /* mapping of payload type to caps */
 428   GHashTable *ptmap;
 429
 430   /* the pads of the session */
 431   GstPad *recv_rtp_sink;
 432   GstPad *recv_rtp_sink_ghost;
 433   GstPad *recv_rtp_src;
 434   GstPad *recv_rtcp_sink;
 435   GstPad *recv_rtcp_sink_ghost;
 436   GstPad *sync_src;
 437   GstPad *send_rtp_sink;
 438   GstPad *send_rtp_sink_ghost;
 439   GstPad *send_rtp_src;
 440   GstPad *send_rtp_src_ghost;
 441   GstPad *send_rtcp_src;
 442   GstPad *send_rtcp_src_ghost;
 443 };
 444
 445 /* Manages the RTP streams that come from one client and should therefore be
 446  * synchronized.
 447  */
 448 struct _GstRtpBinClient
 449 {
 450   /* the common CNAME for the streams */
 451   gchar *cname;
 452   guint cname_len;
 453
 454   /* the streams */
 455   guint nstreams;
 456   GSList *streams;
 457 };
 458
 459 /* find a session with the given id. Must be called with RTP_BIN_LOCK */
 460 static GstRtpBinSession *
 461 find_session_by_id (GstRtpBin * rtpbin, gint id)
 462 {
 463   GSList *walk;
 464
 465   for (walk = rtpbin->sessions; walk; walk = g_slist_next (walk)) {
 466     GstRtpBinSession *sess = (GstRtpBinSession *) walk->data;
 467
 468     if (sess->id == id)
 469       return sess;
 470   }
 471   return NULL;
 472 }
 473
 474 /* find a session with the given request pad. Must be called with RTP_BIN_LOCK */
 475 static GstRtpBinSession *
 476 find_session_by_pad (GstRtpBin * rtpbin, GstPad * pad)
 477 {
 478   GSList *walk;
 479
 480   for (walk = rtpbin->sessions; walk; walk = g_slist_next (walk)) {
 481     GstRtpBinSession *sess = (GstRtpBinSession *) walk->data;
 482
 483     if ((sess->recv_rtp_sink_ghost == pad) ||
 484         (sess->recv_rtcp_sink_ghost == pad) ||
 485         (sess->send_rtp_sink_ghost == pad)
 486         || (sess->send_rtcp_src_ghost == pad))
 487       return sess;
 488   }
 489   return NULL;
 490 }
 491
 492 static void
 493 on_new_ssrc (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 494 {
 495   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_NEW_SSRC], 0,
 496       sess->id, ssrc);
 497 }
 498
 499 static void
 500 on_ssrc_collision (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 501 {
 502   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_COLLISION], 0,
 503       sess->id, ssrc);
 504 }
 505
 506 static void
 507 on_ssrc_validated (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 508 {
 509   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_VALIDATED], 0,
 510       sess->id, ssrc);
 511 }
 512
 513 static void
 514 on_ssrc_active (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 515 {
 516   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_ACTIVE], 0,
 517       sess->id, ssrc);
 518 }
 519
 520 static void
 521 on_ssrc_sdes (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 522 {
 523   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_SDES], 0,
 524       sess->id, ssrc);
 525 }
 526
 527 static void
 528 on_bye_ssrc (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 529 {
 530   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_BYE_SSRC], 0,
 531       sess->id, ssrc);
 532 }
 533
 534 static void
 535 on_bye_timeout (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 536 {
 537   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_BYE_TIMEOUT], 0,
 538       sess->id, ssrc);
 539
 540   if (sess->bin->priv->autoremove)
 541     g_signal_emit_by_name (sess->demux, "clear-ssrc", ssrc, NULL);
 542 }
 543
 544 static void
 545 on_timeout (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 546 {
 547   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_TIMEOUT], 0,
 548       sess->id, ssrc);
 549
 550   if (sess->bin->priv->autoremove)
 551     g_signal_emit_by_name (sess->demux, "clear-ssrc", ssrc, NULL);
 552 }
 553
 554 static void
 555 on_sender_timeout (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 556 {
 557   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SENDER_TIMEOUT], 0,
 558       sess->id, ssrc);
 559 }
 560
 561 static void
 562 on_npt_stop (GstElement * jbuf, GstRtpBinStream * stream)
 563 {
 564   g_signal_emit (stream->bin, gst_rtp_bin_signals[SIGNAL_ON_NPT_STOP], 0,
 565       stream->session->id, stream->ssrc);
 566 }
 567
 568 /* must be called with the SESSION lock */
 569 static GstRtpBinStream *
 570 find_stream_by_ssrc (GstRtpBinSession * session, guint32 ssrc)
 571 {
 572   GSList *walk;
 573
 574   for (walk = session->streams; walk; walk = g_slist_next (walk)) {
 575     GstRtpBinStream *stream = (GstRtpBinStream *) walk->data;
 576
 577     if (stream->ssrc == ssrc)
 578       return stream;
 579   }
 580   return NULL;
 581 }
 582
 583 static void
 584 ssrc_demux_pad_removed (GstElement * element, guint ssrc, GstPad * pad,
 585     GstRtpBinSession * session)
 586 {
 587   GstRtpBinStream *stream = NULL;
 588   GstRtpBin *rtpbin;
 589
 590   rtpbin = session->bin;
 591
 592   GST_RTP_BIN_LOCK (rtpbin);
 593
 594   GST_RTP_SESSION_LOCK (session);
 595   if ((stream = find_stream_by_ssrc (session, ssrc)))
 596     session->streams = g_slist_remove (session->streams, stream);
 597   GST_RTP_SESSION_UNLOCK (session);
 598
 599   if (stream)
 600     free_stream (stream, rtpbin);
 601
 602   GST_RTP_BIN_UNLOCK (rtpbin);
 603 }
 604
 605 /* create a session with the given id.  Must be called with RTP_BIN_LOCK */
 606 static GstRtpBinSession *
 607 create_session (GstRtpBin * rtpbin, gint id)
 608 {
 609   GstRtpBinSession *sess;
 610   GstElement *session, *demux;
 611   GstState target;
 612
 613   if (!(session = gst_element_factory_make ("rtpsession", NULL)))
 614     goto no_session;
 615
 616   if (!(demux = gst_element_factory_make ("rtpssrcdemux", NULL)))
 617     goto no_demux;
 618
 619   sess = g_new0 (GstRtpBinSession, 1);
 620   g_mutex_init (&sess->lock);
 621   sess->id = id;
 622   sess->bin = rtpbin;
 623   sess->session = session;
 624   sess->demux = demux;
 625   sess->ptmap = g_hash_table_new_full (NULL, NULL, NULL,
 626       (GDestroyNotify) gst_caps_unref);
 627   rtpbin->sessions = g_slist_prepend (rtpbin->sessions, sess);
 628
 629   /* configure SDES items */
 630   GST_OBJECT_LOCK (rtpbin);
 631   g_object_set (session, "sdes", rtpbin->sdes, "use-pipeline-clock",
 632       rtpbin->use_pipeline_clock, NULL);
 633   GST_OBJECT_UNLOCK (rtpbin);
 634
 635   /* provide clock_rate to the session manager when needed */
 636   g_signal_connect (session, "request-pt-map",
 637       (GCallback) pt_map_requested, sess);
 638
 639   g_signal_connect (sess->session, "on-new-ssrc",
 640       (GCallback) on_new_ssrc, sess);
 641   g_signal_connect (sess->session, "on-ssrc-collision",
 642       (GCallback) on_ssrc_collision, sess);
 643   g_signal_connect (sess->session, "on-ssrc-validated",
 644       (GCallback) on_ssrc_validated, sess);
 645   g_signal_connect (sess->session, "on-ssrc-active",
 646       (GCallback) on_ssrc_active, sess);
 647   g_signal_connect (sess->session, "on-ssrc-sdes",
 648       (GCallback) on_ssrc_sdes, sess);
 649   g_signal_connect (sess->session, "on-bye-ssrc",
 650       (GCallback) on_bye_ssrc, sess);
 651   g_signal_connect (sess->session, "on-bye-timeout",
 652       (GCallback) on_bye_timeout, sess);
 653   g_signal_connect (sess->session, "on-timeout", (GCallback) on_timeout, sess);
 654   g_signal_connect (sess->session, "on-sender-timeout",
 655       (GCallback) on_sender_timeout, sess);
 656
 657   gst_bin_add (GST_BIN_CAST (rtpbin), session);
 658   gst_bin_add (GST_BIN_CAST (rtpbin), demux);
 659
 660   GST_OBJECT_LOCK (rtpbin);
 661   target = GST_STATE_TARGET (rtpbin);
 662   GST_OBJECT_UNLOCK (rtpbin);
 663
 664   /* change state only to what's needed */
 665   gst_element_set_state (demux, target);
 666   gst_element_set_state (session, target);
 667
 668   return sess;
 669
 670   /* ERRORS */
 671 no_session:
 672   {
 673     g_warning ("rtpbin: could not create rtpsession element");
 674     return NULL;
 675   }
 676 no_demux:
 677   {
 678     gst_object_unref (session);
 679     g_warning ("rtpbin: could not create rtpssrcdemux element");
 680     return NULL;
 681   }
 682 }
 683
 684 static gboolean
 685 bin_manage_element (GstRtpBin * bin, GstElement * element)
 686 {
 687   GstRtpBinPrivate *priv = bin->priv;
 688
 689   if (g_list_find (priv->elements, element)) {
 690     GST_DEBUG_OBJECT (bin, "requested element %p already in bin", element);
 691   } else {
 692     GST_DEBUG_OBJECT (bin, "adding requested element %p", element);
 693     if (!gst_bin_add (GST_BIN_CAST (bin), element))
 694       goto add_failed;
 695     if (!gst_element_sync_state_with_parent (element))
 696       GST_WARNING_OBJECT (bin, "unable to sync element state with rtpbin");
 697   }
 698   /* we add the element multiple times, each we need an equal number of
 699    * removes to really remove the element from the bin */
 700   priv->elements = g_list_prepend (priv->elements, element);
 701
 702   return TRUE;
 703
 704   /* ERRORS */
 705 add_failed:
 706   {
 707     GST_WARNING_OBJECT (bin, "unable to add element");
 708     return FALSE;
 709   }
 710 }
 711
 712 static void
 713 remove_bin_element (GstElement * element, GstRtpBin * bin)
 714 {
 715   GstRtpBinPrivate *priv = bin->priv;
 716   GList *find;
 717
 718   find = g_list_find (priv->elements, element);
 719   if (find) {
 720     priv->elements = g_list_delete_link (priv->elements, find);
 721
 722     if (!g_list_find (priv->elements, element))
 723       gst_bin_remove (GST_BIN_CAST (bin), element);
 724     else
 725       gst_object_unref (element);
 726   }
 727 }
 728
 729 /* called with RTP_BIN_LOCK */
 730 static void
 731 free_session (GstRtpBinSession * sess, GstRtpBin * bin)
 732 {
 733   GST_DEBUG_OBJECT (bin, "freeing session %p", sess);
 734
 735   gst_element_set_locked_state (sess->demux, TRUE);
 736   gst_element_set_locked_state (sess->session, TRUE);
 737
 738   gst_element_set_state (sess->demux, GST_STATE_NULL);
 739   gst_element_set_state (sess->session, GST_STATE_NULL);
 740
 741   remove_recv_rtp (bin, sess);
 742   remove_recv_rtcp (bin, sess);
 743   remove_send_rtp (bin, sess);
 744   remove_rtcp (bin, sess);
 745
 746   gst_bin_remove (GST_BIN_CAST (bin), sess->session);
 747   gst_bin_remove (GST_BIN_CAST (bin), sess->demux);
 748
 749   g_slist_foreach (sess->elements, (GFunc) remove_bin_element, bin);
 750   g_slist_free (sess->elements);
 751
 752   g_slist_foreach (sess->streams, (GFunc) free_stream, bin);
 753   g_slist_free (sess->streams);
 754
 755   g_mutex_clear (&sess->lock);
 756   g_hash_table_destroy (sess->ptmap);
 757
 758   g_free (sess);
 759 }
 760
 761 /* get the payload type caps for the specific payload @pt in @session */
 762 static GstCaps *
 763 get_pt_map (GstRtpBinSession * session, guint pt)
 764 {
 765   GstCaps *caps = NULL;
 766   GstRtpBin *bin;
 767   GValue ret = { 0 };
 768   GValue args[3] = { {0}, {0}, {0} };
 769
 770   GST_DEBUG ("searching pt %d in cache", pt);
 771
 772   GST_RTP_SESSION_LOCK (session);
 773
 774   /* first look in the cache */
 775   caps = g_hash_table_lookup (session->ptmap, GINT_TO_POINTER (pt));
 776   if (caps) {
 777     gst_caps_ref (caps);
 778     goto done;
 779   }
 780
 781   bin = session->bin;
 782
 783   GST_DEBUG ("emiting signal for pt %d in session %d", pt, session->id);
 784
 785   /* not in cache, send signal to request caps */
 786   g_value_init (&args[0], GST_TYPE_ELEMENT);
 787   g_value_set_object (&args[0], bin);
 788   g_value_init (&args[1], G_TYPE_UINT);
 789   g_value_set_uint (&args[1], session->id);
 790   g_value_init (&args[2], G_TYPE_UINT);
 791   g_value_set_uint (&args[2], pt);
 792
 793   g_value_init (&ret, GST_TYPE_CAPS);
 794   g_value_set_boxed (&ret, NULL);
 795
 796   GST_RTP_SESSION_UNLOCK (session);
 797
 798   g_signal_emitv (args, gst_rtp_bin_signals[SIGNAL_REQUEST_PT_MAP], 0, &ret);
 799
 800   GST_RTP_SESSION_LOCK (session);
 801
 802   g_value_unset (&args[0]);
 803   g_value_unset (&args[1]);
 804   g_value_unset (&args[2]);
 805
 806   /* look in the cache again because we let the lock go */
 807   caps = g_hash_table_lookup (session->ptmap, GINT_TO_POINTER (pt));
 808   if (caps) {
 809     gst_caps_ref (caps);
 810     g_value_unset (&ret);
 811     goto done;
 812   }
 813
 814   caps = (GstCaps *) g_value_dup_boxed (&ret);
 815   g_value_unset (&ret);
 816   if (!caps)
 817     goto no_caps;
 818
 819   GST_DEBUG ("caching pt %d as %" GST_PTR_FORMAT, pt, caps);
 820
 821   /* store in cache, take additional ref */
 822   g_hash_table_insert (session->ptmap, GINT_TO_POINTER (pt),
 823       gst_caps_ref (caps));
 824
 825 done:
 826   GST_RTP_SESSION_UNLOCK (session);
 827
 828   return caps;
 829
 830   /* ERRORS */
 831 no_caps:
 832   {
 833     GST_RTP_SESSION_UNLOCK (session);
 834     GST_DEBUG ("no pt map could be obtained");
 835     return NULL;
 836   }
 837 }
 838
 839 static gboolean
 840 return_true (gpointer key, gpointer value, gpointer user_data)
 841 {
 842   return TRUE;
 843 }
 844
 845 static void
 846 gst_rtp_bin_reset_sync (GstRtpBin * rtpbin)
 847 {
 848   GSList *clients, *streams;
 849
 850   GST_DEBUG_OBJECT (rtpbin, "Reset sync on all clients");
 851
 852   GST_RTP_BIN_LOCK (rtpbin);
 853   for (clients = rtpbin->clients; clients; clients = g_slist_next (clients)) {
 854     GstRtpBinClient *client = (GstRtpBinClient *) clients->data;
 855
 856     /* reset sync on all streams for this client */
 857     for (streams = client->streams; streams; streams = g_slist_next (streams)) {
 858       GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
 859
 860       /* make use require a new SR packet for this stream before we attempt new
 861        * lip-sync */
 862       stream->have_sync = FALSE;
 863       stream->rt_delta = 0;
 864       stream->rtp_delta = 0;
 865       stream->clock_base = -100 * GST_SECOND;
 866     }
 867   }
 868   GST_RTP_BIN_UNLOCK (rtpbin);
 869 }
 870
 871 static void
 872 gst_rtp_bin_clear_pt_map (GstRtpBin * bin)
 873 {
 874   GSList *sessions, *streams;
 875
 876   GST_RTP_BIN_LOCK (bin);
 877   GST_DEBUG_OBJECT (bin, "clearing pt map");
 878   for (sessions = bin->sessions; sessions; sessions = g_slist_next (sessions)) {
 879     GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
 880
 881     GST_DEBUG_OBJECT (bin, "clearing session %p", session);
 882     g_signal_emit_by_name (session->session, "clear-pt-map", NULL);
 883
 884     GST_RTP_SESSION_LOCK (session);
 885     g_hash_table_foreach_remove (session->ptmap, return_true, NULL);
 886
 887     for (streams = session->streams; streams; streams = g_slist_next (streams)) {
 888       GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
 889
 890       GST_DEBUG_OBJECT (bin, "clearing stream %p", stream);
 891       g_signal_emit_by_name (stream->buffer, "clear-pt-map", NULL);
 892       if (stream->demux)
 893         g_signal_emit_by_name (stream->demux, "clear-pt-map", NULL);
 894     }
 895     GST_RTP_SESSION_UNLOCK (session);
 896   }
 897   GST_RTP_BIN_UNLOCK (bin);
 898
 899   /* reset sync too */
 900   gst_rtp_bin_reset_sync (bin);
 901 }
 902
 903 static RTPSession *
 904 gst_rtp_bin_get_internal_session (GstRtpBin * bin, guint session_id)
 905 {
 906   RTPSession *internal_session = NULL;
 907   GstRtpBinSession *session;
 908
 909   GST_RTP_BIN_LOCK (bin);
 910   GST_DEBUG_OBJECT (bin, "retrieving internal RTPSession object, index: %d",
 911       session_id);
 912   session = find_session_by_id (bin, (gint) session_id);
 913   if (session) {
 914     g_object_get (session->session, "internal-session", &internal_session,
 915         NULL);
 916   }
 917   GST_RTP_BIN_UNLOCK (bin);
 918
 919   return internal_session;
 920 }
 921
 922 static GstElement *
 923 gst_rtp_bin_request_encoder (GstRtpBin * bin, guint session_id)
 924 {
 925   GST_DEBUG_OBJECT (bin, "return NULL encoder");
 926   return NULL;
 927 }
 928
 929 static GstElement *
 930 gst_rtp_bin_request_decoder (GstRtpBin * bin, guint session_id)
 931 {
 932   GST_DEBUG_OBJECT (bin, "return NULL decoder");
 933   return NULL;
 934 }
 935
 936 static void
 937 gst_rtp_bin_propagate_property_to_jitterbuffer (GstRtpBin * bin,
 938     const gchar * name, const GValue * value)
 939 {
 940   GSList *sessions, *streams;
 941
 942   GST_RTP_BIN_LOCK (bin);
 943   for (sessions = bin->sessions; sessions; sessions = g_slist_next (sessions)) {
 944     GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
 945
 946     GST_RTP_SESSION_LOCK (session);
 947     for (streams = session->streams; streams; streams = g_slist_next (streams)) {
 948       GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
 949
 950       g_object_set_property (G_OBJECT (stream->buffer), name, value);
 951     }
 952     GST_RTP_SESSION_UNLOCK (session);
 953   }
 954   GST_RTP_BIN_UNLOCK (bin);
 955 }
 956
 957 /* get a client with the given SDES name. Must be called with RTP_BIN_LOCK */
 958 static GstRtpBinClient *
 959 get_client (GstRtpBin * bin, guint8 len, guint8 * data, gboolean * created)
 960 {
 961   GstRtpBinClient *result = NULL;
 962   GSList *walk;
 963
 964   for (walk = bin->clients; walk; walk = g_slist_next (walk)) {
 965     GstRtpBinClient *client = (GstRtpBinClient *) walk->data;
 966
 967     if (len != client->cname_len)
 968       continue;
 969
 970     if (!strncmp ((gchar *) data, client->cname, client->cname_len)) {
 971       GST_DEBUG_OBJECT (bin, "found existing client %p with CNAME %s", client,
 972           client->cname);
 973       result = client;
 974       break;
 975     }
 976   }
 977
 978   /* nothing found, create one */
 979   if (result == NULL) {
 980     result = g_new0 (GstRtpBinClient, 1);
 981     result->cname = g_strndup ((gchar *) data, len);
 982     result->cname_len = len;
 983     bin->clients = g_slist_prepend (bin->clients, result);
 984     GST_DEBUG_OBJECT (bin, "created new client %p with CNAME %s", result,
 985         result->cname);
 986   }
 987   return result;
 988 }
 989
 990 static void
 991 free_client (GstRtpBinClient * client, GstRtpBin * bin)
 992 {
 993   GST_DEBUG_OBJECT (bin, "freeing client %p", client);
 994   g_slist_free (client->streams);
 995   g_free (client->cname);
 996   g_free (client);
 997 }
 998
 999 static void
1000 get_current_times (GstRtpBin * bin, GstClockTime * running_time,
1001     guint64 * ntpnstime)
1002 {
1003   guint64 ntpns;
1004   GstClock *clock;
1005   GstClockTime base_time, rt, clock_time;
1006
1007   GST_OBJECT_LOCK (bin);
1008   if ((clock = GST_ELEMENT_CLOCK (bin))) {
1009     base_time = GST_ELEMENT_CAST (bin)->base_time;
1010     gst_object_ref (clock);
1011     GST_OBJECT_UNLOCK (bin);
1012
1013     clock_time = gst_clock_get_time (clock);
1014
1015     if (bin->use_pipeline_clock) {
1016       ntpns = clock_time - base_time;
1017     } else {
1018       GTimeVal current;
1019
1020       /* get current NTP time */
1021       g_get_current_time (&current);
1022       ntpns = GST_TIMEVAL_TO_TIME (current);
1023     }
1024
1025     /* add constant to convert from 1970 based time to 1900 based time */
1026     ntpns += (2208988800LL * GST_SECOND);
1027
1028     /* get current clock time and convert to running time */
1029     rt = clock_time - base_time;
1030
1031     gst_object_unref (clock);
1032   } else {
1033     GST_OBJECT_UNLOCK (bin);
1034     rt = -1;
1035     ntpns = -1;
1036   }
1037   if (running_time)
1038     *running_time = rt;
1039   if (ntpnstime)
1040     *ntpnstime = ntpns;
1041 }
1042
1043 static void
1044 stream_set_ts_offset (GstRtpBin * bin, GstRtpBinStream * stream,
1045     gint64 ts_offset, gboolean check)
1046 {
1047   gint64 prev_ts_offset;
1048
1049   g_object_get (stream->buffer, "ts-offset", &prev_ts_offset, NULL);
1050
1051   /* delta changed, see how much */
1052   if (prev_ts_offset != ts_offset) {
1053     gint64 diff;
1054
1055     diff = prev_ts_offset - ts_offset;
1056
1057     GST_DEBUG_OBJECT (bin,
1058         "ts-offset %" G_GINT64_FORMAT ", prev %" G_GINT64_FORMAT
1059         ", diff: %" G_GINT64_FORMAT, ts_offset, prev_ts_offset, diff);
1060
1061     if (check) {
1062       /* only change diff when it changed more than 4 milliseconds. This
1063        * compensates for rounding errors in NTP to RTP timestamp
1064        * conversions */
1065       if (ABS (diff) < 4 * GST_MSECOND) {
1066         GST_DEBUG_OBJECT (bin, "offset too small, ignoring");
1067         return;
1068       }
1069       if (ABS (diff) > (3 * GST_SECOND)) {
1070         GST_WARNING_OBJECT (bin, "offset unusually large, ignoring");
1071         return;
1072       }
1073     }
1074     g_object_set (stream->buffer, "ts-offset", ts_offset, NULL);
1075   }
1076   GST_DEBUG_OBJECT (bin, "stream SSRC %08x, delta %" G_GINT64_FORMAT,
1077       stream->ssrc, ts_offset);
1078 }
1079
1080 static void
1081 gst_rtp_bin_send_sync_event (GstRtpBinStream * stream)
1082 {
1083   if (stream->bin->send_sync_event) {
1084     GstEvent *event;
1085     GstPad *srcpad;
1086
1087     GST_DEBUG_OBJECT (stream->bin,
1088         "sending GstRTCPSRReceived event downstream");
1089
1090     event = gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM,
1091         gst_structure_new_empty ("GstRTCPSRReceived"));
1092
1093     srcpad = gst_element_get_static_pad (stream->buffer, "src");
1094     gst_pad_push_event (srcpad, event);
1095     gst_object_unref (srcpad);
1096   }
1097 }
1098
1099 /* associate a stream to the given CNAME. This will make sure all streams for
1100  * that CNAME are synchronized together.
1101  * Must be called with GST_RTP_BIN_LOCK */
1102 static void
1103 gst_rtp_bin_associate (GstRtpBin * bin, GstRtpBinStream * stream, guint8 len,
1104     guint8 * data, guint64 ntptime, guint64 last_extrtptime,
1105     guint64 base_rtptime, guint64 base_time, guint clock_rate,
1106     gint64 rtp_clock_base)
1107 {
1108   GstRtpBinClient *client;
1109   gboolean created;
1110   GSList *walk;
1111   guint64 local_rt;
1112   guint64 local_rtp;
1113   GstClockTime running_time;
1114   guint64 ntpnstime;
1115   gint64 ntpdiff, rtdiff;
1116   guint64 last_unix;
1117
1118   /* first find or create the CNAME */
1119   client = get_client (bin, len, data, &created);
1120
1121   /* find stream in the client */
1122   for (walk = client->streams; walk; walk = g_slist_next (walk)) {
1123     GstRtpBinStream *ostream = (GstRtpBinStream *) walk->data;
1124
1125     if (ostream == stream)
1126       break;
1127   }
1128   /* not found, add it to the list */
1129   if (walk == NULL) {
1130     GST_DEBUG_OBJECT (bin,
1131         "new association of SSRC %08x with client %p with CNAME %s",
1132         stream->ssrc, client, client->cname);
1133     client->streams = g_slist_prepend (client->streams, stream);
1134     client->nstreams++;
1135   } else {
1136     GST_DEBUG_OBJECT (bin,
1137         "found association of SSRC %08x with client %p with CNAME %s",
1138         stream->ssrc, client, client->cname);
1139   }
1140
1141   if (!GST_CLOCK_TIME_IS_VALID (last_extrtptime)) {
1142     GST_DEBUG_OBJECT (bin, "invalidated sync data");
1143     if (bin->rtcp_sync == GST_RTP_BIN_RTCP_SYNC_RTP) {
1144       /* we don't need that data, so carry on,
1145        * but make some values look saner */
1146       last_extrtptime = base_rtptime;
1147     } else {
1148       /* nothing we can do with this data in this case */
1149       GST_DEBUG_OBJECT (bin, "bailing out");
1150       return;
1151     }
1152   }
1153
1154   /* Take the extended rtptime we found in the SR packet and map it to the
1155    * local rtptime. The local rtp time is used to construct timestamps on the
1156    * buffers so we will calculate what running_time corresponds to the RTP
1157    * timestamp in the SR packet. */
1158   local_rtp = last_extrtptime - base_rtptime;
1159
1160   GST_DEBUG_OBJECT (bin,
1161       "base %" G_GUINT64_FORMAT ", extrtptime %" G_GUINT64_FORMAT
1162       ", local RTP %" G_GUINT64_FORMAT ", clock-rate %d, "
1163       "clock-base %" G_GINT64_FORMAT, base_rtptime,
1164       last_extrtptime, local_rtp, clock_rate, rtp_clock_base);
1165
1166   /* calculate local RTP time in gstreamer timestamp, we essentially perform the
1167    * same conversion that a jitterbuffer would use to convert an rtp timestamp
1168    * into a corresponding gstreamer timestamp. Note that the base_time also
1169    * contains the drift between sender and receiver. */
1170   local_rt = gst_util_uint64_scale_int (local_rtp, GST_SECOND, clock_rate);
1171   local_rt += base_time;
1172
1173   /* convert ntptime to unix time since 1900 */
1174   last_unix = gst_util_uint64_scale (ntptime, GST_SECOND,
1175       (G_GINT64_CONSTANT (1) << 32));
1176
1177   stream->have_sync = TRUE;
1178
1179   GST_DEBUG_OBJECT (bin,
1180       "local UNIX %" G_GUINT64_FORMAT ", remote UNIX %" G_GUINT64_FORMAT,
1181       local_rt, last_unix);
1182
1183   /* recalc inter stream playout offset, but only if there is more than one
1184    * stream or we're doing NTP sync. */
1185   if (bin->ntp_sync) {
1186     /* For NTP sync we need to first get a snapshot of running_time and NTP
1187      * time. We know at what running_time we play a certain RTP time, we also
1188      * calculated when we would play the RTP time in the SR packet. Now we need
1189      * to know how the running_time and the NTP time relate to eachother. */
1190     get_current_times (bin, &running_time, &ntpnstime);
1191
1192     /* see how far away the NTP time is. This is the difference between the
1193      * current NTP time and the NTP time in the last SR packet. */
1194     ntpdiff = ntpnstime - last_unix;
1195     /* see how far away the running_time is. This is the difference between the
1196      * current running_time and the running_time of the RTP timestamp in the
1197      * last SR packet. */
1198     rtdiff = running_time - local_rt;
1199
1200     GST_DEBUG_OBJECT (bin,
1201         "NTP time %" G_GUINT64_FORMAT ", last unix %" G_GUINT64_FORMAT,
1202         ntpnstime, last_unix);
1203     GST_DEBUG_OBJECT (bin,
1204         "NTP diff %" G_GINT64_FORMAT ", RT diff %" G_GINT64_FORMAT, ntpdiff,
1205         rtdiff);
1206
1207     /* combine to get the final diff to apply to the running_time */
1208     stream->rt_delta = rtdiff - ntpdiff;
1209
1210     stream_set_ts_offset (bin, stream, stream->rt_delta, FALSE);
1211   } else {
1212     gint64 min, rtp_min, clock_base = stream->clock_base;
1213     gboolean all_sync, use_rtp;
1214     gboolean rtcp_sync = g_atomic_int_get (&bin->rtcp_sync);
1215
1216     /* calculate delta between server and receiver. last_unix is created by
1217      * converting the ntptime in the last SR packet to a gstreamer timestamp. This
1218      * delta expresses the difference to our timeline and the server timeline. The
1219      * difference in itself doesn't mean much but we can combine the delta of
1220      * multiple streams to create a stream specific offset. */
1221     stream->rt_delta = last_unix - local_rt;
1222
1223     /* calculate the min of all deltas, ignoring streams that did not yet have a
1224      * valid rt_delta because we did not yet receive an SR packet for those
1225      * streams.
1226      * We calculate the mininum because we would like to only apply positive
1227      * offsets to streams, delaying their playback instead of trying to speed up
1228      * other streams (which might be imposible when we have to create negative
1229      * latencies).
1230      * The stream that has the smallest diff is selected as the reference stream,
1231      * all other streams will have a positive offset to this difference. */
1232
1233     /* some alternative setting allow ignoring RTCP as much as possible,
1234      * for servers generating bogus ntp timeline */
1235     min = rtp_min = G_MAXINT64;
1236     use_rtp = FALSE;
1237     if (rtcp_sync == GST_RTP_BIN_RTCP_SYNC_RTP) {
1238       guint64 ext_base;
1239
1240       use_rtp = TRUE;
1241       /* signed version for convienience */
1242       clock_base = base_rtptime;
1243       /* deal with possible wrap-around */
1244       ext_base = base_rtptime;
1245       rtp_clock_base = gst_rtp_buffer_ext_timestamp (&ext_base, rtp_clock_base);
1246       /* sanity check; base rtp and provided clock_base should be close */
1247       if (rtp_clock_base >= clock_base) {
1248         if (rtp_clock_base - clock_base < 10 * clock_rate) {
1249           rtp_clock_base = base_time +
1250               gst_util_uint64_scale_int (rtp_clock_base - clock_base,
1251               GST_SECOND, clock_rate);
1252         } else {
1253           use_rtp = FALSE;
1254         }
1255       } else {
1256         if (clock_base - rtp_clock_base < 10 * clock_rate) {
1257           rtp_clock_base = base_time -
1258               gst_util_uint64_scale_int (clock_base - rtp_clock_base,
1259               GST_SECOND, clock_rate);
1260         } else {
1261           use_rtp = FALSE;
1262         }
1263       }
1264       /* warn and bail for clarity out if no sane values */
1265       if (!use_rtp) {
1266         GST_WARNING_OBJECT (bin, "unable to sync to provided rtptime");
1267         return;
1268       }
1269       /* store to track changes */
1270       clock_base = rtp_clock_base;
1271       /* generate a fake as before,
1272        * now equating rtptime obtained from RTP-Info,
1273        * where the large time represent the otherwise irrelevant npt/ntp time */
1274       stream->rtp_delta = (GST_SECOND << 28) - rtp_clock_base;
1275     } else {
1276       clock_base = rtp_clock_base;
1277     }
1278
1279     all_sync = TRUE;
1280     for (walk = client->streams; walk; walk = g_slist_next (walk)) {
1281       GstRtpBinStream *ostream = (GstRtpBinStream *) walk->data;
1282
1283       if (!ostream->have_sync) {
1284         all_sync = FALSE;
1285         continue;
1286       }
1287
1288       /* change in current stream's base from previously init'ed value
1289        * leads to reset of all stream's base */
1290       if (stream != ostream && stream->clock_base >= 0 &&
1291           (stream->clock_base != clock_base)) {
1292         GST_DEBUG_OBJECT (bin, "reset upon clock base change");
1293         ostream->clock_base = -100 * GST_SECOND;
1294         ostream->rtp_delta = 0;
1295       }
1296
1297       if (ostream->rt_delta < min)
1298         min = ostream->rt_delta;
1299       if (ostream->rtp_delta < rtp_min)
1300         rtp_min = ostream->rtp_delta;
1301     }
1302
1303     /* arrange to re-sync for each stream upon significant change,
1304      * e.g. post-seek */
1305     all_sync = all_sync && (stream->clock_base == clock_base);
1306     stream->clock_base = clock_base;
1307
1308     /* may need init performed above later on, but nothing more to do now */
1309     if (client->nstreams <= 1)
1310       return;
1311
1312     GST_DEBUG_OBJECT (bin, "client %p min delta %" G_GINT64_FORMAT
1313         " all sync %d", client, min, all_sync);
1314     GST_DEBUG_OBJECT (bin, "rtcp sync mode %d, use_rtp %d", rtcp_sync, use_rtp);
1315
1316     switch (rtcp_sync) {
1317       case GST_RTP_BIN_RTCP_SYNC_RTP:
1318         if (!use_rtp)
1319           break;
1320         GST_DEBUG_OBJECT (bin, "using rtp generated reports; "
1321             "client %p min rtp delta %" G_GINT64_FORMAT, client, rtp_min);
1322         /* fall-through */
1323       case GST_RTP_BIN_RTCP_SYNC_INITIAL:
1324         /* if all have been synced already, do not bother further */
1325         if (all_sync) {
1326           GST_DEBUG_OBJECT (bin, "all streams already synced; done");
1327           return;
1328         }
1329         break;
1330       default:
1331         break;
1332     }
1333
1334     /* bail out if we adjusted recently enough */
1335     if (all_sync && (last_unix - bin->priv->last_unix) <
1336         bin->rtcp_sync_interval * GST_MSECOND) {
1337       GST_DEBUG_OBJECT (bin, "discarding RTCP sender packet for sync; "
1338           "previous sender info too recent "
1339           "(previous UNIX %" G_GUINT64_FORMAT ")", bin->priv->last_unix);
1340       return;
1341     }
1342     bin->priv->last_unix = last_unix;
1343
1344     /* calculate offsets for each stream */
1345     for (walk = client->streams; walk; walk = g_slist_next (walk)) {
1346       GstRtpBinStream *ostream = (GstRtpBinStream *) walk->data;
1347       gint64 ts_offset;
1348
1349       /* ignore streams for which we didn't receive an SR packet yet, we
1350        * can't synchronize them yet. We can however sync other streams just
1351        * fine. */
1352       if (!ostream->have_sync)
1353         continue;
1354
1355       /* calculate offset to our reference stream, this should always give a
1356        * positive number. */
1357       if (use_rtp)
1358         ts_offset = ostream->rtp_delta - rtp_min;
1359       else
1360         ts_offset = ostream->rt_delta - min;
1361
1362       stream_set_ts_offset (bin, ostream, ts_offset, TRUE);
1363     }
1364   }
1365   gst_rtp_bin_send_sync_event (stream);
1366
1367   return;
1368 }
1369
1370 #define GST_RTCP_BUFFER_FOR_PACKETS(b,buffer,packet) \
1371   for ((b) = gst_rtcp_buffer_get_first_packet ((buffer), (packet)); (b); \
1372           (b) = gst_rtcp_packet_move_to_next ((packet)))
1373
1374 #define GST_RTCP_SDES_FOR_ITEMS(b,packet) \
1375   for ((b) = gst_rtcp_packet_sdes_first_item ((packet)); (b); \
1376           (b) = gst_rtcp_packet_sdes_next_item ((packet)))
1377
1378 #define GST_RTCP_SDES_FOR_ENTRIES(b,packet) \
1379   for ((b) = gst_rtcp_packet_sdes_first_entry ((packet)); (b); \
1380           (b) = gst_rtcp_packet_sdes_next_entry ((packet)))
1381
1382 static void
1383 gst_rtp_bin_handle_sync (GstElement * jitterbuffer, GstStructure * s,
1384     GstRtpBinStream * stream)
1385 {
1386   GstRtpBin *bin;
1387   GstRTCPPacket packet;
1388   guint32 ssrc;
1389   guint64 ntptime;
1390   gboolean have_sr, have_sdes;
1391   gboolean more;
1392   guint64 base_rtptime;
1393   guint64 base_time;
1394   guint clock_rate;
1395   guint64 clock_base;
1396   guint64 extrtptime;
1397   GstBuffer *buffer;
1398   GstRTCPBuffer rtcp = { NULL, };
1399
1400   bin = stream->bin;
1401
1402   GST_DEBUG_OBJECT (bin, "sync handler called");
1403
1404   /* get the last relation between the rtp timestamps and the gstreamer
1405    * timestamps. We get this info directly from the jitterbuffer which
1406    * constructs gstreamer timestamps from rtp timestamps and so it know exactly
1407    * what the current situation is. */
1408   base_rtptime =
1409       g_value_get_uint64 (gst_structure_get_value (s, "base-rtptime"));
1410   base_time = g_value_get_uint64 (gst_structure_get_value (s, "base-time"));
1411   clock_rate = g_value_get_uint (gst_structure_get_value (s, "clock-rate"));
1412   clock_base = g_value_get_uint64 (gst_structure_get_value (s, "clock-base"));
1413   extrtptime =
1414       g_value_get_uint64 (gst_structure_get_value (s, "sr-ext-rtptime"));
1415   buffer = gst_value_get_buffer (gst_structure_get_value (s, "sr-buffer"));
1416
1417   have_sr = FALSE;
1418   have_sdes = FALSE;
1419
1420   gst_rtcp_buffer_map (buffer, GST_MAP_READ, &rtcp);
1421
1422   GST_RTCP_BUFFER_FOR_PACKETS (more, &rtcp, &packet) {
1423     /* first packet must be SR or RR or else the validate would have failed */
1424     switch (gst_rtcp_packet_get_type (&packet)) {
1425       case GST_RTCP_TYPE_SR:
1426         /* only parse first. There is only supposed to be one SR in the packet
1427          * but we will deal with malformed packets gracefully */
1428         if (have_sr)
1429           break;
1430         /* get NTP and RTP times */
1431         gst_rtcp_packet_sr_get_sender_info (&packet, &ssrc, &ntptime, NULL,
1432             NULL, NULL);
1433
1434         GST_DEBUG_OBJECT (bin, "received sync packet from SSRC %08x", ssrc);
1435         /* ignore SR that is not ours */
1436         if (ssrc != stream->ssrc)
1437           continue;
1438
1439         have_sr = TRUE;
1440         break;
1441       case GST_RTCP_TYPE_SDES:
1442       {
1443         gboolean more_items, more_entries;
1444
1445         /* only deal with first SDES, there is only supposed to be one SDES in
1446          * the RTCP packet but we deal with bad packets gracefully. Also bail
1447          * out if we have not seen an SR item yet. */
1448         if (have_sdes || !have_sr)
1449           break;
1450
1451         GST_RTCP_SDES_FOR_ITEMS (more_items, &packet) {
1452           /* skip items that are not about the SSRC of the sender */
1453           if (gst_rtcp_packet_sdes_get_ssrc (&packet) != ssrc)
1454             continue;
1455
1456           /* find the CNAME entry */
1457           GST_RTCP_SDES_FOR_ENTRIES (more_entries, &packet) {
1458             GstRTCPSDESType type;
1459             guint8 len;
1460             guint8 *data;
1461
1462             gst_rtcp_packet_sdes_get_entry (&packet, &type, &len, &data);
1463
1464             if (type == GST_RTCP_SDES_CNAME) {
1465               GST_RTP_BIN_LOCK (bin);
1466               /* associate the stream to CNAME */
1467               gst_rtp_bin_associate (bin, stream, len, data,
1468                   ntptime, extrtptime, base_rtptime, base_time, clock_rate,
1469                   clock_base);
1470               GST_RTP_BIN_UNLOCK (bin);
1471             }
1472           }
1473         }
1474         have_sdes = TRUE;
1475         break;
1476       }
1477       default:
1478         /* we can ignore these packets */
1479         break;
1480     }
1481   }
1482   gst_rtcp_buffer_unmap (&rtcp);
1483 }
1484
1485 /* create a new stream with @ssrc in @session. Must be called with
1486  * RTP_SESSION_LOCK. */
1487 static GstRtpBinStream *
1488 create_stream (GstRtpBinSession * session, guint32 ssrc)
1489 {
1490   GstElement *buffer, *demux = NULL;
1491   GstRtpBinStream *stream;
1492   GstRtpBin *rtpbin;
1493   GstState target;
1494
1495   rtpbin = session->bin;
1496
1497   if (!(buffer = gst_element_factory_make ("rtpjitterbuffer", NULL)))
1498     goto no_jitterbuffer;
1499
1500   if (!rtpbin->ignore_pt)
1501     if (!(demux = gst_element_factory_make ("rtpptdemux", NULL)))
1502       goto no_demux;
1503
1504   stream = g_new0 (GstRtpBinStream, 1);
1505   stream->ssrc = ssrc;
1506   stream->bin = rtpbin;
1507   stream->session = session;
1508   stream->buffer = buffer;
1509   stream->demux = demux;
1510
1511   stream->have_sync = FALSE;
1512   stream->rt_delta = 0;
1513   stream->rtp_delta = 0;
1514   stream->percent = 100;
1515   stream->clock_base = -100 * GST_SECOND;
1516   session->streams = g_slist_prepend (session->streams, stream);
1517
1518   /* provide clock_rate to the jitterbuffer when needed */
1519   stream->buffer_ptreq_sig = g_signal_connect (buffer, "request-pt-map",
1520       (GCallback) pt_map_requested, session);
1521   stream->buffer_ntpstop_sig = g_signal_connect (buffer, "on-npt-stop",
1522       (GCallback) on_npt_stop, stream);
1523
1524   g_object_set_data (G_OBJECT (buffer), "GstRTPBin.session", session);
1525   g_object_set_data (G_OBJECT (buffer), "GstRTPBin.stream", stream);
1526
1527   /* configure latency and packet lost */
1528   g_object_set (buffer, "latency", rtpbin->latency_ms, NULL);
1529   g_object_set (buffer, "drop-on-latency", rtpbin->drop_on_latency, NULL);
1530   g_object_set (buffer, "do-lost", rtpbin->do_lost, NULL);
1531   g_object_set (buffer, "mode", rtpbin->buffer_mode, NULL);
1532   g_object_set (buffer, "do-retransmission", rtpbin->do_retransmission, NULL);
1533
1534   g_signal_emit (rtpbin, gst_rtp_bin_signals[SIGNAL_NEW_JITTERBUFFER], 0,
1535       buffer, session->id, ssrc);
1536
1537   if (!rtpbin->ignore_pt)
1538     gst_bin_add (GST_BIN_CAST (rtpbin), demux);
1539   gst_bin_add (GST_BIN_CAST (rtpbin), buffer);
1540
1541   /* link stuff */
1542   if (demux)
1543     gst_element_link_pads_full (buffer, "src", demux, "sink",
1544         GST_PAD_LINK_CHECK_NOTHING);
1545
1546   if (rtpbin->buffering) {
1547     guint64 last_out;
1548
1549     GST_INFO_OBJECT (rtpbin,
1550         "bin is buffering, set jitterbuffer as not active");
1551     g_signal_emit_by_name (buffer, "set-active", FALSE, (gint64) 0, &last_out);
1552   }
1553
1554
1555   GST_OBJECT_LOCK (rtpbin);
1556   target = GST_STATE_TARGET (rtpbin);
1557   GST_OBJECT_UNLOCK (rtpbin);
1558
1559   /* from sink to source */
1560   if (demux)
1561     gst_element_set_state (demux, target);
1562
1563   gst_element_set_state (buffer, target);
1564
1565   return stream;
1566
1567   /* ERRORS */
1568 no_jitterbuffer:
1569   {
1570     g_warning ("rtpbin: could not create rtpjitterbuffer element");
1571     return NULL;
1572   }
1573 no_demux:
1574   {
1575     gst_object_unref (buffer);
1576     g_warning ("rtpbin: could not create rtpptdemux element");
1577     return NULL;
1578   }
1579 }
1580
1581 /* called with RTP_BIN_LOCK */
1582 static void
1583 free_stream (GstRtpBinStream * stream, GstRtpBin * bin)
1584 {
1585   GSList *clients, *next_client;
1586
1587   GST_DEBUG_OBJECT (bin, "freeing stream %p", stream);
1588
1589   if (stream->demux) {
1590     g_signal_handler_disconnect (stream->demux, stream->demux_newpad_sig);
1591     g_signal_handler_disconnect (stream->demux, stream->demux_ptreq_sig);
1592     g_signal_handler_disconnect (stream->demux, stream->demux_ptchange_sig);
1593   }
1594   g_signal_handler_disconnect (stream->buffer, stream->buffer_handlesync_sig);
1595   g_signal_handler_disconnect (stream->buffer, stream->buffer_ptreq_sig);
1596   g_signal_handler_disconnect (stream->buffer, stream->buffer_ntpstop_sig);
1597
1598   if (stream->demux)
1599     gst_element_set_locked_state (stream->demux, TRUE);
1600   gst_element_set_locked_state (stream->buffer, TRUE);
1601
1602   if (stream->demux)
1603     gst_element_set_state (stream->demux, GST_STATE_NULL);
1604   gst_element_set_state (stream->buffer, GST_STATE_NULL);
1605
1606   /* now remove this signal, we need this while going to NULL because it to
1607    * do some cleanups */
1608   if (stream->demux)
1609     g_signal_handler_disconnect (stream->demux, stream->demux_padremoved_sig);
1610
1611   gst_bin_remove (GST_BIN_CAST (bin), stream->buffer);
1612   if (stream->demux)
1613     gst_bin_remove (GST_BIN_CAST (bin), stream->demux);
1614
1615   for (clients = bin->clients; clients; clients = next_client) {
1616     GstRtpBinClient *client = (GstRtpBinClient *) clients->data;
1617     GSList *streams, *next_stream;
1618
1619     next_client = g_slist_next (clients);
1620
1621     for (streams = client->streams; streams; streams = next_stream) {
1622       GstRtpBinStream *ostream = (GstRtpBinStream *) streams->data;
1623
1624       next_stream = g_slist_next (streams);
1625
1626       if (ostream == stream) {
1627         client->streams = g_slist_delete_link (client->streams, streams);
1628         /* If this was the last stream belonging to this client,
1629          * clean up the client. */
1630         if (--client->nstreams == 0) {
1631           bin->clients = g_slist_delete_link (bin->clients, clients);
1632           free_client (client, bin);
1633           break;
1634         }
1635       }
1636     }
1637   }
1638   g_free (stream);
1639 }
1640
1641 /* GObject vmethods */
1642 static void gst_rtp_bin_dispose (GObject * object);
1643 static void gst_rtp_bin_finalize (GObject * object);
1644 static void gst_rtp_bin_set_property (GObject * object, guint prop_id,
1645     const GValue * value, GParamSpec * pspec);
1646 static void gst_rtp_bin_get_property (GObject * object, guint prop_id,
1647     GValue * value, GParamSpec * pspec);
1648
1649 /* GstElement vmethods */
1650 static GstStateChangeReturn gst_rtp_bin_change_state (GstElement * element,
1651     GstStateChange transition);
1652 static GstPad *gst_rtp_bin_request_new_pad (GstElement * element,
1653     GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
1654 static void gst_rtp_bin_release_pad (GstElement * element, GstPad * pad);
1655 static void gst_rtp_bin_handle_message (GstBin * bin, GstMessage * message);
1656
1657 #define gst_rtp_bin_parent_class parent_class
1658 G_DEFINE_TYPE (GstRtpBin, gst_rtp_bin, GST_TYPE_BIN);
1659
1660 static gboolean
1661 _gst_element_accumulator (GSignalInvocationHint * ihint,
1662     GValue * return_accu, const GValue * handler_return, gpointer dummy)
1663 {
1664   GstElement *element;
1665
1666   element = g_value_get_object (handler_return);
1667   GST_DEBUG ("got element %" GST_PTR_FORMAT, element);
1668
1669   if (!(ihint->run_type & G_SIGNAL_RUN_CLEANUP))
1670     g_value_set_object (return_accu, element);
1671
1672   /* stop emission if we have an element */
1673   return (element == NULL);
1674 }
1675
1676 static gboolean
1677 _gst_caps_accumulator (GSignalInvocationHint * ihint,
1678     GValue * return_accu, const GValue * handler_return, gpointer dummy)
1679 {
1680   GstCaps *caps;
1681
1682   caps = g_value_get_boxed (handler_return);
1683   GST_DEBUG ("got caps %" GST_PTR_FORMAT, caps);
1684
1685   if (!(ihint->run_type & G_SIGNAL_RUN_CLEANUP))
1686     g_value_set_boxed (return_accu, caps);
1687
1688   /* stop emission if we have a caps */
1689   return (caps == NULL);
1690 }
1691
1692 static void
1693 gst_rtp_bin_class_init (GstRtpBinClass * klass)
1694 {
1695   GObjectClass *gobject_class;
1696   GstElementClass *gstelement_class;
1697   GstBinClass *gstbin_class;
1698
1699   gobject_class = (GObjectClass *) klass;
1700   gstelement_class = (GstElementClass *) klass;
1701   gstbin_class = (GstBinClass *) klass;
1702
1703   g_type_class_add_private (klass, sizeof (GstRtpBinPrivate));
1704
1705   gobject_class->dispose = gst_rtp_bin_dispose;
1706   gobject_class->finalize = gst_rtp_bin_finalize;
1707   gobject_class->set_property = gst_rtp_bin_set_property;
1708   gobject_class->get_property = gst_rtp_bin_get_property;
1709
1710   g_object_class_install_property (gobject_class, PROP_LATENCY,
1711       g_param_spec_uint ("latency", "Buffer latency in ms",
1712           "Default amount of ms to buffer in the jitterbuffers", 0,
1713           G_MAXUINT, DEFAULT_LATENCY_MS,
1714           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
1715
1716   g_object_class_install_property (gobject_class, PROP_DROP_ON_LATENCY,
1717       g_param_spec_boolean ("drop-on-latency",
1718           "Drop buffers when maximum latency is reached",
1719           "Tells the jitterbuffer to never exceed the given latency in size",
1720           DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
1721
1722   /**
1723    * GstRtpBin::request-pt-map:
1724    * @rtpbin: the object which received the signal
1725    * @session: the session
1726    * @pt: the pt
1727    *
1728    * Request the payload type as #GstCaps for @pt in @session.
1729    */
1730   gst_rtp_bin_signals[SIGNAL_REQUEST_PT_MAP] =
1731       g_signal_new ("request-pt-map", G_TYPE_FROM_CLASS (klass),
1732       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, request_pt_map),
1733       _gst_caps_accumulator, NULL, g_cclosure_marshal_generic, GST_TYPE_CAPS,
1734       2, G_TYPE_UINT, G_TYPE_UINT);
1735
1736     /**
1737    * GstRtpBin::payload-type-change:
1738    * @rtpbin: the object which received the signal
1739    * @session: the session
1740    * @pt: the pt
1741    *
1742    * Signal that the current payload type changed to @pt in @session.
1743    */
1744   gst_rtp_bin_signals[SIGNAL_PAYLOAD_TYPE_CHANGE] =
1745       g_signal_new ("payload-type-change", G_TYPE_FROM_CLASS (klass),
1746       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, payload_type_change),
1747       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1748       G_TYPE_UINT);
1749
1750   /**
1751    * GstRtpBin::clear-pt-map:
1752    * @rtpbin: the object which received the signal
1753    *
1754    * Clear all previously cached pt-mapping obtained with
1755    * #GstRtpBin::request-pt-map.
1756    */
1757   gst_rtp_bin_signals[SIGNAL_CLEAR_PT_MAP] =
1758       g_signal_new ("clear-pt-map", G_TYPE_FROM_CLASS (klass),
1759       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpBinClass,
1760           clear_pt_map), NULL, NULL, g_cclosure_marshal_VOID__VOID, G_TYPE_NONE,
1761       0, G_TYPE_NONE);
1762
1763   /**
1764    * GstRtpBin::reset-sync:
1765    * @rtpbin: the object which received the signal
1766    *
1767    * Reset all currently configured lip-sync parameters and require new SR
1768    * packets for all streams before lip-sync is attempted again.
1769    */
1770   gst_rtp_bin_signals[SIGNAL_RESET_SYNC] =
1771       g_signal_new ("reset-sync", G_TYPE_FROM_CLASS (klass),
1772       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpBinClass,
1773           reset_sync), NULL, NULL, g_cclosure_marshal_VOID__VOID, G_TYPE_NONE,
1774       0, G_TYPE_NONE);
1775
1776   /**
1777    * GstRtpBin::get-internal-session:
1778    * @rtpbin: the object which received the signal
1779    * @id: the session id
1780    *
1781    * Request the internal RTPSession object as #GObject in session @id.
1782    */
1783   gst_rtp_bin_signals[SIGNAL_GET_INTERNAL_SESSION] =
1784       g_signal_new ("get-internal-session", G_TYPE_FROM_CLASS (klass),
1785       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpBinClass,
1786           get_internal_session), NULL, NULL, g_cclosure_marshal_generic,
1787       RTP_TYPE_SESSION, 1, G_TYPE_UINT);
1788
1789   /**
1790    * GstRtpBin::on-new-ssrc:
1791    * @rtpbin: the object which received the signal
1792    * @session: the session
1793    * @ssrc: the SSRC
1794    *
1795    * Notify of a new SSRC that entered @session.
1796    */
1797   gst_rtp_bin_signals[SIGNAL_ON_NEW_SSRC] =
1798       g_signal_new ("on-new-ssrc", G_TYPE_FROM_CLASS (klass),
1799       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_new_ssrc),
1800       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1801       G_TYPE_UINT);
1802   /**
1803    * GstRtpBin::on-ssrc-collision:
1804    * @rtpbin: the object which received the signal
1805    * @session: the session
1806    * @ssrc: the SSRC
1807    *
1808    * Notify when we have an SSRC collision
1809    */
1810   gst_rtp_bin_signals[SIGNAL_ON_SSRC_COLLISION] =
1811       g_signal_new ("on-ssrc-collision", G_TYPE_FROM_CLASS (klass),
1812       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_ssrc_collision),
1813       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1814       G_TYPE_UINT);
1815   /**
1816    * GstRtpBin::on-ssrc-validated:
1817    * @rtpbin: the object which received the signal
1818    * @session: the session
1819    * @ssrc: the SSRC
1820    *
1821    * Notify of a new SSRC that became validated.
1822    */
1823   gst_rtp_bin_signals[SIGNAL_ON_SSRC_VALIDATED] =
1824       g_signal_new ("on-ssrc-validated", G_TYPE_FROM_CLASS (klass),
1825       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_ssrc_validated),
1826       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1827       G_TYPE_UINT);
1828   /**
1829    * GstRtpBin::on-ssrc-active:
1830    * @rtpbin: the object which received the signal
1831    * @session: the session
1832    * @ssrc: the SSRC
1833    *
1834    * Notify of a SSRC that is active, i.e., sending RTCP.
1835    */
1836   gst_rtp_bin_signals[SIGNAL_ON_SSRC_ACTIVE] =
1837       g_signal_new ("on-ssrc-active", G_TYPE_FROM_CLASS (klass),
1838       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_ssrc_active),
1839       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1840       G_TYPE_UINT);
1841   /**
1842    * GstRtpBin::on-ssrc-sdes:
1843    * @rtpbin: the object which received the signal
1844    * @session: the session
1845    * @ssrc: the SSRC
1846    *
1847    * Notify of a SSRC that is active, i.e., sending RTCP.
1848    */
1849   gst_rtp_bin_signals[SIGNAL_ON_SSRC_SDES] =
1850       g_signal_new ("on-ssrc-sdes", G_TYPE_FROM_CLASS (klass),
1851       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_ssrc_sdes),
1852       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1853       G_TYPE_UINT);
1854
1855   /**
1856    * GstRtpBin::on-bye-ssrc:
1857    * @rtpbin: the object which received the signal
1858    * @session: the session
1859    * @ssrc: the SSRC
1860    *
1861    * Notify of an SSRC that became inactive because of a BYE packet.
1862    */
1863   gst_rtp_bin_signals[SIGNAL_ON_BYE_SSRC] =
1864       g_signal_new ("on-bye-ssrc", G_TYPE_FROM_CLASS (klass),
1865       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_bye_ssrc),
1866       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1867       G_TYPE_UINT);
1868   /**
1869    * GstRtpBin::on-bye-timeout:
1870    * @rtpbin: the object which received the signal
1871    * @session: the session
1872    * @ssrc: the SSRC
1873    *
1874    * Notify of an SSRC that has timed out because of BYE
1875    */
1876   gst_rtp_bin_signals[SIGNAL_ON_BYE_TIMEOUT] =
1877       g_signal_new ("on-bye-timeout", G_TYPE_FROM_CLASS (klass),
1878       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_bye_timeout),
1879       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1880       G_TYPE_UINT);
1881   /**
1882    * GstRtpBin::on-timeout:
1883    * @rtpbin: the object which received the signal
1884    * @session: the session
1885    * @ssrc: the SSRC
1886    *
1887    * Notify of an SSRC that has timed out
1888    */
1889   gst_rtp_bin_signals[SIGNAL_ON_TIMEOUT] =
1890       g_signal_new ("on-timeout", G_TYPE_FROM_CLASS (klass),
1891       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_timeout),
1892       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1893       G_TYPE_UINT);
1894   /**
1895    * GstRtpBin::on-sender-timeout:
1896    * @rtpbin: the object which received the signal
1897    * @session: the session
1898    * @ssrc: the SSRC
1899    *
1900    * Notify of a sender SSRC that has timed out and became a receiver
1901    */
1902   gst_rtp_bin_signals[SIGNAL_ON_SENDER_TIMEOUT] =
1903       g_signal_new ("on-sender-timeout", G_TYPE_FROM_CLASS (klass),
1904       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_sender_timeout),
1905       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1906       G_TYPE_UINT);
1907
1908   /**
1909    * GstRtpBin::on-npt-stop:
1910    * @rtpbin: the object which received the signal
1911    * @session: the session
1912    * @ssrc: the SSRC
1913    *
1914    * Notify that SSRC sender has sent data up to the configured NPT stop time.
1915    */
1916   gst_rtp_bin_signals[SIGNAL_ON_NPT_STOP] =
1917       g_signal_new ("on-npt-stop", G_TYPE_FROM_CLASS (klass),
1918       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_npt_stop),
1919       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
1920       G_TYPE_UINT);
1921
1922   /**
1923    * GstRtpBin::request-rtp-encoder:
1924    * @rtpbin: the object which received the signal
1925    * @session: the session
1926    *
1927    * Request an RTP encoder element for the given @session. The encoder
1928    * element will be added to the bin if not previously added.
1929    *
1930    * If no handler is connected, no encoder will be used.
1931    *
1932    * Since: 1.4
1933    */
1934   gst_rtp_bin_signals[SIGNAL_REQUEST_RTP_ENCODER] =
1935       g_signal_new ("request-rtp-encoder", G_TYPE_FROM_CLASS (klass),
1936       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
1937           request_rtp_encoder), _gst_element_accumulator, NULL,
1938       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
1939
1940   /**
1941    * GstRtpBin::request-rtp-decoder:
1942    * @rtpbin: the object which received the signal
1943    * @session: the session
1944    *
1945    * Request an RTP decoder element for the given @session. The decoder
1946    * element will be added to the bin if not previously added.
1947    *
1948    * If no handler is connected, no encoder will be used.
1949    *
1950    * Since: 1.4
1951    */
1952   gst_rtp_bin_signals[SIGNAL_REQUEST_RTP_DECODER] =
1953       g_signal_new ("request-rtp-decoder", G_TYPE_FROM_CLASS (klass),
1954       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
1955           request_rtp_decoder), _gst_element_accumulator, NULL,
1956       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
1957
1958   /**
1959    * GstRtpBin::request-rtcp-encoder:
1960    * @rtpbin: the object which received the signal
1961    * @session: the session
1962    *
1963    * Request an RTCP encoder element for the given @session. The encoder
1964    * element will be added to the bin if not previously added.
1965    *
1966    * If no handler is connected, no encoder will be used.
1967    *
1968    * Since: 1.4
1969    */
1970   gst_rtp_bin_signals[SIGNAL_REQUEST_RTCP_ENCODER] =
1971       g_signal_new ("request-rtcp-encoder", G_TYPE_FROM_CLASS (klass),
1972       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
1973           request_rtcp_encoder), _gst_element_accumulator, NULL,
1974       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
1975
1976   /**
1977    * GstRtpBin::request-rtcp-decoder:
1978    * @rtpbin: the object which received the signal
1979    * @session: the session
1980    *
1981    * Request an RTCP decoder element for the given @session. The decoder
1982    * element will be added to the bin if not previously added.
1983    *
1984    * If no handler is connected, no encoder will be used.
1985    *
1986    * Since: 1.4
1987    */
1988   gst_rtp_bin_signals[SIGNAL_REQUEST_RTCP_DECODER] =
1989       g_signal_new ("request-rtcp-decoder", G_TYPE_FROM_CLASS (klass),
1990       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
1991           request_rtcp_decoder), _gst_element_accumulator, NULL,
1992       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
1993
1994   /**
1995    * GstRtpBin::new-jitterbuffer:
1996    * @rtpbin: the object which received the signal
1997    * @jitterbuffer: the new jitterbuffer
1998    * @session: the session
1999    * @ssrc: the SSRC
2000    *
2001    * Notify that a new @jitterbuffer was created for @session and @ssrc.
2002    * This signal can, for example, be used to configure @jitterbuffer.
2003    *
2004    * Since: 1.4
2005    */
2006   gst_rtp_bin_signals[SIGNAL_NEW_JITTERBUFFER] =
2007       g_signal_new ("new-jitterbuffer", G_TYPE_FROM_CLASS (klass),
2008       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2009           new_jitterbuffer), NULL, NULL, g_cclosure_marshal_generic,
2010       G_TYPE_NONE, 3, GST_TYPE_ELEMENT, G_TYPE_UINT, G_TYPE_UINT);
2011
2012   /**
2013    * GstRtpBin::request-aux-sender:
2014    * @rtpbin: the object which received the signal
2015    * @session: the session
2016    *
2017    * Request an AUX sender element for the given @session. The AUX
2018    * element will be added to the bin.
2019    *
2020    * If no handler is connected, no AUX element will be used.
2021    *
2022    * Since: 1.4
2023    */
2024   gst_rtp_bin_signals[SIGNAL_REQUEST_AUX_SENDER] =
2025       g_signal_new ("request-aux-sender", G_TYPE_FROM_CLASS (klass),
2026       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2027           request_aux_sender), _gst_element_accumulator, NULL,
2028       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2029   /**
2030    * GstRtpBin::request-aux-receiver:
2031    * @rtpbin: the object which received the signal
2032    * @session: the session
2033    *
2034    * Request an AUX receiver element for the given @session. The AUX
2035    * element will be added to the bin.
2036    *
2037    * If no handler is connected, no AUX element will be used.
2038    *
2039    * Since: 1.4
2040    */
2041   gst_rtp_bin_signals[SIGNAL_REQUEST_AUX_RECEIVER] =
2042       g_signal_new ("request-aux-receiver", G_TYPE_FROM_CLASS (klass),
2043       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2044           request_aux_receiver), _gst_element_accumulator, NULL,
2045       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2046
2047   g_object_class_install_property (gobject_class, PROP_SDES,
2048       g_param_spec_boxed ("sdes", "SDES",
2049           "The SDES items of this session",
2050           GST_TYPE_STRUCTURE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2051
2052   g_object_class_install_property (gobject_class, PROP_DO_LOST,
2053       g_param_spec_boolean ("do-lost", "Do Lost",
2054           "Send an event downstream when a packet is lost", DEFAULT_DO_LOST,
2055           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2056
2057   g_object_class_install_property (gobject_class, PROP_AUTOREMOVE,
2058       g_param_spec_boolean ("autoremove", "Auto Remove",
2059           "Automatically remove timed out sources", DEFAULT_AUTOREMOVE,
2060           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2061
2062   g_object_class_install_property (gobject_class, PROP_IGNORE_PT,
2063       g_param_spec_boolean ("ignore-pt", "Ignore PT",
2064           "Do not demultiplex based on PT values", DEFAULT_IGNORE_PT,
2065           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2066
2067   g_object_class_install_property (gobject_class, PROP_USE_PIPELINE_CLOCK,
2068       g_param_spec_boolean ("use-pipeline-clock", "Use pipeline clock",
2069           "Use the pipeline running-time to set the NTP time in the RTCP SR messages",
2070           DEFAULT_USE_PIPELINE_CLOCK,
2071           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2072   /**
2073    * GstRtpBin:buffer-mode:
2074    *
2075    * Control the buffering and timestamping mode used by the jitterbuffer.
2076    */
2077   g_object_class_install_property (gobject_class, PROP_BUFFER_MODE,
2078       g_param_spec_enum ("buffer-mode", "Buffer Mode",
2079           "Control the buffering algorithm in use", RTP_TYPE_JITTER_BUFFER_MODE,
2080           DEFAULT_BUFFER_MODE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2081   /**
2082    * GstRtpBin:ntp-sync:
2083    *
2084    * Set the NTP time from the sender reports as the running-time on the
2085    * buffers. When both the sender and receiver have sychronized
2086    * running-time, i.e. when the clock and base-time is shared
2087    * between the receivers and the and the senders, this option can be
2088    * used to synchronize receivers on multiple machines.
2089    */
2090   g_object_class_install_property (gobject_class, PROP_NTP_SYNC,
2091       g_param_spec_boolean ("ntp-sync", "Sync on NTP clock",
2092           "Synchronize received streams to the NTP clock", DEFAULT_NTP_SYNC,
2093           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2094
2095   /**
2096    * GstRtpBin:rtcp-sync:
2097    *
2098    * If not synchronizing (directly) to the NTP clock, determines how to sync
2099    * the various streams.
2100    */
2101   g_object_class_install_property (gobject_class, PROP_RTCP_SYNC,
2102       g_param_spec_enum ("rtcp-sync", "RTCP Sync",
2103           "Use of RTCP SR in synchronization", GST_RTP_BIN_RTCP_SYNC_TYPE,
2104           DEFAULT_RTCP_SYNC, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2105
2106   /**
2107    * GstRtpBin:rtcp-sync-interval:
2108    *
2109    * Determines how often to sync streams using RTCP data.
2110    */
2111   g_object_class_install_property (gobject_class, PROP_RTCP_SYNC_INTERVAL,
2112       g_param_spec_uint ("rtcp-sync-interval", "RTCP Sync Interval",
2113           "RTCP SR interval synchronization (ms) (0 = always)",
2114           0, G_MAXUINT, DEFAULT_RTCP_SYNC_INTERVAL,
2115           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2116
2117   g_object_class_install_property (gobject_class, PROP_DO_SYNC_EVENT,
2118       g_param_spec_boolean ("do-sync-event", "Do Sync Event",
2119           "Send event downstream when a stream is synchronized to the sender",
2120           DEFAULT_DO_SYNC_EVENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2121
2122   g_object_class_install_property (gobject_class, PROP_DO_RETRANSMISSION,
2123       g_param_spec_boolean ("do-retransmission", "Do retransmission",
2124           "Send an event downstream to request packet retransmission",
2125           DEFAULT_DO_RETRANSMISSION,
2126           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2127
2128   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_rtp_bin_change_state);
2129   gstelement_class->request_new_pad =
2130       GST_DEBUG_FUNCPTR (gst_rtp_bin_request_new_pad);
2131   gstelement_class->release_pad = GST_DEBUG_FUNCPTR (gst_rtp_bin_release_pad);
2132
2133   /* sink pads */
2134   gst_element_class_add_pad_template (gstelement_class,
2135       gst_static_pad_template_get (&rtpbin_recv_rtp_sink_template));
2136   gst_element_class_add_pad_template (gstelement_class,
2137       gst_static_pad_template_get (&rtpbin_recv_rtcp_sink_template));
2138   gst_element_class_add_pad_template (gstelement_class,
2139       gst_static_pad_template_get (&rtpbin_send_rtp_sink_template));
2140
2141   /* src pads */
2142   gst_element_class_add_pad_template (gstelement_class,
2143       gst_static_pad_template_get (&rtpbin_recv_rtp_src_template));
2144   gst_element_class_add_pad_template (gstelement_class,
2145       gst_static_pad_template_get (&rtpbin_send_rtcp_src_template));
2146   gst_element_class_add_pad_template (gstelement_class,
2147       gst_static_pad_template_get (&rtpbin_send_rtp_src_template));
2148
2149   gst_element_class_set_static_metadata (gstelement_class, "RTP Bin",
2150       "Filter/Network/RTP",
2151       "Real-Time Transport Protocol bin",
2152       "Wim Taymans <wim.taymans@gmail.com>");
2153
2154   gstbin_class->handle_message = GST_DEBUG_FUNCPTR (gst_rtp_bin_handle_message);
2155
2156   klass->clear_pt_map = GST_DEBUG_FUNCPTR (gst_rtp_bin_clear_pt_map);
2157   klass->reset_sync = GST_DEBUG_FUNCPTR (gst_rtp_bin_reset_sync);
2158   klass->get_internal_session =
2159       GST_DEBUG_FUNCPTR (gst_rtp_bin_get_internal_session);
2160   klass->request_rtp_encoder = GST_DEBUG_FUNCPTR (gst_rtp_bin_request_encoder);
2161   klass->request_rtp_decoder = GST_DEBUG_FUNCPTR (gst_rtp_bin_request_decoder);
2162   klass->request_rtcp_encoder = GST_DEBUG_FUNCPTR (gst_rtp_bin_request_encoder);
2163   klass->request_rtcp_decoder = GST_DEBUG_FUNCPTR (gst_rtp_bin_request_decoder);
2164
2165   GST_DEBUG_CATEGORY_INIT (gst_rtp_bin_debug, "rtpbin", 0, "RTP bin");
2166 }
2167
2168 static void
2169 gst_rtp_bin_init (GstRtpBin * rtpbin)
2170 {
2171   gchar *cname;
2172
2173   rtpbin->priv = GST_RTP_BIN_GET_PRIVATE (rtpbin);
2174   g_mutex_init (&rtpbin->priv->bin_lock);
2175   g_mutex_init (&rtpbin->priv->dyn_lock);
2176
2177   rtpbin->latency_ms = DEFAULT_LATENCY_MS;
2178   rtpbin->latency_ns = DEFAULT_LATENCY_MS * GST_MSECOND;
2179   rtpbin->drop_on_latency = DEFAULT_DROP_ON_LATENCY;
2180   rtpbin->do_lost = DEFAULT_DO_LOST;
2181   rtpbin->ignore_pt = DEFAULT_IGNORE_PT;
2182   rtpbin->ntp_sync = DEFAULT_NTP_SYNC;
2183   rtpbin->rtcp_sync = DEFAULT_RTCP_SYNC;
2184   rtpbin->rtcp_sync_interval = DEFAULT_RTCP_SYNC_INTERVAL;
2185   rtpbin->priv->autoremove = DEFAULT_AUTOREMOVE;
2186   rtpbin->buffer_mode = DEFAULT_BUFFER_MODE;
2187   rtpbin->use_pipeline_clock = DEFAULT_USE_PIPELINE_CLOCK;
2188   rtpbin->send_sync_event = DEFAULT_DO_SYNC_EVENT;
2189   rtpbin->do_retransmission = DEFAULT_DO_RETRANSMISSION;
2190
2191   /* some default SDES entries */
2192   cname = g_strdup_printf ("user%u@host-%x", g_random_int (), g_random_int ());
2193   rtpbin->sdes = gst_structure_new ("application/x-rtp-source-sdes",
2194       "cname", G_TYPE_STRING, cname, "tool", G_TYPE_STRING, "GStreamer", NULL);
2195   g_free (cname);
2196 }
2197
2198 static void
2199 gst_rtp_bin_dispose (GObject * object)
2200 {
2201   GstRtpBin *rtpbin;
2202
2203   rtpbin = GST_RTP_BIN (object);
2204
2205   GST_RTP_BIN_LOCK (rtpbin);
2206   GST_DEBUG_OBJECT (object, "freeing sessions");
2207   g_slist_foreach (rtpbin->sessions, (GFunc) free_session, rtpbin);
2208   g_slist_free (rtpbin->sessions);
2209   rtpbin->sessions = NULL;
2210   GST_RTP_BIN_UNLOCK (rtpbin);
2211
2212   G_OBJECT_CLASS (parent_class)->dispose (object);
2213 }
2214
2215 static void
2216 gst_rtp_bin_finalize (GObject * object)
2217 {
2218   GstRtpBin *rtpbin;
2219
2220   rtpbin = GST_RTP_BIN (object);
2221
2222   if (rtpbin->sdes)
2223     gst_structure_free (rtpbin->sdes);
2224
2225   g_mutex_clear (&rtpbin->priv->bin_lock);
2226   g_mutex_clear (&rtpbin->priv->dyn_lock);
2227
2228   G_OBJECT_CLASS (parent_class)->finalize (object);
2229 }
2230
2231
2232 static void
2233 gst_rtp_bin_set_sdes_struct (GstRtpBin * bin, const GstStructure * sdes)
2234 {
2235   GSList *item;
2236
2237   if (sdes == NULL)
2238     return;
2239
2240   GST_RTP_BIN_LOCK (bin);
2241
2242   GST_OBJECT_LOCK (bin);
2243   if (bin->sdes)
2244     gst_structure_free (bin->sdes);
2245   bin->sdes = gst_structure_copy (sdes);
2246   GST_OBJECT_UNLOCK (bin);
2247
2248   /* store in all sessions */
2249   for (item = bin->sessions; item; item = g_slist_next (item)) {
2250     GstRtpBinSession *session = item->data;
2251     g_object_set (session->session, "sdes", sdes, NULL);
2252   }
2253
2254   GST_RTP_BIN_UNLOCK (bin);
2255 }
2256
2257 static GstStructure *
2258 gst_rtp_bin_get_sdes_struct (GstRtpBin * bin)
2259 {
2260   GstStructure *result;
2261
2262   GST_OBJECT_LOCK (bin);
2263   result = gst_structure_copy (bin->sdes);
2264   GST_OBJECT_UNLOCK (bin);
2265
2266   return result;
2267 }
2268
2269 static void
2270 gst_rtp_bin_set_property (GObject * object, guint prop_id,
2271     const GValue * value, GParamSpec * pspec)
2272 {
2273   GstRtpBin *rtpbin;
2274
2275   rtpbin = GST_RTP_BIN (object);
2276
2277   switch (prop_id) {
2278     case PROP_LATENCY:
2279       GST_RTP_BIN_LOCK (rtpbin);
2280       rtpbin->latency_ms = g_value_get_uint (value);
2281       rtpbin->latency_ns = rtpbin->latency_ms * GST_MSECOND;
2282       GST_RTP_BIN_UNLOCK (rtpbin);
2283       /* propagate the property down to the jitterbuffer */
2284       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin, "latency", value);
2285       break;
2286     case PROP_DROP_ON_LATENCY:
2287       GST_RTP_BIN_LOCK (rtpbin);
2288       rtpbin->drop_on_latency = g_value_get_boolean (value);
2289       GST_RTP_BIN_UNLOCK (rtpbin);
2290       /* propagate the property down to the jitterbuffer */
2291       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
2292           "drop-on-latency", value);
2293       break;
2294     case PROP_SDES:
2295       gst_rtp_bin_set_sdes_struct (rtpbin, g_value_get_boxed (value));
2296       break;
2297     case PROP_DO_LOST:
2298       GST_RTP_BIN_LOCK (rtpbin);
2299       rtpbin->do_lost = g_value_get_boolean (value);
2300       GST_RTP_BIN_UNLOCK (rtpbin);
2301       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin, "do-lost", value);
2302       break;
2303     case PROP_NTP_SYNC:
2304       rtpbin->ntp_sync = g_value_get_boolean (value);
2305       break;
2306     case PROP_RTCP_SYNC:
2307       g_atomic_int_set (&rtpbin->rtcp_sync, g_value_get_enum (value));
2308       break;
2309     case PROP_RTCP_SYNC_INTERVAL:
2310       rtpbin->rtcp_sync_interval = g_value_get_uint (value);
2311       break;
2312     case PROP_IGNORE_PT:
2313       rtpbin->ignore_pt = g_value_get_boolean (value);
2314       break;
2315     case PROP_AUTOREMOVE:
2316       rtpbin->priv->autoremove = g_value_get_boolean (value);
2317       break;
2318     case PROP_USE_PIPELINE_CLOCK:
2319     {
2320       GSList *sessions;
2321       GST_RTP_BIN_LOCK (rtpbin);
2322       rtpbin->use_pipeline_clock = g_value_get_boolean (value);
2323       for (sessions = rtpbin->sessions; sessions;
2324           sessions = g_slist_next (sessions)) {
2325         GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
2326
2327         g_object_set (G_OBJECT (session->session),
2328             "use-pipeline-clock", rtpbin->use_pipeline_clock, NULL);
2329       }
2330       GST_RTP_BIN_UNLOCK (rtpbin);
2331     }
2332       break;
2333     case PROP_DO_SYNC_EVENT:
2334       rtpbin->send_sync_event = g_value_get_boolean (value);
2335       break;
2336     case PROP_BUFFER_MODE:
2337       GST_RTP_BIN_LOCK (rtpbin);
2338       rtpbin->buffer_mode = g_value_get_enum (value);
2339       GST_RTP_BIN_UNLOCK (rtpbin);
2340       /* propagate the property down to the jitterbuffer */
2341       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin, "mode", value);
2342       break;
2343     case PROP_DO_RETRANSMISSION:
2344       GST_RTP_BIN_LOCK (rtpbin);
2345       rtpbin->do_retransmission = g_value_get_boolean (value);
2346       GST_RTP_BIN_UNLOCK (rtpbin);
2347       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
2348           "do-retransmission", value);
2349       break;
2350     default:
2351       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2352       break;
2353   }
2354 }
2355
2356 static void
2357 gst_rtp_bin_get_property (GObject * object, guint prop_id,
2358     GValue * value, GParamSpec * pspec)
2359 {
2360   GstRtpBin *rtpbin;
2361
2362   rtpbin = GST_RTP_BIN (object);
2363
2364   switch (prop_id) {
2365     case PROP_LATENCY:
2366       GST_RTP_BIN_LOCK (rtpbin);
2367       g_value_set_uint (value, rtpbin->latency_ms);
2368       GST_RTP_BIN_UNLOCK (rtpbin);
2369       break;
2370     case PROP_DROP_ON_LATENCY:
2371       GST_RTP_BIN_LOCK (rtpbin);
2372       g_value_set_boolean (value, rtpbin->drop_on_latency);
2373       GST_RTP_BIN_UNLOCK (rtpbin);
2374       break;
2375     case PROP_SDES:
2376       g_value_take_boxed (value, gst_rtp_bin_get_sdes_struct (rtpbin));
2377       break;
2378     case PROP_DO_LOST:
2379       GST_RTP_BIN_LOCK (rtpbin);
2380       g_value_set_boolean (value, rtpbin->do_lost);
2381       GST_RTP_BIN_UNLOCK (rtpbin);
2382       break;
2383     case PROP_IGNORE_PT:
2384       g_value_set_boolean (value, rtpbin->ignore_pt);
2385       break;
2386     case PROP_NTP_SYNC:
2387       g_value_set_boolean (value, rtpbin->ntp_sync);
2388       break;
2389     case PROP_RTCP_SYNC:
2390       g_value_set_enum (value, g_atomic_int_get (&rtpbin->rtcp_sync));
2391       break;
2392     case PROP_RTCP_SYNC_INTERVAL:
2393       g_value_set_uint (value, rtpbin->rtcp_sync_interval);
2394       break;
2395     case PROP_AUTOREMOVE:
2396       g_value_set_boolean (value, rtpbin->priv->autoremove);
2397       break;
2398     case PROP_BUFFER_MODE:
2399       g_value_set_enum (value, rtpbin->buffer_mode);
2400       break;
2401     case PROP_USE_PIPELINE_CLOCK:
2402       g_value_set_boolean (value, rtpbin->use_pipeline_clock);
2403       break;
2404     case PROP_DO_SYNC_EVENT:
2405       g_value_set_boolean (value, rtpbin->send_sync_event);
2406       break;
2407     case PROP_DO_RETRANSMISSION:
2408       GST_RTP_BIN_LOCK (rtpbin);
2409       g_value_set_boolean (value, rtpbin->do_retransmission);
2410       GST_RTP_BIN_UNLOCK (rtpbin);
2411       break;
2412     default:
2413       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2414       break;
2415   }
2416 }
2417
2418 static void
2419 gst_rtp_bin_handle_message (GstBin * bin, GstMessage * message)
2420 {
2421   GstRtpBin *rtpbin;
2422
2423   rtpbin = GST_RTP_BIN (bin);
2424
2425   switch (GST_MESSAGE_TYPE (message)) {
2426     case GST_MESSAGE_ELEMENT:
2427     {
2428       const GstStructure *s = gst_message_get_structure (message);
2429
2430       /* we change the structure name and add the session ID to it */
2431       if (gst_structure_has_name (s, "application/x-rtp-source-sdes")) {
2432         GstRtpBinSession *sess;
2433
2434         /* find the session we set it as object data */
2435         sess = g_object_get_data (G_OBJECT (GST_MESSAGE_SRC (message)),
2436             "GstRTPBin.session");
2437
2438         if (G_LIKELY (sess)) {
2439           message = gst_message_make_writable (message);
2440           s = gst_message_get_structure (message);
2441           gst_structure_set ((GstStructure *) s, "session", G_TYPE_UINT,
2442               sess->id, NULL);
2443         }
2444       }
2445       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
2446       break;
2447     }
2448     case GST_MESSAGE_BUFFERING:
2449     {
2450       gint percent;
2451       gint min_percent = 100;
2452       GSList *sessions, *streams;
2453       GstRtpBinStream *stream;
2454       gboolean change = FALSE, active = FALSE;
2455       GstClockTime min_out_time;
2456       GstBufferingMode mode;
2457       gint avg_in, avg_out;
2458       gint64 buffering_left;
2459
2460       gst_message_parse_buffering (message, &percent);
2461       gst_message_parse_buffering_stats (message, &mode, &avg_in, &avg_out,
2462           &buffering_left);
2463
2464       stream =
2465           g_object_get_data (G_OBJECT (GST_MESSAGE_SRC (message)),
2466           "GstRTPBin.stream");
2467
2468       GST_DEBUG_OBJECT (bin, "got percent %d from stream %p", percent, stream);
2469
2470       /* get the stream */
2471       if (G_LIKELY (stream)) {
2472         GST_RTP_BIN_LOCK (rtpbin);
2473         /* fill in the percent */
2474         stream->percent = percent;
2475
2476         /* calculate the min value for all streams */
2477         for (sessions = rtpbin->sessions; sessions;
2478             sessions = g_slist_next (sessions)) {
2479           GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
2480
2481           GST_RTP_SESSION_LOCK (session);
2482           if (session->streams) {
2483             for (streams = session->streams; streams;
2484                 streams = g_slist_next (streams)) {
2485               GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
2486
2487               GST_DEBUG_OBJECT (bin, "stream %p percent %d", stream,
2488                   stream->percent);
2489
2490               /* find min percent */
2491               if (min_percent > stream->percent)
2492                 min_percent = stream->percent;
2493             }
2494           } else {
2495             GST_INFO_OBJECT (bin,
2496                 "session has no streams, setting min_percent to 0");
2497             min_percent = 0;
2498           }
2499           GST_RTP_SESSION_UNLOCK (session);
2500         }
2501         GST_DEBUG_OBJECT (bin, "min percent %d", min_percent);
2502
2503         if (rtpbin->buffering) {
2504           if (min_percent == 100) {
2505             rtpbin->buffering = FALSE;
2506             active = TRUE;
2507             change = TRUE;
2508           }
2509         } else {
2510           if (min_percent < 100) {
2511             /* pause the streams */
2512             rtpbin->buffering = TRUE;
2513             active = FALSE;
2514             change = TRUE;
2515           }
2516         }
2517         GST_RTP_BIN_UNLOCK (rtpbin);
2518
2519         gst_message_unref (message);
2520
2521         /* make a new buffering message with the min value */
2522         message =
2523             gst_message_new_buffering (GST_OBJECT_CAST (bin), min_percent);
2524         gst_message_set_buffering_stats (message, mode, avg_in, avg_out,
2525             buffering_left);
2526
2527         if (G_UNLIKELY (change)) {
2528           GstClock *clock;
2529           guint64 running_time = 0;
2530           guint64 offset = 0;
2531
2532           /* figure out the running time when we have a clock */
2533           if (G_LIKELY ((clock =
2534                       gst_element_get_clock (GST_ELEMENT_CAST (bin))))) {
2535             guint64 now, base_time;
2536
2537             now = gst_clock_get_time (clock);
2538             base_time = gst_element_get_base_time (GST_ELEMENT_CAST (bin));
2539             running_time = now - base_time;
2540             gst_object_unref (clock);
2541           }
2542           GST_DEBUG_OBJECT (bin,
2543               "running time now %" GST_TIME_FORMAT,
2544               GST_TIME_ARGS (running_time));
2545
2546           GST_RTP_BIN_LOCK (rtpbin);
2547
2548           /* when we reactivate, calculate the offsets so that all streams have
2549            * an output time that is at least as big as the running_time */
2550           offset = 0;
2551           if (active) {
2552             if (running_time > rtpbin->buffer_start) {
2553               offset = running_time - rtpbin->buffer_start;
2554               if (offset >= rtpbin->latency_ns)
2555                 offset -= rtpbin->latency_ns;
2556               else
2557                 offset = 0;
2558             }
2559           }
2560
2561           /* pause all streams */
2562           min_out_time = -1;
2563           for (sessions = rtpbin->sessions; sessions;
2564               sessions = g_slist_next (sessions)) {
2565             GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
2566
2567             GST_RTP_SESSION_LOCK (session);
2568             for (streams = session->streams; streams;
2569                 streams = g_slist_next (streams)) {
2570               GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
2571               GstElement *element = stream->buffer;
2572               guint64 last_out;
2573
2574               g_signal_emit_by_name (element, "set-active", active, offset,
2575                   &last_out);
2576
2577               if (!active) {
2578                 g_object_get (element, "percent", &stream->percent, NULL);
2579
2580                 if (last_out == -1)
2581                   last_out = 0;
2582                 if (min_out_time == -1 || last_out < min_out_time)
2583                   min_out_time = last_out;
2584               }
2585
2586               GST_DEBUG_OBJECT (bin,
2587                   "setting %p to %d, offset %" GST_TIME_FORMAT ", last %"
2588                   GST_TIME_FORMAT ", percent %d", element, active,
2589                   GST_TIME_ARGS (offset), GST_TIME_ARGS (last_out),
2590                   stream->percent);
2591             }
2592             GST_RTP_SESSION_UNLOCK (session);
2593           }
2594           GST_DEBUG_OBJECT (bin,
2595               "min out time %" GST_TIME_FORMAT, GST_TIME_ARGS (min_out_time));
2596
2597           /* the buffer_start is the min out time of all paused jitterbuffers */
2598           if (!active)
2599             rtpbin->buffer_start = min_out_time;
2600
2601           GST_RTP_BIN_UNLOCK (rtpbin);
2602         }
2603       }
2604       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
2605       break;
2606     }
2607     default:
2608     {
2609       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
2610       break;
2611     }
2612   }
2613 }
2614
2615 static GstStateChangeReturn
2616 gst_rtp_bin_change_state (GstElement * element, GstStateChange transition)
2617 {
2618   GstStateChangeReturn res;
2619   GstRtpBin *rtpbin;
2620   GstRtpBinPrivate *priv;
2621
2622   rtpbin = GST_RTP_BIN (element);
2623   priv = rtpbin->priv;
2624
2625   switch (transition) {
2626     case GST_STATE_CHANGE_NULL_TO_READY:
2627       break;
2628     case GST_STATE_CHANGE_READY_TO_PAUSED:
2629       priv->last_unix = 0;
2630       GST_LOG_OBJECT (rtpbin, "clearing shutdown flag");
2631       g_atomic_int_set (&priv->shutdown, 0);
2632       break;
2633     case GST_STATE_CHANGE_PAUSED_TO_READY:
2634       GST_LOG_OBJECT (rtpbin, "setting shutdown flag");
2635       g_atomic_int_set (&priv->shutdown, 1);
2636       /* wait for all callbacks to end by taking the lock. No new callbacks will
2637        * be able to happen as we set the shutdown flag. */
2638       GST_RTP_BIN_DYN_LOCK (rtpbin);
2639       GST_LOG_OBJECT (rtpbin, "dynamic lock taken, we can continue shutdown");
2640       GST_RTP_BIN_DYN_UNLOCK (rtpbin);
2641       break;
2642     default:
2643       break;
2644   }
2645
2646   res = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
2647
2648   switch (transition) {
2649     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
2650       break;
2651     case GST_STATE_CHANGE_PAUSED_TO_READY:
2652       break;
2653     case GST_STATE_CHANGE_READY_TO_NULL:
2654       break;
2655     default:
2656       break;
2657   }
2658   return res;
2659 }
2660
2661 static GstElement *
2662 session_request_element (GstRtpBinSession * session, guint signal)
2663 {
2664   GstElement *element = NULL;
2665   GstRtpBin *bin = session->bin;
2666
2667   g_signal_emit (bin, gst_rtp_bin_signals[signal], 0, session->id, &element);
2668
2669   if (element) {
2670     if (!bin_manage_element (bin, element))
2671       goto manage_failed;
2672     session->elements = g_slist_prepend (session->elements, element);
2673   }
2674   return element;
2675
2676   /* ERRORS */
2677 manage_failed:
2678   {
2679     GST_WARNING_OBJECT (bin, "unable to manage element");
2680     gst_object_unref (element);
2681     return NULL;
2682   }
2683 }
2684
2685 static gboolean
2686 copy_sticky_events (GstPad * pad, GstEvent ** event, gpointer user_data)
2687 {
2688   GstPad *gpad = GST_PAD_CAST (user_data);
2689
2690   GST_DEBUG_OBJECT (gpad, "store sticky event %" GST_PTR_FORMAT, *event);
2691   gst_pad_store_sticky_event (gpad, *event);
2692
2693   return TRUE;
2694 }
2695
2696 /* a new pad (SSRC) was created in @session. This signal is emited from the
2697  * payload demuxer. */
2698 static void
2699 new_payload_found (GstElement * element, guint pt, GstPad * pad,
2700     GstRtpBinStream * stream)
2701 {
2702   GstRtpBin *rtpbin;
2703   GstElementClass *klass;
2704   GstPadTemplate *templ;
2705   gchar *padname;
2706   GstPad *gpad;
2707
2708   rtpbin = stream->bin;
2709
2710   GST_DEBUG ("new payload pad %d", pt);
2711
2712   GST_RTP_BIN_SHUTDOWN_LOCK (rtpbin, shutdown);
2713
2714   /* ghost the pad to the parent */
2715   klass = GST_ELEMENT_GET_CLASS (rtpbin);
2716   templ = gst_element_class_get_pad_template (klass, "recv_rtp_src_%u_%u_%u");
2717   padname = g_strdup_printf ("recv_rtp_src_%u_%u_%u",
2718       stream->session->id, stream->ssrc, pt);
2719   gpad = gst_ghost_pad_new_from_template (padname, pad, templ);
2720   g_free (padname);
2721   g_object_set_data (G_OBJECT (pad), "GstRTPBin.ghostpad", gpad);
2722
2723   gst_pad_set_active (gpad, TRUE);
2724   GST_RTP_BIN_SHUTDOWN_UNLOCK (rtpbin);
2725
2726   gst_pad_sticky_events_foreach (pad, copy_sticky_events, gpad);
2727   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), gpad);
2728
2729   return;
2730
2731 shutdown:
2732   {
2733     GST_DEBUG ("ignoring, we are shutting down");
2734     return;
2735   }
2736 }
2737
2738 static void
2739 payload_pad_removed (GstElement * element, GstPad * pad,
2740     GstRtpBinStream * stream)
2741 {
2742   GstRtpBin *rtpbin;
2743   GstPad *gpad;
2744
2745   rtpbin = stream->bin;
2746
2747   GST_DEBUG ("payload pad removed");
2748
2749   GST_RTP_BIN_DYN_LOCK (rtpbin);
2750   if ((gpad = g_object_get_data (G_OBJECT (pad), "GstRTPBin.ghostpad"))) {
2751     g_object_set_data (G_OBJECT (pad), "GstRTPBin.ghostpad", NULL);
2752
2753     gst_pad_set_active (gpad, FALSE);
2754     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin), gpad);
2755   }
2756   GST_RTP_BIN_DYN_UNLOCK (rtpbin);
2757 }
2758
2759 static GstCaps *
2760 pt_map_requested (GstElement * element, guint pt, GstRtpBinSession * session)
2761 {
2762   GstRtpBin *rtpbin;
2763   GstCaps *caps;
2764
2765   rtpbin = session->bin;
2766
2767   GST_DEBUG_OBJECT (rtpbin, "payload map requested for pt %d in session %d", pt,
2768       session->id);
2769
2770   caps = get_pt_map (session, pt);
2771   if (!caps)
2772     goto no_caps;
2773
2774   return caps;
2775
2776   /* ERRORS */
2777 no_caps:
2778   {
2779     GST_DEBUG_OBJECT (rtpbin, "could not get caps");
2780     return NULL;
2781   }
2782 }
2783
2784 static void
2785 payload_type_change (GstElement * element, guint pt, GstRtpBinSession * session)
2786 {
2787   GST_DEBUG_OBJECT (session->bin,
2788       "emiting signal for pt type changed to %d in session %d", pt,
2789       session->id);
2790
2791   g_signal_emit (session->bin, gst_rtp_bin_signals[SIGNAL_PAYLOAD_TYPE_CHANGE],
2792       0, session->id, pt);
2793 }
2794
2795 /* emited when caps changed for the session */
2796 static void
2797 caps_changed (GstPad * pad, GParamSpec * pspec, GstRtpBinSession * session)
2798 {
2799   GstRtpBin *bin;
2800   GstCaps *caps;
2801   gint payload;
2802   const GstStructure *s;
2803
2804   bin = session->bin;
2805
2806   g_object_get (pad, "caps", &caps, NULL);
2807
2808   if (caps == NULL)
2809     return;
2810
2811   GST_DEBUG_OBJECT (bin, "got caps %" GST_PTR_FORMAT, caps);
2812
2813   s = gst_caps_get_structure (caps, 0);
2814
2815   /* get payload, finish when it's not there */
2816   if (!gst_structure_get_int (s, "payload", &payload))
2817     return;
2818
2819   GST_RTP_SESSION_LOCK (session);
2820   GST_DEBUG_OBJECT (bin, "insert caps for payload %d", payload);
2821   g_hash_table_insert (session->ptmap, GINT_TO_POINTER (payload), caps);
2822   GST_RTP_SESSION_UNLOCK (session);
2823 }
2824
2825 /* a new pad (SSRC) was created in @session */
2826 static void
2827 new_ssrc_pad_found (GstElement * element, guint ssrc, GstPad * pad,
2828     GstRtpBinSession * session)
2829 {
2830   GstRtpBin *rtpbin;
2831   GstRtpBinStream *stream;
2832   GstPad *sinkpad, *srcpad;
2833   gchar *padname;
2834
2835   rtpbin = session->bin;
2836
2837   GST_DEBUG_OBJECT (rtpbin, "new SSRC pad %08x, %s:%s", ssrc,
2838       GST_DEBUG_PAD_NAME (pad));
2839
2840   GST_RTP_BIN_SHUTDOWN_LOCK (rtpbin, shutdown);
2841
2842   GST_RTP_SESSION_LOCK (session);
2843
2844   /* create new stream */
2845   stream = create_stream (session, ssrc);
2846   if (!stream)
2847     goto no_stream;
2848
2849   /* get pad and link */
2850   GST_DEBUG_OBJECT (rtpbin, "linking jitterbuffer RTP");
2851   padname = g_strdup_printf ("src_%u", ssrc);
2852   srcpad = gst_element_get_static_pad (element, padname);
2853   g_free (padname);
2854   sinkpad = gst_element_get_static_pad (stream->buffer, "sink");
2855   gst_pad_link_full (srcpad, sinkpad, GST_PAD_LINK_CHECK_NOTHING);
2856   gst_object_unref (sinkpad);
2857   gst_object_unref (srcpad);
2858
2859   GST_DEBUG_OBJECT (rtpbin, "linking jitterbuffer RTCP");
2860   padname = g_strdup_printf ("rtcp_src_%u", ssrc);
2861   srcpad = gst_element_get_static_pad (element, padname);
2862   g_free (padname);
2863   sinkpad = gst_element_get_request_pad (stream->buffer, "sink_rtcp");
2864   gst_pad_link_full (srcpad, sinkpad, GST_PAD_LINK_CHECK_NOTHING);
2865   gst_object_unref (sinkpad);
2866   gst_object_unref (srcpad);
2867
2868   /* connect to the RTCP sync signal from the jitterbuffer */
2869   GST_DEBUG_OBJECT (rtpbin, "connecting sync signal");
2870   stream->buffer_handlesync_sig = g_signal_connect (stream->buffer,
2871       "handle-sync", (GCallback) gst_rtp_bin_handle_sync, stream);
2872
2873   if (stream->demux) {
2874     /* connect to the new-pad signal of the payload demuxer, this will expose the
2875      * new pad by ghosting it. */
2876     stream->demux_newpad_sig = g_signal_connect (stream->demux,
2877         "new-payload-type", (GCallback) new_payload_found, stream);
2878     stream->demux_padremoved_sig = g_signal_connect (stream->demux,
2879         "pad-removed", (GCallback) payload_pad_removed, stream);
2880
2881     /* connect to the request-pt-map signal. This signal will be emited by the
2882      * demuxer so that it can apply a proper caps on the buffers for the
2883      * depayloaders. */
2884     stream->demux_ptreq_sig = g_signal_connect (stream->demux,
2885         "request-pt-map", (GCallback) pt_map_requested, session);
2886     /* connect to the  signal so it can be forwarded. */
2887     stream->demux_ptchange_sig = g_signal_connect (stream->demux,
2888         "payload-type-change", (GCallback) payload_type_change, session);
2889   } else {
2890     /* add rtpjitterbuffer src pad to pads */
2891     GstElementClass *klass;
2892     GstPadTemplate *templ;
2893     gchar *padname;
2894     GstPad *gpad, *pad;
2895
2896     pad = gst_element_get_static_pad (stream->buffer, "src");
2897
2898     /* ghost the pad to the parent */
2899     klass = GST_ELEMENT_GET_CLASS (rtpbin);
2900     templ = gst_element_class_get_pad_template (klass, "recv_rtp_src_%u_%u_%u");
2901     padname = g_strdup_printf ("recv_rtp_src_%u_%u_%u",
2902         stream->session->id, stream->ssrc, 255);
2903     gpad = gst_ghost_pad_new_from_template (padname, pad, templ);
2904     g_free (padname);
2905
2906     gst_pad_set_active (gpad, TRUE);
2907     gst_pad_sticky_events_foreach (pad, copy_sticky_events, gpad);
2908     gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), gpad);
2909
2910     gst_object_unref (pad);
2911   }
2912
2913   GST_RTP_SESSION_UNLOCK (session);
2914   GST_RTP_BIN_SHUTDOWN_UNLOCK (rtpbin);
2915
2916   return;
2917
2918   /* ERRORS */
2919 shutdown:
2920   {
2921     GST_DEBUG_OBJECT (rtpbin, "we are shutting down");
2922     return;
2923   }
2924 no_stream:
2925   {
2926     GST_RTP_SESSION_UNLOCK (session);
2927     GST_RTP_BIN_SHUTDOWN_UNLOCK (rtpbin);
2928     GST_DEBUG_OBJECT (rtpbin, "could not create stream");
2929     return;
2930   }
2931 }
2932
2933 static gboolean
2934 complete_session_sink (GstRtpBin * rtpbin, GstRtpBinSession * session)
2935 {
2936   gchar *gname;
2937   guint sessid = session->id;
2938   GstPad *recv_rtp_sink;
2939   GstElement *decoder;
2940   GstElementClass *klass;
2941   GstPadTemplate *templ;
2942
2943   /* get recv_rtp pad and store */
2944   session->recv_rtp_sink =
2945       gst_element_get_request_pad (session->session, "recv_rtp_sink");
2946   if (session->recv_rtp_sink == NULL)
2947     goto pad_failed;
2948
2949   g_signal_connect (session->recv_rtp_sink, "notify::caps",
2950       (GCallback) caps_changed, session);
2951
2952   GST_DEBUG_OBJECT (rtpbin, "requesting RTP decoder");
2953   decoder = session_request_element (session, SIGNAL_REQUEST_RTP_DECODER);
2954   if (decoder) {
2955     GstPad *decsrc, *decsink;
2956     GstPadLinkReturn ret;
2957
2958     GST_DEBUG_OBJECT (rtpbin, "linking RTP decoder");
2959     decsink = gst_element_get_static_pad (decoder, "rtp_sink");
2960     if (decsink == NULL)
2961       goto dec_sink_failed;
2962
2963     recv_rtp_sink = decsink;
2964
2965     decsrc = gst_element_get_static_pad (decoder, "rtp_src");
2966     if (decsrc == NULL)
2967       goto dec_src_failed;
2968
2969     ret = gst_pad_link (decsrc, session->recv_rtp_sink);
2970     gst_object_unref (decsrc);
2971
2972     if (ret != GST_PAD_LINK_OK)
2973       goto dec_link_failed;
2974
2975   } else {
2976     GST_DEBUG_OBJECT (rtpbin, "no RTP decoder given");
2977     recv_rtp_sink = gst_object_ref (session->recv_rtp_sink);
2978   }
2979
2980   GST_DEBUG_OBJECT (rtpbin, "ghosting session sink pad");
2981   klass = GST_ELEMENT_GET_CLASS (rtpbin);
2982   gname = g_strdup_printf ("recv_rtp_sink_%u", sessid);
2983   templ = gst_element_class_get_pad_template (klass, "recv_rtp_sink_%u");
2984   session->recv_rtp_sink_ghost =
2985       gst_ghost_pad_new_from_template (gname, recv_rtp_sink, templ);
2986   gst_object_unref (recv_rtp_sink);
2987   gst_pad_set_active (session->recv_rtp_sink_ghost, TRUE);
2988   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), session->recv_rtp_sink_ghost);
2989   g_free (gname);
2990
2991   return TRUE;
2992
2993   /* ERRORS */
2994 pad_failed:
2995   {
2996     g_warning ("rtpbin: failed to get session recv_rtp_sink pad");
2997     return FALSE;
2998   }
2999 dec_sink_failed:
3000   {
3001     g_warning ("rtpbin: failed to get decoder sink pad for session %d", sessid);
3002     return FALSE;
3003   }
3004 dec_src_failed:
3005   {
3006     g_warning ("rtpbin: failed to get decoder src pad for session %d", sessid);
3007     gst_object_unref (recv_rtp_sink);
3008     return FALSE;
3009   }
3010 dec_link_failed:
3011   {
3012     g_warning ("rtpbin: failed to link rtp decoder for session %d", sessid);
3013     gst_object_unref (recv_rtp_sink);
3014     return FALSE;
3015   }
3016 }
3017
3018 /* Create a pad for receiving RTP for the session in @name. Must be called with
3019  * RTP_BIN_LOCK.
3020  */
3021 static GstPad *
3022 create_recv_rtp (GstRtpBin * rtpbin, GstPadTemplate * templ, const gchar * name)
3023 {
3024   guint sessid;
3025   GstElement *aux;
3026   GstPad *recv_rtp_src;
3027   GstRtpBinSession *session;
3028
3029   /* first get the session number */
3030   if (name == NULL || sscanf (name, "recv_rtp_sink_%u", &sessid) != 1)
3031     goto no_name;
3032
3033   GST_DEBUG_OBJECT (rtpbin, "finding session %d", sessid);
3034
3035   /* get or create session */
3036   session = find_session_by_id (rtpbin, sessid);
3037   if (!session) {
3038     GST_DEBUG_OBJECT (rtpbin, "creating session %d", sessid);
3039     /* create session now */
3040     session = create_session (rtpbin, sessid);
3041     if (session == NULL)
3042       goto create_error;
3043   }
3044
3045   /* check if pad was requested */
3046   if (session->recv_rtp_sink_ghost != NULL)
3047     return session->recv_rtp_sink_ghost;
3048
3049   /* setup the session sink pad */
3050   if (!complete_session_sink (rtpbin, session))
3051     goto session_sink_failed;
3052
3053   session->recv_rtp_src =
3054       gst_element_get_static_pad (session->session, "recv_rtp_src");
3055   if (session->recv_rtp_src == NULL)
3056     goto pad_failed;
3057
3058   /* find out if we need AUX elements or if we can go into the SSRC demuxer
3059    * directly */
3060   aux = session_request_element (session, SIGNAL_REQUEST_AUX_RECEIVER);
3061   if (aux) {
3062     gchar *pname;
3063     GstPad *auxsink;
3064     GstPadLinkReturn ret;
3065
3066     GST_DEBUG_OBJECT (rtpbin, "linking AUX receiver");
3067
3068     pname = g_strdup_printf ("sink_%d", sessid);
3069     auxsink = gst_element_get_static_pad (aux, pname);
3070     g_free (pname);
3071     if (auxsink == NULL)
3072       goto aux_sink_failed;
3073
3074     ret = gst_pad_link (session->recv_rtp_src, auxsink);
3075     gst_object_unref (auxsink);
3076     if (ret != GST_PAD_LINK_OK)
3077       goto aux_link_failed;
3078
3079     /* this can be NULL when this AUX element is not to be linked to
3080      * an SSRC demuxer */
3081     pname = g_strdup_printf ("src_%d", sessid);
3082     recv_rtp_src = gst_element_get_static_pad (aux, pname);
3083     g_free (pname);
3084   } else {
3085     recv_rtp_src = gst_object_ref (session->recv_rtp_src);
3086   }
3087
3088   if (recv_rtp_src) {
3089     GstPad *sinkdpad;
3090
3091     GST_DEBUG_OBJECT (rtpbin, "getting demuxer RTP sink pad");
3092     sinkdpad = gst_element_get_static_pad (session->demux, "sink");
3093     GST_DEBUG_OBJECT (rtpbin, "linking demuxer RTP sink pad");
3094     gst_pad_link_full (recv_rtp_src, sinkdpad, GST_PAD_LINK_CHECK_NOTHING);
3095     gst_object_unref (recv_rtp_src);
3096     gst_object_unref (sinkdpad);
3097
3098     /* connect to the new-ssrc-pad signal of the SSRC demuxer */
3099     session->demux_newpad_sig = g_signal_connect (session->demux,
3100         "new-ssrc-pad", (GCallback) new_ssrc_pad_found, session);
3101     session->demux_padremoved_sig = g_signal_connect (session->demux,
3102         "removed-ssrc-pad", (GCallback) ssrc_demux_pad_removed, session);
3103   }
3104   return session->recv_rtp_sink_ghost;
3105
3106   /* ERRORS */
3107 no_name:
3108   {
3109     g_warning ("rtpbin: invalid name given");
3110     return NULL;
3111   }
3112 create_error:
3113   {
3114     /* create_session already warned */
3115     return NULL;
3116   }
3117 session_sink_failed:
3118   {
3119     /* warning already done */
3120     return NULL;
3121   }
3122 pad_failed:
3123   {
3124     g_warning ("rtpbin: failed to get session recv_rtp_src pad");
3125     return NULL;
3126   }
3127 aux_sink_failed:
3128   {
3129     g_warning ("rtpbin: failed to get AUX sink pad for session %d", sessid);
3130     return NULL;
3131   }
3132 aux_link_failed:
3133   {
3134     g_warning ("rtpbin: failed to link AUX pad to session %d", sessid);
3135     return NULL;
3136   }
3137 }
3138
3139 static void
3140 remove_recv_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session)
3141 {
3142   if (session->demux_newpad_sig) {
3143     g_signal_handler_disconnect (session->demux, session->demux_newpad_sig);
3144     session->demux_newpad_sig = 0;
3145   }
3146   if (session->demux_padremoved_sig) {
3147     g_signal_handler_disconnect (session->demux, session->demux_padremoved_sig);
3148     session->demux_padremoved_sig = 0;
3149   }
3150   if (session->recv_rtp_src) {
3151     gst_object_unref (session->recv_rtp_src);
3152     session->recv_rtp_src = NULL;
3153   }
3154   if (session->recv_rtp_sink) {
3155     gst_element_release_request_pad (session->session, session->recv_rtp_sink);
3156     gst_object_unref (session->recv_rtp_sink);
3157     session->recv_rtp_sink = NULL;
3158   }
3159   if (session->recv_rtp_sink_ghost) {
3160     gst_pad_set_active (session->recv_rtp_sink_ghost, FALSE);
3161     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
3162         session->recv_rtp_sink_ghost);
3163     session->recv_rtp_sink_ghost = NULL;
3164   }
3165 }
3166
3167 /* Create a pad for receiving RTCP for the session in @name. Must be called with
3168  * RTP_BIN_LOCK.
3169  */
3170 static GstPad *
3171 create_recv_rtcp (GstRtpBin * rtpbin, GstPadTemplate * templ,
3172     const gchar * name)
3173 {
3174   guint sessid;
3175   GstElement *decoder;
3176   GstRtpBinSession *session;
3177   GstPad *sinkdpad, *decsink;
3178
3179   /* first get the session number */
3180   if (name == NULL || sscanf (name, "recv_rtcp_sink_%u", &sessid) != 1)
3181     goto no_name;
3182
3183   GST_DEBUG_OBJECT (rtpbin, "finding session %d", sessid);
3184
3185   /* get or create the session */
3186   session = find_session_by_id (rtpbin, sessid);
3187   if (!session) {
3188     GST_DEBUG_OBJECT (rtpbin, "creating session %d", sessid);
3189     /* create session now */
3190     session = create_session (rtpbin, sessid);
3191     if (session == NULL)
3192       goto create_error;
3193   }
3194
3195   /* check if pad was requested */
3196   if (session->recv_rtcp_sink_ghost != NULL)
3197     return session->recv_rtcp_sink_ghost;
3198
3199   /* get recv_rtp pad and store */
3200   GST_DEBUG_OBJECT (rtpbin, "getting RTCP sink pad");
3201   session->recv_rtcp_sink =
3202       gst_element_get_request_pad (session->session, "recv_rtcp_sink");
3203   if (session->recv_rtcp_sink == NULL)
3204     goto pad_failed;
3205
3206   GST_DEBUG_OBJECT (rtpbin, "getting RTCP decoder");
3207   decoder = session_request_element (session, SIGNAL_REQUEST_RTCP_DECODER);
3208   if (decoder) {
3209     GstPad *decsrc;
3210     GstPadLinkReturn ret;
3211
3212     GST_DEBUG_OBJECT (rtpbin, "linking RTCP decoder");
3213     decsink = gst_element_get_static_pad (decoder, "rtcp_sink");
3214     decsrc = gst_element_get_static_pad (decoder, "rtcp_src");
3215
3216     if (decsink == NULL)
3217       goto dec_sink_failed;
3218
3219     if (decsrc == NULL)
3220       goto dec_src_failed;
3221
3222     ret = gst_pad_link (decsrc, session->recv_rtcp_sink);
3223     gst_object_unref (decsrc);
3224
3225     if (ret != GST_PAD_LINK_OK)
3226       goto dec_link_failed;
3227   } else {
3228     GST_DEBUG_OBJECT (rtpbin, "no RTCP decoder given");
3229     decsink = gst_object_ref (session->recv_rtcp_sink);
3230   }
3231
3232   /* get srcpad, link to SSRCDemux */
3233   GST_DEBUG_OBJECT (rtpbin, "getting sync src pad");
3234   session->sync_src = gst_element_get_static_pad (session->session, "sync_src");
3235   if (session->sync_src == NULL)
3236     goto src_pad_failed;
3237
3238   GST_DEBUG_OBJECT (rtpbin, "getting demuxer RTCP sink pad");
3239   sinkdpad = gst_element_get_static_pad (session->demux, "rtcp_sink");
3240   gst_pad_link_full (session->sync_src, sinkdpad, GST_PAD_LINK_CHECK_NOTHING);
3241   gst_object_unref (sinkdpad);
3242
3243   session->recv_rtcp_sink_ghost =
3244       gst_ghost_pad_new_from_template (name, decsink, templ);
3245   gst_object_unref (decsink);
3246   gst_pad_set_active (session->recv_rtcp_sink_ghost, TRUE);
3247   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin),
3248       session->recv_rtcp_sink_ghost);
3249
3250   return session->recv_rtcp_sink_ghost;
3251
3252   /* ERRORS */
3253 no_name:
3254   {
3255     g_warning ("rtpbin: invalid name given");
3256     return NULL;
3257   }
3258 create_error:
3259   {
3260     /* create_session already warned */
3261     return NULL;
3262   }
3263 pad_failed:
3264   {
3265     g_warning ("rtpbin: failed to get session rtcp_sink pad");
3266     return NULL;
3267   }
3268 dec_sink_failed:
3269   {
3270     g_warning ("rtpbin: failed to get decoder sink pad for session %d", sessid);
3271     return NULL;
3272   }
3273 dec_src_failed:
3274   {
3275     g_warning ("rtpbin: failed to get decoder src pad for session %d", sessid);
3276     gst_object_unref (decsink);
3277     return NULL;
3278   }
3279 dec_link_failed:
3280   {
3281     g_warning ("rtpbin: failed to link rtcp decoder for session %d", sessid);
3282     gst_object_unref (decsink);
3283     return NULL;
3284   }
3285 src_pad_failed:
3286   {
3287     g_warning ("rtpbin: failed to get session sync_src pad");
3288     gst_object_unref (decsink);
3289     return NULL;
3290   }
3291 }
3292
3293 static void
3294 remove_recv_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session)
3295 {
3296   if (session->recv_rtcp_sink_ghost) {
3297     gst_pad_set_active (session->recv_rtcp_sink_ghost, FALSE);
3298     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
3299         session->recv_rtcp_sink_ghost);
3300     session->recv_rtcp_sink_ghost = NULL;
3301   }
3302   if (session->sync_src) {
3303     /* releasing the request pad should also unref the sync pad */
3304     gst_object_unref (session->sync_src);
3305     session->sync_src = NULL;
3306   }
3307   if (session->recv_rtcp_sink) {
3308     gst_element_release_request_pad (session->session, session->recv_rtcp_sink);
3309     gst_object_unref (session->recv_rtcp_sink);
3310     session->recv_rtcp_sink = NULL;
3311   }
3312 }
3313
3314 static gboolean
3315 complete_session_src (GstRtpBin * rtpbin, GstRtpBinSession * session)
3316 {
3317   gchar *gname;
3318   guint sessid = session->id;
3319   GstPad *send_rtp_src;
3320   GstElement *encoder;
3321   GstElementClass *klass;
3322   GstPadTemplate *templ;
3323
3324   /* get srcpad */
3325   session->send_rtp_src =
3326       gst_element_get_static_pad (session->session, "send_rtp_src");
3327   if (session->send_rtp_src == NULL)
3328     goto no_srcpad;
3329
3330   GST_DEBUG_OBJECT (rtpbin, "getting RTP encoder");
3331   encoder = session_request_element (session, SIGNAL_REQUEST_RTP_ENCODER);
3332   if (encoder) {
3333     gchar *ename;
3334     GstPad *encsrc, *encsink;
3335     GstPadLinkReturn ret;
3336
3337     GST_DEBUG_OBJECT (rtpbin, "linking RTP encoder");
3338     ename = g_strdup_printf ("rtp_src_%d", sessid);
3339     encsrc = gst_element_get_static_pad (encoder, ename);
3340     g_free (ename);
3341
3342     if (encsrc == NULL)
3343       goto enc_src_failed;
3344
3345     send_rtp_src = encsrc;
3346
3347     ename = g_strdup_printf ("rtp_sink_%d", sessid);
3348     encsink = gst_element_get_static_pad (encoder, ename);
3349     g_free (ename);
3350     if (encsink == NULL)
3351       goto enc_sink_failed;
3352
3353     ret = gst_pad_link (session->send_rtp_src, encsink);
3354     gst_object_unref (encsink);
3355
3356     if (ret != GST_PAD_LINK_OK)
3357       goto enc_link_failed;
3358   } else {
3359     GST_DEBUG_OBJECT (rtpbin, "no RTP encoder given");
3360     send_rtp_src = gst_object_ref (session->send_rtp_src);
3361   }
3362
3363   /* ghost the new source pad */
3364   klass = GST_ELEMENT_GET_CLASS (rtpbin);
3365   gname = g_strdup_printf ("send_rtp_src_%u", sessid);
3366   templ = gst_element_class_get_pad_template (klass, "send_rtp_src_%u");
3367   session->send_rtp_src_ghost =
3368       gst_ghost_pad_new_from_template (gname, send_rtp_src, templ);
3369   gst_object_unref (send_rtp_src);
3370   gst_pad_set_active (session->send_rtp_src_ghost, TRUE);
3371   gst_pad_sticky_events_foreach (send_rtp_src, copy_sticky_events,
3372       session->send_rtp_src_ghost);
3373   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), session->send_rtp_src_ghost);
3374   g_free (gname);
3375
3376   return TRUE;
3377
3378   /* ERRORS */
3379 no_srcpad:
3380   {
3381     g_warning ("rtpbin: failed to get rtp source pad for session %d", sessid);
3382     return FALSE;
3383   }
3384 enc_src_failed:
3385   {
3386     g_warning ("rtpbin: failed to get encoder src pad for session %d", sessid);
3387     return FALSE;
3388   }
3389 enc_sink_failed:
3390   {
3391     g_warning ("rtpbin: failed to get encoder sink pad for session %d", sessid);
3392     gst_object_unref (send_rtp_src);
3393     return FALSE;
3394   }
3395 enc_link_failed:
3396   {
3397     g_warning ("rtpbin: failed to link rtp encoder for session %d", sessid);
3398     gst_object_unref (send_rtp_src);
3399     return FALSE;
3400   }
3401 }
3402
3403 static gboolean
3404 setup_aux_sender_fold (const GValue * item, GValue * result, gpointer user_data)
3405 {
3406   GstPad *pad;
3407   gchar *name;
3408   guint sessid;
3409   GstRtpBinSession *session = user_data, *newsess;
3410   GstRtpBin *rtpbin = session->bin;
3411   GstPadLinkReturn ret;
3412
3413   pad = g_value_get_object (item);
3414   name = gst_pad_get_name (pad);
3415
3416   if (name == NULL || sscanf (name, "src_%u", &sessid) != 1)
3417     goto no_name;
3418
3419   g_free (name);
3420
3421   newsess = find_session_by_id (rtpbin, sessid);
3422   if (newsess == NULL) {
3423     /* create new session */
3424     newsess = create_session (rtpbin, sessid);
3425     if (newsess == NULL)
3426       goto create_error;
3427   } else if (newsess->send_rtp_sink != NULL)
3428     goto existing_session;
3429
3430   /* get send_rtp pad and store */
3431   newsess->send_rtp_sink =
3432       gst_element_get_request_pad (newsess->session, "send_rtp_sink");
3433   if (newsess->send_rtp_sink == NULL)
3434     goto pad_failed;
3435
3436   ret = gst_pad_link (pad, newsess->send_rtp_sink);
3437   if (ret != GST_PAD_LINK_OK)
3438     goto aux_link_failed;
3439
3440   if (!complete_session_src (rtpbin, newsess))
3441     goto session_src_failed;
3442
3443   return TRUE;
3444
3445   /* ERRORS */
3446 no_name:
3447   {
3448     GST_WARNING ("ignoring invalid pad name %s", GST_STR_NULL (name));
3449     g_free (name);
3450     return TRUE;
3451   }
3452 create_error:
3453   {
3454     /* create_session already warned */
3455     return FALSE;
3456   }
3457 existing_session:
3458   {
3459     g_warning ("rtpbin: session %d is already a sender", sessid);
3460     return FALSE;
3461   }
3462 pad_failed:
3463   {
3464     g_warning ("rtpbin: failed to get session pad for session %d", sessid);
3465     return FALSE;
3466   }
3467 aux_link_failed:
3468   {
3469     g_warning ("rtpbin: failed to link AUX for session %d", sessid);
3470     return FALSE;
3471   }
3472 session_src_failed:
3473   {
3474     g_warning ("rtpbin: failed to complete AUX for session %d", sessid);
3475     return FALSE;
3476   }
3477 }
3478
3479 static gboolean
3480 setup_aux_sender (GstRtpBin * rtpbin, GstRtpBinSession * session,
3481     GstElement * aux)
3482 {
3483   GstIterator *it;
3484   GValue result = { 0, };
3485   GstIteratorResult res;
3486
3487   it = gst_element_iterate_src_pads (aux);
3488   res = gst_iterator_fold (it, setup_aux_sender_fold, &result, session);
3489   gst_iterator_free (it);
3490
3491   return res == GST_ITERATOR_DONE;
3492 }
3493
3494 /* Create a pad for sending RTP for the session in @name. Must be called with
3495  * RTP_BIN_LOCK.
3496  */
3497 static GstPad *
3498 create_send_rtp (GstRtpBin * rtpbin, GstPadTemplate * templ, const gchar * name)
3499 {
3500   gchar *pname;
3501   guint sessid;
3502   GstPad *send_rtp_sink;
3503   GstElement *aux;
3504   GstRtpBinSession *session;
3505
3506   /* first get the session number */
3507   if (name == NULL || sscanf (name, "send_rtp_sink_%u", &sessid) != 1)
3508     goto no_name;
3509
3510   /* get or create session */
3511   session = find_session_by_id (rtpbin, sessid);
3512   if (!session) {
3513     /* create session now */
3514     session = create_session (rtpbin, sessid);
3515     if (session == NULL)
3516       goto create_error;
3517   }
3518
3519   /* check if pad was requested */
3520   if (session->send_rtp_sink_ghost != NULL)
3521     return session->send_rtp_sink_ghost;
3522
3523   /* check if we are already using this session as a sender */
3524   if (session->send_rtp_sink != NULL)
3525     goto existing_session;
3526
3527   GST_DEBUG_OBJECT (rtpbin, "getting RTP AUX sender");
3528   aux = session_request_element (session, SIGNAL_REQUEST_AUX_SENDER);
3529   if (aux) {
3530     GST_DEBUG_OBJECT (rtpbin, "linking AUX sender");
3531     if (!setup_aux_sender (rtpbin, session, aux))
3532       goto aux_session_failed;
3533
3534     pname = g_strdup_printf ("sink_%d", sessid);
3535     send_rtp_sink = gst_element_get_static_pad (aux, pname);
3536     g_free (pname);
3537
3538     if (send_rtp_sink == NULL)
3539       goto aux_sink_failed;
3540   } else {
3541     /* get send_rtp pad and store */
3542     session->send_rtp_sink =
3543         gst_element_get_request_pad (session->session, "send_rtp_sink");
3544     if (session->send_rtp_sink == NULL)
3545       goto pad_failed;
3546
3547     if (!complete_session_src (rtpbin, session))
3548       goto session_src_failed;
3549
3550     send_rtp_sink = gst_object_ref (session->send_rtp_sink);
3551   }
3552
3553   session->send_rtp_sink_ghost =
3554       gst_ghost_pad_new_from_template (name, send_rtp_sink, templ);
3555   gst_object_unref (send_rtp_sink);
3556   gst_pad_set_active (session->send_rtp_sink_ghost, TRUE);
3557   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), session->send_rtp_sink_ghost);
3558
3559   return session->send_rtp_sink_ghost;
3560
3561   /* ERRORS */
3562 no_name:
3563   {
3564     g_warning ("rtpbin: invalid name given");
3565     return NULL;
3566   }
3567 create_error:
3568   {
3569     /* create_session already warned */
3570     return NULL;
3571   }
3572 existing_session:
3573   {
3574     g_warning ("rtpbin: session %d is already in use", sessid);
3575     return NULL;
3576   }
3577 aux_session_failed:
3578   {
3579     g_warning ("rtpbin: failed to get AUX sink pad for session %d", sessid);
3580     return NULL;
3581   }
3582 aux_sink_failed:
3583   {
3584     g_warning ("rtpbin: failed to get AUX sink pad for session %d", sessid);
3585     return NULL;
3586   }
3587 pad_failed:
3588   {
3589     g_warning ("rtpbin: failed to get session pad for session %d", sessid);
3590     return NULL;
3591   }
3592 session_src_failed:
3593   {
3594     g_warning ("rtpbin: failed to setup source pads for session %d", sessid);
3595     return NULL;
3596   }
3597 }
3598
3599 static void
3600 remove_send_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session)
3601 {
3602   if (session->send_rtp_src_ghost) {
3603     gst_pad_set_active (session->send_rtp_src_ghost, FALSE);
3604     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
3605         session->send_rtp_src_ghost);
3606     session->send_rtp_src_ghost = NULL;
3607   }
3608   if (session->send_rtp_src) {
3609     gst_object_unref (session->send_rtp_src);
3610     session->send_rtp_src = NULL;
3611   }
3612   if (session->send_rtp_sink) {
3613     gst_element_release_request_pad (GST_ELEMENT_CAST (session->session),
3614         session->send_rtp_sink);
3615     gst_object_unref (session->send_rtp_sink);
3616     session->send_rtp_sink = NULL;
3617   }
3618   if (session->send_rtp_sink_ghost) {
3619     gst_pad_set_active (session->send_rtp_sink_ghost, FALSE);
3620     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
3621         session->send_rtp_sink_ghost);
3622     session->send_rtp_sink_ghost = NULL;
3623   }
3624 }
3625
3626 /* Create a pad for sending RTCP for the session in @name. Must be called with
3627  * RTP_BIN_LOCK.
3628  */
3629 static GstPad *
3630 create_rtcp (GstRtpBin * rtpbin, GstPadTemplate * templ, const gchar * name)
3631 {
3632   guint sessid;
3633   GstPad *encsrc;
3634   GstElement *encoder;
3635   GstRtpBinSession *session;
3636
3637   /* first get the session number */
3638   if (name == NULL || sscanf (name, "send_rtcp_src_%u", &sessid) != 1)
3639     goto no_name;
3640
3641   /* get or create session */
3642   session = find_session_by_id (rtpbin, sessid);
3643   if (!session)
3644     goto no_session;
3645
3646   /* check if pad was requested */
3647   if (session->send_rtcp_src_ghost != NULL)
3648     return session->send_rtcp_src_ghost;
3649
3650   /* get rtcp_src pad and store */
3651   session->send_rtcp_src =
3652       gst_element_get_request_pad (session->session, "send_rtcp_src");
3653   if (session->send_rtcp_src == NULL)
3654     goto pad_failed;
3655
3656   GST_DEBUG_OBJECT (rtpbin, "getting RTCP encoder");
3657   encoder = session_request_element (session, SIGNAL_REQUEST_RTCP_ENCODER);
3658   if (encoder) {
3659     gchar *ename;
3660     GstPad *encsink;
3661     GstPadLinkReturn ret;
3662
3663     GST_DEBUG_OBJECT (rtpbin, "linking RTCP encoder");
3664     ename = g_strdup_printf ("rtcp_sink_%d", sessid);
3665     encsink = gst_element_get_static_pad (encoder, ename);
3666     g_free (ename);
3667     ename = g_strdup_printf ("rtcp_src_%d", sessid);
3668     encsrc = gst_element_get_static_pad (encoder, ename);
3669     g_free (ename);
3670
3671     if (encsrc == NULL)
3672       goto enc_src_failed;
3673
3674     if (encsink == NULL)
3675       goto enc_sink_failed;
3676
3677     ret = gst_pad_link (session->send_rtcp_src, encsink);
3678     gst_object_unref (encsink);
3679
3680     if (ret != GST_PAD_LINK_OK)
3681       goto enc_link_failed;
3682   } else {
3683     GST_DEBUG_OBJECT (rtpbin, "no RTCP encoder given");
3684     encsrc = gst_object_ref (session->send_rtcp_src);
3685   }
3686
3687   session->send_rtcp_src_ghost =
3688       gst_ghost_pad_new_from_template (name, encsrc, templ);
3689   gst_object_unref (encsrc);
3690   gst_pad_set_active (session->send_rtcp_src_ghost, TRUE);
3691   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), session->send_rtcp_src_ghost);
3692
3693   return session->send_rtcp_src_ghost;
3694
3695   /* ERRORS */
3696 no_name:
3697   {
3698     g_warning ("rtpbin: invalid name given");
3699     return NULL;
3700   }
3701 no_session:
3702   {
3703     g_warning ("rtpbin: session with id %d does not exist", sessid);
3704     return NULL;
3705   }
3706 pad_failed:
3707   {
3708     g_warning ("rtpbin: failed to get rtcp pad for session %d", sessid);
3709     return NULL;
3710   }
3711 enc_src_failed:
3712   {
3713     g_warning ("rtpbin: failed to get encoder src pad for session %d", sessid);
3714     return NULL;
3715   }
3716 enc_sink_failed:
3717   {
3718     g_warning ("rtpbin: failed to get encoder sink pad for session %d", sessid);
3719     gst_object_unref (encsrc);
3720     return NULL;
3721   }
3722 enc_link_failed:
3723   {
3724     g_warning ("rtpbin: failed to link rtcp encoder for session %d", sessid);
3725     gst_object_unref (encsrc);
3726     return NULL;
3727   }
3728 }
3729
3730 static void
3731 remove_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session)
3732 {
3733   if (session->send_rtcp_src_ghost) {
3734     gst_pad_set_active (session->send_rtcp_src_ghost, FALSE);
3735     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
3736         session->send_rtcp_src_ghost);
3737     session->send_rtcp_src_ghost = NULL;
3738   }
3739   if (session->send_rtcp_src) {
3740     gst_element_release_request_pad (session->session, session->send_rtcp_src);
3741     gst_object_unref (session->send_rtcp_src);
3742     session->send_rtcp_src = NULL;
3743   }
3744 }
3745
3746 /* If the requested name is NULL we should create a name with
3747  * the session number assuming we want the lowest posible session
3748  * with a free pad like the template */
3749 static gchar *
3750 gst_rtp_bin_get_free_pad_name (GstElement * element, GstPadTemplate * templ)
3751 {
3752   gboolean name_found = FALSE;
3753   gint session = 0;
3754   GstIterator *pad_it = NULL;
3755   gchar *pad_name = NULL;
3756   GValue data = { 0, };
3757
3758   GST_DEBUG_OBJECT (element, "find a free pad name for template");
3759   while (!name_found) {
3760     gboolean done = FALSE;
3761
3762     g_free (pad_name);
3763     pad_name = g_strdup_printf (templ->name_template, session++);
3764     pad_it = gst_element_iterate_pads (GST_ELEMENT (element));
3765     name_found = TRUE;
3766     while (!done) {
3767       switch (gst_iterator_next (pad_it, &data)) {
3768         case GST_ITERATOR_OK:
3769         {
3770           GstPad *pad;
3771           gchar *name;
3772
3773           pad = g_value_get_object (&data);
3774           name = gst_pad_get_name (pad);
3775
3776           if (strcmp (name, pad_name) == 0) {
3777             done = TRUE;
3778             name_found = FALSE;
3779           }
3780           g_free (name);
3781           g_value_reset (&data);
3782           break;
3783         }
3784         case GST_ITERATOR_ERROR:
3785         case GST_ITERATOR_RESYNC:
3786           /* restart iteration */
3787           done = TRUE;
3788           name_found = FALSE;
3789           session = 0;
3790           break;
3791         case GST_ITERATOR_DONE:
3792           done = TRUE;
3793           break;
3794       }
3795     }
3796     g_value_unset (&data);
3797     gst_iterator_free (pad_it);
3798   }
3799
3800   GST_DEBUG_OBJECT (element, "free pad name found: '%s'", pad_name);
3801   return pad_name;
3802 }
3803
3804 /*
3805  */
3806 static GstPad *
3807 gst_rtp_bin_request_new_pad (GstElement * element,
3808     GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
3809 {
3810   GstRtpBin *rtpbin;
3811   GstElementClass *klass;
3812   GstPad *result;
3813
3814   gchar *pad_name = NULL;
3815
3816   g_return_val_if_fail (templ != NULL, NULL);
3817   g_return_val_if_fail (GST_IS_RTP_BIN (element), NULL);
3818
3819   rtpbin = GST_RTP_BIN (element);
3820   klass = GST_ELEMENT_GET_CLASS (element);
3821
3822   GST_RTP_BIN_LOCK (rtpbin);
3823
3824   if (name == NULL) {
3825     /* use a free pad name */
3826     pad_name = gst_rtp_bin_get_free_pad_name (element, templ);
3827   } else {
3828     /* use the provided name */
3829     pad_name = g_strdup (name);
3830   }
3831
3832   GST_DEBUG_OBJECT (rtpbin, "Trying to request a pad with name %s", pad_name);
3833
3834   /* figure out the template */
3835   if (templ == gst_element_class_get_pad_template (klass, "recv_rtp_sink_%u")) {
3836     result = create_recv_rtp (rtpbin, templ, pad_name);
3837   } else if (templ == gst_element_class_get_pad_template (klass,
3838           "recv_rtcp_sink_%u")) {
3839     result = create_recv_rtcp (rtpbin, templ, pad_name);
3840   } else if (templ == gst_element_class_get_pad_template (klass,
3841           "send_rtp_sink_%u")) {
3842     result = create_send_rtp (rtpbin, templ, pad_name);
3843   } else if (templ == gst_element_class_get_pad_template (klass,
3844           "send_rtcp_src_%u")) {
3845     result = create_rtcp (rtpbin, templ, pad_name);
3846   } else
3847     goto wrong_template;
3848
3849   g_free (pad_name);
3850   GST_RTP_BIN_UNLOCK (rtpbin);
3851
3852   return result;
3853
3854   /* ERRORS */
3855 wrong_template:
3856   {
3857     g_free (pad_name);
3858     GST_RTP_BIN_UNLOCK (rtpbin);
3859     g_warning ("rtpbin: this is not our template");
3860     return NULL;
3861   }
3862 }
3863
3864 static void
3865 gst_rtp_bin_release_pad (GstElement * element, GstPad * pad)
3866 {
3867   GstRtpBinSession *session;
3868   GstRtpBin *rtpbin;
3869
3870   g_return_if_fail (GST_IS_GHOST_PAD (pad));
3871   g_return_if_fail (GST_IS_RTP_BIN (element));
3872
3873   rtpbin = GST_RTP_BIN (element);
3874
3875   GST_RTP_BIN_LOCK (rtpbin);
3876   GST_DEBUG_OBJECT (rtpbin, "Trying to release pad %s:%s",
3877       GST_DEBUG_PAD_NAME (pad));
3878
3879   if (!(session = find_session_by_pad (rtpbin, pad)))
3880     goto unknown_pad;
3881
3882   if (session->recv_rtp_sink_ghost == pad) {
3883     remove_recv_rtp (rtpbin, session);
3884   } else if (session->recv_rtcp_sink_ghost == pad) {
3885     remove_recv_rtcp (rtpbin, session);
3886   } else if (session->send_rtp_sink_ghost == pad) {
3887     remove_send_rtp (rtpbin, session);
3888   } else if (session->send_rtcp_src_ghost == pad) {
3889     remove_rtcp (rtpbin, session);
3890   }
3891
3892   /* no more request pads, free the complete session */
3893   if (session->recv_rtp_sink_ghost == NULL
3894       && session->recv_rtcp_sink_ghost == NULL
3895       && session->send_rtp_sink_ghost == NULL
3896       && session->send_rtcp_src_ghost == NULL) {
3897     GST_DEBUG_OBJECT (rtpbin, "no more pads for session %p", session);
3898     rtpbin->sessions = g_slist_remove (rtpbin->sessions, session);
3899     free_session (session, rtpbin);
3900   }
3901   GST_RTP_BIN_UNLOCK (rtpbin);
3902
3903   return;
3904
3905   /* ERROR */
3906 unknown_pad:
3907   {
3908     GST_RTP_BIN_UNLOCK (rtpbin);
3909     g_warning ("rtpbin: %s:%s is not one of our request pads",
3910         GST_DEBUG_PAD_NAME (pad));
3911     return;
3912   }
3913 }