gst/rtpmanager/gstrtpbin.c

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