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  * #GstRtpBin can also demultiplex incoming bundled streams. The first
  57  * #GstRtpSession will have a #GstRtpSsrcDemux element splitting the streams
  58  * based on their SSRC and potentially dispatched to a different #GstRtpSession.
  59  * Because retransmission SSRCs need to be merged with the corresponding media
  60  * stream the #GstRtpBin::on-bundled-ssrc signal is emitted so that the
  61  * application can find out to which session the SSRC belongs.
  62  *
  63  * The session manager needs the clock-rate of the payload types it is handling
  64  * and will signal the #GstRtpSession::request-pt-map signal when it needs such a
  65  * mapping. One can clear the cached values with the #GstRtpSession::clear-pt-map
  66  * signal.
  67  *
  68  * Access to the internal statistics of rtpbin is provided with the
  69  * get-internal-session property. This action signal gives access to the
  70  * RTPSession object which further provides action signals to retrieve the
  71  * internal source and other sources.
  72  *
  73  * #GstRtpBin also has signals (#GstRtpBin::request-rtp-encoder,
  74  * #GstRtpBin::request-rtp-decoder, #GstRtpBin::request-rtcp-encoder and
  75  * #GstRtpBin::request-rtp-decoder) to dynamically request for RTP and RTCP encoders
  76  * and decoders in order to support SRTP. The encoders must provide the pads
  77  * rtp_sink_\%u and rtp_src_\%u for RTP and rtcp_sink_\%u and rtcp_src_\%u for
  78  * RTCP. The session number will be used in the pad name. The decoders must provide
  79  * rtp_sink and rtp_src for RTP and rtcp_sink and rtcp_src for RTCP. The decoders will
  80  * be placed before the #GstRtpSession element, thus they must support SSRC demuxing
  81  * internally.
  82  *
  83  * #GstRtpBin has signals (#GstRtpBin::request-aux-sender and
  84  * #GstRtpBin::request-aux-receiver to dynamically request an element that can be
  85  * used to create or merge additional RTP streams. AUX elements are needed to
  86  * implement FEC or retransmission (such as RFC 4588). An AUX sender must have one
  87  * sink_\%u pad that matches the sessionid in the signal and it should have 1 or
  88  * more src_\%u pads. For each src_%\u pad, a session will be made (if needed)
  89  * and the pad will be linked to the session send_rtp_sink pad. Each session will
  90  * then expose its source pad as send_rtp_src_\%u on #GstRtpBin.
  91  * An AUX receiver has 1 src_\%u pad that much match the sessionid in the signal
  92  * and 1 or more sink_\%u pads. A session will be made for each sink_\%u pad
  93  * when the corresponding recv_rtp_sink_\%u pad is requested on #GstRtpBin.
  94  *
  95  * <refsect2>
  96  * <title>Example pipelines</title>
  97  * |[
  98  * gst-launch-1.0 udpsrc port=5000 caps="application/x-rtp, ..." ! .recv_rtp_sink_0 \
  99  *     rtpbin ! rtptheoradepay ! theoradec ! xvimagesink
 100  * ]| Receive RTP data from port 5000 and send to the session 0 in rtpbin.
 101  * |[
 102  * gst-launch-1.0 rtpbin name=rtpbin \
 103  *         v4l2src ! videoconvert ! ffenc_h263 ! rtph263ppay ! rtpbin.send_rtp_sink_0 \
 104  *                   rtpbin.send_rtp_src_0 ! udpsink port=5000                            \
 105  *                   rtpbin.send_rtcp_src_0 ! udpsink port=5001 sync=false async=false    \
 106  *                   udpsrc port=5005 ! rtpbin.recv_rtcp_sink_0                           \
 107  *         audiotestsrc ! amrnbenc ! rtpamrpay ! rtpbin.send_rtp_sink_1                   \
 108  *                   rtpbin.send_rtp_src_1 ! udpsink port=5002                            \
 109  *                   rtpbin.send_rtcp_src_1 ! udpsink port=5003 sync=false async=false    \
 110  *                   udpsrc port=5007 ! rtpbin.recv_rtcp_sink_1
 111  * ]| Encode and payload H263 video captured from a v4l2src. Encode and payload AMR
 112  * audio generated from audiotestsrc. The video is sent to session 0 in rtpbin
 113  * and the audio is sent to session 1. Video packets are sent on UDP port 5000
 114  * and audio packets on port 5002. The video RTCP packets for session 0 are sent
 115  * on port 5001 and the audio RTCP packets for session 0 are sent on port 5003.
 116  * RTCP packets for session 0 are received on port 5005 and RTCP for session 1
 117  * is received on port 5007. Since RTCP packets from the sender should be sent
 118  * as soon as possible and do not participate in preroll, sync=false and
 119  * async=false is configured on udpsink
 120  * |[
 121  * gst-launch-1.0 -v rtpbin name=rtpbin                                          \
 122  *     udpsrc caps="application/x-rtp,media=(string)video,clock-rate=(int)90000,encoding-name=(string)H263-1998" \
 123  *             port=5000 ! rtpbin.recv_rtp_sink_0                                \
 124  *         rtpbin. ! rtph263pdepay ! ffdec_h263 ! xvimagesink                    \
 125  *      udpsrc port=5001 ! rtpbin.recv_rtcp_sink_0                               \
 126  *      rtpbin.send_rtcp_src_0 ! udpsink port=5005 sync=false async=false        \
 127  *     udpsrc caps="application/x-rtp,media=(string)audio,clock-rate=(int)8000,encoding-name=(string)AMR,encoding-params=(string)1,octet-align=(string)1" \
 128  *             port=5002 ! rtpbin.recv_rtp_sink_1                                \
 129  *         rtpbin. ! rtpamrdepay ! amrnbdec ! alsasink                           \
 130  *      udpsrc port=5003 ! rtpbin.recv_rtcp_sink_1                               \
 131  *      rtpbin.send_rtcp_src_1 ! udpsink port=5007 sync=false async=false
 132  * ]| Receive H263 on port 5000, send it through rtpbin in session 0, depayload,
 133  * decode and display the video.
 134  * Receive AMR on port 5002, send it through rtpbin in session 1, depayload,
 135  * decode and play the audio.
 136  * Receive server RTCP packets for session 0 on port 5001 and RTCP packets for
 137  * session 1 on port 5003. These packets will be used for session management and
 138  * synchronisation.
 139  * Send RTCP reports for session 0 on port 5005 and RTCP reports for session 1
 140  * on port 5007.
 141  * </refsect2>
 142  */
 143
 144 #ifdef HAVE_CONFIG_H
 145 #include "config.h"
 146 #endif
 147 #include <stdio.h>
 148 #include <string.h>
 149
 150 #include <gst/rtp/gstrtpbuffer.h>
 151 #include <gst/rtp/gstrtcpbuffer.h>
 152
 153 #include "gstrtpbin.h"
 154 #include "rtpsession.h"
 155 #include "gstrtpsession.h"
 156 #include "gstrtpjitterbuffer.h"
 157
 158 #include <gst/glib-compat-private.h>
 159
 160 GST_DEBUG_CATEGORY_STATIC (gst_rtp_bin_debug);
 161 #define GST_CAT_DEFAULT gst_rtp_bin_debug
 162
 163 /* sink pads */
 164 static GstStaticPadTemplate rtpbin_recv_rtp_sink_template =
 165     GST_STATIC_PAD_TEMPLATE ("recv_rtp_sink_%u",
 166     GST_PAD_SINK,
 167     GST_PAD_REQUEST,
 168     GST_STATIC_CAPS ("application/x-rtp;application/x-srtp")
 169     );
 170
 171 static GstStaticPadTemplate rtpbin_recv_rtcp_sink_template =
 172     GST_STATIC_PAD_TEMPLATE ("recv_rtcp_sink_%u",
 173     GST_PAD_SINK,
 174     GST_PAD_REQUEST,
 175     GST_STATIC_CAPS ("application/x-rtcp;application/x-srtcp")
 176     );
 177
 178 static GstStaticPadTemplate rtpbin_send_rtp_sink_template =
 179 GST_STATIC_PAD_TEMPLATE ("send_rtp_sink_%u",
 180     GST_PAD_SINK,
 181     GST_PAD_REQUEST,
 182     GST_STATIC_CAPS ("application/x-rtp")
 183     );
 184
 185 /* src pads */
 186 static GstStaticPadTemplate rtpbin_recv_rtp_src_template =
 187 GST_STATIC_PAD_TEMPLATE ("recv_rtp_src_%u_%u_%u",
 188     GST_PAD_SRC,
 189     GST_PAD_SOMETIMES,
 190     GST_STATIC_CAPS ("application/x-rtp")
 191     );
 192
 193 static GstStaticPadTemplate rtpbin_send_rtcp_src_template =
 194     GST_STATIC_PAD_TEMPLATE ("send_rtcp_src_%u",
 195     GST_PAD_SRC,
 196     GST_PAD_REQUEST,
 197     GST_STATIC_CAPS ("application/x-rtcp;application/x-srtcp")
 198     );
 199
 200 static GstStaticPadTemplate rtpbin_send_rtp_src_template =
 201     GST_STATIC_PAD_TEMPLATE ("send_rtp_src_%u",
 202     GST_PAD_SRC,
 203     GST_PAD_SOMETIMES,
 204     GST_STATIC_CAPS ("application/x-rtp;application/x-srtp")
 205     );
 206
 207 #define GST_RTP_BIN_GET_PRIVATE(obj)  \
 208    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTP_BIN, GstRtpBinPrivate))
 209
 210 #define GST_RTP_BIN_LOCK(bin)   g_mutex_lock (&(bin)->priv->bin_lock)
 211 #define GST_RTP_BIN_UNLOCK(bin) g_mutex_unlock (&(bin)->priv->bin_lock)
 212
 213 /* lock to protect dynamic callbacks, like pad-added and new ssrc. */
 214 #define GST_RTP_BIN_DYN_LOCK(bin)    g_mutex_lock (&(bin)->priv->dyn_lock)
 215 #define GST_RTP_BIN_DYN_UNLOCK(bin)  g_mutex_unlock (&(bin)->priv->dyn_lock)
 216
 217 /* lock for shutdown */
 218 #define GST_RTP_BIN_SHUTDOWN_LOCK(bin,label)     \
 219 G_STMT_START {                                   \
 220   if (g_atomic_int_get (&bin->priv->shutdown))   \
 221     goto label;                                  \
 222   GST_RTP_BIN_DYN_LOCK (bin);                    \
 223   if (g_atomic_int_get (&bin->priv->shutdown)) { \
 224     GST_RTP_BIN_DYN_UNLOCK (bin);                \
 225     goto label;                                  \
 226   }                                              \
 227 } G_STMT_END
 228
 229 /* unlock for shutdown */
 230 #define GST_RTP_BIN_SHUTDOWN_UNLOCK(bin)         \
 231   GST_RTP_BIN_DYN_UNLOCK (bin);                  \
 232
 233 /* Minimum time offset to apply. This compensates for rounding errors in NTP to
 234  * RTP timestamp conversions */
 235 #define MIN_TS_OFFSET (4 * GST_MSECOND)
 236
 237 struct _GstRtpBinPrivate
 238 {
 239   GMutex bin_lock;
 240
 241   /* lock protecting dynamic adding/removing */
 242   GMutex dyn_lock;
 243
 244   /* if we are shutting down or not */
 245   gint shutdown;
 246
 247   gboolean autoremove;
 248
 249   /* NTP time in ns of last SR sync used */
 250   guint64 last_ntpnstime;
 251
 252   /* list of extra elements */
 253   GList *elements;
 254 };
 255
 256 /* signals and args */
 257 enum
 258 {
 259   SIGNAL_REQUEST_PT_MAP,
 260   SIGNAL_PAYLOAD_TYPE_CHANGE,
 261   SIGNAL_CLEAR_PT_MAP,
 262   SIGNAL_RESET_SYNC,
 263   SIGNAL_GET_SESSION,
 264   SIGNAL_GET_INTERNAL_SESSION,
 265   SIGNAL_GET_STORAGE,
 266   SIGNAL_GET_INTERNAL_STORAGE,
 267
 268   SIGNAL_ON_NEW_SSRC,
 269   SIGNAL_ON_SSRC_COLLISION,
 270   SIGNAL_ON_SSRC_VALIDATED,
 271   SIGNAL_ON_SSRC_ACTIVE,
 272   SIGNAL_ON_SSRC_SDES,
 273   SIGNAL_ON_BYE_SSRC,
 274   SIGNAL_ON_BYE_TIMEOUT,
 275   SIGNAL_ON_TIMEOUT,
 276   SIGNAL_ON_SENDER_TIMEOUT,
 277   SIGNAL_ON_NPT_STOP,
 278
 279   SIGNAL_REQUEST_RTP_ENCODER,
 280   SIGNAL_REQUEST_RTP_DECODER,
 281   SIGNAL_REQUEST_RTCP_ENCODER,
 282   SIGNAL_REQUEST_RTCP_DECODER,
 283
 284   SIGNAL_REQUEST_FEC_DECODER,
 285   SIGNAL_REQUEST_FEC_ENCODER,
 286
 287   SIGNAL_NEW_JITTERBUFFER,
 288   SIGNAL_NEW_STORAGE,
 289
 290   SIGNAL_REQUEST_AUX_SENDER,
 291   SIGNAL_REQUEST_AUX_RECEIVER,
 292
 293   SIGNAL_ON_NEW_SENDER_SSRC,
 294   SIGNAL_ON_SENDER_SSRC_ACTIVE,
 295
 296   SIGNAL_ON_BUNDLED_SSRC,
 297
 298   LAST_SIGNAL
 299 };
 300
 301 #define DEFAULT_LATENCY_MS           200
 302 #define DEFAULT_DROP_ON_LATENCY      FALSE
 303 #define DEFAULT_SDES                 NULL
 304 #define DEFAULT_DO_LOST              FALSE
 305 #define DEFAULT_IGNORE_PT            FALSE
 306 #define DEFAULT_NTP_SYNC             FALSE
 307 #define DEFAULT_AUTOREMOVE           FALSE
 308 #define DEFAULT_BUFFER_MODE          RTP_JITTER_BUFFER_MODE_SLAVE
 309 #define DEFAULT_USE_PIPELINE_CLOCK   FALSE
 310 #define DEFAULT_RTCP_SYNC            GST_RTP_BIN_RTCP_SYNC_ALWAYS
 311 #define DEFAULT_RTCP_SYNC_INTERVAL   0
 312 #define DEFAULT_DO_SYNC_EVENT        FALSE
 313 #define DEFAULT_DO_RETRANSMISSION    FALSE
 314 #define DEFAULT_RTP_PROFILE          GST_RTP_PROFILE_AVP
 315 #define DEFAULT_NTP_TIME_SOURCE      GST_RTP_NTP_TIME_SOURCE_NTP
 316 #define DEFAULT_RTCP_SYNC_SEND_TIME  TRUE
 317 #define DEFAULT_MAX_RTCP_RTP_TIME_DIFF 1000
 318 #define DEFAULT_MAX_DROPOUT_TIME     60000
 319 #define DEFAULT_MAX_MISORDER_TIME    2000
 320 #define DEFAULT_RFC7273_SYNC         FALSE
 321 #define DEFAULT_MAX_STREAMS          G_MAXUINT
 322 #define DEFAULT_MAX_TS_OFFSET_ADJUSTMENT G_GUINT64_CONSTANT(0)
 323 #define DEFAULT_MAX_TS_OFFSET        G_GINT64_CONSTANT(3000000000)
 324
 325 enum
 326 {
 327   PROP_0,
 328   PROP_LATENCY,
 329   PROP_DROP_ON_LATENCY,
 330   PROP_SDES,
 331   PROP_DO_LOST,
 332   PROP_IGNORE_PT,
 333   PROP_NTP_SYNC,
 334   PROP_RTCP_SYNC,
 335   PROP_RTCP_SYNC_INTERVAL,
 336   PROP_AUTOREMOVE,
 337   PROP_BUFFER_MODE,
 338   PROP_USE_PIPELINE_CLOCK,
 339   PROP_DO_SYNC_EVENT,
 340   PROP_DO_RETRANSMISSION,
 341   PROP_RTP_PROFILE,
 342   PROP_NTP_TIME_SOURCE,
 343   PROP_RTCP_SYNC_SEND_TIME,
 344   PROP_MAX_RTCP_RTP_TIME_DIFF,
 345   PROP_MAX_DROPOUT_TIME,
 346   PROP_MAX_MISORDER_TIME,
 347   PROP_RFC7273_SYNC,
 348   PROP_MAX_STREAMS,
 349   PROP_MAX_TS_OFFSET_ADJUSTMENT,
 350   PROP_MAX_TS_OFFSET,
 351 };
 352
 353 #define GST_RTP_BIN_RTCP_SYNC_TYPE (gst_rtp_bin_rtcp_sync_get_type())
 354 static GType
 355 gst_rtp_bin_rtcp_sync_get_type (void)
 356 {
 357   static GType rtcp_sync_type = 0;
 358   static const GEnumValue rtcp_sync_types[] = {
 359     {GST_RTP_BIN_RTCP_SYNC_ALWAYS, "always", "always"},
 360     {GST_RTP_BIN_RTCP_SYNC_INITIAL, "initial", "initial"},
 361     {GST_RTP_BIN_RTCP_SYNC_RTP, "rtp-info", "rtp-info"},
 362     {0, NULL, NULL},
 363   };
 364
 365   if (!rtcp_sync_type) {
 366     rtcp_sync_type = g_enum_register_static ("GstRTCPSync", rtcp_sync_types);
 367   }
 368   return rtcp_sync_type;
 369 }
 370
 371 /* helper objects */
 372 typedef struct _GstRtpBinSession GstRtpBinSession;
 373 typedef struct _GstRtpBinStream GstRtpBinStream;
 374 typedef struct _GstRtpBinClient GstRtpBinClient;
 375
 376 static guint gst_rtp_bin_signals[LAST_SIGNAL] = { 0 };
 377
 378 static GstCaps *pt_map_requested (GstElement * element, guint pt,
 379     GstRtpBinSession * session);
 380 static void payload_type_change (GstElement * element, guint pt,
 381     GstRtpBinSession * session);
 382 static void remove_recv_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session);
 383 static void remove_recv_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session);
 384 static void remove_send_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session);
 385 static void remove_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session);
 386 static void free_client (GstRtpBinClient * client, GstRtpBin * bin);
 387 static void free_stream (GstRtpBinStream * stream, GstRtpBin * bin);
 388 static GstRtpBinSession *create_session (GstRtpBin * rtpbin, gint id);
 389 static GstPad *complete_session_sink (GstRtpBin * rtpbin,
 390     GstRtpBinSession * session, gboolean bundle_demuxer_needed);
 391 static void
 392 complete_session_receiver (GstRtpBin * rtpbin, GstRtpBinSession * session,
 393     guint sessid);
 394 static GstPad *complete_session_rtcp (GstRtpBin * rtpbin,
 395     GstRtpBinSession * session, guint sessid, gboolean bundle_demuxer_needed);
 396
 397 /* Manages the RTP stream for one SSRC.
 398  *
 399  * We pipe the stream (comming from the SSRC demuxer) into a jitterbuffer.
 400  * If we see an SDES RTCP packet that links multiple SSRCs together based on a
 401  * common CNAME, we create a GstRtpBinClient structure to group the SSRCs
 402  * together (see below).
 403  */
 404 struct _GstRtpBinStream
 405 {
 406   /* the SSRC of this stream */
 407   guint32 ssrc;
 408
 409   /* parent bin */
 410   GstRtpBin *bin;
 411
 412   /* the session this SSRC belongs to */
 413   GstRtpBinSession *session;
 414
 415   /* the jitterbuffer of the SSRC */
 416   GstElement *buffer;
 417   gulong buffer_handlesync_sig;
 418   gulong buffer_ptreq_sig;
 419   gulong buffer_ntpstop_sig;
 420   gint percent;
 421
 422   /* the PT demuxer of the SSRC */
 423   GstElement *demux;
 424   gulong demux_newpad_sig;
 425   gulong demux_padremoved_sig;
 426   gulong demux_ptreq_sig;
 427   gulong demux_ptchange_sig;
 428
 429   /* if we have calculated a valid rt_delta for this stream */
 430   gboolean have_sync;
 431   /* mapping to local RTP and NTP time */
 432   gint64 rt_delta;
 433   gint64 rtp_delta;
 434   /* base rtptime in gst time */
 435   gint64 clock_base;
 436 };
 437
 438 #define GST_RTP_SESSION_LOCK(sess)   g_mutex_lock (&(sess)->lock)
 439 #define GST_RTP_SESSION_UNLOCK(sess) g_mutex_unlock (&(sess)->lock)
 440
 441 /* Manages the receiving end of the packets.
 442  *
 443  * There is one such structure for each RTP session (audio/video/...).
 444  * We get the RTP/RTCP packets and stuff them into the session manager. From
 445  * there they are pushed into an SSRC demuxer that splits the stream based on
 446  * SSRC. Each of the SSRC streams go into their own jitterbuffer (managed with
 447  * the GstRtpBinStream above).
 448  *
 449  * Before the SSRC demuxer, a storage element may be inserted for the purpose
 450  * of Forward Error Correction.
 451  */
 452 struct _GstRtpBinSession
 453 {
 454   /* session id */
 455   gint id;
 456   /* the parent bin */
 457   GstRtpBin *bin;
 458   /* the session element */
 459   GstElement *session;
 460   /* the SSRC demuxer */
 461   GstElement *demux;
 462   gulong demux_newpad_sig;
 463   gulong demux_padremoved_sig;
 464
 465   /* Fec support */
 466   GstElement *storage;
 467
 468   /* Bundling support */
 469   GstElement *rtp_funnel;
 470   GstElement *rtcp_funnel;
 471   GstElement *bundle_demux;
 472   gulong bundle_demux_newpad_sig;
 473
 474   GMutex lock;
 475
 476   /* list of GstRtpBinStream */
 477   GSList *streams;
 478
 479   /* list of elements */
 480   GSList *elements;
 481
 482   /* mapping of payload type to caps */
 483   GHashTable *ptmap;
 484
 485   /* the pads of the session */
 486   GstPad *recv_rtp_sink;
 487   GstPad *recv_rtp_sink_ghost;
 488   GstPad *recv_rtp_src;
 489   GstPad *recv_rtcp_sink;
 490   GstPad *recv_rtcp_sink_ghost;
 491   GstPad *sync_src;
 492   GstPad *send_rtp_sink;
 493   GstPad *send_rtp_sink_ghost;
 494   GstPad *send_rtp_src_ghost;
 495   GstPad *send_rtcp_src;
 496   GstPad *send_rtcp_src_ghost;
 497 };
 498
 499 /* Manages the RTP streams that come from one client and should therefore be
 500  * synchronized.
 501  */
 502 struct _GstRtpBinClient
 503 {
 504   /* the common CNAME for the streams */
 505   gchar *cname;
 506   guint cname_len;
 507
 508   /* the streams */
 509   guint nstreams;
 510   GSList *streams;
 511 };
 512
 513 /* find a session with the given id. Must be called with RTP_BIN_LOCK */
 514 static GstRtpBinSession *
 515 find_session_by_id (GstRtpBin * rtpbin, gint id)
 516 {
 517   GSList *walk;
 518
 519   for (walk = rtpbin->sessions; walk; walk = g_slist_next (walk)) {
 520     GstRtpBinSession *sess = (GstRtpBinSession *) walk->data;
 521
 522     if (sess->id == id)
 523       return sess;
 524   }
 525   return NULL;
 526 }
 527
 528 /* find a session with the given request pad. Must be called with RTP_BIN_LOCK */
 529 static GstRtpBinSession *
 530 find_session_by_pad (GstRtpBin * rtpbin, GstPad * pad)
 531 {
 532   GSList *walk;
 533
 534   for (walk = rtpbin->sessions; walk; walk = g_slist_next (walk)) {
 535     GstRtpBinSession *sess = (GstRtpBinSession *) walk->data;
 536
 537     if ((sess->recv_rtp_sink_ghost == pad) ||
 538         (sess->recv_rtcp_sink_ghost == pad) ||
 539         (sess->send_rtp_sink_ghost == pad)
 540         || (sess->send_rtcp_src_ghost == pad))
 541       return sess;
 542   }
 543   return NULL;
 544 }
 545
 546 static void
 547 on_new_ssrc (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 548 {
 549   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_NEW_SSRC], 0,
 550       sess->id, ssrc);
 551 }
 552
 553 static void
 554 on_ssrc_collision (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 555 {
 556   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_COLLISION], 0,
 557       sess->id, ssrc);
 558 }
 559
 560 static void
 561 on_ssrc_validated (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 562 {
 563   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_VALIDATED], 0,
 564       sess->id, ssrc);
 565 }
 566
 567 static void
 568 on_ssrc_active (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 569 {
 570   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_ACTIVE], 0,
 571       sess->id, ssrc);
 572 }
 573
 574 static void
 575 on_ssrc_sdes (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 576 {
 577   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SSRC_SDES], 0,
 578       sess->id, ssrc);
 579 }
 580
 581 static void
 582 on_bye_ssrc (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 583 {
 584   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_BYE_SSRC], 0,
 585       sess->id, ssrc);
 586 }
 587
 588 static void
 589 on_bye_timeout (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 590 {
 591   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_BYE_TIMEOUT], 0,
 592       sess->id, ssrc);
 593
 594   if (sess->bin->priv->autoremove)
 595     g_signal_emit_by_name (sess->demux, "clear-ssrc", ssrc, NULL);
 596 }
 597
 598 static void
 599 on_timeout (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 600 {
 601   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_TIMEOUT], 0,
 602       sess->id, ssrc);
 603
 604   if (sess->bin->priv->autoremove)
 605     g_signal_emit_by_name (sess->demux, "clear-ssrc", ssrc, NULL);
 606 }
 607
 608 static void
 609 on_sender_timeout (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 610 {
 611   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SENDER_TIMEOUT], 0,
 612       sess->id, ssrc);
 613 }
 614
 615 static void
 616 on_npt_stop (GstElement * jbuf, GstRtpBinStream * stream)
 617 {
 618   g_signal_emit (stream->bin, gst_rtp_bin_signals[SIGNAL_ON_NPT_STOP], 0,
 619       stream->session->id, stream->ssrc);
 620 }
 621
 622 static void
 623 on_new_sender_ssrc (GstElement * session, guint32 ssrc, GstRtpBinSession * sess)
 624 {
 625   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_NEW_SENDER_SSRC], 0,
 626       sess->id, ssrc);
 627 }
 628
 629 static void
 630 on_sender_ssrc_active (GstElement * session, guint32 ssrc,
 631     GstRtpBinSession * sess)
 632 {
 633   g_signal_emit (sess->bin, gst_rtp_bin_signals[SIGNAL_ON_SENDER_SSRC_ACTIVE],
 634       0, sess->id, ssrc);
 635 }
 636
 637 /* must be called with the SESSION lock */
 638 static GstRtpBinStream *
 639 find_stream_by_ssrc (GstRtpBinSession * session, guint32 ssrc)
 640 {
 641   GSList *walk;
 642
 643   for (walk = session->streams; walk; walk = g_slist_next (walk)) {
 644     GstRtpBinStream *stream = (GstRtpBinStream *) walk->data;
 645
 646     if (stream->ssrc == ssrc)
 647       return stream;
 648   }
 649   return NULL;
 650 }
 651
 652 static void
 653 ssrc_demux_pad_removed (GstElement * element, guint ssrc, GstPad * pad,
 654     GstRtpBinSession * session)
 655 {
 656   GstRtpBinStream *stream = NULL;
 657   GstRtpBin *rtpbin;
 658
 659   rtpbin = session->bin;
 660
 661   GST_RTP_BIN_LOCK (rtpbin);
 662
 663   GST_RTP_SESSION_LOCK (session);
 664   if ((stream = find_stream_by_ssrc (session, ssrc)))
 665     session->streams = g_slist_remove (session->streams, stream);
 666   GST_RTP_SESSION_UNLOCK (session);
 667
 668   if (stream)
 669     free_stream (stream, rtpbin);
 670
 671   GST_RTP_BIN_UNLOCK (rtpbin);
 672 }
 673
 674 static void
 675 new_bundled_ssrc_pad_found (GstElement * element, guint ssrc, GstPad * pad,
 676     GstRtpBinSession * session)
 677 {
 678   GValue result = G_VALUE_INIT;
 679   GValue params[2] = { G_VALUE_INIT, G_VALUE_INIT };
 680   guint session_id = 0;
 681   GstRtpBinSession *target_session = NULL;
 682   GstRtpBin *rtpbin = session->bin;
 683   gchar *name;
 684   GstPad *src_pad;
 685   GstPad *recv_rtp_sink = NULL;
 686   GstPad *recv_rtcp_sink = NULL;
 687   GstPadLinkReturn ret;
 688
 689   GST_RTP_BIN_DYN_LOCK (rtpbin);
 690   GST_DEBUG_OBJECT (rtpbin, "new bundled SSRC pad %08x, %s:%s", ssrc,
 691       GST_DEBUG_PAD_NAME (pad));
 692
 693   g_value_init (&result, G_TYPE_UINT);
 694   g_value_init (&params[0], GST_TYPE_ELEMENT);
 695   g_value_set_object (&params[0], rtpbin);
 696   g_value_init (&params[1], G_TYPE_UINT);
 697   g_value_set_uint (&params[1], ssrc);
 698
 699   g_signal_emitv (params,
 700       gst_rtp_bin_signals[SIGNAL_ON_BUNDLED_SSRC], 0, &result);
 701   g_value_unset (&params[0]);
 702
 703   session_id = g_value_get_uint (&result);
 704   if (session_id == 0) {
 705     target_session = session;
 706   } else {
 707     target_session = find_session_by_id (rtpbin, (gint) session_id);
 708     if (!target_session) {
 709       target_session = create_session (rtpbin, session_id);
 710     }
 711     if (!target_session) {
 712       /* create_session() warned already */
 713       GST_RTP_BIN_DYN_UNLOCK (rtpbin);
 714       return;
 715     }
 716
 717     if (!target_session->recv_rtp_sink) {
 718       recv_rtp_sink = complete_session_sink (rtpbin, target_session, FALSE);
 719     }
 720
 721     if (!target_session->recv_rtp_src)
 722       complete_session_receiver (rtpbin, target_session, session_id);
 723
 724     if (!target_session->recv_rtcp_sink) {
 725       recv_rtcp_sink =
 726           complete_session_rtcp (rtpbin, target_session, session_id, FALSE);
 727     }
 728   }
 729
 730   GST_DEBUG_OBJECT (rtpbin, "Assigning bundled ssrc %u to session %u", ssrc,
 731       session_id);
 732
 733   if (!recv_rtp_sink) {
 734     recv_rtp_sink =
 735         gst_element_get_request_pad (target_session->rtp_funnel, "sink_%u");
 736   }
 737
 738   if (!recv_rtcp_sink) {
 739     recv_rtcp_sink =
 740         gst_element_get_request_pad (target_session->rtcp_funnel, "sink_%u");
 741   }
 742
 743   name = g_strdup_printf ("src_%u", ssrc);
 744   src_pad = gst_element_get_static_pad (element, name);
 745   ret = gst_pad_link (src_pad, recv_rtp_sink);
 746   g_free (name);
 747   gst_object_unref (src_pad);
 748   gst_object_unref (recv_rtp_sink);
 749   if (ret != GST_PAD_LINK_OK) {
 750     g_warning
 751         ("rtpbin: failed to link bundle demuxer to receive rtp funnel for session %u",
 752         session_id);
 753   }
 754
 755   name = g_strdup_printf ("rtcp_src_%u", ssrc);
 756   src_pad = gst_element_get_static_pad (element, name);
 757   gst_pad_link (src_pad, recv_rtcp_sink);
 758   g_free (name);
 759   gst_object_unref (src_pad);
 760   gst_object_unref (recv_rtcp_sink);
 761   if (ret != GST_PAD_LINK_OK) {
 762     g_warning
 763         ("rtpbin: failed to link bundle demuxer to receive rtcp sink pad for session %u",
 764         session_id);
 765   }
 766
 767   GST_RTP_BIN_DYN_UNLOCK (rtpbin);
 768 }
 769
 770 /* create a session with the given id.  Must be called with RTP_BIN_LOCK */
 771 static GstRtpBinSession *
 772 create_session (GstRtpBin * rtpbin, gint id)
 773 {
 774   GstRtpBinSession *sess;
 775   GstElement *session, *demux;
 776   GstElement *storage = NULL;
 777   GstState target;
 778
 779   if (!(session = gst_element_factory_make ("rtpsession", NULL)))
 780     goto no_session;
 781
 782   if (!(demux = gst_element_factory_make ("rtpssrcdemux", NULL)))
 783     goto no_demux;
 784
 785   if (!(storage = gst_element_factory_make ("rtpstorage", NULL)))
 786     goto no_storage;
 787
 788   g_signal_emit (rtpbin, gst_rtp_bin_signals[SIGNAL_NEW_STORAGE], 0, storage,
 789       id);
 790
 791   sess = g_new0 (GstRtpBinSession, 1);
 792   g_mutex_init (&sess->lock);
 793   sess->id = id;
 794   sess->bin = rtpbin;
 795   sess->session = session;
 796   sess->demux = demux;
 797   sess->storage = storage;
 798
 799   sess->rtp_funnel = gst_element_factory_make ("funnel", NULL);
 800   sess->rtcp_funnel = gst_element_factory_make ("funnel", NULL);
 801
 802   sess->ptmap = g_hash_table_new_full (NULL, NULL, NULL,
 803       (GDestroyNotify) gst_caps_unref);
 804   rtpbin->sessions = g_slist_prepend (rtpbin->sessions, sess);
 805
 806   /* configure SDES items */
 807   GST_OBJECT_LOCK (rtpbin);
 808   g_object_set (session, "sdes", rtpbin->sdes, "rtp-profile",
 809       rtpbin->rtp_profile, "rtcp-sync-send-time", rtpbin->rtcp_sync_send_time,
 810       NULL);
 811   if (rtpbin->use_pipeline_clock)
 812     g_object_set (session, "use-pipeline-clock", rtpbin->use_pipeline_clock,
 813         NULL);
 814   else
 815     g_object_set (session, "ntp-time-source", rtpbin->ntp_time_source, NULL);
 816
 817   g_object_set (session, "max-dropout-time", rtpbin->max_dropout_time,
 818       "max-misorder-time", rtpbin->max_misorder_time, NULL);
 819   GST_OBJECT_UNLOCK (rtpbin);
 820
 821   /* provide clock_rate to the session manager when needed */
 822   g_signal_connect (session, "request-pt-map",
 823       (GCallback) pt_map_requested, sess);
 824
 825   g_signal_connect (sess->session, "on-new-ssrc",
 826       (GCallback) on_new_ssrc, sess);
 827   g_signal_connect (sess->session, "on-ssrc-collision",
 828       (GCallback) on_ssrc_collision, sess);
 829   g_signal_connect (sess->session, "on-ssrc-validated",
 830       (GCallback) on_ssrc_validated, sess);
 831   g_signal_connect (sess->session, "on-ssrc-active",
 832       (GCallback) on_ssrc_active, sess);
 833   g_signal_connect (sess->session, "on-ssrc-sdes",
 834       (GCallback) on_ssrc_sdes, sess);
 835   g_signal_connect (sess->session, "on-bye-ssrc",
 836       (GCallback) on_bye_ssrc, sess);
 837   g_signal_connect (sess->session, "on-bye-timeout",
 838       (GCallback) on_bye_timeout, sess);
 839   g_signal_connect (sess->session, "on-timeout", (GCallback) on_timeout, sess);
 840   g_signal_connect (sess->session, "on-sender-timeout",
 841       (GCallback) on_sender_timeout, sess);
 842   g_signal_connect (sess->session, "on-new-sender-ssrc",
 843       (GCallback) on_new_sender_ssrc, sess);
 844   g_signal_connect (sess->session, "on-sender-ssrc-active",
 845       (GCallback) on_sender_ssrc_active, sess);
 846
 847   gst_bin_add (GST_BIN_CAST (rtpbin), session);
 848   gst_bin_add (GST_BIN_CAST (rtpbin), demux);
 849   gst_bin_add (GST_BIN_CAST (rtpbin), sess->rtp_funnel);
 850   gst_bin_add (GST_BIN_CAST (rtpbin), sess->rtcp_funnel);
 851   gst_bin_add (GST_BIN_CAST (rtpbin), storage);
 852
 853   GST_OBJECT_LOCK (rtpbin);
 854   target = GST_STATE_TARGET (rtpbin);
 855   GST_OBJECT_UNLOCK (rtpbin);
 856
 857   /* change state only to what's needed */
 858   gst_element_set_state (demux, target);
 859   gst_element_set_state (session, target);
 860   gst_element_set_state (sess->rtp_funnel, target);
 861   gst_element_set_state (sess->rtcp_funnel, target);
 862   gst_element_set_state (storage, target);
 863
 864   return sess;
 865
 866   /* ERRORS */
 867 no_session:
 868   {
 869     g_warning ("rtpbin: could not create rtpsession element");
 870     return NULL;
 871   }
 872 no_demux:
 873   {
 874     gst_object_unref (session);
 875     g_warning ("rtpbin: could not create rtpssrcdemux element");
 876     return NULL;
 877   }
 878 no_storage:
 879   {
 880     gst_object_unref (session);
 881     gst_object_unref (demux);
 882     g_warning ("rtpbin: could not create rtpstorage element");
 883     return NULL;
 884   }
 885 }
 886
 887 static gboolean
 888 bin_manage_element (GstRtpBin * bin, GstElement * element)
 889 {
 890   GstRtpBinPrivate *priv = bin->priv;
 891
 892   if (g_list_find (priv->elements, element)) {
 893     GST_DEBUG_OBJECT (bin, "requested element %p already in bin", element);
 894   } else {
 895     GST_DEBUG_OBJECT (bin, "adding requested element %p", element);
 896
 897     if (g_object_is_floating (element))
 898       element = gst_object_ref_sink (element);
 899
 900     if (!gst_bin_add (GST_BIN_CAST (bin), element))
 901       goto add_failed;
 902     if (!gst_element_sync_state_with_parent (element))
 903       GST_WARNING_OBJECT (bin, "unable to sync element state with rtpbin");
 904   }
 905   /* we add the element multiple times, each we need an equal number of
 906    * removes to really remove the element from the bin */
 907   priv->elements = g_list_prepend (priv->elements, element);
 908
 909   return TRUE;
 910
 911   /* ERRORS */
 912 add_failed:
 913   {
 914     GST_WARNING_OBJECT (bin, "unable to add element");
 915     gst_object_unref (element);
 916     return FALSE;
 917   }
 918 }
 919
 920 static void
 921 remove_bin_element (GstElement * element, GstRtpBin * bin)
 922 {
 923   GstRtpBinPrivate *priv = bin->priv;
 924   GList *find;
 925
 926   find = g_list_find (priv->elements, element);
 927   if (find) {
 928     priv->elements = g_list_delete_link (priv->elements, find);
 929
 930     if (!g_list_find (priv->elements, element)) {
 931       gst_element_set_locked_state (element, TRUE);
 932       gst_bin_remove (GST_BIN_CAST (bin), element);
 933       gst_element_set_state (element, GST_STATE_NULL);
 934     }
 935
 936     gst_object_unref (element);
 937   }
 938 }
 939
 940 /* called with RTP_BIN_LOCK */
 941 static void
 942 free_session (GstRtpBinSession * sess, GstRtpBin * bin)
 943 {
 944   GST_DEBUG_OBJECT (bin, "freeing session %p", sess);
 945
 946   gst_element_set_locked_state (sess->demux, TRUE);
 947   gst_element_set_locked_state (sess->session, TRUE);
 948
 949   gst_element_set_state (sess->demux, GST_STATE_NULL);
 950   gst_element_set_state (sess->session, GST_STATE_NULL);
 951
 952   remove_recv_rtp (bin, sess);
 953   remove_recv_rtcp (bin, sess);
 954   remove_send_rtp (bin, sess);
 955   remove_rtcp (bin, sess);
 956
 957   gst_bin_remove (GST_BIN_CAST (bin), sess->session);
 958   gst_bin_remove (GST_BIN_CAST (bin), sess->demux);
 959
 960   g_slist_foreach (sess->elements, (GFunc) remove_bin_element, bin);
 961   g_slist_free (sess->elements);
 962
 963   g_slist_foreach (sess->streams, (GFunc) free_stream, bin);
 964   g_slist_free (sess->streams);
 965
 966   g_mutex_clear (&sess->lock);
 967   g_hash_table_destroy (sess->ptmap);
 968
 969   g_free (sess);
 970 }
 971
 972 /* get the payload type caps for the specific payload @pt in @session */
 973 static GstCaps *
 974 get_pt_map (GstRtpBinSession * session, guint pt)
 975 {
 976   GstCaps *caps = NULL;
 977   GstRtpBin *bin;
 978   GValue ret = { 0 };
 979   GValue args[3] = { {0}, {0}, {0} };
 980
 981   GST_DEBUG ("searching pt %u in cache", pt);
 982
 983   GST_RTP_SESSION_LOCK (session);
 984
 985   /* first look in the cache */
 986   caps = g_hash_table_lookup (session->ptmap, GINT_TO_POINTER (pt));
 987   if (caps) {
 988     gst_caps_ref (caps);
 989     goto done;
 990   }
 991
 992   bin = session->bin;
 993
 994   GST_DEBUG ("emiting signal for pt %u in session %u", pt, session->id);
 995
 996   /* not in cache, send signal to request caps */
 997   g_value_init (&args[0], GST_TYPE_ELEMENT);
 998   g_value_set_object (&args[0], bin);
 999   g_value_init (&args[1], G_TYPE_UINT);
1000   g_value_set_uint (&args[1], session->id);
1001   g_value_init (&args[2], G_TYPE_UINT);
1002   g_value_set_uint (&args[2], pt);
1003
1004   g_value_init (&ret, GST_TYPE_CAPS);
1005   g_value_set_boxed (&ret, NULL);
1006
1007   GST_RTP_SESSION_UNLOCK (session);
1008
1009   g_signal_emitv (args, gst_rtp_bin_signals[SIGNAL_REQUEST_PT_MAP], 0, &ret);
1010
1011   GST_RTP_SESSION_LOCK (session);
1012
1013   g_value_unset (&args[0]);
1014   g_value_unset (&args[1]);
1015   g_value_unset (&args[2]);
1016
1017   /* look in the cache again because we let the lock go */
1018   caps = g_hash_table_lookup (session->ptmap, GINT_TO_POINTER (pt));
1019   if (caps) {
1020     gst_caps_ref (caps);
1021     g_value_unset (&ret);
1022     goto done;
1023   }
1024
1025   caps = (GstCaps *) g_value_dup_boxed (&ret);
1026   g_value_unset (&ret);
1027   if (!caps)
1028     goto no_caps;
1029
1030   GST_DEBUG ("caching pt %u as %" GST_PTR_FORMAT, pt, caps);
1031
1032   /* store in cache, take additional ref */
1033   g_hash_table_insert (session->ptmap, GINT_TO_POINTER (pt),
1034       gst_caps_ref (caps));
1035
1036 done:
1037   GST_RTP_SESSION_UNLOCK (session);
1038
1039   return caps;
1040
1041   /* ERRORS */
1042 no_caps:
1043   {
1044     GST_RTP_SESSION_UNLOCK (session);
1045     GST_DEBUG ("no pt map could be obtained");
1046     return NULL;
1047   }
1048 }
1049
1050 static gboolean
1051 return_true (gpointer key, gpointer value, gpointer user_data)
1052 {
1053   return TRUE;
1054 }
1055
1056 static void
1057 gst_rtp_bin_reset_sync (GstRtpBin * rtpbin)
1058 {
1059   GSList *clients, *streams;
1060
1061   GST_DEBUG_OBJECT (rtpbin, "Reset sync on all clients");
1062
1063   GST_RTP_BIN_LOCK (rtpbin);
1064   for (clients = rtpbin->clients; clients; clients = g_slist_next (clients)) {
1065     GstRtpBinClient *client = (GstRtpBinClient *) clients->data;
1066
1067     /* reset sync on all streams for this client */
1068     for (streams = client->streams; streams; streams = g_slist_next (streams)) {
1069       GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
1070
1071       /* make use require a new SR packet for this stream before we attempt new
1072        * lip-sync */
1073       stream->have_sync = FALSE;
1074       stream->rt_delta = 0;
1075       stream->rtp_delta = 0;
1076       stream->clock_base = -100 * GST_SECOND;
1077     }
1078   }
1079   GST_RTP_BIN_UNLOCK (rtpbin);
1080 }
1081
1082 static void
1083 gst_rtp_bin_clear_pt_map (GstRtpBin * bin)
1084 {
1085   GSList *sessions, *streams;
1086
1087   GST_RTP_BIN_LOCK (bin);
1088   GST_DEBUG_OBJECT (bin, "clearing pt map");
1089   for (sessions = bin->sessions; sessions; sessions = g_slist_next (sessions)) {
1090     GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
1091
1092     GST_DEBUG_OBJECT (bin, "clearing session %p", session);
1093     g_signal_emit_by_name (session->session, "clear-pt-map", NULL);
1094
1095     GST_RTP_SESSION_LOCK (session);
1096     g_hash_table_foreach_remove (session->ptmap, return_true, NULL);
1097
1098     for (streams = session->streams; streams; streams = g_slist_next (streams)) {
1099       GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
1100
1101       GST_DEBUG_OBJECT (bin, "clearing stream %p", stream);
1102       g_signal_emit_by_name (stream->buffer, "clear-pt-map", NULL);
1103       if (stream->demux)
1104         g_signal_emit_by_name (stream->demux, "clear-pt-map", NULL);
1105     }
1106     GST_RTP_SESSION_UNLOCK (session);
1107   }
1108   GST_RTP_BIN_UNLOCK (bin);
1109
1110   /* reset sync too */
1111   gst_rtp_bin_reset_sync (bin);
1112 }
1113
1114 static GstElement *
1115 gst_rtp_bin_get_session (GstRtpBin * bin, guint session_id)
1116 {
1117   GstRtpBinSession *session;
1118   GstElement *ret = NULL;
1119
1120   GST_RTP_BIN_LOCK (bin);
1121   GST_DEBUG_OBJECT (bin, "retrieving GstRtpSession, index: %u", session_id);
1122   session = find_session_by_id (bin, (gint) session_id);
1123   if (session) {
1124     ret = gst_object_ref (session->session);
1125   }
1126   GST_RTP_BIN_UNLOCK (bin);
1127
1128   return ret;
1129 }
1130
1131 static RTPSession *
1132 gst_rtp_bin_get_internal_session (GstRtpBin * bin, guint session_id)
1133 {
1134   RTPSession *internal_session = NULL;
1135   GstRtpBinSession *session;
1136
1137   GST_RTP_BIN_LOCK (bin);
1138   GST_DEBUG_OBJECT (bin, "retrieving internal RTPSession object, index: %u",
1139       session_id);
1140   session = find_session_by_id (bin, (gint) session_id);
1141   if (session) {
1142     g_object_get (session->session, "internal-session", &internal_session,
1143         NULL);
1144   }
1145   GST_RTP_BIN_UNLOCK (bin);
1146
1147   return internal_session;
1148 }
1149
1150 static GstElement *
1151 gst_rtp_bin_get_storage (GstRtpBin * bin, guint session_id)
1152 {
1153   GstRtpBinSession *session;
1154   GstElement *res = NULL;
1155
1156   GST_RTP_BIN_LOCK (bin);
1157   GST_DEBUG_OBJECT (bin, "retrieving internal storage object, index: %u",
1158       session_id);
1159   session = find_session_by_id (bin, (gint) session_id);
1160   if (session && session->storage) {
1161     res = gst_object_ref (session->storage);
1162   }
1163   GST_RTP_BIN_UNLOCK (bin);
1164
1165   return res;
1166 }
1167
1168 static GObject *
1169 gst_rtp_bin_get_internal_storage (GstRtpBin * bin, guint session_id)
1170 {
1171   GObject *internal_storage = NULL;
1172   GstRtpBinSession *session;
1173
1174   GST_RTP_BIN_LOCK (bin);
1175   GST_DEBUG_OBJECT (bin, "retrieving internal storage object, index: %u",
1176       session_id);
1177   session = find_session_by_id (bin, (gint) session_id);
1178   if (session && session->storage) {
1179     g_object_get (session->storage, "internal-storage", &internal_storage,
1180         NULL);
1181   }
1182   GST_RTP_BIN_UNLOCK (bin);
1183
1184   return internal_storage;
1185 }
1186
1187 static GstElement *
1188 gst_rtp_bin_request_encoder (GstRtpBin * bin, guint session_id)
1189 {
1190   GST_DEBUG_OBJECT (bin, "return NULL encoder");
1191   return NULL;
1192 }
1193
1194 static GstElement *
1195 gst_rtp_bin_request_decoder (GstRtpBin * bin, guint session_id)
1196 {
1197   GST_DEBUG_OBJECT (bin, "return NULL decoder");
1198   return NULL;
1199 }
1200
1201 static void
1202 gst_rtp_bin_propagate_property_to_jitterbuffer (GstRtpBin * bin,
1203     const gchar * name, const GValue * value)
1204 {
1205   GSList *sessions, *streams;
1206
1207   GST_RTP_BIN_LOCK (bin);
1208   for (sessions = bin->sessions; sessions; sessions = g_slist_next (sessions)) {
1209     GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
1210
1211     GST_RTP_SESSION_LOCK (session);
1212     for (streams = session->streams; streams; streams = g_slist_next (streams)) {
1213       GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
1214
1215       g_object_set_property (G_OBJECT (stream->buffer), name, value);
1216     }
1217     GST_RTP_SESSION_UNLOCK (session);
1218   }
1219   GST_RTP_BIN_UNLOCK (bin);
1220 }
1221
1222 static void
1223 gst_rtp_bin_propagate_property_to_session (GstRtpBin * bin,
1224     const gchar * name, const GValue * value)
1225 {
1226   GSList *sessions;
1227
1228   GST_RTP_BIN_LOCK (bin);
1229   for (sessions = bin->sessions; sessions; sessions = g_slist_next (sessions)) {
1230     GstRtpBinSession *sess = (GstRtpBinSession *) sessions->data;
1231
1232     g_object_set_property (G_OBJECT (sess->session), name, value);
1233   }
1234   GST_RTP_BIN_UNLOCK (bin);
1235 }
1236
1237 /* get a client with the given SDES name. Must be called with RTP_BIN_LOCK */
1238 static GstRtpBinClient *
1239 get_client (GstRtpBin * bin, guint8 len, guint8 * data, gboolean * created)
1240 {
1241   GstRtpBinClient *result = NULL;
1242   GSList *walk;
1243
1244   for (walk = bin->clients; walk; walk = g_slist_next (walk)) {
1245     GstRtpBinClient *client = (GstRtpBinClient *) walk->data;
1246
1247     if (len != client->cname_len)
1248       continue;
1249
1250     if (!strncmp ((gchar *) data, client->cname, client->cname_len)) {
1251       GST_DEBUG_OBJECT (bin, "found existing client %p with CNAME %s", client,
1252           client->cname);
1253       result = client;
1254       break;
1255     }
1256   }
1257
1258   /* nothing found, create one */
1259   if (result == NULL) {
1260     result = g_new0 (GstRtpBinClient, 1);
1261     result->cname = g_strndup ((gchar *) data, len);
1262     result->cname_len = len;
1263     bin->clients = g_slist_prepend (bin->clients, result);
1264     GST_DEBUG_OBJECT (bin, "created new client %p with CNAME %s", result,
1265         result->cname);
1266   }
1267   return result;
1268 }
1269
1270 static void
1271 free_client (GstRtpBinClient * client, GstRtpBin * bin)
1272 {
1273   GST_DEBUG_OBJECT (bin, "freeing client %p", client);
1274   g_slist_free (client->streams);
1275   g_free (client->cname);
1276   g_free (client);
1277 }
1278
1279 static void
1280 get_current_times (GstRtpBin * bin, GstClockTime * running_time,
1281     guint64 * ntpnstime)
1282 {
1283   guint64 ntpns = -1;
1284   GstClock *clock;
1285   GstClockTime base_time, rt, clock_time;
1286
1287   GST_OBJECT_LOCK (bin);
1288   if ((clock = GST_ELEMENT_CLOCK (bin))) {
1289     base_time = GST_ELEMENT_CAST (bin)->base_time;
1290     gst_object_ref (clock);
1291     GST_OBJECT_UNLOCK (bin);
1292
1293     /* get current clock time and convert to running time */
1294     clock_time = gst_clock_get_time (clock);
1295     rt = clock_time - base_time;
1296
1297     if (bin->use_pipeline_clock) {
1298       ntpns = rt;
1299       /* add constant to convert from 1970 based time to 1900 based time */
1300       ntpns += (2208988800LL * GST_SECOND);
1301     } else {
1302       switch (bin->ntp_time_source) {
1303         case GST_RTP_NTP_TIME_SOURCE_NTP:
1304         case GST_RTP_NTP_TIME_SOURCE_UNIX:{
1305           GTimeVal current;
1306
1307           /* get current NTP time */
1308           g_get_current_time (&current);
1309           ntpns = GST_TIMEVAL_TO_TIME (current);
1310
1311           /* add constant to convert from 1970 based time to 1900 based time */
1312           if (bin->ntp_time_source == GST_RTP_NTP_TIME_SOURCE_NTP)
1313             ntpns += (2208988800LL * GST_SECOND);
1314           break;
1315         }
1316         case GST_RTP_NTP_TIME_SOURCE_RUNNING_TIME:
1317           ntpns = rt;
1318           break;
1319         case GST_RTP_NTP_TIME_SOURCE_CLOCK_TIME:
1320           ntpns = clock_time;
1321           break;
1322         default:
1323           ntpns = -1;           /* Fix uninited compiler warning */
1324           g_assert_not_reached ();
1325           break;
1326       }
1327     }
1328
1329     gst_object_unref (clock);
1330   } else {
1331     GST_OBJECT_UNLOCK (bin);
1332     rt = -1;
1333     ntpns = -1;
1334   }
1335   if (running_time)
1336     *running_time = rt;
1337   if (ntpnstime)
1338     *ntpnstime = ntpns;
1339 }
1340
1341 static void
1342 stream_set_ts_offset (GstRtpBin * bin, GstRtpBinStream * stream,
1343     gint64 ts_offset, gint64 max_ts_offset, gint64 min_ts_offset,
1344     gboolean allow_positive_ts_offset)
1345 {
1346   gint64 prev_ts_offset;
1347
1348   g_object_get (stream->buffer, "ts-offset", &prev_ts_offset, NULL);
1349
1350   /* delta changed, see how much */
1351   if (prev_ts_offset != ts_offset) {
1352     gint64 diff;
1353
1354     diff = prev_ts_offset - ts_offset;
1355
1356     GST_DEBUG_OBJECT (bin,
1357         "ts-offset %" G_GINT64_FORMAT ", prev %" G_GINT64_FORMAT
1358         ", diff: %" G_GINT64_FORMAT, ts_offset, prev_ts_offset, diff);
1359
1360     /* ignore minor offsets */
1361     if (ABS (diff) < min_ts_offset) {
1362       GST_DEBUG_OBJECT (bin, "offset too small, ignoring");
1363       return;
1364     }
1365
1366     /* sanity check offset */
1367     if (max_ts_offset > 0) {
1368       if (ts_offset > 0 && !allow_positive_ts_offset) {
1369         GST_DEBUG_OBJECT (bin,
1370             "offset is positive (clocks are out of sync), ignoring");
1371         return;
1372       }
1373       if (ABS (ts_offset) > max_ts_offset) {
1374         GST_DEBUG_OBJECT (bin, "offset too large, ignoring");
1375         return;
1376       }
1377     }
1378
1379     g_object_set (stream->buffer, "ts-offset", ts_offset, NULL);
1380   }
1381   GST_DEBUG_OBJECT (bin, "stream SSRC %08x, delta %" G_GINT64_FORMAT,
1382       stream->ssrc, ts_offset);
1383 }
1384
1385 static void
1386 gst_rtp_bin_send_sync_event (GstRtpBinStream * stream)
1387 {
1388   if (stream->bin->send_sync_event) {
1389     GstEvent *event;
1390     GstPad *srcpad;
1391
1392     GST_DEBUG_OBJECT (stream->bin,
1393         "sending GstRTCPSRReceived event downstream");
1394
1395     event = gst_event_new_custom (GST_EVENT_CUSTOM_DOWNSTREAM,
1396         gst_structure_new_empty ("GstRTCPSRReceived"));
1397
1398     srcpad = gst_element_get_static_pad (stream->buffer, "src");
1399     gst_pad_push_event (srcpad, event);
1400     gst_object_unref (srcpad);
1401   }
1402 }
1403
1404 /* associate a stream to the given CNAME. This will make sure all streams for
1405  * that CNAME are synchronized together.
1406  * Must be called with GST_RTP_BIN_LOCK */
1407 static void
1408 gst_rtp_bin_associate (GstRtpBin * bin, GstRtpBinStream * stream, guint8 len,
1409     guint8 * data, guint64 ntptime, guint64 last_extrtptime,
1410     guint64 base_rtptime, guint64 base_time, guint clock_rate,
1411     gint64 rtp_clock_base)
1412 {
1413   GstRtpBinClient *client;
1414   gboolean created;
1415   GSList *walk;
1416   GstClockTime running_time, running_time_rtp;
1417   guint64 ntpnstime;
1418
1419   /* first find or create the CNAME */
1420   client = get_client (bin, len, data, &created);
1421
1422   /* find stream in the client */
1423   for (walk = client->streams; walk; walk = g_slist_next (walk)) {
1424     GstRtpBinStream *ostream = (GstRtpBinStream *) walk->data;
1425
1426     if (ostream == stream)
1427       break;
1428   }
1429   /* not found, add it to the list */
1430   if (walk == NULL) {
1431     GST_DEBUG_OBJECT (bin,
1432         "new association of SSRC %08x with client %p with CNAME %s",
1433         stream->ssrc, client, client->cname);
1434     client->streams = g_slist_prepend (client->streams, stream);
1435     client->nstreams++;
1436   } else {
1437     GST_DEBUG_OBJECT (bin,
1438         "found association of SSRC %08x with client %p with CNAME %s",
1439         stream->ssrc, client, client->cname);
1440   }
1441
1442   if (!GST_CLOCK_TIME_IS_VALID (last_extrtptime)) {
1443     GST_DEBUG_OBJECT (bin, "invalidated sync data");
1444     if (bin->rtcp_sync == GST_RTP_BIN_RTCP_SYNC_RTP) {
1445       /* we don't need that data, so carry on,
1446        * but make some values look saner */
1447       last_extrtptime = base_rtptime;
1448     } else {
1449       /* nothing we can do with this data in this case */
1450       GST_DEBUG_OBJECT (bin, "bailing out");
1451       return;
1452     }
1453   }
1454
1455   /* Take the extended rtptime we found in the SR packet and map it to the
1456    * local rtptime. The local rtp time is used to construct timestamps on the
1457    * buffers so we will calculate what running_time corresponds to the RTP
1458    * timestamp in the SR packet. */
1459   running_time_rtp = last_extrtptime - base_rtptime;
1460
1461   GST_DEBUG_OBJECT (bin,
1462       "base %" G_GUINT64_FORMAT ", extrtptime %" G_GUINT64_FORMAT
1463       ", local RTP %" G_GUINT64_FORMAT ", clock-rate %d, "
1464       "clock-base %" G_GINT64_FORMAT, base_rtptime,
1465       last_extrtptime, running_time_rtp, clock_rate, rtp_clock_base);
1466
1467   /* calculate local RTP time in gstreamer timestamp, we essentially perform the
1468    * same conversion that a jitterbuffer would use to convert an rtp timestamp
1469    * into a corresponding gstreamer timestamp. Note that the base_time also
1470    * contains the drift between sender and receiver. */
1471   running_time =
1472       gst_util_uint64_scale_int (running_time_rtp, GST_SECOND, clock_rate);
1473   running_time += base_time;
1474
1475   /* convert ntptime to nanoseconds */
1476   ntpnstime = gst_util_uint64_scale (ntptime, GST_SECOND,
1477       (G_GINT64_CONSTANT (1) << 32));
1478
1479   stream->have_sync = TRUE;
1480
1481   GST_DEBUG_OBJECT (bin,
1482       "SR RTP running time %" G_GUINT64_FORMAT ", SR NTP %" G_GUINT64_FORMAT,
1483       running_time, ntpnstime);
1484
1485   /* recalc inter stream playout offset, but only if there is more than one
1486    * stream or we're doing NTP sync. */
1487   if (bin->ntp_sync) {
1488     gint64 ntpdiff, rtdiff;
1489     guint64 local_ntpnstime;
1490     GstClockTime local_running_time;
1491
1492     /* For NTP sync we need to first get a snapshot of running_time and NTP
1493      * time. We know at what running_time we play a certain RTP time, we also
1494      * calculated when we would play the RTP time in the SR packet. Now we need
1495      * to know how the running_time and the NTP time relate to eachother. */
1496     get_current_times (bin, &local_running_time, &local_ntpnstime);
1497
1498     /* see how far away the NTP time is. This is the difference between the
1499      * current NTP time and the NTP time in the last SR packet. */
1500     ntpdiff = local_ntpnstime - ntpnstime;
1501     /* see how far away the running_time is. This is the difference between the
1502      * current running_time and the running_time of the RTP timestamp in the
1503      * last SR packet. */
1504     rtdiff = local_running_time - running_time;
1505
1506     GST_DEBUG_OBJECT (bin,
1507         "local NTP time %" G_GUINT64_FORMAT ", SR NTP time %" G_GUINT64_FORMAT,
1508         local_ntpnstime, ntpnstime);
1509     GST_DEBUG_OBJECT (bin,
1510         "local running time %" G_GUINT64_FORMAT ", SR RTP running time %"
1511         G_GUINT64_FORMAT, local_running_time, running_time);
1512     GST_DEBUG_OBJECT (bin,
1513         "NTP diff %" G_GINT64_FORMAT ", RT diff %" G_GINT64_FORMAT, ntpdiff,
1514         rtdiff);
1515
1516     /* combine to get the final diff to apply to the running_time */
1517     stream->rt_delta = rtdiff - ntpdiff;
1518
1519     stream_set_ts_offset (bin, stream, stream->rt_delta, bin->max_ts_offset,
1520         0, FALSE);
1521   } else {
1522     gint64 min, rtp_min, clock_base = stream->clock_base;
1523     gboolean all_sync, use_rtp;
1524     gboolean rtcp_sync = g_atomic_int_get (&bin->rtcp_sync);
1525
1526     /* calculate delta between server and receiver. ntpnstime is created by
1527      * converting the ntptime in the last SR packet to a gstreamer timestamp. This
1528      * delta expresses the difference to our timeline and the server timeline. The
1529      * difference in itself doesn't mean much but we can combine the delta of
1530      * multiple streams to create a stream specific offset. */
1531     stream->rt_delta = ntpnstime - running_time;
1532
1533     /* calculate the min of all deltas, ignoring streams that did not yet have a
1534      * valid rt_delta because we did not yet receive an SR packet for those
1535      * streams.
1536      * We calculate the mininum because we would like to only apply positive
1537      * offsets to streams, delaying their playback instead of trying to speed up
1538      * other streams (which might be imposible when we have to create negative
1539      * latencies).
1540      * The stream that has the smallest diff is selected as the reference stream,
1541      * all other streams will have a positive offset to this difference. */
1542
1543     /* some alternative setting allow ignoring RTCP as much as possible,
1544      * for servers generating bogus ntp timeline */
1545     min = rtp_min = G_MAXINT64;
1546     use_rtp = FALSE;
1547     if (rtcp_sync == GST_RTP_BIN_RTCP_SYNC_RTP) {
1548       guint64 ext_base;
1549
1550       use_rtp = TRUE;
1551       /* signed version for convienience */
1552       clock_base = base_rtptime;
1553       /* deal with possible wrap-around */
1554       ext_base = base_rtptime;
1555       rtp_clock_base = gst_rtp_buffer_ext_timestamp (&ext_base, rtp_clock_base);
1556       /* sanity check; base rtp and provided clock_base should be close */
1557       if (rtp_clock_base >= clock_base) {
1558         if (rtp_clock_base - clock_base < 10 * clock_rate) {
1559           rtp_clock_base = base_time +
1560               gst_util_uint64_scale_int (rtp_clock_base - clock_base,
1561               GST_SECOND, clock_rate);
1562         } else {
1563           use_rtp = FALSE;
1564         }
1565       } else {
1566         if (clock_base - rtp_clock_base < 10 * clock_rate) {
1567           rtp_clock_base = base_time -
1568               gst_util_uint64_scale_int (clock_base - rtp_clock_base,
1569               GST_SECOND, clock_rate);
1570         } else {
1571           use_rtp = FALSE;
1572         }
1573       }
1574       /* warn and bail for clarity out if no sane values */
1575       if (!use_rtp) {
1576         GST_WARNING_OBJECT (bin, "unable to sync to provided rtptime");
1577         return;
1578       }
1579       /* store to track changes */
1580       clock_base = rtp_clock_base;
1581       /* generate a fake as before,
1582        * now equating rtptime obtained from RTP-Info,
1583        * where the large time represent the otherwise irrelevant npt/ntp time */
1584       stream->rtp_delta = (GST_SECOND << 28) - rtp_clock_base;
1585     } else {
1586       clock_base = rtp_clock_base;
1587     }
1588
1589     all_sync = TRUE;
1590     for (walk = client->streams; walk; walk = g_slist_next (walk)) {
1591       GstRtpBinStream *ostream = (GstRtpBinStream *) walk->data;
1592
1593       if (!ostream->have_sync) {
1594         all_sync = FALSE;
1595         continue;
1596       }
1597
1598       /* change in current stream's base from previously init'ed value
1599        * leads to reset of all stream's base */
1600       if (stream != ostream && stream->clock_base >= 0 &&
1601           (stream->clock_base != clock_base)) {
1602         GST_DEBUG_OBJECT (bin, "reset upon clock base change");
1603         ostream->clock_base = -100 * GST_SECOND;
1604         ostream->rtp_delta = 0;
1605       }
1606
1607       if (ostream->rt_delta < min)
1608         min = ostream->rt_delta;
1609       if (ostream->rtp_delta < rtp_min)
1610         rtp_min = ostream->rtp_delta;
1611     }
1612
1613     /* arrange to re-sync for each stream upon significant change,
1614      * e.g. post-seek */
1615     all_sync = all_sync && (stream->clock_base == clock_base);
1616     stream->clock_base = clock_base;
1617
1618     /* may need init performed above later on, but nothing more to do now */
1619     if (client->nstreams <= 1)
1620       return;
1621
1622     GST_DEBUG_OBJECT (bin, "client %p min delta %" G_GINT64_FORMAT
1623         " all sync %d", client, min, all_sync);
1624     GST_DEBUG_OBJECT (bin, "rtcp sync mode %d, use_rtp %d", rtcp_sync, use_rtp);
1625
1626     switch (rtcp_sync) {
1627       case GST_RTP_BIN_RTCP_SYNC_RTP:
1628         if (!use_rtp)
1629           break;
1630         GST_DEBUG_OBJECT (bin, "using rtp generated reports; "
1631             "client %p min rtp delta %" G_GINT64_FORMAT, client, rtp_min);
1632         /* fall-through */
1633       case GST_RTP_BIN_RTCP_SYNC_INITIAL:
1634         /* if all have been synced already, do not bother further */
1635         if (all_sync) {
1636           GST_DEBUG_OBJECT (bin, "all streams already synced; done");
1637           return;
1638         }
1639         break;
1640       default:
1641         break;
1642     }
1643
1644     /* bail out if we adjusted recently enough */
1645     if (all_sync && (ntpnstime - bin->priv->last_ntpnstime) <
1646         bin->rtcp_sync_interval * GST_MSECOND) {
1647       GST_DEBUG_OBJECT (bin, "discarding RTCP sender packet for sync; "
1648           "previous sender info too recent "
1649           "(previous NTP %" G_GUINT64_FORMAT ")", bin->priv->last_ntpnstime);
1650       return;
1651     }
1652     bin->priv->last_ntpnstime = ntpnstime;
1653
1654     /* calculate offsets for each stream */
1655     for (walk = client->streams; walk; walk = g_slist_next (walk)) {
1656       GstRtpBinStream *ostream = (GstRtpBinStream *) walk->data;
1657       gint64 ts_offset;
1658
1659       /* ignore streams for which we didn't receive an SR packet yet, we
1660        * can't synchronize them yet. We can however sync other streams just
1661        * fine. */
1662       if (!ostream->have_sync)
1663         continue;
1664
1665       /* calculate offset to our reference stream, this should always give a
1666        * positive number. */
1667       if (use_rtp)
1668         ts_offset = ostream->rtp_delta - rtp_min;
1669       else
1670         ts_offset = ostream->rt_delta - min;
1671
1672       stream_set_ts_offset (bin, ostream, ts_offset, bin->max_ts_offset,
1673           MIN_TS_OFFSET, TRUE);
1674     }
1675   }
1676   gst_rtp_bin_send_sync_event (stream);
1677
1678   return;
1679 }
1680
1681 #define GST_RTCP_BUFFER_FOR_PACKETS(b,buffer,packet) \
1682   for ((b) = gst_rtcp_buffer_get_first_packet ((buffer), (packet)); (b); \
1683           (b) = gst_rtcp_packet_move_to_next ((packet)))
1684
1685 #define GST_RTCP_SDES_FOR_ITEMS(b,packet) \
1686   for ((b) = gst_rtcp_packet_sdes_first_item ((packet)); (b); \
1687           (b) = gst_rtcp_packet_sdes_next_item ((packet)))
1688
1689 #define GST_RTCP_SDES_FOR_ENTRIES(b,packet) \
1690   for ((b) = gst_rtcp_packet_sdes_first_entry ((packet)); (b); \
1691           (b) = gst_rtcp_packet_sdes_next_entry ((packet)))
1692
1693 static void
1694 gst_rtp_bin_handle_sync (GstElement * jitterbuffer, GstStructure * s,
1695     GstRtpBinStream * stream)
1696 {
1697   GstRtpBin *bin;
1698   GstRTCPPacket packet;
1699   guint32 ssrc;
1700   guint64 ntptime;
1701   gboolean have_sr, have_sdes;
1702   gboolean more;
1703   guint64 base_rtptime;
1704   guint64 base_time;
1705   guint clock_rate;
1706   guint64 clock_base;
1707   guint64 extrtptime;
1708   GstBuffer *buffer;
1709   GstRTCPBuffer rtcp = { NULL, };
1710
1711   bin = stream->bin;
1712
1713   GST_DEBUG_OBJECT (bin, "sync handler called");
1714
1715   /* get the last relation between the rtp timestamps and the gstreamer
1716    * timestamps. We get this info directly from the jitterbuffer which
1717    * constructs gstreamer timestamps from rtp timestamps and so it know exactly
1718    * what the current situation is. */
1719   base_rtptime =
1720       g_value_get_uint64 (gst_structure_get_value (s, "base-rtptime"));
1721   base_time = g_value_get_uint64 (gst_structure_get_value (s, "base-time"));
1722   clock_rate = g_value_get_uint (gst_structure_get_value (s, "clock-rate"));
1723   clock_base = g_value_get_uint64 (gst_structure_get_value (s, "clock-base"));
1724   extrtptime =
1725       g_value_get_uint64 (gst_structure_get_value (s, "sr-ext-rtptime"));
1726   buffer = gst_value_get_buffer (gst_structure_get_value (s, "sr-buffer"));
1727
1728   have_sr = FALSE;
1729   have_sdes = FALSE;
1730
1731   gst_rtcp_buffer_map (buffer, GST_MAP_READ, &rtcp);
1732
1733   GST_RTCP_BUFFER_FOR_PACKETS (more, &rtcp, &packet) {
1734     /* first packet must be SR or RR or else the validate would have failed */
1735     switch (gst_rtcp_packet_get_type (&packet)) {
1736       case GST_RTCP_TYPE_SR:
1737         /* only parse first. There is only supposed to be one SR in the packet
1738          * but we will deal with malformed packets gracefully */
1739         if (have_sr)
1740           break;
1741         /* get NTP and RTP times */
1742         gst_rtcp_packet_sr_get_sender_info (&packet, &ssrc, &ntptime, NULL,
1743             NULL, NULL);
1744
1745         GST_DEBUG_OBJECT (bin, "received sync packet from SSRC %08x", ssrc);
1746         /* ignore SR that is not ours */
1747         if (ssrc != stream->ssrc)
1748           continue;
1749
1750         have_sr = TRUE;
1751         break;
1752       case GST_RTCP_TYPE_SDES:
1753       {
1754         gboolean more_items, more_entries;
1755
1756         /* only deal with first SDES, there is only supposed to be one SDES in
1757          * the RTCP packet but we deal with bad packets gracefully. Also bail
1758          * out if we have not seen an SR item yet. */
1759         if (have_sdes || !have_sr)
1760           break;
1761
1762         GST_RTCP_SDES_FOR_ITEMS (more_items, &packet) {
1763           /* skip items that are not about the SSRC of the sender */
1764           if (gst_rtcp_packet_sdes_get_ssrc (&packet) != ssrc)
1765             continue;
1766
1767           /* find the CNAME entry */
1768           GST_RTCP_SDES_FOR_ENTRIES (more_entries, &packet) {
1769             GstRTCPSDESType type;
1770             guint8 len;
1771             guint8 *data;
1772
1773             gst_rtcp_packet_sdes_get_entry (&packet, &type, &len, &data);
1774
1775             if (type == GST_RTCP_SDES_CNAME) {
1776               GST_RTP_BIN_LOCK (bin);
1777               /* associate the stream to CNAME */
1778               gst_rtp_bin_associate (bin, stream, len, data,
1779                   ntptime, extrtptime, base_rtptime, base_time, clock_rate,
1780                   clock_base);
1781               GST_RTP_BIN_UNLOCK (bin);
1782             }
1783           }
1784         }
1785         have_sdes = TRUE;
1786         break;
1787       }
1788       default:
1789         /* we can ignore these packets */
1790         break;
1791     }
1792   }
1793   gst_rtcp_buffer_unmap (&rtcp);
1794 }
1795
1796 /* create a new stream with @ssrc in @session. Must be called with
1797  * RTP_SESSION_LOCK. */
1798 static GstRtpBinStream *
1799 create_stream (GstRtpBinSession * session, guint32 ssrc)
1800 {
1801   GstElement *buffer, *demux = NULL;
1802   GstRtpBinStream *stream;
1803   GstRtpBin *rtpbin;
1804   GstState target;
1805
1806   rtpbin = session->bin;
1807
1808   if (g_slist_length (session->streams) >= rtpbin->max_streams)
1809     goto max_streams;
1810
1811   if (!(buffer = gst_element_factory_make ("rtpjitterbuffer", NULL)))
1812     goto no_jitterbuffer;
1813
1814   if (!rtpbin->ignore_pt) {
1815     if (!(demux = gst_element_factory_make ("rtpptdemux", NULL)))
1816       goto no_demux;
1817   }
1818
1819   stream = g_new0 (GstRtpBinStream, 1);
1820   stream->ssrc = ssrc;
1821   stream->bin = rtpbin;
1822   stream->session = session;
1823   stream->buffer = buffer;
1824   stream->demux = demux;
1825
1826   stream->have_sync = FALSE;
1827   stream->rt_delta = 0;
1828   stream->rtp_delta = 0;
1829   stream->percent = 100;
1830   stream->clock_base = -100 * GST_SECOND;
1831   session->streams = g_slist_prepend (session->streams, stream);
1832
1833   /* provide clock_rate to the jitterbuffer when needed */
1834   stream->buffer_ptreq_sig = g_signal_connect (buffer, "request-pt-map",
1835       (GCallback) pt_map_requested, session);
1836   stream->buffer_ntpstop_sig = g_signal_connect (buffer, "on-npt-stop",
1837       (GCallback) on_npt_stop, stream);
1838
1839   g_object_set_data (G_OBJECT (buffer), "GstRTPBin.session", session);
1840   g_object_set_data (G_OBJECT (buffer), "GstRTPBin.stream", stream);
1841
1842   /* configure latency and packet lost */
1843   g_object_set (buffer, "latency", rtpbin->latency_ms, NULL);
1844   g_object_set (buffer, "drop-on-latency", rtpbin->drop_on_latency, NULL);
1845   g_object_set (buffer, "do-lost", rtpbin->do_lost, NULL);
1846   g_object_set (buffer, "mode", rtpbin->buffer_mode, NULL);
1847   g_object_set (buffer, "do-retransmission", rtpbin->do_retransmission, NULL);
1848   g_object_set (buffer, "max-rtcp-rtp-time-diff",
1849       rtpbin->max_rtcp_rtp_time_diff, NULL);
1850   g_object_set (buffer, "max-dropout-time", rtpbin->max_dropout_time,
1851       "max-misorder-time", rtpbin->max_misorder_time, NULL);
1852   g_object_set (buffer, "rfc7273-sync", rtpbin->rfc7273_sync, NULL);
1853   g_object_set (buffer, "max-ts-offset-adjustment",
1854       rtpbin->max_ts_offset_adjustment, NULL);
1855
1856   g_signal_emit (rtpbin, gst_rtp_bin_signals[SIGNAL_NEW_JITTERBUFFER], 0,
1857       buffer, session->id, ssrc);
1858
1859   if (!rtpbin->ignore_pt)
1860     gst_bin_add (GST_BIN_CAST (rtpbin), demux);
1861   gst_bin_add (GST_BIN_CAST (rtpbin), buffer);
1862
1863   /* link stuff */
1864   if (demux)
1865     gst_element_link_pads_full (buffer, "src", demux, "sink",
1866         GST_PAD_LINK_CHECK_NOTHING);
1867
1868   if (rtpbin->buffering) {
1869     guint64 last_out;
1870
1871     GST_INFO_OBJECT (rtpbin,
1872         "bin is buffering, set jitterbuffer as not active");
1873     g_signal_emit_by_name (buffer, "set-active", FALSE, (gint64) 0, &last_out);
1874   }
1875
1876
1877   GST_OBJECT_LOCK (rtpbin);
1878   target = GST_STATE_TARGET (rtpbin);
1879   GST_OBJECT_UNLOCK (rtpbin);
1880
1881   /* from sink to source */
1882   if (demux)
1883     gst_element_set_state (demux, target);
1884
1885   gst_element_set_state (buffer, target);
1886
1887   return stream;
1888
1889   /* ERRORS */
1890 max_streams:
1891   {
1892     GST_WARNING_OBJECT (rtpbin, "stream exeeds maximum (%d)",
1893         rtpbin->max_streams);
1894     return NULL;
1895   }
1896 no_jitterbuffer:
1897   {
1898     g_warning ("rtpbin: could not create rtpjitterbuffer element");
1899     return NULL;
1900   }
1901 no_demux:
1902   {
1903     gst_object_unref (buffer);
1904     g_warning ("rtpbin: could not create rtpptdemux element");
1905     return NULL;
1906   }
1907 }
1908
1909 /* called with RTP_BIN_LOCK */
1910 static void
1911 free_stream (GstRtpBinStream * stream, GstRtpBin * bin)
1912 {
1913   GSList *clients, *next_client;
1914
1915   GST_DEBUG_OBJECT (bin, "freeing stream %p", stream);
1916
1917   if (stream->demux) {
1918     g_signal_handler_disconnect (stream->demux, stream->demux_newpad_sig);
1919     g_signal_handler_disconnect (stream->demux, stream->demux_ptreq_sig);
1920     g_signal_handler_disconnect (stream->demux, stream->demux_ptchange_sig);
1921   }
1922   g_signal_handler_disconnect (stream->buffer, stream->buffer_handlesync_sig);
1923   g_signal_handler_disconnect (stream->buffer, stream->buffer_ptreq_sig);
1924   g_signal_handler_disconnect (stream->buffer, stream->buffer_ntpstop_sig);
1925
1926   if (stream->demux)
1927     gst_element_set_locked_state (stream->demux, TRUE);
1928   gst_element_set_locked_state (stream->buffer, TRUE);
1929
1930   if (stream->demux)
1931     gst_element_set_state (stream->demux, GST_STATE_NULL);
1932   gst_element_set_state (stream->buffer, GST_STATE_NULL);
1933
1934   /* now remove this signal, we need this while going to NULL because it to
1935    * do some cleanups */
1936   if (stream->demux)
1937     g_signal_handler_disconnect (stream->demux, stream->demux_padremoved_sig);
1938
1939   gst_bin_remove (GST_BIN_CAST (bin), stream->buffer);
1940   if (stream->demux)
1941     gst_bin_remove (GST_BIN_CAST (bin), stream->demux);
1942
1943   for (clients = bin->clients; clients; clients = next_client) {
1944     GstRtpBinClient *client = (GstRtpBinClient *) clients->data;
1945     GSList *streams, *next_stream;
1946
1947     next_client = g_slist_next (clients);
1948
1949     for (streams = client->streams; streams; streams = next_stream) {
1950       GstRtpBinStream *ostream = (GstRtpBinStream *) streams->data;
1951
1952       next_stream = g_slist_next (streams);
1953
1954       if (ostream == stream) {
1955         client->streams = g_slist_delete_link (client->streams, streams);
1956         /* If this was the last stream belonging to this client,
1957          * clean up the client. */
1958         if (--client->nstreams == 0) {
1959           bin->clients = g_slist_delete_link (bin->clients, clients);
1960           free_client (client, bin);
1961           break;
1962         }
1963       }
1964     }
1965   }
1966   g_free (stream);
1967 }
1968
1969 /* GObject vmethods */
1970 static void gst_rtp_bin_dispose (GObject * object);
1971 static void gst_rtp_bin_finalize (GObject * object);
1972 static void gst_rtp_bin_set_property (GObject * object, guint prop_id,
1973     const GValue * value, GParamSpec * pspec);
1974 static void gst_rtp_bin_get_property (GObject * object, guint prop_id,
1975     GValue * value, GParamSpec * pspec);
1976
1977 /* GstElement vmethods */
1978 static GstStateChangeReturn gst_rtp_bin_change_state (GstElement * element,
1979     GstStateChange transition);
1980 static GstPad *gst_rtp_bin_request_new_pad (GstElement * element,
1981     GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
1982 static void gst_rtp_bin_release_pad (GstElement * element, GstPad * pad);
1983 static void gst_rtp_bin_handle_message (GstBin * bin, GstMessage * message);
1984
1985 #define gst_rtp_bin_parent_class parent_class
1986 G_DEFINE_TYPE (GstRtpBin, gst_rtp_bin, GST_TYPE_BIN);
1987
1988 static gboolean
1989 _gst_element_accumulator (GSignalInvocationHint * ihint,
1990     GValue * return_accu, const GValue * handler_return, gpointer dummy)
1991 {
1992   GstElement *element;
1993
1994   element = g_value_get_object (handler_return);
1995   GST_DEBUG ("got element %" GST_PTR_FORMAT, element);
1996
1997   if (!(ihint->run_type & G_SIGNAL_RUN_CLEANUP))
1998     g_value_set_object (return_accu, element);
1999
2000   /* stop emission if we have an element */
2001   return (element == NULL);
2002 }
2003
2004 static gboolean
2005 _gst_caps_accumulator (GSignalInvocationHint * ihint,
2006     GValue * return_accu, const GValue * handler_return, gpointer dummy)
2007 {
2008   GstCaps *caps;
2009
2010   caps = g_value_get_boxed (handler_return);
2011   GST_DEBUG ("got caps %" GST_PTR_FORMAT, caps);
2012
2013   if (!(ihint->run_type & G_SIGNAL_RUN_CLEANUP))
2014     g_value_set_boxed (return_accu, caps);
2015
2016   /* stop emission if we have a caps */
2017   return (caps == NULL);
2018 }
2019
2020 static void
2021 gst_rtp_bin_class_init (GstRtpBinClass * klass)
2022 {
2023   GObjectClass *gobject_class;
2024   GstElementClass *gstelement_class;
2025   GstBinClass *gstbin_class;
2026
2027   gobject_class = (GObjectClass *) klass;
2028   gstelement_class = (GstElementClass *) klass;
2029   gstbin_class = (GstBinClass *) klass;
2030
2031   g_type_class_add_private (klass, sizeof (GstRtpBinPrivate));
2032
2033   gobject_class->dispose = gst_rtp_bin_dispose;
2034   gobject_class->finalize = gst_rtp_bin_finalize;
2035   gobject_class->set_property = gst_rtp_bin_set_property;
2036   gobject_class->get_property = gst_rtp_bin_get_property;
2037
2038   g_object_class_install_property (gobject_class, PROP_LATENCY,
2039       g_param_spec_uint ("latency", "Buffer latency in ms",
2040           "Default amount of ms to buffer in the jitterbuffers", 0,
2041           G_MAXUINT, DEFAULT_LATENCY_MS,
2042           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2043
2044   g_object_class_install_property (gobject_class, PROP_DROP_ON_LATENCY,
2045       g_param_spec_boolean ("drop-on-latency",
2046           "Drop buffers when maximum latency is reached",
2047           "Tells the jitterbuffer to never exceed the given latency in size",
2048           DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2049
2050   /**
2051    * GstRtpBin::request-pt-map:
2052    * @rtpbin: the object which received the signal
2053    * @session: the session
2054    * @pt: the pt
2055    *
2056    * Request the payload type as #GstCaps for @pt in @session.
2057    */
2058   gst_rtp_bin_signals[SIGNAL_REQUEST_PT_MAP] =
2059       g_signal_new ("request-pt-map", G_TYPE_FROM_CLASS (klass),
2060       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, request_pt_map),
2061       _gst_caps_accumulator, NULL, g_cclosure_marshal_generic, GST_TYPE_CAPS,
2062       2, G_TYPE_UINT, G_TYPE_UINT);
2063
2064     /**
2065    * GstRtpBin::payload-type-change:
2066    * @rtpbin: the object which received the signal
2067    * @session: the session
2068    * @pt: the pt
2069    *
2070    * Signal that the current payload type changed to @pt in @session.
2071    */
2072   gst_rtp_bin_signals[SIGNAL_PAYLOAD_TYPE_CHANGE] =
2073       g_signal_new ("payload-type-change", G_TYPE_FROM_CLASS (klass),
2074       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, payload_type_change),
2075       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2076       G_TYPE_UINT);
2077
2078   /**
2079    * GstRtpBin::clear-pt-map:
2080    * @rtpbin: the object which received the signal
2081    *
2082    * Clear all previously cached pt-mapping obtained with
2083    * #GstRtpBin::request-pt-map.
2084    */
2085   gst_rtp_bin_signals[SIGNAL_CLEAR_PT_MAP] =
2086       g_signal_new ("clear-pt-map", G_TYPE_FROM_CLASS (klass),
2087       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpBinClass,
2088           clear_pt_map), NULL, NULL, g_cclosure_marshal_VOID__VOID, G_TYPE_NONE,
2089       0, G_TYPE_NONE);
2090
2091   /**
2092    * GstRtpBin::reset-sync:
2093    * @rtpbin: the object which received the signal
2094    *
2095    * Reset all currently configured lip-sync parameters and require new SR
2096    * packets for all streams before lip-sync is attempted again.
2097    */
2098   gst_rtp_bin_signals[SIGNAL_RESET_SYNC] =
2099       g_signal_new ("reset-sync", G_TYPE_FROM_CLASS (klass),
2100       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpBinClass,
2101           reset_sync), NULL, NULL, g_cclosure_marshal_VOID__VOID, G_TYPE_NONE,
2102       0, G_TYPE_NONE);
2103
2104   /**
2105    * GstRtpBin::get-session:
2106    * @rtpbin: the object which received the signal
2107    * @id: the session id
2108    *
2109    * Request the related GstRtpSession as #GstElement related with session @id.
2110    *
2111    * Since: 1.8
2112    */
2113   gst_rtp_bin_signals[SIGNAL_GET_SESSION] =
2114       g_signal_new ("get-session", G_TYPE_FROM_CLASS (klass),
2115       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpBinClass,
2116           get_session), NULL, NULL, g_cclosure_marshal_generic,
2117       GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2118
2119   /**
2120    * GstRtpBin::get-internal-session:
2121    * @rtpbin: the object which received the signal
2122    * @id: the session id
2123    *
2124    * Request the internal RTPSession object as #GObject in session @id.
2125    */
2126   gst_rtp_bin_signals[SIGNAL_GET_INTERNAL_SESSION] =
2127       g_signal_new ("get-internal-session", G_TYPE_FROM_CLASS (klass),
2128       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpBinClass,
2129           get_internal_session), NULL, NULL, g_cclosure_marshal_generic,
2130       RTP_TYPE_SESSION, 1, G_TYPE_UINT);
2131
2132   /**
2133    * GstRtpBin::get-internal-storage:
2134    * @rtpbin: the object which received the signal
2135    * @id: the session id
2136    *
2137    * Request the internal RTPStorage object as #GObject in session @id.
2138    *
2139    * Since: 1.14
2140    */
2141   gst_rtp_bin_signals[SIGNAL_GET_INTERNAL_STORAGE] =
2142       g_signal_new ("get-internal-storage", G_TYPE_FROM_CLASS (klass),
2143       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpBinClass,
2144           get_internal_storage), NULL, NULL, g_cclosure_marshal_generic,
2145       G_TYPE_OBJECT, 1, G_TYPE_UINT);
2146
2147   /**
2148    * GstRtpBin::get-storage:
2149    * @rtpbin: the object which received the signal
2150    * @id: the session id
2151    *
2152    * Request the RTPStorage element as #GObject in session @id.
2153    *
2154    * Since: 1.16
2155    */
2156   gst_rtp_bin_signals[SIGNAL_GET_STORAGE] =
2157       g_signal_new ("get-storage", G_TYPE_FROM_CLASS (klass),
2158       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpBinClass,
2159           get_storage), NULL, NULL, g_cclosure_marshal_generic,
2160       GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2161
2162   /**
2163    * GstRtpBin::on-new-ssrc:
2164    * @rtpbin: the object which received the signal
2165    * @session: the session
2166    * @ssrc: the SSRC
2167    *
2168    * Notify of a new SSRC that entered @session.
2169    */
2170   gst_rtp_bin_signals[SIGNAL_ON_NEW_SSRC] =
2171       g_signal_new ("on-new-ssrc", G_TYPE_FROM_CLASS (klass),
2172       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_new_ssrc),
2173       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2174       G_TYPE_UINT);
2175   /**
2176    * GstRtpBin::on-ssrc-collision:
2177    * @rtpbin: the object which received the signal
2178    * @session: the session
2179    * @ssrc: the SSRC
2180    *
2181    * Notify when we have an SSRC collision
2182    */
2183   gst_rtp_bin_signals[SIGNAL_ON_SSRC_COLLISION] =
2184       g_signal_new ("on-ssrc-collision", G_TYPE_FROM_CLASS (klass),
2185       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_ssrc_collision),
2186       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2187       G_TYPE_UINT);
2188   /**
2189    * GstRtpBin::on-ssrc-validated:
2190    * @rtpbin: the object which received the signal
2191    * @session: the session
2192    * @ssrc: the SSRC
2193    *
2194    * Notify of a new SSRC that became validated.
2195    */
2196   gst_rtp_bin_signals[SIGNAL_ON_SSRC_VALIDATED] =
2197       g_signal_new ("on-ssrc-validated", G_TYPE_FROM_CLASS (klass),
2198       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_ssrc_validated),
2199       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2200       G_TYPE_UINT);
2201   /**
2202    * GstRtpBin::on-ssrc-active:
2203    * @rtpbin: the object which received the signal
2204    * @session: the session
2205    * @ssrc: the SSRC
2206    *
2207    * Notify of a SSRC that is active, i.e., sending RTCP.
2208    */
2209   gst_rtp_bin_signals[SIGNAL_ON_SSRC_ACTIVE] =
2210       g_signal_new ("on-ssrc-active", G_TYPE_FROM_CLASS (klass),
2211       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_ssrc_active),
2212       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2213       G_TYPE_UINT);
2214   /**
2215    * GstRtpBin::on-ssrc-sdes:
2216    * @rtpbin: the object which received the signal
2217    * @session: the session
2218    * @ssrc: the SSRC
2219    *
2220    * Notify of a SSRC that is active, i.e., sending RTCP.
2221    */
2222   gst_rtp_bin_signals[SIGNAL_ON_SSRC_SDES] =
2223       g_signal_new ("on-ssrc-sdes", G_TYPE_FROM_CLASS (klass),
2224       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_ssrc_sdes),
2225       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2226       G_TYPE_UINT);
2227
2228   /**
2229    * GstRtpBin::on-bye-ssrc:
2230    * @rtpbin: the object which received the signal
2231    * @session: the session
2232    * @ssrc: the SSRC
2233    *
2234    * Notify of an SSRC that became inactive because of a BYE packet.
2235    */
2236   gst_rtp_bin_signals[SIGNAL_ON_BYE_SSRC] =
2237       g_signal_new ("on-bye-ssrc", G_TYPE_FROM_CLASS (klass),
2238       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_bye_ssrc),
2239       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2240       G_TYPE_UINT);
2241   /**
2242    * GstRtpBin::on-bye-timeout:
2243    * @rtpbin: the object which received the signal
2244    * @session: the session
2245    * @ssrc: the SSRC
2246    *
2247    * Notify of an SSRC that has timed out because of BYE
2248    */
2249   gst_rtp_bin_signals[SIGNAL_ON_BYE_TIMEOUT] =
2250       g_signal_new ("on-bye-timeout", G_TYPE_FROM_CLASS (klass),
2251       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_bye_timeout),
2252       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2253       G_TYPE_UINT);
2254   /**
2255    * GstRtpBin::on-timeout:
2256    * @rtpbin: the object which received the signal
2257    * @session: the session
2258    * @ssrc: the SSRC
2259    *
2260    * Notify of an SSRC that has timed out
2261    */
2262   gst_rtp_bin_signals[SIGNAL_ON_TIMEOUT] =
2263       g_signal_new ("on-timeout", G_TYPE_FROM_CLASS (klass),
2264       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_timeout),
2265       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2266       G_TYPE_UINT);
2267   /**
2268    * GstRtpBin::on-sender-timeout:
2269    * @rtpbin: the object which received the signal
2270    * @session: the session
2271    * @ssrc: the SSRC
2272    *
2273    * Notify of a sender SSRC that has timed out and became a receiver
2274    */
2275   gst_rtp_bin_signals[SIGNAL_ON_SENDER_TIMEOUT] =
2276       g_signal_new ("on-sender-timeout", G_TYPE_FROM_CLASS (klass),
2277       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_sender_timeout),
2278       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2279       G_TYPE_UINT);
2280
2281   /**
2282    * GstRtpBin::on-npt-stop:
2283    * @rtpbin: the object which received the signal
2284    * @session: the session
2285    * @ssrc: the SSRC
2286    *
2287    * Notify that SSRC sender has sent data up to the configured NPT stop time.
2288    */
2289   gst_rtp_bin_signals[SIGNAL_ON_NPT_STOP] =
2290       g_signal_new ("on-npt-stop", G_TYPE_FROM_CLASS (klass),
2291       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_npt_stop),
2292       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2293       G_TYPE_UINT);
2294
2295   /**
2296    * GstRtpBin::request-rtp-encoder:
2297    * @rtpbin: the object which received the signal
2298    * @session: the session
2299    *
2300    * Request an RTP encoder element for the given @session. The encoder
2301    * element will be added to the bin if not previously added.
2302    *
2303    * If no handler is connected, no encoder will be used.
2304    *
2305    * Since: 1.4
2306    */
2307   gst_rtp_bin_signals[SIGNAL_REQUEST_RTP_ENCODER] =
2308       g_signal_new ("request-rtp-encoder", G_TYPE_FROM_CLASS (klass),
2309       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2310           request_rtp_encoder), _gst_element_accumulator, NULL,
2311       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2312
2313   /**
2314    * GstRtpBin::request-rtp-decoder:
2315    * @rtpbin: the object which received the signal
2316    * @session: the session
2317    *
2318    * Request an RTP decoder element for the given @session. The decoder
2319    * element will be added to the bin if not previously added.
2320    *
2321    * If no handler is connected, no encoder will be used.
2322    *
2323    * Since: 1.4
2324    */
2325   gst_rtp_bin_signals[SIGNAL_REQUEST_RTP_DECODER] =
2326       g_signal_new ("request-rtp-decoder", G_TYPE_FROM_CLASS (klass),
2327       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2328           request_rtp_decoder), _gst_element_accumulator, NULL,
2329       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2330
2331   /**
2332    * GstRtpBin::request-rtcp-encoder:
2333    * @rtpbin: the object which received the signal
2334    * @session: the session
2335    *
2336    * Request an RTCP encoder element for the given @session. The encoder
2337    * element will be added to the bin if not previously added.
2338    *
2339    * If no handler is connected, no encoder will be used.
2340    *
2341    * Since: 1.4
2342    */
2343   gst_rtp_bin_signals[SIGNAL_REQUEST_RTCP_ENCODER] =
2344       g_signal_new ("request-rtcp-encoder", G_TYPE_FROM_CLASS (klass),
2345       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2346           request_rtcp_encoder), _gst_element_accumulator, NULL,
2347       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2348
2349   /**
2350    * GstRtpBin::request-rtcp-decoder:
2351    * @rtpbin: the object which received the signal
2352    * @session: the session
2353    *
2354    * Request an RTCP decoder element for the given @session. The decoder
2355    * element will be added to the bin if not previously added.
2356    *
2357    * If no handler is connected, no encoder will be used.
2358    *
2359    * Since: 1.4
2360    */
2361   gst_rtp_bin_signals[SIGNAL_REQUEST_RTCP_DECODER] =
2362       g_signal_new ("request-rtcp-decoder", G_TYPE_FROM_CLASS (klass),
2363       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2364           request_rtcp_decoder), _gst_element_accumulator, NULL,
2365       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2366
2367   /**
2368    * GstRtpBin::new-jitterbuffer:
2369    * @rtpbin: the object which received the signal
2370    * @jitterbuffer: the new jitterbuffer
2371    * @session: the session
2372    * @ssrc: the SSRC
2373    *
2374    * Notify that a new @jitterbuffer was created for @session and @ssrc.
2375    * This signal can, for example, be used to configure @jitterbuffer.
2376    *
2377    * Since: 1.4
2378    */
2379   gst_rtp_bin_signals[SIGNAL_NEW_JITTERBUFFER] =
2380       g_signal_new ("new-jitterbuffer", G_TYPE_FROM_CLASS (klass),
2381       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2382           new_jitterbuffer), NULL, NULL, g_cclosure_marshal_generic,
2383       G_TYPE_NONE, 3, GST_TYPE_ELEMENT, G_TYPE_UINT, G_TYPE_UINT);
2384
2385   /**
2386    * GstRtpBin::new-storage:
2387    * @rtpbin: the object which received the signal
2388    * @storage: the new storage
2389    * @session: the session
2390    *
2391    * Notify that a new @storage was created for @session.
2392    * This signal can, for example, be used to configure @storage.
2393    *
2394    * Since: 1.14
2395    */
2396   gst_rtp_bin_signals[SIGNAL_NEW_STORAGE] =
2397       g_signal_new ("new-storage", G_TYPE_FROM_CLASS (klass),
2398       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2399           new_storage), NULL, NULL, g_cclosure_marshal_generic,
2400       G_TYPE_NONE, 2, GST_TYPE_ELEMENT, G_TYPE_UINT);
2401
2402   /**
2403    * GstRtpBin::request-aux-sender:
2404    * @rtpbin: the object which received the signal
2405    * @session: the session
2406    *
2407    * Request an AUX sender element for the given @session. The AUX
2408    * element will be added to the bin.
2409    *
2410    * If no handler is connected, no AUX element will be used.
2411    *
2412    * Since: 1.4
2413    */
2414   gst_rtp_bin_signals[SIGNAL_REQUEST_AUX_SENDER] =
2415       g_signal_new ("request-aux-sender", G_TYPE_FROM_CLASS (klass),
2416       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2417           request_aux_sender), _gst_element_accumulator, NULL,
2418       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2419
2420   /**
2421    * GstRtpBin::request-aux-receiver:
2422    * @rtpbin: the object which received the signal
2423    * @session: the session
2424    *
2425    * Request an AUX receiver element for the given @session. The AUX
2426    * element will be added to the bin.
2427    *
2428    * If no handler is connected, no AUX element will be used.
2429    *
2430    * Since: 1.4
2431    */
2432   gst_rtp_bin_signals[SIGNAL_REQUEST_AUX_RECEIVER] =
2433       g_signal_new ("request-aux-receiver", G_TYPE_FROM_CLASS (klass),
2434       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2435           request_aux_receiver), _gst_element_accumulator, NULL,
2436       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2437
2438   /**
2439    * GstRtpBin::request-fec-decoder:
2440    * @rtpbin: the object which received the signal
2441    * @session: the session index
2442    *
2443    * Request a FEC decoder element for the given @session. The element
2444    * will be added to the bin after the pt demuxer.
2445    *
2446    * If no handler is connected, no FEC decoder will be used.
2447    *
2448    * Since: 1.14
2449    */
2450   gst_rtp_bin_signals[SIGNAL_REQUEST_FEC_DECODER] =
2451       g_signal_new ("request-fec-decoder", G_TYPE_FROM_CLASS (klass),
2452       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2453           request_fec_decoder), _gst_element_accumulator, NULL,
2454       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2455
2456   /**
2457    * GstRtpBin::request-fec-encoder:
2458    * @rtpbin: the object which received the signal
2459    * @session: the session index
2460    *
2461    * Request a FEC encoder element for the given @session. The element
2462    * will be added to the bin after the RTPSession.
2463    *
2464    * If no handler is connected, no FEC encoder will be used.
2465    *
2466    * Since: 1.14
2467    */
2468   gst_rtp_bin_signals[SIGNAL_REQUEST_FEC_ENCODER] =
2469       g_signal_new ("request-fec-encoder", G_TYPE_FROM_CLASS (klass),
2470       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2471           request_fec_encoder), _gst_element_accumulator, NULL,
2472       g_cclosure_marshal_generic, GST_TYPE_ELEMENT, 1, G_TYPE_UINT);
2473
2474   /**
2475    * GstRtpBin::on-new-sender-ssrc:
2476    * @rtpbin: the object which received the signal
2477    * @session: the session
2478    * @ssrc: the sender SSRC
2479    *
2480    * Notify of a new sender SSRC that entered @session.
2481    *
2482    * Since: 1.8
2483    */
2484   gst_rtp_bin_signals[SIGNAL_ON_NEW_SENDER_SSRC] =
2485       g_signal_new ("on-new-sender-ssrc", G_TYPE_FROM_CLASS (klass),
2486       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass, on_new_sender_ssrc),
2487       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_UINT,
2488       G_TYPE_UINT);
2489   /**
2490    * GstRtpBin::on-sender-ssrc-active:
2491    * @rtpbin: the object which received the signal
2492    * @session: the session
2493    * @ssrc: the sender SSRC
2494    *
2495    * Notify of a sender SSRC that is active, i.e., sending RTCP.
2496    *
2497    * Since: 1.8
2498    */
2499   gst_rtp_bin_signals[SIGNAL_ON_SENDER_SSRC_ACTIVE] =
2500       g_signal_new ("on-sender-ssrc-active", G_TYPE_FROM_CLASS (klass),
2501       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2502           on_sender_ssrc_active), NULL, NULL, g_cclosure_marshal_generic,
2503       G_TYPE_NONE, 2, G_TYPE_UINT, G_TYPE_UINT);
2504
2505
2506   /**
2507    * GstRtpBin::on-bundled-ssrc:
2508    * @rtpbin: the object which received the signal
2509    * @ssrc: the bundled SSRC
2510    *
2511    * Notify of a new incoming bundled SSRC. If no handler is connected to the
2512    * signal then the #GstRtpSession created for the recv_rtp_sink_\%u
2513    * request pad will be managing this new SSRC. However if there is a handler
2514    * connected then the application can decided to dispatch this new stream to
2515    * another session by providing its ID as return value of the handler. This
2516    * can be particularly useful to keep retransmission SSRCs grouped with the
2517    * session for which they handle retransmission.
2518    *
2519    * Since: 1.12
2520    */
2521   gst_rtp_bin_signals[SIGNAL_ON_BUNDLED_SSRC] =
2522       g_signal_new ("on-bundled-ssrc", G_TYPE_FROM_CLASS (klass),
2523       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpBinClass,
2524           on_bundled_ssrc), NULL, NULL,
2525       g_cclosure_marshal_generic, G_TYPE_UINT, 1, G_TYPE_UINT);
2526
2527
2528   g_object_class_install_property (gobject_class, PROP_SDES,
2529       g_param_spec_boxed ("sdes", "SDES",
2530           "The SDES items of this session",
2531           GST_TYPE_STRUCTURE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2532
2533   g_object_class_install_property (gobject_class, PROP_DO_LOST,
2534       g_param_spec_boolean ("do-lost", "Do Lost",
2535           "Send an event downstream when a packet is lost", DEFAULT_DO_LOST,
2536           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2537
2538   g_object_class_install_property (gobject_class, PROP_AUTOREMOVE,
2539       g_param_spec_boolean ("autoremove", "Auto Remove",
2540           "Automatically remove timed out sources", DEFAULT_AUTOREMOVE,
2541           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2542
2543   g_object_class_install_property (gobject_class, PROP_IGNORE_PT,
2544       g_param_spec_boolean ("ignore-pt", "Ignore PT",
2545           "Do not demultiplex based on PT values", DEFAULT_IGNORE_PT,
2546           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2547
2548   g_object_class_install_property (gobject_class, PROP_USE_PIPELINE_CLOCK,
2549       g_param_spec_boolean ("use-pipeline-clock", "Use pipeline clock",
2550           "Use the pipeline running-time to set the NTP time in the RTCP SR messages "
2551           "(DEPRECATED: Use ntp-time-source property)",
2552           DEFAULT_USE_PIPELINE_CLOCK,
2553           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_DEPRECATED));
2554   /**
2555    * GstRtpBin:buffer-mode:
2556    *
2557    * Control the buffering and timestamping mode used by the jitterbuffer.
2558    */
2559   g_object_class_install_property (gobject_class, PROP_BUFFER_MODE,
2560       g_param_spec_enum ("buffer-mode", "Buffer Mode",
2561           "Control the buffering algorithm in use", RTP_TYPE_JITTER_BUFFER_MODE,
2562           DEFAULT_BUFFER_MODE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2563   /**
2564    * GstRtpBin:ntp-sync:
2565    *
2566    * Set the NTP time from the sender reports as the running-time on the
2567    * buffers. When both the sender and receiver have sychronized
2568    * running-time, i.e. when the clock and base-time is shared
2569    * between the receivers and the and the senders, this option can be
2570    * used to synchronize receivers on multiple machines.
2571    */
2572   g_object_class_install_property (gobject_class, PROP_NTP_SYNC,
2573       g_param_spec_boolean ("ntp-sync", "Sync on NTP clock",
2574           "Synchronize received streams to the NTP clock", DEFAULT_NTP_SYNC,
2575           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2576
2577   /**
2578    * GstRtpBin:rtcp-sync:
2579    *
2580    * If not synchronizing (directly) to the NTP clock, determines how to sync
2581    * the various streams.
2582    */
2583   g_object_class_install_property (gobject_class, PROP_RTCP_SYNC,
2584       g_param_spec_enum ("rtcp-sync", "RTCP Sync",
2585           "Use of RTCP SR in synchronization", GST_RTP_BIN_RTCP_SYNC_TYPE,
2586           DEFAULT_RTCP_SYNC, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2587
2588   /**
2589    * GstRtpBin:rtcp-sync-interval:
2590    *
2591    * Determines how often to sync streams using RTCP data.
2592    */
2593   g_object_class_install_property (gobject_class, PROP_RTCP_SYNC_INTERVAL,
2594       g_param_spec_uint ("rtcp-sync-interval", "RTCP Sync Interval",
2595           "RTCP SR interval synchronization (ms) (0 = always)",
2596           0, G_MAXUINT, DEFAULT_RTCP_SYNC_INTERVAL,
2597           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2598
2599   g_object_class_install_property (gobject_class, PROP_DO_SYNC_EVENT,
2600       g_param_spec_boolean ("do-sync-event", "Do Sync Event",
2601           "Send event downstream when a stream is synchronized to the sender",
2602           DEFAULT_DO_SYNC_EVENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2603
2604   /**
2605    * GstRtpBin:do-retransmission:
2606    *
2607    * Enables RTP retransmission on all streams. To control retransmission on
2608    * a per-SSRC basis, connect to the #GstRtpBin::new-jitterbuffer signal and
2609    * set the #GstRtpJitterBuffer::do-retransmission property on the
2610    * #GstRtpJitterBuffer object instead.
2611    */
2612   g_object_class_install_property (gobject_class, PROP_DO_RETRANSMISSION,
2613       g_param_spec_boolean ("do-retransmission", "Do retransmission",
2614           "Enable retransmission on all streams",
2615           DEFAULT_DO_RETRANSMISSION,
2616           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2617
2618   /**
2619    * GstRtpBin:rtp-profile:
2620    *
2621    * Sets the default RTP profile of newly created RTP sessions. The
2622    * profile can be changed afterwards on a per-session basis.
2623    */
2624   g_object_class_install_property (gobject_class, PROP_RTP_PROFILE,
2625       g_param_spec_enum ("rtp-profile", "RTP Profile",
2626           "Default RTP profile of newly created sessions",
2627           GST_TYPE_RTP_PROFILE, DEFAULT_RTP_PROFILE,
2628           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2629
2630   g_object_class_install_property (gobject_class, PROP_NTP_TIME_SOURCE,
2631       g_param_spec_enum ("ntp-time-source", "NTP Time Source",
2632           "NTP time source for RTCP packets",
2633           gst_rtp_ntp_time_source_get_type (), DEFAULT_NTP_TIME_SOURCE,
2634           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2635
2636   g_object_class_install_property (gobject_class, PROP_RTCP_SYNC_SEND_TIME,
2637       g_param_spec_boolean ("rtcp-sync-send-time", "RTCP Sync Send Time",
2638           "Use send time or capture time for RTCP sync "
2639           "(TRUE = send time, FALSE = capture time)",
2640           DEFAULT_RTCP_SYNC_SEND_TIME,
2641           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2642
2643   g_object_class_install_property (gobject_class, PROP_MAX_RTCP_RTP_TIME_DIFF,
2644       g_param_spec_int ("max-rtcp-rtp-time-diff", "Max RTCP RTP Time Diff",
2645           "Maximum amount of time in ms that the RTP time in RTCP SRs "
2646           "is allowed to be ahead (-1 disabled)", -1, G_MAXINT,
2647           DEFAULT_MAX_RTCP_RTP_TIME_DIFF,
2648           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2649
2650   g_object_class_install_property (gobject_class, PROP_MAX_DROPOUT_TIME,
2651       g_param_spec_uint ("max-dropout-time", "Max dropout time",
2652           "The maximum time (milliseconds) of missing packets tolerated.",
2653           0, G_MAXUINT, DEFAULT_MAX_DROPOUT_TIME,
2654           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2655
2656   g_object_class_install_property (gobject_class, PROP_MAX_MISORDER_TIME,
2657       g_param_spec_uint ("max-misorder-time", "Max misorder time",
2658           "The maximum time (milliseconds) of misordered packets tolerated.",
2659           0, G_MAXUINT, DEFAULT_MAX_MISORDER_TIME,
2660           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2661
2662   g_object_class_install_property (gobject_class, PROP_RFC7273_SYNC,
2663       g_param_spec_boolean ("rfc7273-sync", "Sync on RFC7273 clock",
2664           "Synchronize received streams to the RFC7273 clock "
2665           "(requires clock and offset to be provided)", DEFAULT_RFC7273_SYNC,
2666           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2667
2668   g_object_class_install_property (gobject_class, PROP_MAX_STREAMS,
2669       g_param_spec_uint ("max-streams", "Max Streams",
2670           "The maximum number of streams to create for one session",
2671           0, G_MAXUINT, DEFAULT_MAX_STREAMS,
2672           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2673
2674   /**
2675    * GstRtpBin:max-ts-offset-adjustment:
2676    *
2677    * Syncing time stamps to NTP time adds a time offset. This parameter
2678    * specifies the maximum number of nanoseconds per frame that this time offset
2679    * may be adjusted with. This is used to avoid sudden large changes to time
2680    * stamps.
2681    *
2682    * Since: 1.14
2683    */
2684   g_object_class_install_property (gobject_class, PROP_MAX_TS_OFFSET_ADJUSTMENT,
2685       g_param_spec_uint64 ("max-ts-offset-adjustment",
2686           "Max Timestamp Offset Adjustment",
2687           "The maximum number of nanoseconds per frame that time stamp offsets "
2688           "may be adjusted (0 = no limit).", 0, G_MAXUINT64,
2689           DEFAULT_MAX_TS_OFFSET_ADJUSTMENT, G_PARAM_READWRITE |
2690           G_PARAM_STATIC_STRINGS));
2691
2692   /**
2693    * GstRtpBin:max-ts-offset:
2694    *
2695    * Used to set an upper limit of how large a time offset may be. This
2696    * is used to protect against unrealistic values as a result of either
2697    * client,server or clock issues.
2698    *
2699    * Since: 1.14
2700    */
2701   g_object_class_install_property (gobject_class, PROP_MAX_TS_OFFSET,
2702       g_param_spec_int64 ("max-ts-offset", "Max TS Offset",
2703           "The maximum absolute value of the time offset in (nanoseconds). "
2704           "Note, if the ntp-sync parameter is set the default value is "
2705           "changed to 0 (no limit)", 0, G_MAXINT64, DEFAULT_MAX_TS_OFFSET,
2706           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
2707
2708   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_rtp_bin_change_state);
2709   gstelement_class->request_new_pad =
2710       GST_DEBUG_FUNCPTR (gst_rtp_bin_request_new_pad);
2711   gstelement_class->release_pad = GST_DEBUG_FUNCPTR (gst_rtp_bin_release_pad);
2712
2713   /* sink pads */
2714   gst_element_class_add_static_pad_template (gstelement_class,
2715       &rtpbin_recv_rtp_sink_template);
2716   gst_element_class_add_static_pad_template (gstelement_class,
2717       &rtpbin_recv_rtcp_sink_template);
2718   gst_element_class_add_static_pad_template (gstelement_class,
2719       &rtpbin_send_rtp_sink_template);
2720
2721   /* src pads */
2722   gst_element_class_add_static_pad_template (gstelement_class,
2723       &rtpbin_recv_rtp_src_template);
2724   gst_element_class_add_static_pad_template (gstelement_class,
2725       &rtpbin_send_rtcp_src_template);
2726   gst_element_class_add_static_pad_template (gstelement_class,
2727       &rtpbin_send_rtp_src_template);
2728
2729   gst_element_class_set_static_metadata (gstelement_class, "RTP Bin",
2730       "Filter/Network/RTP",
2731       "Real-Time Transport Protocol bin",
2732       "Wim Taymans <wim.taymans@gmail.com>");
2733
2734   gstbin_class->handle_message = GST_DEBUG_FUNCPTR (gst_rtp_bin_handle_message);
2735
2736   klass->clear_pt_map = GST_DEBUG_FUNCPTR (gst_rtp_bin_clear_pt_map);
2737   klass->reset_sync = GST_DEBUG_FUNCPTR (gst_rtp_bin_reset_sync);
2738   klass->get_session = GST_DEBUG_FUNCPTR (gst_rtp_bin_get_session);
2739   klass->get_internal_session =
2740       GST_DEBUG_FUNCPTR (gst_rtp_bin_get_internal_session);
2741   klass->get_storage = GST_DEBUG_FUNCPTR (gst_rtp_bin_get_storage);
2742   klass->get_internal_storage =
2743       GST_DEBUG_FUNCPTR (gst_rtp_bin_get_internal_storage);
2744   klass->request_rtp_encoder = GST_DEBUG_FUNCPTR (gst_rtp_bin_request_encoder);
2745   klass->request_rtp_decoder = GST_DEBUG_FUNCPTR (gst_rtp_bin_request_decoder);
2746   klass->request_rtcp_encoder = GST_DEBUG_FUNCPTR (gst_rtp_bin_request_encoder);
2747   klass->request_rtcp_decoder = GST_DEBUG_FUNCPTR (gst_rtp_bin_request_decoder);
2748
2749   GST_DEBUG_CATEGORY_INIT (gst_rtp_bin_debug, "rtpbin", 0, "RTP bin");
2750 }
2751
2752 static void
2753 gst_rtp_bin_init (GstRtpBin * rtpbin)
2754 {
2755   gchar *cname;
2756
2757   rtpbin->priv = GST_RTP_BIN_GET_PRIVATE (rtpbin);
2758   g_mutex_init (&rtpbin->priv->bin_lock);
2759   g_mutex_init (&rtpbin->priv->dyn_lock);
2760
2761   rtpbin->latency_ms = DEFAULT_LATENCY_MS;
2762   rtpbin->latency_ns = DEFAULT_LATENCY_MS * GST_MSECOND;
2763   rtpbin->drop_on_latency = DEFAULT_DROP_ON_LATENCY;
2764   rtpbin->do_lost = DEFAULT_DO_LOST;
2765   rtpbin->ignore_pt = DEFAULT_IGNORE_PT;
2766   rtpbin->ntp_sync = DEFAULT_NTP_SYNC;
2767   rtpbin->rtcp_sync = DEFAULT_RTCP_SYNC;
2768   rtpbin->rtcp_sync_interval = DEFAULT_RTCP_SYNC_INTERVAL;
2769   rtpbin->priv->autoremove = DEFAULT_AUTOREMOVE;
2770   rtpbin->buffer_mode = DEFAULT_BUFFER_MODE;
2771   rtpbin->use_pipeline_clock = DEFAULT_USE_PIPELINE_CLOCK;
2772   rtpbin->send_sync_event = DEFAULT_DO_SYNC_EVENT;
2773   rtpbin->do_retransmission = DEFAULT_DO_RETRANSMISSION;
2774   rtpbin->rtp_profile = DEFAULT_RTP_PROFILE;
2775   rtpbin->ntp_time_source = DEFAULT_NTP_TIME_SOURCE;
2776   rtpbin->rtcp_sync_send_time = DEFAULT_RTCP_SYNC_SEND_TIME;
2777   rtpbin->max_rtcp_rtp_time_diff = DEFAULT_MAX_RTCP_RTP_TIME_DIFF;
2778   rtpbin->max_dropout_time = DEFAULT_MAX_DROPOUT_TIME;
2779   rtpbin->max_misorder_time = DEFAULT_MAX_MISORDER_TIME;
2780   rtpbin->rfc7273_sync = DEFAULT_RFC7273_SYNC;
2781   rtpbin->max_streams = DEFAULT_MAX_STREAMS;
2782   rtpbin->max_ts_offset_adjustment = DEFAULT_MAX_TS_OFFSET_ADJUSTMENT;
2783   rtpbin->max_ts_offset = DEFAULT_MAX_TS_OFFSET;
2784   rtpbin->max_ts_offset_is_set = FALSE;
2785
2786   /* some default SDES entries */
2787   cname = g_strdup_printf ("user%u@host-%x", g_random_int (), g_random_int ());
2788   rtpbin->sdes = gst_structure_new ("application/x-rtp-source-sdes",
2789       "cname", G_TYPE_STRING, cname, "tool", G_TYPE_STRING, "GStreamer", NULL);
2790   g_free (cname);
2791 }
2792
2793 static void
2794 gst_rtp_bin_dispose (GObject * object)
2795 {
2796   GstRtpBin *rtpbin;
2797
2798   rtpbin = GST_RTP_BIN (object);
2799
2800   GST_RTP_BIN_LOCK (rtpbin);
2801   GST_DEBUG_OBJECT (object, "freeing sessions");
2802   g_slist_foreach (rtpbin->sessions, (GFunc) free_session, rtpbin);
2803   g_slist_free (rtpbin->sessions);
2804   rtpbin->sessions = NULL;
2805   GST_RTP_BIN_UNLOCK (rtpbin);
2806
2807   G_OBJECT_CLASS (parent_class)->dispose (object);
2808 }
2809
2810 static void
2811 gst_rtp_bin_finalize (GObject * object)
2812 {
2813   GstRtpBin *rtpbin;
2814
2815   rtpbin = GST_RTP_BIN (object);
2816
2817   if (rtpbin->sdes)
2818     gst_structure_free (rtpbin->sdes);
2819
2820   g_mutex_clear (&rtpbin->priv->bin_lock);
2821   g_mutex_clear (&rtpbin->priv->dyn_lock);
2822
2823   G_OBJECT_CLASS (parent_class)->finalize (object);
2824 }
2825
2826
2827 static void
2828 gst_rtp_bin_set_sdes_struct (GstRtpBin * bin, const GstStructure * sdes)
2829 {
2830   GSList *item;
2831
2832   if (sdes == NULL)
2833     return;
2834
2835   GST_RTP_BIN_LOCK (bin);
2836
2837   GST_OBJECT_LOCK (bin);
2838   if (bin->sdes)
2839     gst_structure_free (bin->sdes);
2840   bin->sdes = gst_structure_copy (sdes);
2841   GST_OBJECT_UNLOCK (bin);
2842
2843   /* store in all sessions */
2844   for (item = bin->sessions; item; item = g_slist_next (item)) {
2845     GstRtpBinSession *session = item->data;
2846     g_object_set (session->session, "sdes", sdes, NULL);
2847   }
2848
2849   GST_RTP_BIN_UNLOCK (bin);
2850 }
2851
2852 static GstStructure *
2853 gst_rtp_bin_get_sdes_struct (GstRtpBin * bin)
2854 {
2855   GstStructure *result;
2856
2857   GST_OBJECT_LOCK (bin);
2858   result = gst_structure_copy (bin->sdes);
2859   GST_OBJECT_UNLOCK (bin);
2860
2861   return result;
2862 }
2863
2864 static void
2865 gst_rtp_bin_set_property (GObject * object, guint prop_id,
2866     const GValue * value, GParamSpec * pspec)
2867 {
2868   GstRtpBin *rtpbin;
2869
2870   rtpbin = GST_RTP_BIN (object);
2871
2872   switch (prop_id) {
2873     case PROP_LATENCY:
2874       GST_RTP_BIN_LOCK (rtpbin);
2875       rtpbin->latency_ms = g_value_get_uint (value);
2876       rtpbin->latency_ns = rtpbin->latency_ms * GST_MSECOND;
2877       GST_RTP_BIN_UNLOCK (rtpbin);
2878       /* propagate the property down to the jitterbuffer */
2879       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin, "latency", value);
2880       break;
2881     case PROP_DROP_ON_LATENCY:
2882       GST_RTP_BIN_LOCK (rtpbin);
2883       rtpbin->drop_on_latency = g_value_get_boolean (value);
2884       GST_RTP_BIN_UNLOCK (rtpbin);
2885       /* propagate the property down to the jitterbuffer */
2886       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
2887           "drop-on-latency", value);
2888       break;
2889     case PROP_SDES:
2890       gst_rtp_bin_set_sdes_struct (rtpbin, g_value_get_boxed (value));
2891       break;
2892     case PROP_DO_LOST:
2893       GST_RTP_BIN_LOCK (rtpbin);
2894       rtpbin->do_lost = g_value_get_boolean (value);
2895       GST_RTP_BIN_UNLOCK (rtpbin);
2896       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin, "do-lost", value);
2897       break;
2898     case PROP_NTP_SYNC:
2899       rtpbin->ntp_sync = g_value_get_boolean (value);
2900       /* The default value of max_ts_offset depends on ntp_sync. If user
2901        * hasn't set it then change default value */
2902       if (!rtpbin->max_ts_offset_is_set) {
2903         if (rtpbin->ntp_sync) {
2904           rtpbin->max_ts_offset = 0;
2905         } else {
2906           rtpbin->max_ts_offset = DEFAULT_MAX_TS_OFFSET;
2907         }
2908       }
2909       break;
2910     case PROP_RTCP_SYNC:
2911       g_atomic_int_set (&rtpbin->rtcp_sync, g_value_get_enum (value));
2912       break;
2913     case PROP_RTCP_SYNC_INTERVAL:
2914       rtpbin->rtcp_sync_interval = g_value_get_uint (value);
2915       break;
2916     case PROP_IGNORE_PT:
2917       rtpbin->ignore_pt = g_value_get_boolean (value);
2918       break;
2919     case PROP_AUTOREMOVE:
2920       rtpbin->priv->autoremove = g_value_get_boolean (value);
2921       break;
2922     case PROP_USE_PIPELINE_CLOCK:
2923     {
2924       GSList *sessions;
2925       GST_RTP_BIN_LOCK (rtpbin);
2926       rtpbin->use_pipeline_clock = g_value_get_boolean (value);
2927       for (sessions = rtpbin->sessions; sessions;
2928           sessions = g_slist_next (sessions)) {
2929         GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
2930
2931         g_object_set (G_OBJECT (session->session),
2932             "use-pipeline-clock", rtpbin->use_pipeline_clock, NULL);
2933       }
2934       GST_RTP_BIN_UNLOCK (rtpbin);
2935     }
2936       break;
2937     case PROP_DO_SYNC_EVENT:
2938       rtpbin->send_sync_event = g_value_get_boolean (value);
2939       break;
2940     case PROP_BUFFER_MODE:
2941       GST_RTP_BIN_LOCK (rtpbin);
2942       rtpbin->buffer_mode = g_value_get_enum (value);
2943       GST_RTP_BIN_UNLOCK (rtpbin);
2944       /* propagate the property down to the jitterbuffer */
2945       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin, "mode", value);
2946       break;
2947     case PROP_DO_RETRANSMISSION:
2948       GST_RTP_BIN_LOCK (rtpbin);
2949       rtpbin->do_retransmission = g_value_get_boolean (value);
2950       GST_RTP_BIN_UNLOCK (rtpbin);
2951       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
2952           "do-retransmission", value);
2953       break;
2954     case PROP_RTP_PROFILE:
2955       rtpbin->rtp_profile = g_value_get_enum (value);
2956       break;
2957     case PROP_NTP_TIME_SOURCE:{
2958       GSList *sessions;
2959       GST_RTP_BIN_LOCK (rtpbin);
2960       rtpbin->ntp_time_source = g_value_get_enum (value);
2961       for (sessions = rtpbin->sessions; sessions;
2962           sessions = g_slist_next (sessions)) {
2963         GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
2964
2965         g_object_set (G_OBJECT (session->session),
2966             "ntp-time-source", rtpbin->ntp_time_source, NULL);
2967       }
2968       GST_RTP_BIN_UNLOCK (rtpbin);
2969       break;
2970     }
2971     case PROP_RTCP_SYNC_SEND_TIME:{
2972       GSList *sessions;
2973       GST_RTP_BIN_LOCK (rtpbin);
2974       rtpbin->rtcp_sync_send_time = g_value_get_boolean (value);
2975       for (sessions = rtpbin->sessions; sessions;
2976           sessions = g_slist_next (sessions)) {
2977         GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
2978
2979         g_object_set (G_OBJECT (session->session),
2980             "rtcp-sync-send-time", rtpbin->rtcp_sync_send_time, NULL);
2981       }
2982       GST_RTP_BIN_UNLOCK (rtpbin);
2983       break;
2984     }
2985     case PROP_MAX_RTCP_RTP_TIME_DIFF:
2986       GST_RTP_BIN_LOCK (rtpbin);
2987       rtpbin->max_rtcp_rtp_time_diff = g_value_get_int (value);
2988       GST_RTP_BIN_UNLOCK (rtpbin);
2989       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
2990           "max-rtcp-rtp-time-diff", value);
2991       break;
2992     case PROP_MAX_DROPOUT_TIME:
2993       GST_RTP_BIN_LOCK (rtpbin);
2994       rtpbin->max_dropout_time = g_value_get_uint (value);
2995       GST_RTP_BIN_UNLOCK (rtpbin);
2996       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
2997           "max-dropout-time", value);
2998       gst_rtp_bin_propagate_property_to_session (rtpbin, "max-dropout-time",
2999           value);
3000       break;
3001     case PROP_MAX_MISORDER_TIME:
3002       GST_RTP_BIN_LOCK (rtpbin);
3003       rtpbin->max_misorder_time = g_value_get_uint (value);
3004       GST_RTP_BIN_UNLOCK (rtpbin);
3005       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
3006           "max-misorder-time", value);
3007       gst_rtp_bin_propagate_property_to_session (rtpbin, "max-misorder-time",
3008           value);
3009       break;
3010     case PROP_RFC7273_SYNC:
3011       rtpbin->rfc7273_sync = g_value_get_boolean (value);
3012       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
3013           "rfc7273-sync", value);
3014       break;
3015     case PROP_MAX_STREAMS:
3016       rtpbin->max_streams = g_value_get_uint (value);
3017       break;
3018     case PROP_MAX_TS_OFFSET_ADJUSTMENT:
3019       rtpbin->max_ts_offset_adjustment = g_value_get_uint64 (value);
3020       gst_rtp_bin_propagate_property_to_jitterbuffer (rtpbin,
3021           "max-ts-offset-adjustment", value);
3022       break;
3023     case PROP_MAX_TS_OFFSET:
3024       rtpbin->max_ts_offset = g_value_get_int64 (value);
3025       rtpbin->max_ts_offset_is_set = TRUE;
3026       break;
3027     default:
3028       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
3029       break;
3030   }
3031 }
3032
3033 static void
3034 gst_rtp_bin_get_property (GObject * object, guint prop_id,
3035     GValue * value, GParamSpec * pspec)
3036 {
3037   GstRtpBin *rtpbin;
3038
3039   rtpbin = GST_RTP_BIN (object);
3040
3041   switch (prop_id) {
3042     case PROP_LATENCY:
3043       GST_RTP_BIN_LOCK (rtpbin);
3044       g_value_set_uint (value, rtpbin->latency_ms);
3045       GST_RTP_BIN_UNLOCK (rtpbin);
3046       break;
3047     case PROP_DROP_ON_LATENCY:
3048       GST_RTP_BIN_LOCK (rtpbin);
3049       g_value_set_boolean (value, rtpbin->drop_on_latency);
3050       GST_RTP_BIN_UNLOCK (rtpbin);
3051       break;
3052     case PROP_SDES:
3053       g_value_take_boxed (value, gst_rtp_bin_get_sdes_struct (rtpbin));
3054       break;
3055     case PROP_DO_LOST:
3056       GST_RTP_BIN_LOCK (rtpbin);
3057       g_value_set_boolean (value, rtpbin->do_lost);
3058       GST_RTP_BIN_UNLOCK (rtpbin);
3059       break;
3060     case PROP_IGNORE_PT:
3061       g_value_set_boolean (value, rtpbin->ignore_pt);
3062       break;
3063     case PROP_NTP_SYNC:
3064       g_value_set_boolean (value, rtpbin->ntp_sync);
3065       break;
3066     case PROP_RTCP_SYNC:
3067       g_value_set_enum (value, g_atomic_int_get (&rtpbin->rtcp_sync));
3068       break;
3069     case PROP_RTCP_SYNC_INTERVAL:
3070       g_value_set_uint (value, rtpbin->rtcp_sync_interval);
3071       break;
3072     case PROP_AUTOREMOVE:
3073       g_value_set_boolean (value, rtpbin->priv->autoremove);
3074       break;
3075     case PROP_BUFFER_MODE:
3076       g_value_set_enum (value, rtpbin->buffer_mode);
3077       break;
3078     case PROP_USE_PIPELINE_CLOCK:
3079       g_value_set_boolean (value, rtpbin->use_pipeline_clock);
3080       break;
3081     case PROP_DO_SYNC_EVENT:
3082       g_value_set_boolean (value, rtpbin->send_sync_event);
3083       break;
3084     case PROP_DO_RETRANSMISSION:
3085       GST_RTP_BIN_LOCK (rtpbin);
3086       g_value_set_boolean (value, rtpbin->do_retransmission);
3087       GST_RTP_BIN_UNLOCK (rtpbin);
3088       break;
3089     case PROP_RTP_PROFILE:
3090       g_value_set_enum (value, rtpbin->rtp_profile);
3091       break;
3092     case PROP_NTP_TIME_SOURCE:
3093       g_value_set_enum (value, rtpbin->ntp_time_source);
3094       break;
3095     case PROP_RTCP_SYNC_SEND_TIME:
3096       g_value_set_boolean (value, rtpbin->rtcp_sync_send_time);
3097       break;
3098     case PROP_MAX_RTCP_RTP_TIME_DIFF:
3099       GST_RTP_BIN_LOCK (rtpbin);
3100       g_value_set_int (value, rtpbin->max_rtcp_rtp_time_diff);
3101       GST_RTP_BIN_UNLOCK (rtpbin);
3102       break;
3103     case PROP_MAX_DROPOUT_TIME:
3104       g_value_set_uint (value, rtpbin->max_dropout_time);
3105       break;
3106     case PROP_MAX_MISORDER_TIME:
3107       g_value_set_uint (value, rtpbin->max_misorder_time);
3108       break;
3109     case PROP_RFC7273_SYNC:
3110       g_value_set_boolean (value, rtpbin->rfc7273_sync);
3111       break;
3112     case PROP_MAX_STREAMS:
3113       g_value_set_uint (value, rtpbin->max_streams);
3114       break;
3115     case PROP_MAX_TS_OFFSET_ADJUSTMENT:
3116       g_value_set_uint64 (value, rtpbin->max_ts_offset_adjustment);
3117       break;
3118     case PROP_MAX_TS_OFFSET:
3119       g_value_set_int64 (value, rtpbin->max_ts_offset);
3120       break;
3121     default:
3122       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
3123       break;
3124   }
3125 }
3126
3127 static void
3128 gst_rtp_bin_handle_message (GstBin * bin, GstMessage * message)
3129 {
3130   GstRtpBin *rtpbin;
3131
3132   rtpbin = GST_RTP_BIN (bin);
3133
3134   switch (GST_MESSAGE_TYPE (message)) {
3135     case GST_MESSAGE_ELEMENT:
3136     {
3137       const GstStructure *s = gst_message_get_structure (message);
3138
3139       /* we change the structure name and add the session ID to it */
3140       if (gst_structure_has_name (s, "application/x-rtp-source-sdes")) {
3141         GstRtpBinSession *sess;
3142
3143         /* find the session we set it as object data */
3144         sess = g_object_get_data (G_OBJECT (GST_MESSAGE_SRC (message)),
3145             "GstRTPBin.session");
3146
3147         if (G_LIKELY (sess)) {
3148           message = gst_message_make_writable (message);
3149           s = gst_message_get_structure (message);
3150           gst_structure_set ((GstStructure *) s, "session", G_TYPE_UINT,
3151               sess->id, NULL);
3152         }
3153       }
3154       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
3155       break;
3156     }
3157     case GST_MESSAGE_BUFFERING:
3158     {
3159       gint percent;
3160       gint min_percent = 100;
3161       GSList *sessions, *streams;
3162       GstRtpBinStream *stream;
3163       gboolean change = FALSE, active = FALSE;
3164       GstClockTime min_out_time;
3165       GstBufferingMode mode;
3166       gint avg_in, avg_out;
3167       gint64 buffering_left;
3168
3169       gst_message_parse_buffering (message, &percent);
3170       gst_message_parse_buffering_stats (message, &mode, &avg_in, &avg_out,
3171           &buffering_left);
3172
3173       stream =
3174           g_object_get_data (G_OBJECT (GST_MESSAGE_SRC (message)),
3175           "GstRTPBin.stream");
3176
3177       GST_DEBUG_OBJECT (bin, "got percent %d from stream %p", percent, stream);
3178
3179       /* get the stream */
3180       if (G_LIKELY (stream)) {
3181         GST_RTP_BIN_LOCK (rtpbin);
3182         /* fill in the percent */
3183         stream->percent = percent;
3184
3185         /* calculate the min value for all streams */
3186         for (sessions = rtpbin->sessions; sessions;
3187             sessions = g_slist_next (sessions)) {
3188           GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
3189
3190           GST_RTP_SESSION_LOCK (session);
3191           if (session->streams) {
3192             for (streams = session->streams; streams;
3193                 streams = g_slist_next (streams)) {
3194               GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
3195
3196               GST_DEBUG_OBJECT (bin, "stream %p percent %d", stream,
3197                   stream->percent);
3198
3199               /* find min percent */
3200               if (min_percent > stream->percent)
3201                 min_percent = stream->percent;
3202             }
3203           } else {
3204             GST_INFO_OBJECT (bin,
3205                 "session has no streams, setting min_percent to 0");
3206             min_percent = 0;
3207           }
3208           GST_RTP_SESSION_UNLOCK (session);
3209         }
3210         GST_DEBUG_OBJECT (bin, "min percent %d", min_percent);
3211
3212         if (rtpbin->buffering) {
3213           if (min_percent == 100) {
3214             rtpbin->buffering = FALSE;
3215             active = TRUE;
3216             change = TRUE;
3217           }
3218         } else {
3219           if (min_percent < 100) {
3220             /* pause the streams */
3221             rtpbin->buffering = TRUE;
3222             active = FALSE;
3223             change = TRUE;
3224           }
3225         }
3226         GST_RTP_BIN_UNLOCK (rtpbin);
3227
3228         gst_message_unref (message);
3229
3230         /* make a new buffering message with the min value */
3231         message =
3232             gst_message_new_buffering (GST_OBJECT_CAST (bin), min_percent);
3233         gst_message_set_buffering_stats (message, mode, avg_in, avg_out,
3234             buffering_left);
3235
3236         if (G_UNLIKELY (change)) {
3237           GstClock *clock;
3238           guint64 running_time = 0;
3239           guint64 offset = 0;
3240
3241           /* figure out the running time when we have a clock */
3242           if (G_LIKELY ((clock =
3243                       gst_element_get_clock (GST_ELEMENT_CAST (bin))))) {
3244             guint64 now, base_time;
3245
3246             now = gst_clock_get_time (clock);
3247             base_time = gst_element_get_base_time (GST_ELEMENT_CAST (bin));
3248             running_time = now - base_time;
3249             gst_object_unref (clock);
3250           }
3251           GST_DEBUG_OBJECT (bin,
3252               "running time now %" GST_TIME_FORMAT,
3253               GST_TIME_ARGS (running_time));
3254
3255           GST_RTP_BIN_LOCK (rtpbin);
3256
3257           /* when we reactivate, calculate the offsets so that all streams have
3258            * an output time that is at least as big as the running_time */
3259           offset = 0;
3260           if (active) {
3261             if (running_time > rtpbin->buffer_start) {
3262               offset = running_time - rtpbin->buffer_start;
3263               if (offset >= rtpbin->latency_ns)
3264                 offset -= rtpbin->latency_ns;
3265               else
3266                 offset = 0;
3267             }
3268           }
3269
3270           /* pause all streams */
3271           min_out_time = -1;
3272           for (sessions = rtpbin->sessions; sessions;
3273               sessions = g_slist_next (sessions)) {
3274             GstRtpBinSession *session = (GstRtpBinSession *) sessions->data;
3275
3276             GST_RTP_SESSION_LOCK (session);
3277             for (streams = session->streams; streams;
3278                 streams = g_slist_next (streams)) {
3279               GstRtpBinStream *stream = (GstRtpBinStream *) streams->data;
3280               GstElement *element = stream->buffer;
3281               guint64 last_out;
3282
3283               g_signal_emit_by_name (element, "set-active", active, offset,
3284                   &last_out);
3285
3286               if (!active) {
3287                 g_object_get (element, "percent", &stream->percent, NULL);
3288
3289                 if (last_out == -1)
3290                   last_out = 0;
3291                 if (min_out_time == -1 || last_out < min_out_time)
3292                   min_out_time = last_out;
3293               }
3294
3295               GST_DEBUG_OBJECT (bin,
3296                   "setting %p to %d, offset %" GST_TIME_FORMAT ", last %"
3297                   GST_TIME_FORMAT ", percent %d", element, active,
3298                   GST_TIME_ARGS (offset), GST_TIME_ARGS (last_out),
3299                   stream->percent);
3300             }
3301             GST_RTP_SESSION_UNLOCK (session);
3302           }
3303           GST_DEBUG_OBJECT (bin,
3304               "min out time %" GST_TIME_FORMAT, GST_TIME_ARGS (min_out_time));
3305
3306           /* the buffer_start is the min out time of all paused jitterbuffers */
3307           if (!active)
3308             rtpbin->buffer_start = min_out_time;
3309
3310           GST_RTP_BIN_UNLOCK (rtpbin);
3311         }
3312       }
3313       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
3314       break;
3315     }
3316     default:
3317     {
3318       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
3319       break;
3320     }
3321   }
3322 }
3323
3324 static GstStateChangeReturn
3325 gst_rtp_bin_change_state (GstElement * element, GstStateChange transition)
3326 {
3327   GstStateChangeReturn res;
3328   GstRtpBin *rtpbin;
3329   GstRtpBinPrivate *priv;
3330
3331   rtpbin = GST_RTP_BIN (element);
3332   priv = rtpbin->priv;
3333
3334   switch (transition) {
3335     case GST_STATE_CHANGE_NULL_TO_READY:
3336       break;
3337     case GST_STATE_CHANGE_READY_TO_PAUSED:
3338       priv->last_ntpnstime = 0;
3339       GST_LOG_OBJECT (rtpbin, "clearing shutdown flag");
3340       g_atomic_int_set (&priv->shutdown, 0);
3341       break;
3342     case GST_STATE_CHANGE_PAUSED_TO_READY:
3343       GST_LOG_OBJECT (rtpbin, "setting shutdown flag");
3344       g_atomic_int_set (&priv->shutdown, 1);
3345       /* wait for all callbacks to end by taking the lock. No new callbacks will
3346        * be able to happen as we set the shutdown flag. */
3347       GST_RTP_BIN_DYN_LOCK (rtpbin);
3348       GST_LOG_OBJECT (rtpbin, "dynamic lock taken, we can continue shutdown");
3349       GST_RTP_BIN_DYN_UNLOCK (rtpbin);
3350       break;
3351     default:
3352       break;
3353   }
3354
3355   res = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
3356
3357   switch (transition) {
3358     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
3359       break;
3360     case GST_STATE_CHANGE_PAUSED_TO_READY:
3361       break;
3362     case GST_STATE_CHANGE_READY_TO_NULL:
3363       break;
3364     default:
3365       break;
3366   }
3367   return res;
3368 }
3369
3370 static GstElement *
3371 session_request_element (GstRtpBinSession * session, guint signal)
3372 {
3373   GstElement *element = NULL;
3374   GstRtpBin *bin = session->bin;
3375
3376   g_signal_emit (bin, gst_rtp_bin_signals[signal], 0, session->id, &element);
3377
3378   if (element) {
3379     if (!bin_manage_element (bin, element))
3380       goto manage_failed;
3381     session->elements = g_slist_prepend (session->elements, element);
3382   }
3383   return element;
3384
3385   /* ERRORS */
3386 manage_failed:
3387   {
3388     GST_WARNING_OBJECT (bin, "unable to manage element");
3389     gst_object_unref (element);
3390     return NULL;
3391   }
3392 }
3393
3394 static gboolean
3395 copy_sticky_events (GstPad * pad, GstEvent ** event, gpointer user_data)
3396 {
3397   GstPad *gpad = GST_PAD_CAST (user_data);
3398
3399   GST_DEBUG_OBJECT (gpad, "store sticky event %" GST_PTR_FORMAT, *event);
3400   gst_pad_store_sticky_event (gpad, *event);
3401
3402   return TRUE;
3403 }
3404
3405 /* a new pad (SSRC) was created in @session. This signal is emitted from the
3406  * payload demuxer. */
3407 static void
3408 new_payload_found (GstElement * element, guint pt, GstPad * pad,
3409     GstRtpBinStream * stream)
3410 {
3411   GstRtpBin *rtpbin;
3412   GstElementClass *klass;
3413   GstPadTemplate *templ;
3414   gchar *padname;
3415   GstPad *gpad;
3416
3417   rtpbin = stream->bin;
3418
3419   GST_DEBUG_OBJECT (rtpbin, "new payload pad %u", pt);
3420
3421   pad = gst_object_ref (pad);
3422
3423   if (stream->session->storage) {
3424     GstElement *fec_decoder =
3425         session_request_element (stream->session, SIGNAL_REQUEST_FEC_DECODER);
3426
3427     if (fec_decoder) {
3428       GstPad *sinkpad, *srcpad;
3429       GstPadLinkReturn ret;
3430
3431       sinkpad = gst_element_get_static_pad (fec_decoder, "sink");
3432
3433       if (!sinkpad)
3434         goto fec_decoder_sink_failed;
3435
3436       ret = gst_pad_link (pad, sinkpad);
3437       gst_object_unref (sinkpad);
3438
3439       if (ret != GST_PAD_LINK_OK)
3440         goto fec_decoder_link_failed;
3441
3442       srcpad = gst_element_get_static_pad (fec_decoder, "src");
3443
3444       if (!srcpad)
3445         goto fec_decoder_src_failed;
3446
3447       gst_pad_sticky_events_foreach (pad, copy_sticky_events, srcpad);
3448       gst_object_unref (pad);
3449       pad = srcpad;
3450     }
3451   }
3452
3453   GST_RTP_BIN_SHUTDOWN_LOCK (rtpbin, shutdown);
3454
3455   /* ghost the pad to the parent */
3456   klass = GST_ELEMENT_GET_CLASS (rtpbin);
3457   templ = gst_element_class_get_pad_template (klass, "recv_rtp_src_%u_%u_%u");
3458   padname = g_strdup_printf ("recv_rtp_src_%u_%u_%u",
3459       stream->session->id, stream->ssrc, pt);
3460   gpad = gst_ghost_pad_new_from_template (padname, pad, templ);
3461   g_free (padname);
3462   g_object_set_data (G_OBJECT (pad), "GstRTPBin.ghostpad", gpad);
3463
3464   gst_pad_set_active (gpad, TRUE);
3465   GST_RTP_BIN_SHUTDOWN_UNLOCK (rtpbin);
3466
3467   gst_pad_sticky_events_foreach (pad, copy_sticky_events, gpad);
3468   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), gpad);
3469
3470 done:
3471   gst_object_unref (pad);
3472
3473   return;
3474
3475 shutdown:
3476   {
3477     GST_DEBUG ("ignoring, we are shutting down");
3478     goto done;
3479   }
3480 fec_decoder_sink_failed:
3481   {
3482     g_warning ("rtpbin: failed to get fec encoder sink pad for session %u",
3483         stream->session->id);
3484     goto done;
3485   }
3486 fec_decoder_src_failed:
3487   {
3488     g_warning ("rtpbin: failed to get fec encoder src pad for session %u",
3489         stream->session->id);
3490     goto done;
3491   }
3492 fec_decoder_link_failed:
3493   {
3494     g_warning ("rtpbin: failed to link fec decoder for session %u",
3495         stream->session->id);
3496     goto done;
3497   }
3498 }
3499
3500 static void
3501 payload_pad_removed (GstElement * element, GstPad * pad,
3502     GstRtpBinStream * stream)
3503 {
3504   GstRtpBin *rtpbin;
3505   GstPad *gpad;
3506
3507   rtpbin = stream->bin;
3508
3509   GST_DEBUG ("payload pad removed");
3510
3511   GST_RTP_BIN_DYN_LOCK (rtpbin);
3512   if ((gpad = g_object_get_data (G_OBJECT (pad), "GstRTPBin.ghostpad"))) {
3513     g_object_set_data (G_OBJECT (pad), "GstRTPBin.ghostpad", NULL);
3514
3515     gst_pad_set_active (gpad, FALSE);
3516     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin), gpad);
3517   }
3518   GST_RTP_BIN_DYN_UNLOCK (rtpbin);
3519 }
3520
3521 static GstCaps *
3522 pt_map_requested (GstElement * element, guint pt, GstRtpBinSession * session)
3523 {
3524   GstRtpBin *rtpbin;
3525   GstCaps *caps;
3526
3527   rtpbin = session->bin;
3528
3529   GST_DEBUG_OBJECT (rtpbin, "payload map requested for pt %u in session %u", pt,
3530       session->id);
3531
3532   caps = get_pt_map (session, pt);
3533   if (!caps)
3534     goto no_caps;
3535
3536   return caps;
3537
3538   /* ERRORS */
3539 no_caps:
3540   {
3541     GST_DEBUG_OBJECT (rtpbin, "could not get caps");
3542     return NULL;
3543   }
3544 }
3545
3546 static GstCaps *
3547 ptdemux_pt_map_requested (GstElement * element, guint pt,
3548     GstRtpBinSession * session)
3549 {
3550   GstCaps *ret = pt_map_requested (element, pt, session);
3551
3552   if (ret && gst_caps_get_size (ret) == 1) {
3553     const GstStructure *s = gst_caps_get_structure (ret, 0);
3554     gboolean is_fec;
3555
3556     if (gst_structure_get_boolean (s, "is-fec", &is_fec) && is_fec) {
3557       GValue v = G_VALUE_INIT;
3558       GValue v2 = G_VALUE_INIT;
3559
3560       GST_INFO_OBJECT (session->bin, "Will ignore FEC pt %u in session %u", pt,
3561           session->id);
3562       g_value_init (&v, GST_TYPE_ARRAY);
3563       g_value_init (&v2, G_TYPE_INT);
3564       g_object_get_property (G_OBJECT (element), "ignored-payload-types", &v);
3565       g_value_set_int (&v2, pt);
3566       gst_value_array_append_value (&v, &v2);
3567       g_value_unset (&v2);
3568       g_object_set_property (G_OBJECT (element), "ignored-payload-types", &v);
3569       g_value_unset (&v);
3570     }
3571   }
3572
3573   return ret;
3574 }
3575
3576 static void
3577 payload_type_change (GstElement * element, guint pt, GstRtpBinSession * session)
3578 {
3579   GST_DEBUG_OBJECT (session->bin,
3580       "emiting signal for pt type changed to %u in session %u", pt,
3581       session->id);
3582
3583   g_signal_emit (session->bin, gst_rtp_bin_signals[SIGNAL_PAYLOAD_TYPE_CHANGE],
3584       0, session->id, pt);
3585 }
3586
3587 /* emitted when caps changed for the session */
3588 static void
3589 caps_changed (GstPad * pad, GParamSpec * pspec, GstRtpBinSession * session)
3590 {
3591   GstRtpBin *bin;
3592   GstCaps *caps;
3593   gint payload;
3594   const GstStructure *s;
3595
3596   bin = session->bin;
3597
3598   g_object_get (pad, "caps", &caps, NULL);
3599
3600   if (caps == NULL)
3601     return;
3602
3603   GST_DEBUG_OBJECT (bin, "got caps %" GST_PTR_FORMAT, caps);
3604
3605   s = gst_caps_get_structure (caps, 0);
3606
3607   /* get payload, finish when it's not there */
3608   if (!gst_structure_get_int (s, "payload", &payload)) {
3609     gst_caps_unref (caps);
3610     return;
3611   }
3612
3613   GST_RTP_SESSION_LOCK (session);
3614   GST_DEBUG_OBJECT (bin, "insert caps for payload %d", payload);
3615   g_hash_table_insert (session->ptmap, GINT_TO_POINTER (payload), caps);
3616   GST_RTP_SESSION_UNLOCK (session);
3617 }
3618
3619 /* a new pad (SSRC) was created in @session */
3620 static void
3621 new_ssrc_pad_found (GstElement * element, guint ssrc, GstPad * pad,
3622     GstRtpBinSession * session)
3623 {
3624   GstRtpBin *rtpbin;
3625   GstRtpBinStream *stream;
3626   GstPad *sinkpad, *srcpad;
3627   gchar *padname;
3628
3629   rtpbin = session->bin;
3630
3631   GST_DEBUG_OBJECT (rtpbin, "new SSRC pad %08x, %s:%s", ssrc,
3632       GST_DEBUG_PAD_NAME (pad));
3633
3634   GST_RTP_BIN_SHUTDOWN_LOCK (rtpbin, shutdown);
3635
3636   GST_RTP_SESSION_LOCK (session);
3637
3638   /* create new stream */
3639   stream = create_stream (session, ssrc);
3640   if (!stream)
3641     goto no_stream;
3642
3643   /* get pad and link */
3644   GST_DEBUG_OBJECT (rtpbin, "linking jitterbuffer RTP");
3645   padname = g_strdup_printf ("src_%u", ssrc);
3646   srcpad = gst_element_get_static_pad (element, padname);
3647   g_free (padname);
3648   sinkpad = gst_element_get_static_pad (stream->buffer, "sink");
3649   gst_pad_link_full (srcpad, sinkpad, GST_PAD_LINK_CHECK_NOTHING);
3650   gst_object_unref (sinkpad);
3651   gst_object_unref (srcpad);
3652
3653   GST_DEBUG_OBJECT (rtpbin, "linking jitterbuffer RTCP");
3654   padname = g_strdup_printf ("rtcp_src_%u", ssrc);
3655   srcpad = gst_element_get_static_pad (element, padname);
3656   g_free (padname);
3657   sinkpad = gst_element_get_request_pad (stream->buffer, "sink_rtcp");
3658   gst_pad_link_full (srcpad, sinkpad, GST_PAD_LINK_CHECK_NOTHING);
3659   gst_object_unref (sinkpad);
3660   gst_object_unref (srcpad);
3661
3662   /* connect to the RTCP sync signal from the jitterbuffer */
3663   GST_DEBUG_OBJECT (rtpbin, "connecting sync signal");
3664   stream->buffer_handlesync_sig = g_signal_connect (stream->buffer,
3665       "handle-sync", (GCallback) gst_rtp_bin_handle_sync, stream);
3666
3667   if (stream->demux) {
3668     /* connect to the new-pad signal of the payload demuxer, this will expose the
3669      * new pad by ghosting it. */
3670     stream->demux_newpad_sig = g_signal_connect (stream->demux,
3671         "new-payload-type", (GCallback) new_payload_found, stream);
3672     stream->demux_padremoved_sig = g_signal_connect (stream->demux,
3673         "pad-removed", (GCallback) payload_pad_removed, stream);
3674
3675     /* connect to the request-pt-map signal. This signal will be emitted by the
3676      * demuxer so that it can apply a proper caps on the buffers for the
3677      * depayloaders. */
3678     stream->demux_ptreq_sig = g_signal_connect (stream->demux,
3679         "request-pt-map", (GCallback) ptdemux_pt_map_requested, session);
3680     /* connect to the  signal so it can be forwarded. */
3681     stream->demux_ptchange_sig = g_signal_connect (stream->demux,
3682         "payload-type-change", (GCallback) payload_type_change, session);
3683   } else {
3684     /* add rtpjitterbuffer src pad to pads */
3685     GstElementClass *klass;
3686     GstPadTemplate *templ;
3687     gchar *padname;
3688     GstPad *gpad, *pad;
3689
3690     pad = gst_element_get_static_pad (stream->buffer, "src");
3691
3692     /* ghost the pad to the parent */
3693     klass = GST_ELEMENT_GET_CLASS (rtpbin);
3694     templ = gst_element_class_get_pad_template (klass, "recv_rtp_src_%u_%u_%u");
3695     padname = g_strdup_printf ("recv_rtp_src_%u_%u_%u",
3696         stream->session->id, stream->ssrc, 255);
3697     gpad = gst_ghost_pad_new_from_template (padname, pad, templ);
3698     g_free (padname);
3699
3700     gst_pad_set_active (gpad, TRUE);
3701     gst_pad_sticky_events_foreach (pad, copy_sticky_events, gpad);
3702     gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), gpad);
3703
3704     gst_object_unref (pad);
3705   }
3706
3707   GST_RTP_SESSION_UNLOCK (session);
3708   GST_RTP_BIN_SHUTDOWN_UNLOCK (rtpbin);
3709
3710   return;
3711
3712   /* ERRORS */
3713 shutdown:
3714   {
3715     GST_DEBUG_OBJECT (rtpbin, "we are shutting down");
3716     return;
3717   }
3718 no_stream:
3719   {
3720     GST_RTP_SESSION_UNLOCK (session);
3721     GST_RTP_BIN_SHUTDOWN_UNLOCK (rtpbin);
3722     GST_DEBUG_OBJECT (rtpbin, "could not create stream");
3723     return;
3724   }
3725 }
3726
3727 static void
3728 session_maybe_create_bundle_demuxer (GstRtpBinSession * session)
3729 {
3730   GstRtpBin *rtpbin;
3731
3732   if (session->bundle_demux)
3733     return;
3734
3735   rtpbin = session->bin;
3736   if (g_signal_has_handler_pending (rtpbin,
3737           gst_rtp_bin_signals[SIGNAL_ON_BUNDLED_SSRC], 0, TRUE)) {
3738     GST_DEBUG_OBJECT (rtpbin, "Adding a bundle SSRC demuxer to session %u",
3739         session->id);
3740     session->bundle_demux = gst_element_factory_make ("rtpssrcdemux", NULL);
3741     session->bundle_demux_newpad_sig = g_signal_connect (session->bundle_demux,
3742         "new-ssrc-pad", (GCallback) new_bundled_ssrc_pad_found, session);
3743
3744     gst_bin_add (GST_BIN_CAST (rtpbin), session->bundle_demux);
3745     gst_element_sync_state_with_parent (session->bundle_demux);
3746   } else {
3747     GST_DEBUG_OBJECT (rtpbin,
3748         "No handler for the on-bundled-ssrc signal so no need for a bundle SSRC demuxer in session %u",
3749         session->id);
3750   }
3751 }
3752
3753 static GstPad *
3754 complete_session_sink (GstRtpBin * rtpbin, GstRtpBinSession * session,
3755     gboolean bundle_demuxer_needed)
3756 {
3757   guint sessid = session->id;
3758   GstPad *recv_rtp_sink;
3759   GstPad *funnel_src;
3760   GstElement *decoder;
3761
3762   g_assert (!session->recv_rtp_sink);
3763
3764   /* get recv_rtp pad and store */
3765   session->recv_rtp_sink =
3766       gst_element_get_request_pad (session->session, "recv_rtp_sink");
3767   if (session->recv_rtp_sink == NULL)
3768     goto pad_failed;
3769
3770   g_signal_connect (session->recv_rtp_sink, "notify::caps",
3771       (GCallback) caps_changed, session);
3772
3773   if (bundle_demuxer_needed)
3774     session_maybe_create_bundle_demuxer (session);
3775
3776   GST_DEBUG_OBJECT (rtpbin, "requesting RTP decoder");
3777   decoder = session_request_element (session, SIGNAL_REQUEST_RTP_DECODER);
3778   if (decoder) {
3779     GstPad *decsrc, *decsink;
3780     GstPadLinkReturn ret;
3781
3782     GST_DEBUG_OBJECT (rtpbin, "linking RTP decoder");
3783     decsink = gst_element_get_static_pad (decoder, "rtp_sink");
3784     if (decsink == NULL)
3785       goto dec_sink_failed;
3786
3787     recv_rtp_sink = decsink;
3788
3789     decsrc = gst_element_get_static_pad (decoder, "rtp_src");
3790     if (decsrc == NULL)
3791       goto dec_src_failed;
3792
3793     if (session->bundle_demux) {
3794       GstPad *demux_sink;
3795       demux_sink = gst_element_get_static_pad (session->bundle_demux, "sink");
3796       ret = gst_pad_link (decsrc, demux_sink);
3797       gst_object_unref (demux_sink);
3798     } else {
3799       ret = gst_pad_link (decsrc, session->recv_rtp_sink);
3800     }
3801     gst_object_unref (decsrc);
3802
3803     if (ret != GST_PAD_LINK_OK)
3804       goto dec_link_failed;
3805
3806   } else {
3807     GST_DEBUG_OBJECT (rtpbin, "no RTP decoder given");
3808     if (session->bundle_demux) {
3809       recv_rtp_sink =
3810           gst_element_get_static_pad (session->bundle_demux, "sink");
3811     } else {
3812       recv_rtp_sink =
3813           gst_element_get_request_pad (session->rtp_funnel, "sink_%u");
3814     }
3815   }
3816
3817   funnel_src = gst_element_get_static_pad (session->rtp_funnel, "src");
3818   gst_pad_link (funnel_src, session->recv_rtp_sink);
3819   gst_object_unref (funnel_src);
3820
3821   return recv_rtp_sink;
3822
3823   /* ERRORS */
3824 pad_failed:
3825   {
3826     g_warning ("rtpbin: failed to get session recv_rtp_sink pad");
3827     return NULL;
3828   }
3829 dec_sink_failed:
3830   {
3831     g_warning ("rtpbin: failed to get decoder sink pad for session %u", sessid);
3832     return NULL;
3833   }
3834 dec_src_failed:
3835   {
3836     g_warning ("rtpbin: failed to get decoder src pad for session %u", sessid);
3837     gst_object_unref (recv_rtp_sink);
3838     return NULL;
3839   }
3840 dec_link_failed:
3841   {
3842     g_warning ("rtpbin: failed to link rtp decoder for session %u", sessid);
3843     gst_object_unref (recv_rtp_sink);
3844     return NULL;
3845   }
3846 }
3847
3848 static void
3849 complete_session_receiver (GstRtpBin * rtpbin, GstRtpBinSession * session,
3850     guint sessid)
3851 {
3852   GstElement *aux;
3853   GstPad *recv_rtp_src;
3854
3855   g_assert (!session->recv_rtp_src);
3856
3857   session->recv_rtp_src =
3858       gst_element_get_static_pad (session->session, "recv_rtp_src");
3859   if (session->recv_rtp_src == NULL)
3860     goto pad_failed;
3861
3862   /* find out if we need AUX elements */
3863   aux = session_request_element (session, SIGNAL_REQUEST_AUX_RECEIVER);
3864   if (aux) {
3865     gchar *pname;
3866     GstPad *auxsink;
3867     GstPadLinkReturn ret;
3868
3869     GST_DEBUG_OBJECT (rtpbin, "linking AUX receiver");
3870
3871     pname = g_strdup_printf ("sink_%u", sessid);
3872     auxsink = gst_element_get_static_pad (aux, pname);
3873     g_free (pname);
3874     if (auxsink == NULL)
3875       goto aux_sink_failed;
3876
3877     ret = gst_pad_link (session->recv_rtp_src, auxsink);
3878     gst_object_unref (auxsink);
3879     if (ret != GST_PAD_LINK_OK)
3880       goto aux_link_failed;
3881
3882     /* this can be NULL when this AUX element is not to be linked any further */
3883     pname = g_strdup_printf ("src_%u", sessid);
3884     recv_rtp_src = gst_element_get_static_pad (aux, pname);
3885     g_free (pname);
3886   } else {
3887     recv_rtp_src = gst_object_ref (session->recv_rtp_src);
3888   }
3889
3890   /* Add a storage element if needed */
3891   if (recv_rtp_src && session->storage) {
3892     GstPadLinkReturn ret;
3893     GstPad *sinkpad = gst_element_get_static_pad (session->storage, "sink");
3894
3895     ret = gst_pad_link (recv_rtp_src, sinkpad);
3896
3897     gst_object_unref (sinkpad);
3898     gst_object_unref (recv_rtp_src);
3899
3900     if (ret != GST_PAD_LINK_OK)
3901       goto storage_link_failed;
3902
3903     recv_rtp_src = gst_element_get_static_pad (session->storage, "src");
3904   }
3905
3906   if (recv_rtp_src) {
3907     GstPad *sinkdpad;
3908
3909     GST_DEBUG_OBJECT (rtpbin, "getting demuxer RTP sink pad");
3910     sinkdpad = gst_element_get_static_pad (session->demux, "sink");
3911     GST_DEBUG_OBJECT (rtpbin, "linking demuxer RTP sink pad");
3912     gst_pad_link_full (recv_rtp_src, sinkdpad, GST_PAD_LINK_CHECK_NOTHING);
3913     gst_object_unref (sinkdpad);
3914     gst_object_unref (recv_rtp_src);
3915
3916     /* connect to the new-ssrc-pad signal of the SSRC demuxer */
3917     session->demux_newpad_sig = g_signal_connect (session->demux,
3918         "new-ssrc-pad", (GCallback) new_ssrc_pad_found, session);
3919     session->demux_padremoved_sig = g_signal_connect (session->demux,
3920         "removed-ssrc-pad", (GCallback) ssrc_demux_pad_removed, session);
3921   }
3922
3923   return;
3924
3925 pad_failed:
3926   {
3927     g_warning ("rtpbin: failed to get session recv_rtp_src pad");
3928     return;
3929   }
3930 aux_sink_failed:
3931   {
3932     g_warning ("rtpbin: failed to get AUX sink pad for session %u", sessid);
3933     return;
3934   }
3935 aux_link_failed:
3936   {
3937     g_warning ("rtpbin: failed to link AUX pad to session %u", sessid);
3938     return;
3939   }
3940 storage_link_failed:
3941   {
3942     g_warning ("rtpbin: failed to link storage");
3943     return;
3944   }
3945 }
3946
3947 /* Create a pad for receiving RTP for the session in @name. Must be called with
3948  * RTP_BIN_LOCK.
3949  */
3950 static GstPad *
3951 create_recv_rtp (GstRtpBin * rtpbin, GstPadTemplate * templ, const gchar * name)
3952 {
3953   guint sessid;
3954   GstRtpBinSession *session;
3955   GstPad *recv_rtp_sink;
3956
3957   /* first get the session number */
3958   if (name == NULL || sscanf (name, "recv_rtp_sink_%u", &sessid) != 1)
3959     goto no_name;
3960
3961   GST_DEBUG_OBJECT (rtpbin, "finding session %u", sessid);
3962
3963   /* get or create session */
3964   session = find_session_by_id (rtpbin, sessid);
3965   if (!session) {
3966     GST_DEBUG_OBJECT (rtpbin, "creating session %u", sessid);
3967     /* create session now */
3968     session = create_session (rtpbin, sessid);
3969     if (session == NULL)
3970       goto create_error;
3971   }
3972
3973   /* check if pad was requested */
3974   if (session->recv_rtp_sink_ghost != NULL)
3975     return session->recv_rtp_sink_ghost;
3976
3977   /* setup the session sink pad */
3978   recv_rtp_sink = complete_session_sink (rtpbin, session, TRUE);
3979   if (!recv_rtp_sink)
3980     goto session_sink_failed;
3981
3982
3983   GST_DEBUG_OBJECT (rtpbin, "ghosting session sink pad");
3984   session->recv_rtp_sink_ghost =
3985       gst_ghost_pad_new_from_template (name, recv_rtp_sink, templ);
3986   gst_object_unref (recv_rtp_sink);
3987   gst_pad_set_active (session->recv_rtp_sink_ghost, TRUE);
3988   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), session->recv_rtp_sink_ghost);
3989
3990   complete_session_receiver (rtpbin, session, sessid);
3991
3992   return session->recv_rtp_sink_ghost;
3993
3994   /* ERRORS */
3995 no_name:
3996   {
3997     g_warning ("rtpbin: invalid name given");
3998     return NULL;
3999   }
4000 create_error:
4001   {
4002     /* create_session already warned */
4003     return NULL;
4004   }
4005 session_sink_failed:
4006   {
4007     /* warning already done */
4008     return NULL;
4009   }
4010 }
4011
4012 static void
4013 remove_recv_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session)
4014 {
4015   if (session->demux_newpad_sig) {
4016     g_signal_handler_disconnect (session->demux, session->demux_newpad_sig);
4017     session->demux_newpad_sig = 0;
4018   }
4019   if (session->demux_padremoved_sig) {
4020     g_signal_handler_disconnect (session->demux, session->demux_padremoved_sig);
4021     session->demux_padremoved_sig = 0;
4022   }
4023   if (session->bundle_demux_newpad_sig) {
4024     g_signal_handler_disconnect (session->bundle_demux,
4025         session->bundle_demux_newpad_sig);
4026     session->bundle_demux_newpad_sig = 0;
4027   }
4028   if (session->recv_rtp_src) {
4029     gst_object_unref (session->recv_rtp_src);
4030     session->recv_rtp_src = NULL;
4031   }
4032   if (session->recv_rtp_sink) {
4033     gst_element_release_request_pad (session->session, session->recv_rtp_sink);
4034     gst_object_unref (session->recv_rtp_sink);
4035     session->recv_rtp_sink = NULL;
4036   }
4037   if (session->recv_rtp_sink_ghost) {
4038     gst_pad_set_active (session->recv_rtp_sink_ghost, FALSE);
4039     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
4040         session->recv_rtp_sink_ghost);
4041     session->recv_rtp_sink_ghost = NULL;
4042   }
4043 }
4044
4045 static GstPad *
4046 complete_session_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session,
4047     guint sessid, gboolean bundle_demuxer_needed)
4048 {
4049   GstElement *decoder;
4050   GstPad *sinkdpad;
4051   GstPad *decsink = NULL;
4052   GstPad *funnel_src;
4053
4054   /* get recv_rtp pad and store */
4055   GST_DEBUG_OBJECT (rtpbin, "getting RTCP sink pad");
4056   session->recv_rtcp_sink =
4057       gst_element_get_request_pad (session->session, "recv_rtcp_sink");
4058   if (session->recv_rtcp_sink == NULL)
4059     goto pad_failed;
4060
4061   if (bundle_demuxer_needed)
4062     session_maybe_create_bundle_demuxer (session);
4063
4064   GST_DEBUG_OBJECT (rtpbin, "getting RTCP decoder");
4065   decoder = session_request_element (session, SIGNAL_REQUEST_RTCP_DECODER);
4066   if (decoder) {
4067     GstPad *decsrc;
4068     GstPadLinkReturn ret;
4069
4070     GST_DEBUG_OBJECT (rtpbin, "linking RTCP decoder");
4071     decsink = gst_element_get_static_pad (decoder, "rtcp_sink");
4072     decsrc = gst_element_get_static_pad (decoder, "rtcp_src");
4073
4074     if (decsink == NULL)
4075       goto dec_sink_failed;
4076
4077     if (decsrc == NULL)
4078       goto dec_src_failed;
4079
4080     if (session->bundle_demux) {
4081       GstPad *demux_sink;
4082       demux_sink =
4083           gst_element_get_static_pad (session->bundle_demux, "rtcp_sink");
4084       ret = gst_pad_link (decsrc, demux_sink);
4085       gst_object_unref (demux_sink);
4086     } else {
4087       ret = gst_pad_link (decsrc, session->recv_rtcp_sink);
4088     }
4089     gst_object_unref (decsrc);
4090
4091     if (ret != GST_PAD_LINK_OK)
4092       goto dec_link_failed;
4093   } else {
4094     GST_DEBUG_OBJECT (rtpbin, "no RTCP decoder given");
4095     if (session->bundle_demux) {
4096       decsink = gst_element_get_static_pad (session->bundle_demux, "rtcp_sink");
4097     } else {
4098       decsink = gst_element_get_request_pad (session->rtcp_funnel, "sink_%u");
4099     }
4100   }
4101
4102   /* get srcpad, link to SSRCDemux */
4103   GST_DEBUG_OBJECT (rtpbin, "getting sync src pad");
4104   session->sync_src = gst_element_get_static_pad (session->session, "sync_src");
4105   if (session->sync_src == NULL)
4106     goto src_pad_failed;
4107
4108   GST_DEBUG_OBJECT (rtpbin, "getting demuxer RTCP sink pad");
4109   sinkdpad = gst_element_get_static_pad (session->demux, "rtcp_sink");
4110   gst_pad_link_full (session->sync_src, sinkdpad, GST_PAD_LINK_CHECK_NOTHING);
4111   gst_object_unref (sinkdpad);
4112
4113   funnel_src = gst_element_get_static_pad (session->rtcp_funnel, "src");
4114   gst_pad_link (funnel_src, session->recv_rtcp_sink);
4115   gst_object_unref (funnel_src);
4116
4117   return decsink;
4118
4119 pad_failed:
4120   {
4121     g_warning ("rtpbin: failed to get session rtcp_sink pad");
4122     return NULL;
4123   }
4124 dec_sink_failed:
4125   {
4126     g_warning ("rtpbin: failed to get decoder sink pad for session %u", sessid);
4127     return NULL;
4128   }
4129 dec_src_failed:
4130   {
4131     g_warning ("rtpbin: failed to get decoder src pad for session %u", sessid);
4132     goto cleanup;
4133   }
4134 dec_link_failed:
4135   {
4136     g_warning ("rtpbin: failed to link rtcp decoder for session %u", sessid);
4137     goto cleanup;
4138   }
4139 src_pad_failed:
4140   {
4141     g_warning ("rtpbin: failed to get session sync_src pad");
4142   }
4143
4144 cleanup:
4145   gst_object_unref (decsink);
4146   return NULL;
4147 }
4148
4149 /* Create a pad for receiving RTCP for the session in @name. Must be called with
4150  * RTP_BIN_LOCK.
4151  */
4152 static GstPad *
4153 create_recv_rtcp (GstRtpBin * rtpbin, GstPadTemplate * templ,
4154     const gchar * name)
4155 {
4156   guint sessid;
4157   GstRtpBinSession *session;
4158   GstPad *decsink = NULL;
4159
4160   /* first get the session number */
4161   if (name == NULL || sscanf (name, "recv_rtcp_sink_%u", &sessid) != 1)
4162     goto no_name;
4163
4164   GST_DEBUG_OBJECT (rtpbin, "finding session %u", sessid);
4165
4166   /* get or create the session */
4167   session = find_session_by_id (rtpbin, sessid);
4168   if (!session) {
4169     GST_DEBUG_OBJECT (rtpbin, "creating session %u", sessid);
4170     /* create session now */
4171     session = create_session (rtpbin, sessid);
4172     if (session == NULL)
4173       goto create_error;
4174   }
4175
4176   /* check if pad was requested */
4177   if (session->recv_rtcp_sink_ghost != NULL)
4178     return session->recv_rtcp_sink_ghost;
4179
4180   decsink = complete_session_rtcp (rtpbin, session, sessid, TRUE);
4181   if (!decsink)
4182     goto create_error;
4183
4184   session->recv_rtcp_sink_ghost =
4185       gst_ghost_pad_new_from_template (name, decsink, templ);
4186   gst_object_unref (decsink);
4187   gst_pad_set_active (session->recv_rtcp_sink_ghost, TRUE);
4188   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin),
4189       session->recv_rtcp_sink_ghost);
4190
4191   return session->recv_rtcp_sink_ghost;
4192
4193   /* ERRORS */
4194 no_name:
4195   {
4196     g_warning ("rtpbin: invalid name given");
4197     return NULL;
4198   }
4199 create_error:
4200   {
4201     /* create_session already warned */
4202     return NULL;
4203   }
4204 }
4205
4206 static void
4207 remove_recv_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session)
4208 {
4209   if (session->recv_rtcp_sink_ghost) {
4210     gst_pad_set_active (session->recv_rtcp_sink_ghost, FALSE);
4211     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
4212         session->recv_rtcp_sink_ghost);
4213     session->recv_rtcp_sink_ghost = NULL;
4214   }
4215   if (session->sync_src) {
4216     /* releasing the request pad should also unref the sync pad */
4217     gst_object_unref (session->sync_src);
4218     session->sync_src = NULL;
4219   }
4220   if (session->recv_rtcp_sink) {
4221     gst_element_release_request_pad (session->session, session->recv_rtcp_sink);
4222     gst_object_unref (session->recv_rtcp_sink);
4223     session->recv_rtcp_sink = NULL;
4224   }
4225 }
4226
4227 static gboolean
4228 complete_session_src (GstRtpBin * rtpbin, GstRtpBinSession * session)
4229 {
4230   gchar *gname;
4231   guint sessid = session->id;
4232   GstPad *send_rtp_src;
4233   GstElement *encoder;
4234   GstElementClass *klass;
4235   GstPadTemplate *templ;
4236   gboolean ret = FALSE;
4237
4238   /* get srcpad */
4239   send_rtp_src = gst_element_get_static_pad (session->session, "send_rtp_src");
4240
4241   if (send_rtp_src == NULL)
4242     goto no_srcpad;
4243
4244   GST_DEBUG_OBJECT (rtpbin, "getting RTP encoder");
4245   encoder = session_request_element (session, SIGNAL_REQUEST_RTP_ENCODER);
4246   if (encoder) {
4247     gchar *ename;
4248     GstPad *encsrc, *encsink;
4249     GstPadLinkReturn ret;
4250
4251     GST_DEBUG_OBJECT (rtpbin, "linking RTP encoder");
4252     ename = g_strdup_printf ("rtp_src_%u", sessid);
4253     encsrc = gst_element_get_static_pad (encoder, ename);
4254     g_free (ename);
4255
4256     if (encsrc == NULL)
4257       goto enc_src_failed;
4258
4259     ename = g_strdup_printf ("rtp_sink_%u", sessid);
4260     encsink = gst_element_get_static_pad (encoder, ename);
4261     g_free (ename);
4262     if (encsink == NULL)
4263       goto enc_sink_failed;
4264
4265     ret = gst_pad_link (send_rtp_src, encsink);
4266     gst_object_unref (encsink);
4267     gst_object_unref (send_rtp_src);
4268
4269     send_rtp_src = encsrc;
4270
4271     if (ret != GST_PAD_LINK_OK)
4272       goto enc_link_failed;
4273   } else {
4274     GST_DEBUG_OBJECT (rtpbin, "no RTP encoder given");
4275   }
4276
4277   /* ghost the new source pad */
4278   klass = GST_ELEMENT_GET_CLASS (rtpbin);
4279   gname = g_strdup_printf ("send_rtp_src_%u", sessid);
4280   templ = gst_element_class_get_pad_template (klass, "send_rtp_src_%u");
4281   session->send_rtp_src_ghost =
4282       gst_ghost_pad_new_from_template (gname, send_rtp_src, templ);
4283   gst_pad_set_active (session->send_rtp_src_ghost, TRUE);
4284   gst_pad_sticky_events_foreach (send_rtp_src, copy_sticky_events,
4285       session->send_rtp_src_ghost);
4286   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), session->send_rtp_src_ghost);
4287   g_free (gname);
4288
4289   ret = TRUE;
4290
4291 done:
4292   if (send_rtp_src)
4293     gst_object_unref (send_rtp_src);
4294
4295   return ret;
4296
4297   /* ERRORS */
4298 no_srcpad:
4299   {
4300     g_warning ("rtpbin: failed to get rtp source pad for session %u", sessid);
4301     goto done;
4302   }
4303 enc_src_failed:
4304   {
4305     g_warning ("rtpbin: failed to get %" GST_PTR_FORMAT
4306         " src pad for session %u", encoder, sessid);
4307     goto done;
4308   }
4309 enc_sink_failed:
4310   {
4311     g_warning ("rtpbin: failed to get %" GST_PTR_FORMAT
4312         " sink pad for session %u", encoder, sessid);
4313     goto done;
4314   }
4315 enc_link_failed:
4316   {
4317     g_warning ("rtpbin: failed to link %" GST_PTR_FORMAT " for session %u",
4318         encoder, sessid);
4319     goto done;
4320   }
4321 }
4322
4323 static gboolean
4324 setup_aux_sender_fold (const GValue * item, GValue * result, gpointer user_data)
4325 {
4326   GstPad *pad;
4327   gchar *name;
4328   guint sessid;
4329   GstRtpBinSession *session = user_data, *newsess;
4330   GstRtpBin *rtpbin = session->bin;
4331   GstPadLinkReturn ret;
4332
4333   pad = g_value_get_object (item);
4334   name = gst_pad_get_name (pad);
4335
4336   if (name == NULL || sscanf (name, "src_%u", &sessid) != 1)
4337     goto no_name;
4338
4339   g_free (name);
4340
4341   newsess = find_session_by_id (rtpbin, sessid);
4342   if (newsess == NULL) {
4343     /* create new session */
4344     newsess = create_session (rtpbin, sessid);
4345     if (newsess == NULL)
4346       goto create_error;
4347   } else if (newsess->send_rtp_sink != NULL)
4348     goto existing_session;
4349
4350   /* get send_rtp pad and store */
4351   newsess->send_rtp_sink =
4352       gst_element_get_request_pad (newsess->session, "send_rtp_sink");
4353   if (newsess->send_rtp_sink == NULL)
4354     goto pad_failed;
4355
4356   ret = gst_pad_link (pad, newsess->send_rtp_sink);
4357   if (ret != GST_PAD_LINK_OK)
4358     goto aux_link_failed;
4359
4360   if (!complete_session_src (rtpbin, newsess))
4361     goto session_src_failed;
4362
4363   return TRUE;
4364
4365   /* ERRORS */
4366 no_name:
4367   {
4368     GST_WARNING ("ignoring invalid pad name %s", GST_STR_NULL (name));
4369     g_free (name);
4370     return TRUE;
4371   }
4372 create_error:
4373   {
4374     /* create_session already warned */
4375     return FALSE;
4376   }
4377 existing_session:
4378   {
4379     g_warning ("rtpbin: session %u is already a sender", sessid);
4380     return FALSE;
4381   }
4382 pad_failed:
4383   {
4384     g_warning ("rtpbin: failed to get session pad for session %u", sessid);
4385     return FALSE;
4386   }
4387 aux_link_failed:
4388   {
4389     g_warning ("rtpbin: failed to link AUX for session %u", sessid);
4390     return FALSE;
4391   }
4392 session_src_failed:
4393   {
4394     g_warning ("rtpbin: failed to complete AUX for session %u", sessid);
4395     return FALSE;
4396   }
4397 }
4398
4399 static gboolean
4400 setup_aux_sender (GstRtpBin * rtpbin, GstRtpBinSession * session,
4401     GstElement * aux)
4402 {
4403   GstIterator *it;
4404   GValue result = { 0, };
4405   GstIteratorResult res;
4406
4407   it = gst_element_iterate_src_pads (aux);
4408   res = gst_iterator_fold (it, setup_aux_sender_fold, &result, session);
4409   gst_iterator_free (it);
4410
4411   return res == GST_ITERATOR_DONE;
4412 }
4413
4414 /* Create a pad for sending RTP for the session in @name. Must be called with
4415  * RTP_BIN_LOCK.
4416  */
4417 static GstPad *
4418 create_send_rtp (GstRtpBin * rtpbin, GstPadTemplate * templ, const gchar * name)
4419 {
4420   gchar *pname;
4421   guint sessid;
4422   GstPad *send_rtp_sink;
4423   GstElement *aux;
4424   GstElement *encoder;
4425   GstElement *prev = NULL;
4426   GstRtpBinSession *session;
4427
4428   /* first get the session number */
4429   if (name == NULL || sscanf (name, "send_rtp_sink_%u", &sessid) != 1)
4430     goto no_name;
4431
4432   /* get or create session */
4433   session = find_session_by_id (rtpbin, sessid);
4434   if (!session) {
4435     /* create session now */
4436     session = create_session (rtpbin, sessid);
4437     if (session == NULL)
4438       goto create_error;
4439   }
4440
4441   /* check if pad was requested */
4442   if (session->send_rtp_sink_ghost != NULL)
4443     return session->send_rtp_sink_ghost;
4444
4445   /* check if we are already using this session as a sender */
4446   if (session->send_rtp_sink != NULL)
4447     goto existing_session;
4448
4449   encoder = session_request_element (session, SIGNAL_REQUEST_FEC_ENCODER);
4450
4451   if (encoder) {
4452     GST_DEBUG_OBJECT (rtpbin, "Linking FEC encoder");
4453
4454     send_rtp_sink = gst_element_get_static_pad (encoder, "sink");
4455
4456     if (!send_rtp_sink)
4457       goto enc_sink_failed;
4458
4459     prev = encoder;
4460   }
4461
4462   GST_DEBUG_OBJECT (rtpbin, "getting RTP AUX sender");
4463   aux = session_request_element (session, SIGNAL_REQUEST_AUX_SENDER);
4464   if (aux) {
4465     GstPad *sinkpad;
4466     GST_DEBUG_OBJECT (rtpbin, "linking AUX sender");
4467     if (!setup_aux_sender (rtpbin, session, aux))
4468       goto aux_session_failed;
4469
4470     pname = g_strdup_printf ("sink_%u", sessid);
4471     sinkpad = gst_element_get_static_pad (aux, pname);
4472     g_free (pname);
4473
4474     if (sinkpad == NULL)
4475       goto aux_sink_failed;
4476
4477     if (!prev) {
4478       send_rtp_sink = sinkpad;
4479     } else {
4480       GstPad *srcpad = gst_element_get_static_pad (prev, "src");
4481       GstPadLinkReturn ret;
4482
4483       ret = gst_pad_link (srcpad, sinkpad);
4484       gst_object_unref (srcpad);
4485       if (ret != GST_PAD_LINK_OK) {
4486         goto aux_link_failed;
4487       }
4488     }
4489     prev = aux;
4490   } else {
4491     /* get send_rtp pad and store */
4492     session->send_rtp_sink =
4493         gst_element_get_request_pad (session->session, "send_rtp_sink");
4494     if (session->send_rtp_sink == NULL)
4495       goto pad_failed;
4496
4497     if (!complete_session_src (rtpbin, session))
4498       goto session_src_failed;
4499
4500     if (!prev) {
4501       send_rtp_sink = gst_object_ref (session->send_rtp_sink);
4502     } else {
4503       GstPad *srcpad = gst_element_get_static_pad (prev, "src");
4504       GstPadLinkReturn ret;
4505
4506       ret = gst_pad_link (srcpad, session->send_rtp_sink);
4507       gst_object_unref (srcpad);
4508       if (ret != GST_PAD_LINK_OK)
4509         goto session_link_failed;
4510     }
4511   }
4512
4513   session->send_rtp_sink_ghost =
4514       gst_ghost_pad_new_from_template (name, send_rtp_sink, templ);
4515   gst_object_unref (send_rtp_sink);
4516   gst_pad_set_active (session->send_rtp_sink_ghost, TRUE);
4517   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), session->send_rtp_sink_ghost);
4518
4519   return session->send_rtp_sink_ghost;
4520
4521   /* ERRORS */
4522 no_name:
4523   {
4524     g_warning ("rtpbin: invalid name given");
4525     return NULL;
4526   }
4527 create_error:
4528   {
4529     /* create_session already warned */
4530     return NULL;
4531   }
4532 existing_session:
4533   {
4534     g_warning ("rtpbin: session %u is already in use", sessid);
4535     return NULL;
4536   }
4537 aux_session_failed:
4538   {
4539     g_warning ("rtpbin: failed to get AUX sink pad for session %u", sessid);
4540     return NULL;
4541   }
4542 aux_sink_failed:
4543   {
4544     g_warning ("rtpbin: failed to get AUX sink pad for session %u", sessid);
4545     return NULL;
4546   }
4547 aux_link_failed:
4548   {
4549     g_warning ("rtpbin: failed to link %" GST_PTR_FORMAT " for session %u",
4550         aux, sessid);
4551     return NULL;
4552   }
4553 pad_failed:
4554   {
4555     g_warning ("rtpbin: failed to get session pad for session %u", sessid);
4556     return NULL;
4557   }
4558 session_src_failed:
4559   {
4560     g_warning ("rtpbin: failed to setup source pads for session %u", sessid);
4561     return NULL;
4562   }
4563 session_link_failed:
4564   {
4565     g_warning ("rtpbin: failed to link %" GST_PTR_FORMAT " for session %u",
4566         session, sessid);
4567     return NULL;
4568   }
4569 enc_sink_failed:
4570   {
4571     g_warning ("rtpbin: failed to get %" GST_PTR_FORMAT
4572         " sink pad for session %u", encoder, sessid);
4573     return NULL;
4574   }
4575 }
4576
4577 static void
4578 remove_send_rtp (GstRtpBin * rtpbin, GstRtpBinSession * session)
4579 {
4580   if (session->send_rtp_src_ghost) {
4581     gst_pad_set_active (session->send_rtp_src_ghost, FALSE);
4582     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
4583         session->send_rtp_src_ghost);
4584     session->send_rtp_src_ghost = NULL;
4585   }
4586   if (session->send_rtp_sink) {
4587     gst_element_release_request_pad (GST_ELEMENT_CAST (session->session),
4588         session->send_rtp_sink);
4589     gst_object_unref (session->send_rtp_sink);
4590     session->send_rtp_sink = NULL;
4591   }
4592   if (session->send_rtp_sink_ghost) {
4593     gst_pad_set_active (session->send_rtp_sink_ghost, FALSE);
4594     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
4595         session->send_rtp_sink_ghost);
4596     session->send_rtp_sink_ghost = NULL;
4597   }
4598 }
4599
4600 /* Create a pad for sending RTCP for the session in @name. Must be called with
4601  * RTP_BIN_LOCK.
4602  */
4603 static GstPad *
4604 create_send_rtcp (GstRtpBin * rtpbin, GstPadTemplate * templ,
4605     const gchar * name)
4606 {
4607   guint sessid;
4608   GstPad *encsrc;
4609   GstElement *encoder;
4610   GstRtpBinSession *session;
4611
4612   /* first get the session number */
4613   if (name == NULL || sscanf (name, "send_rtcp_src_%u", &sessid) != 1)
4614     goto no_name;
4615
4616   /* get or create session */
4617   session = find_session_by_id (rtpbin, sessid);
4618   if (!session) {
4619     GST_DEBUG_OBJECT (rtpbin, "creating session %u", sessid);
4620     /* create session now */
4621     session = create_session (rtpbin, sessid);
4622     if (session == NULL)
4623       goto create_error;
4624   }
4625
4626   /* check if pad was requested */
4627   if (session->send_rtcp_src_ghost != NULL)
4628     return session->send_rtcp_src_ghost;
4629
4630   /* get rtcp_src pad and store */
4631   session->send_rtcp_src =
4632       gst_element_get_request_pad (session->session, "send_rtcp_src");
4633   if (session->send_rtcp_src == NULL)
4634     goto pad_failed;
4635
4636   GST_DEBUG_OBJECT (rtpbin, "getting RTCP encoder");
4637   encoder = session_request_element (session, SIGNAL_REQUEST_RTCP_ENCODER);
4638   if (encoder) {
4639     gchar *ename;
4640     GstPad *encsink;
4641     GstPadLinkReturn ret;
4642
4643     GST_DEBUG_OBJECT (rtpbin, "linking RTCP encoder");
4644
4645     ename = g_strdup_printf ("rtcp_src_%u", sessid);
4646     encsrc = gst_element_get_static_pad (encoder, ename);
4647     g_free (ename);
4648     if (encsrc == NULL)
4649       goto enc_src_failed;
4650
4651     ename = g_strdup_printf ("rtcp_sink_%u", sessid);
4652     encsink = gst_element_get_static_pad (encoder, ename);
4653     g_free (ename);
4654     if (encsink == NULL)
4655       goto enc_sink_failed;
4656
4657     ret = gst_pad_link (session->send_rtcp_src, encsink);
4658     gst_object_unref (encsink);
4659
4660     if (ret != GST_PAD_LINK_OK)
4661       goto enc_link_failed;
4662   } else {
4663     GST_DEBUG_OBJECT (rtpbin, "no RTCP encoder given");
4664     encsrc = gst_object_ref (session->send_rtcp_src);
4665   }
4666
4667   session->send_rtcp_src_ghost =
4668       gst_ghost_pad_new_from_template (name, encsrc, templ);
4669   gst_object_unref (encsrc);
4670   gst_pad_set_active (session->send_rtcp_src_ghost, TRUE);
4671   gst_element_add_pad (GST_ELEMENT_CAST (rtpbin), session->send_rtcp_src_ghost);
4672
4673   return session->send_rtcp_src_ghost;
4674
4675   /* ERRORS */
4676 no_name:
4677   {
4678     g_warning ("rtpbin: invalid name given");
4679     return NULL;
4680   }
4681 create_error:
4682   {
4683     /* create_session already warned */
4684     return NULL;
4685   }
4686 pad_failed:
4687   {
4688     g_warning ("rtpbin: failed to get rtcp pad for session %u", sessid);
4689     return NULL;
4690   }
4691 enc_src_failed:
4692   {
4693     g_warning ("rtpbin: failed to get encoder src pad for session %u", sessid);
4694     return NULL;
4695   }
4696 enc_sink_failed:
4697   {
4698     g_warning ("rtpbin: failed to get encoder sink pad for session %u", sessid);
4699     gst_object_unref (encsrc);
4700     return NULL;
4701   }
4702 enc_link_failed:
4703   {
4704     g_warning ("rtpbin: failed to link rtcp encoder for session %u", sessid);
4705     gst_object_unref (encsrc);
4706     return NULL;
4707   }
4708 }
4709
4710 static void
4711 remove_rtcp (GstRtpBin * rtpbin, GstRtpBinSession * session)
4712 {
4713   if (session->send_rtcp_src_ghost) {
4714     gst_pad_set_active (session->send_rtcp_src_ghost, FALSE);
4715     gst_element_remove_pad (GST_ELEMENT_CAST (rtpbin),
4716         session->send_rtcp_src_ghost);
4717     session->send_rtcp_src_ghost = NULL;
4718   }
4719   if (session->send_rtcp_src) {
4720     gst_element_release_request_pad (session->session, session->send_rtcp_src);
4721     gst_object_unref (session->send_rtcp_src);
4722     session->send_rtcp_src = NULL;
4723   }
4724 }
4725
4726 /* If the requested name is NULL we should create a name with
4727  * the session number assuming we want the lowest posible session
4728  * with a free pad like the template */
4729 static gchar *
4730 gst_rtp_bin_get_free_pad_name (GstElement * element, GstPadTemplate * templ)
4731 {
4732   gboolean name_found = FALSE;
4733   gint session = 0;
4734   GstIterator *pad_it = NULL;
4735   gchar *pad_name = NULL;
4736   GValue data = { 0, };
4737
4738   GST_DEBUG_OBJECT (element, "find a free pad name for template");
4739   while (!name_found) {
4740     gboolean done = FALSE;
4741
4742     g_free (pad_name);
4743     pad_name = g_strdup_printf (templ->name_template, session++);
4744     pad_it = gst_element_iterate_pads (GST_ELEMENT (element));
4745     name_found = TRUE;
4746     while (!done) {
4747       switch (gst_iterator_next (pad_it, &data)) {
4748         case GST_ITERATOR_OK:
4749         {
4750           GstPad *pad;
4751           gchar *name;
4752
4753           pad = g_value_get_object (&data);
4754           name = gst_pad_get_name (pad);
4755
4756           if (strcmp (name, pad_name) == 0) {
4757             done = TRUE;
4758             name_found = FALSE;
4759           }
4760           g_free (name);
4761           g_value_reset (&data);
4762           break;
4763         }
4764         case GST_ITERATOR_ERROR:
4765         case GST_ITERATOR_RESYNC:
4766           /* restart iteration */
4767           done = TRUE;
4768           name_found = FALSE;
4769           session = 0;
4770           break;
4771         case GST_ITERATOR_DONE:
4772           done = TRUE;
4773           break;
4774       }
4775     }
4776     g_value_unset (&data);
4777     gst_iterator_free (pad_it);
4778   }
4779
4780   GST_DEBUG_OBJECT (element, "free pad name found: '%s'", pad_name);
4781   return pad_name;
4782 }
4783
4784 /*
4785  */
4786 static GstPad *
4787 gst_rtp_bin_request_new_pad (GstElement * element,
4788     GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
4789 {
4790   GstRtpBin *rtpbin;
4791   GstElementClass *klass;
4792   GstPad *result;
4793
4794   gchar *pad_name = NULL;
4795
4796   g_return_val_if_fail (templ != NULL, NULL);
4797   g_return_val_if_fail (GST_IS_RTP_BIN (element), NULL);
4798
4799   rtpbin = GST_RTP_BIN (element);
4800   klass = GST_ELEMENT_GET_CLASS (element);
4801
4802   GST_RTP_BIN_LOCK (rtpbin);
4803
4804   if (name == NULL) {
4805     /* use a free pad name */
4806     pad_name = gst_rtp_bin_get_free_pad_name (element, templ);
4807   } else {
4808     /* use the provided name */
4809     pad_name = g_strdup (name);
4810   }
4811
4812   GST_DEBUG_OBJECT (rtpbin, "Trying to request a pad with name %s", pad_name);
4813
4814   /* figure out the template */
4815   if (templ == gst_element_class_get_pad_template (klass, "recv_rtp_sink_%u")) {
4816     result = create_recv_rtp (rtpbin, templ, pad_name);
4817   } else if (templ == gst_element_class_get_pad_template (klass,
4818           "recv_rtcp_sink_%u")) {
4819     result = create_recv_rtcp (rtpbin, templ, pad_name);
4820   } else if (templ == gst_element_class_get_pad_template (klass,
4821           "send_rtp_sink_%u")) {
4822     result = create_send_rtp (rtpbin, templ, pad_name);
4823   } else if (templ == gst_element_class_get_pad_template (klass,
4824           "send_rtcp_src_%u")) {
4825     result = create_send_rtcp (rtpbin, templ, pad_name);
4826   } else
4827     goto wrong_template;
4828
4829   g_free (pad_name);
4830   GST_RTP_BIN_UNLOCK (rtpbin);
4831
4832   return result;
4833
4834   /* ERRORS */
4835 wrong_template:
4836   {
4837     g_free (pad_name);
4838     GST_RTP_BIN_UNLOCK (rtpbin);
4839     g_warning ("rtpbin: this is not our template");
4840     return NULL;
4841   }
4842 }
4843
4844 static void
4845 gst_rtp_bin_release_pad (GstElement * element, GstPad * pad)
4846 {
4847   GstRtpBinSession *session;
4848   GstRtpBin *rtpbin;
4849
4850   g_return_if_fail (GST_IS_GHOST_PAD (pad));
4851   g_return_if_fail (GST_IS_RTP_BIN (element));
4852
4853   rtpbin = GST_RTP_BIN (element);
4854
4855   GST_RTP_BIN_LOCK (rtpbin);
4856   GST_DEBUG_OBJECT (rtpbin, "Trying to release pad %s:%s",
4857       GST_DEBUG_PAD_NAME (pad));
4858
4859   if (!(session = find_session_by_pad (rtpbin, pad)))
4860     goto unknown_pad;
4861
4862   if (session->recv_rtp_sink_ghost == pad) {
4863     remove_recv_rtp (rtpbin, session);
4864   } else if (session->recv_rtcp_sink_ghost == pad) {
4865     remove_recv_rtcp (rtpbin, session);
4866   } else if (session->send_rtp_sink_ghost == pad) {
4867     remove_send_rtp (rtpbin, session);
4868   } else if (session->send_rtcp_src_ghost == pad) {
4869     remove_rtcp (rtpbin, session);
4870   }
4871
4872   /* no more request pads, free the complete session */
4873   if (session->recv_rtp_sink_ghost == NULL
4874       && session->recv_rtcp_sink_ghost == NULL
4875       && session->send_rtp_sink_ghost == NULL
4876       && session->send_rtcp_src_ghost == NULL) {
4877     GST_DEBUG_OBJECT (rtpbin, "no more pads for session %p", session);
4878     rtpbin->sessions = g_slist_remove (rtpbin->sessions, session);
4879     free_session (session, rtpbin);
4880   }
4881   GST_RTP_BIN_UNLOCK (rtpbin);
4882
4883   return;
4884
4885   /* ERROR */
4886 unknown_pad:
4887   {
4888     GST_RTP_BIN_UNLOCK (rtpbin);
4889     g_warning ("rtpbin: %s:%s is not one of our request pads",
4890         GST_DEBUG_PAD_NAME (pad));
4891     return;
4892   }
4893 }