gst/rtpmanager/gstrtpsession.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-rtpsession
  22  * @see_also: rtpjitterbuffer, rtpbin, rtpptdemux, rtpssrcdemux
  23  *
  24  * The RTP session manager models participants with unique SSRC in an RTP
  25  * session. This session can be used to send and receive RTP and RTCP packets.
  26  * Based on what REQUEST pads are requested from the session manager, specific
  27  * functionality can be activated.
  28  *
  29  * The session manager currently implements RFC 3550 including:
  30  * <itemizedlist>
  31  *   <listitem>
  32  *     <para>RTP packet validation based on consecutive sequence numbers.</para>
  33  *   </listitem>
  34  *   <listitem>
  35  *     <para>Maintainance of the SSRC participant database.</para>
  36  *   </listitem>
  37  *   <listitem>
  38  *     <para>Keeping per participant statistics based on received RTCP packets.</para>
  39  *   </listitem>
  40  *   <listitem>
  41  *     <para>Scheduling of RR/SR RTCP packets.</para>
  42  *   </listitem>
  43  *   <listitem>
  44  *     <para>Support for multiple sender SSRC.</para>
  45  *   </listitem>
  46  * </itemizedlist>
  47  *
  48  * The rtpsession will not demux packets based on SSRC or payload type, nor will
  49  * it correct for packet reordering and jitter. Use #GstRtpsSrcDemux,
  50  * #GstRtpPtDemux and GstRtpJitterBuffer in addition to #GstRtpSession to
  51  * perform these tasks. It is usually a good idea to use #GstRtpBin, which
  52  * combines all these features in one element.
  53  *
  54  * To use #GstRtpSession as an RTP receiver, request a recv_rtp_sink pad, which will
  55  * automatically create recv_rtp_src pad. Data received on the recv_rtp_sink pad
  56  * will be processed in the session and after being validated forwarded on the
  57  * recv_rtp_src pad.
  58  *
  59  * To also use #GstRtpSession as an RTCP receiver, request a recv_rtcp_sink pad,
  60  * which will automatically create a sync_src pad. Packets received on the RTCP
  61  * pad will be used by the session manager to update the stats and database of
  62  * the other participants. SR packets will be forwarded on the sync_src pad
  63  * so that they can be used to perform inter-stream synchronisation when needed.
  64  *
  65  * If you want the session manager to generate and send RTCP packets, request
  66  * the send_rtcp_src pad. Packet pushed on this pad contain SR/RR RTCP reports
  67  * that should be sent to all participants in the session.
  68  *
  69  * To use #GstRtpSession as a sender, request a send_rtp_sink pad, which will
  70  * automatically create a send_rtp_src pad. The session manager will
  71  * forward the packets on the send_rtp_src pad after updating its internal state.
  72  *
  73  * The session manager needs the clock-rate of the payload types it is handling
  74  * and will signal the #GstRtpSession::request-pt-map signal when it needs such a
  75  * mapping. One can clear the cached values with the #GstRtpSession::clear-pt-map
  76  * signal.
  77  *
  78  * <refsect2>
  79  * <title>Example pipelines</title>
  80  * |[
  81  * gst-launch-1.0 udpsrc port=5000 caps="application/x-rtp, ..." ! .recv_rtp_sink rtpsession .recv_rtp_src ! rtptheoradepay ! theoradec ! xvimagesink
  82  * ]| Receive theora RTP packets from port 5000 and send them to the depayloader,
  83  * decoder and display. Note that the application/x-rtp caps on udpsrc should be
  84  * configured based on some negotiation process such as RTSP for this pipeline
  85  * to work correctly.
  86  * |[
  87  * gst-launch-1.0 udpsrc port=5000 caps="application/x-rtp, ..." ! .recv_rtp_sink rtpsession name=session \
  88  *        .recv_rtp_src ! rtptheoradepay ! theoradec ! xvimagesink \
  89  *     udpsrc port=5001 caps="application/x-rtcp" ! session.recv_rtcp_sink
  90  * ]| Receive theora RTP packets from port 5000 and send them to the depayloader,
  91  * decoder and display. Receive RTCP packets from port 5001 and process them in
  92  * the session manager.
  93  * Note that the application/x-rtp caps on udpsrc should be
  94  * configured based on some negotiation process such as RTSP for this pipeline
  95  * to work correctly.
  96  * |[
  97  * gst-launch-1.0 videotestsrc ! theoraenc ! rtptheorapay ! .send_rtp_sink rtpsession .send_rtp_src ! udpsink port=5000
  98  * ]| Send theora RTP packets through the session manager and out on UDP port
  99  * 5000.
 100  * |[
 101  * gst-launch-1.0 videotestsrc ! theoraenc ! rtptheorapay ! .send_rtp_sink rtpsession name=session .send_rtp_src \
 102  *     ! udpsink port=5000  session.send_rtcp_src ! udpsink port=5001
 103  * ]| Send theora RTP packets through the session manager and out on UDP port
 104  * 5000. Send RTCP packets on port 5001. Note that this pipeline will not preroll
 105  * correctly because the second udpsink will not preroll correctly (no RTCP
 106  * packets are sent in the PAUSED state). Applications should manually set and
 107  * keep (see gst_element_set_locked_state()) the RTCP udpsink to the PLAYING state.
 108  * </refsect2>
 109  */
 110
 111 #ifdef HAVE_CONFIG_H
 112 #include "config.h"
 113 #endif
 114
 115 #include <gst/rtp/gstrtpbuffer.h>
 116
 117 #include <gst/glib-compat-private.h>
 118
 119 #include "gstrtpsession.h"
 120 #include "rtpsession.h"
 121
 122 GST_DEBUG_CATEGORY_STATIC (gst_rtp_session_debug);
 123 #define GST_CAT_DEFAULT gst_rtp_session_debug
 124
 125 GType
 126 gst_rtp_ntp_time_source_get_type (void)
 127 {
 128   static GType type = 0;
 129   static const GEnumValue values[] = {
 130     {GST_RTP_NTP_TIME_SOURCE_NTP, "NTP time based on realtime clock", "ntp"},
 131     {GST_RTP_NTP_TIME_SOURCE_UNIX, "UNIX time based on realtime clock", "unix"},
 132     {GST_RTP_NTP_TIME_SOURCE_RUNNING_TIME,
 133           "Running time based on pipeline clock",
 134         "running-time"},
 135     {GST_RTP_NTP_TIME_SOURCE_CLOCK_TIME, "Pipeline clock time", "clock-time"},
 136     {0, NULL, NULL},
 137   };
 138
 139   if (!type) {
 140     type = g_enum_register_static ("GstRtpNtpTimeSource", values);
 141   }
 142   return type;
 143 }
 144
 145 /* sink pads */
 146 static GstStaticPadTemplate rtpsession_recv_rtp_sink_template =
 147 GST_STATIC_PAD_TEMPLATE ("recv_rtp_sink",
 148     GST_PAD_SINK,
 149     GST_PAD_REQUEST,
 150     GST_STATIC_CAPS ("application/x-rtp")
 151     );
 152
 153 static GstStaticPadTemplate rtpsession_recv_rtcp_sink_template =
 154 GST_STATIC_PAD_TEMPLATE ("recv_rtcp_sink",
 155     GST_PAD_SINK,
 156     GST_PAD_REQUEST,
 157     GST_STATIC_CAPS ("application/x-rtcp")
 158     );
 159
 160 static GstStaticPadTemplate rtpsession_send_rtp_sink_template =
 161 GST_STATIC_PAD_TEMPLATE ("send_rtp_sink",
 162     GST_PAD_SINK,
 163     GST_PAD_REQUEST,
 164     GST_STATIC_CAPS ("application/x-rtp")
 165     );
 166
 167 /* src pads */
 168 static GstStaticPadTemplate rtpsession_recv_rtp_src_template =
 169 GST_STATIC_PAD_TEMPLATE ("recv_rtp_src",
 170     GST_PAD_SRC,
 171     GST_PAD_SOMETIMES,
 172     GST_STATIC_CAPS ("application/x-rtp")
 173     );
 174
 175 static GstStaticPadTemplate rtpsession_sync_src_template =
 176 GST_STATIC_PAD_TEMPLATE ("sync_src",
 177     GST_PAD_SRC,
 178     GST_PAD_SOMETIMES,
 179     GST_STATIC_CAPS ("application/x-rtcp")
 180     );
 181
 182 static GstStaticPadTemplate rtpsession_send_rtp_src_template =
 183 GST_STATIC_PAD_TEMPLATE ("send_rtp_src",
 184     GST_PAD_SRC,
 185     GST_PAD_SOMETIMES,
 186     GST_STATIC_CAPS ("application/x-rtp")
 187     );
 188
 189 static GstStaticPadTemplate rtpsession_send_rtcp_src_template =
 190 GST_STATIC_PAD_TEMPLATE ("send_rtcp_src",
 191     GST_PAD_SRC,
 192     GST_PAD_REQUEST,
 193     GST_STATIC_CAPS ("application/x-rtcp")
 194     );
 195
 196 /* signals and args */
 197 enum
 198 {
 199   SIGNAL_REQUEST_PT_MAP,
 200   SIGNAL_CLEAR_PT_MAP,
 201
 202   SIGNAL_ON_NEW_SSRC,
 203   SIGNAL_ON_SSRC_COLLISION,
 204   SIGNAL_ON_SSRC_VALIDATED,
 205   SIGNAL_ON_SSRC_ACTIVE,
 206   SIGNAL_ON_SSRC_SDES,
 207   SIGNAL_ON_BYE_SSRC,
 208   SIGNAL_ON_BYE_TIMEOUT,
 209   SIGNAL_ON_TIMEOUT,
 210   SIGNAL_ON_SENDER_TIMEOUT,
 211   SIGNAL_ON_NEW_SENDER_SSRC,
 212   SIGNAL_ON_SENDER_SSRC_ACTIVE,
 213   LAST_SIGNAL
 214 };
 215
 216 #define DEFAULT_BANDWIDTH            0
 217 #define DEFAULT_RTCP_FRACTION        RTP_STATS_RTCP_FRACTION
 218 #define DEFAULT_RTCP_RR_BANDWIDTH    -1
 219 #define DEFAULT_RTCP_RS_BANDWIDTH    -1
 220 #define DEFAULT_SDES                 NULL
 221 #define DEFAULT_NUM_SOURCES          0
 222 #define DEFAULT_NUM_ACTIVE_SOURCES   0
 223 #define DEFAULT_USE_PIPELINE_CLOCK   FALSE
 224 #define DEFAULT_RTCP_MIN_INTERVAL    (RTP_STATS_MIN_INTERVAL * GST_SECOND)
 225 #define DEFAULT_PROBATION            RTP_DEFAULT_PROBATION
 226 #define DEFAULT_MAX_DROPOUT_TIME     60000
 227 #define DEFAULT_MAX_MISORDER_TIME    2000
 228 #define DEFAULT_RTP_PROFILE          GST_RTP_PROFILE_AVP
 229 #define DEFAULT_NTP_TIME_SOURCE      GST_RTP_NTP_TIME_SOURCE_NTP
 230 #define DEFAULT_RTCP_SYNC_SEND_TIME  TRUE
 231
 232 enum
 233 {
 234   PROP_0,
 235   PROP_BANDWIDTH,
 236   PROP_RTCP_FRACTION,
 237   PROP_RTCP_RR_BANDWIDTH,
 238   PROP_RTCP_RS_BANDWIDTH,
 239   PROP_SDES,
 240   PROP_NUM_SOURCES,
 241   PROP_NUM_ACTIVE_SOURCES,
 242   PROP_INTERNAL_SESSION,
 243   PROP_USE_PIPELINE_CLOCK,
 244   PROP_RTCP_MIN_INTERVAL,
 245   PROP_PROBATION,
 246   PROP_MAX_DROPOUT_TIME,
 247   PROP_MAX_MISORDER_TIME,
 248   PROP_STATS,
 249   PROP_RTP_PROFILE,
 250   PROP_NTP_TIME_SOURCE,
 251   PROP_RTCP_SYNC_SEND_TIME
 252 };
 253
 254 #define GST_RTP_SESSION_GET_PRIVATE(obj)  \
 255            (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTP_SESSION, GstRtpSessionPrivate))
 256
 257 #define GST_RTP_SESSION_LOCK(sess)   g_mutex_lock (&(sess)->priv->lock)
 258 #define GST_RTP_SESSION_UNLOCK(sess) g_mutex_unlock (&(sess)->priv->lock)
 259
 260 #define GST_RTP_SESSION_WAIT(sess)   g_cond_wait (&(sess)->priv->cond, &(sess)->priv->lock)
 261 #define GST_RTP_SESSION_SIGNAL(sess) g_cond_signal (&(sess)->priv->cond)
 262
 263 struct _GstRtpSessionPrivate
 264 {
 265   GMutex lock;
 266   GCond cond;
 267   GstClock *sysclock;
 268
 269   RTPSession *session;
 270
 271   /* thread for sending out RTCP */
 272   GstClockID id;
 273   gboolean stop_thread;
 274   GThread *thread;
 275   gboolean thread_stopped;
 276   gboolean wait_send;
 277
 278   /* caps mapping */
 279   GHashTable *ptmap;
 280
 281   GstClockTime send_latency;
 282
 283   gboolean use_pipeline_clock;
 284   GstRtpNtpTimeSource ntp_time_source;
 285   gboolean rtcp_sync_send_time;
 286
 287   guint rtx_count;
 288 };
 289
 290 /* callbacks to handle actions from the session manager */
 291 static GstFlowReturn gst_rtp_session_process_rtp (RTPSession * sess,
 292     RTPSource * src, GstBuffer * buffer, gpointer user_data);
 293 static GstFlowReturn gst_rtp_session_send_rtp (RTPSession * sess,
 294     RTPSource * src, gpointer data, gpointer user_data);
 295 static GstFlowReturn gst_rtp_session_send_rtcp (RTPSession * sess,
 296     RTPSource * src, GstBuffer * buffer, gboolean eos, gpointer user_data);
 297 static GstFlowReturn gst_rtp_session_sync_rtcp (RTPSession * sess,
 298     GstBuffer * buffer, gpointer user_data);
 299 static gint gst_rtp_session_clock_rate (RTPSession * sess, guint8 payload,
 300     gpointer user_data);
 301 static void gst_rtp_session_reconsider (RTPSession * sess, gpointer user_data);
 302 static void gst_rtp_session_request_key_unit (RTPSession * sess,
 303     gboolean all_headers, gpointer user_data);
 304 static GstClockTime gst_rtp_session_request_time (RTPSession * session,
 305     gpointer user_data);
 306 static void gst_rtp_session_notify_nack (RTPSession * sess,
 307     guint16 seqnum, guint16 blp, guint32 ssrc, gpointer user_data);
 308 static void gst_rtp_session_reconfigure (RTPSession * sess, gpointer user_data);
 309
 310 static RTPSessionCallbacks callbacks = {
 311   gst_rtp_session_process_rtp,
 312   gst_rtp_session_send_rtp,
 313   gst_rtp_session_sync_rtcp,
 314   gst_rtp_session_send_rtcp,
 315   gst_rtp_session_clock_rate,
 316   gst_rtp_session_reconsider,
 317   gst_rtp_session_request_key_unit,
 318   gst_rtp_session_request_time,
 319   gst_rtp_session_notify_nack,
 320   gst_rtp_session_reconfigure
 321 };
 322
 323 /* GObject vmethods */
 324 static void gst_rtp_session_finalize (GObject * object);
 325 static void gst_rtp_session_set_property (GObject * object, guint prop_id,
 326     const GValue * value, GParamSpec * pspec);
 327 static void gst_rtp_session_get_property (GObject * object, guint prop_id,
 328     GValue * value, GParamSpec * pspec);
 329
 330 /* GstElement vmethods */
 331 static GstStateChangeReturn gst_rtp_session_change_state (GstElement * element,
 332     GstStateChange transition);
 333 static GstPad *gst_rtp_session_request_new_pad (GstElement * element,
 334     GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
 335 static void gst_rtp_session_release_pad (GstElement * element, GstPad * pad);
 336
 337 static gboolean gst_rtp_session_sink_setcaps (GstPad * pad,
 338     GstRtpSession * rtpsession, GstCaps * caps);
 339 static gboolean gst_rtp_session_setcaps_send_rtp (GstPad * pad,
 340     GstRtpSession * rtpsession, GstCaps * caps);
 341
 342 static void gst_rtp_session_clear_pt_map (GstRtpSession * rtpsession);
 343
 344 static GstStructure *gst_rtp_session_create_stats (GstRtpSession * rtpsession);
 345
 346 static guint gst_rtp_session_signals[LAST_SIGNAL] = { 0 };
 347
 348 static void
 349 on_new_ssrc (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 350 {
 351   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_NEW_SSRC], 0,
 352       src->ssrc);
 353 }
 354
 355 static void
 356 on_ssrc_collision (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 357 {
 358   GstPad *send_rtp_sink;
 359
 360   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_SSRC_COLLISION], 0,
 361       src->ssrc);
 362
 363   GST_RTP_SESSION_LOCK (sess);
 364   if ((send_rtp_sink = sess->send_rtp_sink))
 365     gst_object_ref (send_rtp_sink);
 366   GST_RTP_SESSION_UNLOCK (sess);
 367
 368   if (send_rtp_sink) {
 369     GstStructure *structure;
 370     GstEvent *event;
 371     RTPSource *internal_src;
 372     guint32 suggested_ssrc;
 373
 374     structure = gst_structure_new ("GstRTPCollision", "ssrc", G_TYPE_UINT,
 375         (guint) src->ssrc, NULL);
 376
 377     /* if there is no source using the suggested ssrc, most probably because
 378      * this ssrc has just collided, suggest upstream to use it */
 379     suggested_ssrc = rtp_session_suggest_ssrc (session, NULL);
 380     internal_src = rtp_session_get_source_by_ssrc (session, suggested_ssrc);
 381     if (!internal_src)
 382       gst_structure_set (structure, "suggested-ssrc", G_TYPE_UINT,
 383           (guint) suggested_ssrc, NULL);
 384     else
 385       g_object_unref (internal_src);
 386
 387     event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM, structure);
 388     gst_pad_push_event (send_rtp_sink, event);
 389     gst_object_unref (send_rtp_sink);
 390   }
 391 }
 392
 393 static void
 394 on_ssrc_validated (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 395 {
 396   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_SSRC_VALIDATED], 0,
 397       src->ssrc);
 398 }
 399
 400 static void
 401 on_ssrc_active (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 402 {
 403   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_SSRC_ACTIVE], 0,
 404       src->ssrc);
 405 }
 406
 407 static void
 408 on_ssrc_sdes (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 409 {
 410   GstStructure *s;
 411   GstMessage *m;
 412
 413   /* convert the new SDES info into a message */
 414   RTP_SESSION_LOCK (session);
 415   g_object_get (src, "sdes", &s, NULL);
 416   RTP_SESSION_UNLOCK (session);
 417
 418   m = gst_message_new_custom (GST_MESSAGE_ELEMENT, GST_OBJECT (sess), s);
 419   gst_element_post_message (GST_ELEMENT_CAST (sess), m);
 420
 421   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_SSRC_SDES], 0,
 422       src->ssrc);
 423 }
 424
 425 static void
 426 on_bye_ssrc (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 427 {
 428   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_BYE_SSRC], 0,
 429       src->ssrc);
 430 }
 431
 432 static void
 433 on_bye_timeout (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 434 {
 435   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_BYE_TIMEOUT], 0,
 436       src->ssrc);
 437 }
 438
 439 static void
 440 on_timeout (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 441 {
 442   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_TIMEOUT], 0,
 443       src->ssrc);
 444 }
 445
 446 static void
 447 on_sender_timeout (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 448 {
 449   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_SENDER_TIMEOUT], 0,
 450       src->ssrc);
 451 }
 452
 453 static void
 454 on_new_sender_ssrc (RTPSession * session, RTPSource * src, GstRtpSession * sess)
 455 {
 456   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_NEW_SENDER_SSRC], 0,
 457       src->ssrc);
 458 }
 459
 460 static void
 461 on_sender_ssrc_active (RTPSession * session, RTPSource * src,
 462     GstRtpSession * sess)
 463 {
 464   g_signal_emit (sess, gst_rtp_session_signals[SIGNAL_ON_SENDER_SSRC_ACTIVE], 0,
 465       src->ssrc);
 466 }
 467
 468 static void
 469 on_notify_stats (RTPSession * session, GParamSpec * spec,
 470     GstRtpSession * rtpsession)
 471 {
 472   g_object_notify (G_OBJECT (rtpsession), "stats");
 473 }
 474
 475 #define gst_rtp_session_parent_class parent_class
 476 G_DEFINE_TYPE (GstRtpSession, gst_rtp_session, GST_TYPE_ELEMENT);
 477
 478 static void
 479 gst_rtp_session_class_init (GstRtpSessionClass * klass)
 480 {
 481   GObjectClass *gobject_class;
 482   GstElementClass *gstelement_class;
 483
 484   gobject_class = (GObjectClass *) klass;
 485   gstelement_class = (GstElementClass *) klass;
 486
 487   g_type_class_add_private (klass, sizeof (GstRtpSessionPrivate));
 488
 489   gobject_class->finalize = gst_rtp_session_finalize;
 490   gobject_class->set_property = gst_rtp_session_set_property;
 491   gobject_class->get_property = gst_rtp_session_get_property;
 492
 493   /**
 494    * GstRtpSession::request-pt-map:
 495    * @sess: the object which received the signal
 496    * @pt: the pt
 497    *
 498    * Request the payload type as #GstCaps for @pt.
 499    */
 500   gst_rtp_session_signals[SIGNAL_REQUEST_PT_MAP] =
 501       g_signal_new ("request-pt-map", G_TYPE_FROM_CLASS (klass),
 502       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass, request_pt_map),
 503       NULL, NULL, g_cclosure_marshal_generic, GST_TYPE_CAPS, 1, G_TYPE_UINT);
 504   /**
 505    * GstRtpSession::clear-pt-map:
 506    * @sess: the object which received the signal
 507    *
 508    * Clear the cached pt-maps requested with #GstRtpSession::request-pt-map.
 509    */
 510   gst_rtp_session_signals[SIGNAL_CLEAR_PT_MAP] =
 511       g_signal_new ("clear-pt-map", G_TYPE_FROM_CLASS (klass),
 512       G_SIGNAL_ACTION, G_STRUCT_OFFSET (GstRtpSessionClass, clear_pt_map),
 513       NULL, NULL, g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0, G_TYPE_NONE);
 514
 515   /**
 516    * GstRtpSession::on-new-ssrc:
 517    * @sess: the object which received the signal
 518    * @ssrc: the SSRC
 519    *
 520    * Notify of a new SSRC that entered @session.
 521    */
 522   gst_rtp_session_signals[SIGNAL_ON_NEW_SSRC] =
 523       g_signal_new ("on-new-ssrc", G_TYPE_FROM_CLASS (klass),
 524       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass, on_new_ssrc),
 525       NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
 526   /**
 527    * GstRtpSession::on-ssrc_collision:
 528    * @sess: the object which received the signal
 529    * @ssrc: the SSRC
 530    *
 531    * Notify when we have an SSRC collision
 532    */
 533   gst_rtp_session_signals[SIGNAL_ON_SSRC_COLLISION] =
 534       g_signal_new ("on-ssrc-collision", G_TYPE_FROM_CLASS (klass),
 535       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass,
 536           on_ssrc_collision), NULL, NULL, g_cclosure_marshal_VOID__UINT,
 537       G_TYPE_NONE, 1, G_TYPE_UINT);
 538   /**
 539    * GstRtpSession::on-ssrc_validated:
 540    * @sess: the object which received the signal
 541    * @ssrc: the SSRC
 542    *
 543    * Notify of a new SSRC that became validated.
 544    */
 545   gst_rtp_session_signals[SIGNAL_ON_SSRC_VALIDATED] =
 546       g_signal_new ("on-ssrc-validated", G_TYPE_FROM_CLASS (klass),
 547       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass,
 548           on_ssrc_validated), NULL, NULL, g_cclosure_marshal_VOID__UINT,
 549       G_TYPE_NONE, 1, G_TYPE_UINT);
 550   /**
 551    * GstRtpSession::on-ssrc-active:
 552    * @sess: the object which received the signal
 553    * @ssrc: the SSRC
 554    *
 555    * Notify of a SSRC that is active, i.e., sending RTCP.
 556    */
 557   gst_rtp_session_signals[SIGNAL_ON_SSRC_ACTIVE] =
 558       g_signal_new ("on-ssrc-active", G_TYPE_FROM_CLASS (klass),
 559       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass,
 560           on_ssrc_active), NULL, NULL, g_cclosure_marshal_VOID__UINT,
 561       G_TYPE_NONE, 1, G_TYPE_UINT);
 562   /**
 563    * GstRtpSession::on-ssrc-sdes:
 564    * @session: the object which received the signal
 565    * @src: the SSRC
 566    *
 567    * Notify that a new SDES was received for SSRC.
 568    */
 569   gst_rtp_session_signals[SIGNAL_ON_SSRC_SDES] =
 570       g_signal_new ("on-ssrc-sdes", G_TYPE_FROM_CLASS (klass),
 571       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass, on_ssrc_sdes),
 572       NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
 573
 574   /**
 575    * GstRtpSession::on-bye-ssrc:
 576    * @sess: the object which received the signal
 577    * @ssrc: the SSRC
 578    *
 579    * Notify of an SSRC that became inactive because of a BYE packet.
 580    */
 581   gst_rtp_session_signals[SIGNAL_ON_BYE_SSRC] =
 582       g_signal_new ("on-bye-ssrc", G_TYPE_FROM_CLASS (klass),
 583       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass, on_bye_ssrc),
 584       NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
 585   /**
 586    * GstRtpSession::on-bye-timeout:
 587    * @sess: the object which received the signal
 588    * @ssrc: the SSRC
 589    *
 590    * Notify of an SSRC that has timed out because of BYE
 591    */
 592   gst_rtp_session_signals[SIGNAL_ON_BYE_TIMEOUT] =
 593       g_signal_new ("on-bye-timeout", G_TYPE_FROM_CLASS (klass),
 594       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass, on_bye_timeout),
 595       NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
 596   /**
 597    * GstRtpSession::on-timeout:
 598    * @sess: the object which received the signal
 599    * @ssrc: the SSRC
 600    *
 601    * Notify of an SSRC that has timed out
 602    */
 603   gst_rtp_session_signals[SIGNAL_ON_TIMEOUT] =
 604       g_signal_new ("on-timeout", G_TYPE_FROM_CLASS (klass),
 605       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass, on_timeout),
 606       NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
 607   /**
 608    * GstRtpSession::on-sender-timeout:
 609    * @sess: the object which received the signal
 610    * @ssrc: the SSRC
 611    *
 612    * Notify of a sender SSRC that has timed out and became a receiver
 613    */
 614   gst_rtp_session_signals[SIGNAL_ON_SENDER_TIMEOUT] =
 615       g_signal_new ("on-sender-timeout", G_TYPE_FROM_CLASS (klass),
 616       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass,
 617           on_sender_timeout), NULL, NULL, g_cclosure_marshal_VOID__UINT,
 618       G_TYPE_NONE, 1, G_TYPE_UINT);
 619
 620   /**
 621    * GstRtpSession::on-new-sender-ssrc:
 622    * @sess: the object which received the signal
 623    * @ssrc: the sender SSRC
 624    *
 625    * Since: 1.8
 626    *
 627    * Notify of a new sender SSRC that entered @session.
 628    */
 629   gst_rtp_session_signals[SIGNAL_ON_NEW_SENDER_SSRC] =
 630       g_signal_new ("on-new-sender-ssrc", G_TYPE_FROM_CLASS (klass),
 631       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass, on_new_ssrc),
 632       NULL, NULL, g_cclosure_marshal_VOID__UINT, G_TYPE_NONE, 1, G_TYPE_UINT);
 633
 634   /**
 635    * GstRtpSession::on-sender-ssrc-active:
 636    * @sess: the object which received the signal
 637    * @ssrc: the sender SSRC
 638    *
 639    * Since: 1.8
 640    *
 641    * Notify of a sender SSRC that is active, i.e., sending RTCP.
 642    */
 643   gst_rtp_session_signals[SIGNAL_ON_SENDER_SSRC_ACTIVE] =
 644       g_signal_new ("on-sender-ssrc-active", G_TYPE_FROM_CLASS (klass),
 645       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstRtpSessionClass,
 646           on_ssrc_active), NULL, NULL, g_cclosure_marshal_VOID__UINT,
 647       G_TYPE_NONE, 1, G_TYPE_UINT);
 648
 649   g_object_class_install_property (gobject_class, PROP_BANDWIDTH,
 650       g_param_spec_double ("bandwidth", "Bandwidth",
 651           "The bandwidth of the session in bytes per second (0 for auto-discover)",
 652           0.0, G_MAXDOUBLE, DEFAULT_BANDWIDTH,
 653           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 654
 655   g_object_class_install_property (gobject_class, PROP_RTCP_FRACTION,
 656       g_param_spec_double ("rtcp-fraction", "RTCP Fraction",
 657           "The RTCP bandwidth of the session in bytes per second "
 658           "(or as a real fraction of the RTP bandwidth if < 1.0)",
 659           0.0, G_MAXDOUBLE, DEFAULT_RTCP_FRACTION,
 660           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 661
 662   g_object_class_install_property (gobject_class, PROP_RTCP_RR_BANDWIDTH,
 663       g_param_spec_int ("rtcp-rr-bandwidth", "RTCP RR bandwidth",
 664           "The RTCP bandwidth used for receivers in bytes per second (-1 = default)",
 665           -1, G_MAXINT, DEFAULT_RTCP_RR_BANDWIDTH,
 666           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 667
 668   g_object_class_install_property (gobject_class, PROP_RTCP_RS_BANDWIDTH,
 669       g_param_spec_int ("rtcp-rs-bandwidth", "RTCP RS bandwidth",
 670           "The RTCP bandwidth used for senders in bytes per second (-1 = default)",
 671           -1, G_MAXINT, DEFAULT_RTCP_RS_BANDWIDTH,
 672           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 673
 674   g_object_class_install_property (gobject_class, PROP_SDES,
 675       g_param_spec_boxed ("sdes", "SDES",
 676           "The SDES items of this session",
 677           GST_TYPE_STRUCTURE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 678
 679   g_object_class_install_property (gobject_class, PROP_NUM_SOURCES,
 680       g_param_spec_uint ("num-sources", "Num Sources",
 681           "The number of sources in the session", 0, G_MAXUINT,
 682           DEFAULT_NUM_SOURCES, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 683
 684   g_object_class_install_property (gobject_class, PROP_NUM_ACTIVE_SOURCES,
 685       g_param_spec_uint ("num-active-sources", "Num Active Sources",
 686           "The number of active sources in the session", 0, G_MAXUINT,
 687           DEFAULT_NUM_ACTIVE_SOURCES,
 688           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 689
 690   g_object_class_install_property (gobject_class, PROP_INTERNAL_SESSION,
 691       g_param_spec_object ("internal-session", "Internal Session",
 692           "The internal RTPSession object", RTP_TYPE_SESSION,
 693           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 694
 695   g_object_class_install_property (gobject_class, PROP_USE_PIPELINE_CLOCK,
 696       g_param_spec_boolean ("use-pipeline-clock", "Use pipeline clock",
 697           "Use the pipeline running-time to set the NTP time in the RTCP SR messages "
 698           "(DEPRECATED: Use ntp-time-source property)",
 699           DEFAULT_USE_PIPELINE_CLOCK,
 700           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_DEPRECATED));
 701
 702   g_object_class_install_property (gobject_class, PROP_RTCP_MIN_INTERVAL,
 703       g_param_spec_uint64 ("rtcp-min-interval", "Minimum RTCP interval",
 704           "Minimum interval between Regular RTCP packet (in ns)",
 705           0, G_MAXUINT64, DEFAULT_RTCP_MIN_INTERVAL,
 706           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 707
 708   g_object_class_install_property (gobject_class, PROP_PROBATION,
 709       g_param_spec_uint ("probation", "Number of probations",
 710           "Consecutive packet sequence numbers to accept the source",
 711           0, G_MAXUINT, DEFAULT_PROBATION,
 712           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 713
 714   g_object_class_install_property (gobject_class, PROP_MAX_DROPOUT_TIME,
 715       g_param_spec_uint ("max-dropout-time", "Max dropout time",
 716           "The maximum time (milliseconds) of missing packets tolerated.",
 717           0, G_MAXUINT, DEFAULT_MAX_DROPOUT_TIME,
 718           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 719
 720   g_object_class_install_property (gobject_class, PROP_MAX_MISORDER_TIME,
 721       g_param_spec_uint ("max-misorder-time", "Max misorder time",
 722           "The maximum time (milliseconds) of misordered packets tolerated.",
 723           0, G_MAXUINT, DEFAULT_MAX_MISORDER_TIME,
 724           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 725
 726   /**
 727    * GstRtpSession::stats:
 728    *
 729    * Various session statistics. This property returns a GstStructure
 730    * with name application/x-rtp-session-stats with the following fields:
 731    *
 732    *  "rtx-count"       G_TYPE_UINT   The number of retransmission events
 733    *      received from downstream (in receiver mode)
 734    *  "rtx-drop-count"  G_TYPE_UINT   The number of retransmission events
 735    *      dropped (due to bandwidth constraints)
 736    *  "sent-nack-count" G_TYPE_UINT   Number of NACKs sent
 737    *  "recv-nack-count" G_TYPE_UINT   Number of NACKs received
 738    *  "source-stats"    G_TYPE_BOXED  GValueArray of #RTPSource::stats for all
 739    *      RTP sources (Since 1.8)
 740    *
 741    * Since: 1.4
 742    */
 743   g_object_class_install_property (gobject_class, PROP_STATS,
 744       g_param_spec_boxed ("stats", "Statistics",
 745           "Various statistics", GST_TYPE_STRUCTURE,
 746           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 747
 748   g_object_class_install_property (gobject_class, PROP_RTP_PROFILE,
 749       g_param_spec_enum ("rtp-profile", "RTP Profile",
 750           "RTP profile to use", GST_TYPE_RTP_PROFILE, DEFAULT_RTP_PROFILE,
 751           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 752
 753   g_object_class_install_property (gobject_class, PROP_NTP_TIME_SOURCE,
 754       g_param_spec_enum ("ntp-time-source", "NTP Time Source",
 755           "NTP time source for RTCP packets",
 756           gst_rtp_ntp_time_source_get_type (), DEFAULT_NTP_TIME_SOURCE,
 757           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 758
 759   g_object_class_install_property (gobject_class, PROP_RTCP_SYNC_SEND_TIME,
 760       g_param_spec_boolean ("rtcp-sync-send-time", "RTCP Sync Send Time",
 761           "Use send time or capture time for RTCP sync "
 762           "(TRUE = send time, FALSE = capture time)",
 763           DEFAULT_RTCP_SYNC_SEND_TIME,
 764           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 765
 766   gstelement_class->change_state =
 767       GST_DEBUG_FUNCPTR (gst_rtp_session_change_state);
 768   gstelement_class->request_new_pad =
 769       GST_DEBUG_FUNCPTR (gst_rtp_session_request_new_pad);
 770   gstelement_class->release_pad =
 771       GST_DEBUG_FUNCPTR (gst_rtp_session_release_pad);
 772
 773   klass->clear_pt_map = GST_DEBUG_FUNCPTR (gst_rtp_session_clear_pt_map);
 774
 775   /* sink pads */
 776   gst_element_class_add_static_pad_template (gstelement_class,
 777       &rtpsession_recv_rtp_sink_template);
 778   gst_element_class_add_static_pad_template (gstelement_class,
 779       &rtpsession_recv_rtcp_sink_template);
 780   gst_element_class_add_static_pad_template (gstelement_class,
 781       &rtpsession_send_rtp_sink_template);
 782
 783   /* src pads */
 784   gst_element_class_add_static_pad_template (gstelement_class,
 785       &rtpsession_recv_rtp_src_template);
 786   gst_element_class_add_static_pad_template (gstelement_class,
 787       &rtpsession_sync_src_template);
 788   gst_element_class_add_static_pad_template (gstelement_class,
 789       &rtpsession_send_rtp_src_template);
 790   gst_element_class_add_static_pad_template (gstelement_class,
 791       &rtpsession_send_rtcp_src_template);
 792
 793   gst_element_class_set_static_metadata (gstelement_class, "RTP Session",
 794       "Filter/Network/RTP",
 795       "Implement an RTP session", "Wim Taymans <wim.taymans@gmail.com>");
 796
 797   GST_DEBUG_CATEGORY_INIT (gst_rtp_session_debug,
 798       "rtpsession", 0, "RTP Session");
 799 }
 800
 801 static void
 802 gst_rtp_session_init (GstRtpSession * rtpsession)
 803 {
 804   rtpsession->priv = GST_RTP_SESSION_GET_PRIVATE (rtpsession);
 805   g_mutex_init (&rtpsession->priv->lock);
 806   g_cond_init (&rtpsession->priv->cond);
 807   rtpsession->priv->sysclock = gst_system_clock_obtain ();
 808   rtpsession->priv->session = rtp_session_new ();
 809   rtpsession->priv->use_pipeline_clock = DEFAULT_USE_PIPELINE_CLOCK;
 810   rtpsession->priv->rtcp_sync_send_time = DEFAULT_RTCP_SYNC_SEND_TIME;
 811
 812   /* configure callbacks */
 813   rtp_session_set_callbacks (rtpsession->priv->session, &callbacks, rtpsession);
 814   /* configure signals */
 815   g_signal_connect (rtpsession->priv->session, "on-new-ssrc",
 816       (GCallback) on_new_ssrc, rtpsession);
 817   g_signal_connect (rtpsession->priv->session, "on-ssrc-collision",
 818       (GCallback) on_ssrc_collision, rtpsession);
 819   g_signal_connect (rtpsession->priv->session, "on-ssrc-validated",
 820       (GCallback) on_ssrc_validated, rtpsession);
 821   g_signal_connect (rtpsession->priv->session, "on-ssrc-active",
 822       (GCallback) on_ssrc_active, rtpsession);
 823   g_signal_connect (rtpsession->priv->session, "on-ssrc-sdes",
 824       (GCallback) on_ssrc_sdes, rtpsession);
 825   g_signal_connect (rtpsession->priv->session, "on-bye-ssrc",
 826       (GCallback) on_bye_ssrc, rtpsession);
 827   g_signal_connect (rtpsession->priv->session, "on-bye-timeout",
 828       (GCallback) on_bye_timeout, rtpsession);
 829   g_signal_connect (rtpsession->priv->session, "on-timeout",
 830       (GCallback) on_timeout, rtpsession);
 831   g_signal_connect (rtpsession->priv->session, "on-sender-timeout",
 832       (GCallback) on_sender_timeout, rtpsession);
 833   g_signal_connect (rtpsession->priv->session, "on-new-sender-ssrc",
 834       (GCallback) on_new_sender_ssrc, rtpsession);
 835   g_signal_connect (rtpsession->priv->session, "on-sender-ssrc-active",
 836       (GCallback) on_sender_ssrc_active, rtpsession);
 837   g_signal_connect (rtpsession->priv->session, "notify::stats",
 838       (GCallback) on_notify_stats, rtpsession);
 839   rtpsession->priv->ptmap = g_hash_table_new_full (NULL, NULL, NULL,
 840       (GDestroyNotify) gst_caps_unref);
 841
 842   gst_segment_init (&rtpsession->recv_rtp_seg, GST_FORMAT_UNDEFINED);
 843   gst_segment_init (&rtpsession->send_rtp_seg, GST_FORMAT_UNDEFINED);
 844
 845   rtpsession->priv->thread_stopped = TRUE;
 846
 847   rtpsession->priv->rtx_count = 0;
 848
 849   rtpsession->priv->ntp_time_source = DEFAULT_NTP_TIME_SOURCE;
 850 }
 851
 852 static void
 853 gst_rtp_session_finalize (GObject * object)
 854 {
 855   GstRtpSession *rtpsession;
 856
 857   rtpsession = GST_RTP_SESSION (object);
 858
 859   g_hash_table_destroy (rtpsession->priv->ptmap);
 860   g_mutex_clear (&rtpsession->priv->lock);
 861   g_cond_clear (&rtpsession->priv->cond);
 862   g_object_unref (rtpsession->priv->sysclock);
 863   g_object_unref (rtpsession->priv->session);
 864
 865   G_OBJECT_CLASS (parent_class)->finalize (object);
 866 }
 867
 868 static void
 869 gst_rtp_session_set_property (GObject * object, guint prop_id,
 870     const GValue * value, GParamSpec * pspec)
 871 {
 872   GstRtpSession *rtpsession;
 873   GstRtpSessionPrivate *priv;
 874
 875   rtpsession = GST_RTP_SESSION (object);
 876   priv = rtpsession->priv;
 877
 878   switch (prop_id) {
 879     case PROP_BANDWIDTH:
 880       g_object_set_property (G_OBJECT (priv->session), "bandwidth", value);
 881       break;
 882     case PROP_RTCP_FRACTION:
 883       g_object_set_property (G_OBJECT (priv->session), "rtcp-fraction", value);
 884       break;
 885     case PROP_RTCP_RR_BANDWIDTH:
 886       g_object_set_property (G_OBJECT (priv->session), "rtcp-rr-bandwidth",
 887           value);
 888       break;
 889     case PROP_RTCP_RS_BANDWIDTH:
 890       g_object_set_property (G_OBJECT (priv->session), "rtcp-rs-bandwidth",
 891           value);
 892       break;
 893     case PROP_SDES:
 894       rtp_session_set_sdes_struct (priv->session, g_value_get_boxed (value));
 895       break;
 896     case PROP_USE_PIPELINE_CLOCK:
 897       priv->use_pipeline_clock = g_value_get_boolean (value);
 898       break;
 899     case PROP_RTCP_MIN_INTERVAL:
 900       g_object_set_property (G_OBJECT (priv->session), "rtcp-min-interval",
 901           value);
 902       break;
 903     case PROP_PROBATION:
 904       g_object_set_property (G_OBJECT (priv->session), "probation", value);
 905       break;
 906     case PROP_MAX_DROPOUT_TIME:
 907       g_object_set_property (G_OBJECT (priv->session), "max-dropout-time",
 908           value);
 909       break;
 910     case PROP_MAX_MISORDER_TIME:
 911       g_object_set_property (G_OBJECT (priv->session), "max-misorder-time",
 912           value);
 913       break;
 914     case PROP_RTP_PROFILE:
 915       g_object_set_property (G_OBJECT (priv->session), "rtp-profile", value);
 916       break;
 917     case PROP_NTP_TIME_SOURCE:
 918       priv->ntp_time_source = g_value_get_enum (value);
 919       break;
 920     case PROP_RTCP_SYNC_SEND_TIME:
 921       priv->rtcp_sync_send_time = g_value_get_boolean (value);
 922       break;
 923     default:
 924       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 925       break;
 926   }
 927 }
 928
 929 static void
 930 gst_rtp_session_get_property (GObject * object, guint prop_id,
 931     GValue * value, GParamSpec * pspec)
 932 {
 933   GstRtpSession *rtpsession;
 934   GstRtpSessionPrivate *priv;
 935
 936   rtpsession = GST_RTP_SESSION (object);
 937   priv = rtpsession->priv;
 938
 939   switch (prop_id) {
 940     case PROP_BANDWIDTH:
 941       g_object_get_property (G_OBJECT (priv->session), "bandwidth", value);
 942       break;
 943     case PROP_RTCP_FRACTION:
 944       g_object_get_property (G_OBJECT (priv->session), "rtcp-fraction", value);
 945       break;
 946     case PROP_RTCP_RR_BANDWIDTH:
 947       g_object_get_property (G_OBJECT (priv->session), "rtcp-rr-bandwidth",
 948           value);
 949       break;
 950     case PROP_RTCP_RS_BANDWIDTH:
 951       g_object_get_property (G_OBJECT (priv->session), "rtcp-rs-bandwidth",
 952           value);
 953       break;
 954     case PROP_SDES:
 955       g_value_take_boxed (value, rtp_session_get_sdes_struct (priv->session));
 956       break;
 957     case PROP_NUM_SOURCES:
 958       g_value_set_uint (value, rtp_session_get_num_sources (priv->session));
 959       break;
 960     case PROP_NUM_ACTIVE_SOURCES:
 961       g_value_set_uint (value,
 962           rtp_session_get_num_active_sources (priv->session));
 963       break;
 964     case PROP_INTERNAL_SESSION:
 965       g_value_set_object (value, priv->session);
 966       break;
 967     case PROP_USE_PIPELINE_CLOCK:
 968       g_value_set_boolean (value, priv->use_pipeline_clock);
 969       break;
 970     case PROP_RTCP_MIN_INTERVAL:
 971       g_object_get_property (G_OBJECT (priv->session), "rtcp-min-interval",
 972           value);
 973       break;
 974     case PROP_PROBATION:
 975       g_object_get_property (G_OBJECT (priv->session), "probation", value);
 976       break;
 977     case PROP_MAX_DROPOUT_TIME:
 978       g_object_get_property (G_OBJECT (priv->session), "max-dropout-time",
 979           value);
 980       break;
 981     case PROP_MAX_MISORDER_TIME:
 982       g_object_get_property (G_OBJECT (priv->session), "max-misorder-time",
 983           value);
 984       break;
 985     case PROP_STATS:
 986       g_value_take_boxed (value, gst_rtp_session_create_stats (rtpsession));
 987       break;
 988     case PROP_RTP_PROFILE:
 989       g_object_get_property (G_OBJECT (priv->session), "rtp-profile", value);
 990       break;
 991     case PROP_NTP_TIME_SOURCE:
 992       g_value_set_enum (value, priv->ntp_time_source);
 993       break;
 994     case PROP_RTCP_SYNC_SEND_TIME:
 995       g_value_set_boolean (value, priv->rtcp_sync_send_time);
 996       break;
 997     default:
 998       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 999       break;
1000   }
1001 }
1002
1003 static GstStructure *
1004 gst_rtp_session_create_stats (GstRtpSession * rtpsession)
1005 {
1006   GstStructure *s;
1007
1008   g_object_get (rtpsession->priv->session, "stats", &s, NULL);
1009   gst_structure_set (s, "rtx-count", G_TYPE_UINT, rtpsession->priv->rtx_count,
1010       NULL);
1011
1012   return s;
1013 }
1014
1015 static void
1016 get_current_times (GstRtpSession * rtpsession, GstClockTime * running_time,
1017     guint64 * ntpnstime)
1018 {
1019   guint64 ntpns = -1;
1020   GstClock *clock;
1021   GstClockTime base_time, rt, clock_time;
1022
1023   GST_OBJECT_LOCK (rtpsession);
1024   if ((clock = GST_ELEMENT_CLOCK (rtpsession))) {
1025     base_time = GST_ELEMENT_CAST (rtpsession)->base_time;
1026     gst_object_ref (clock);
1027     GST_OBJECT_UNLOCK (rtpsession);
1028
1029     /* get current clock time and convert to running time */
1030     clock_time = gst_clock_get_time (clock);
1031     rt = clock_time - base_time;
1032
1033     if (rtpsession->priv->use_pipeline_clock) {
1034       ntpns = rt;
1035       /* add constant to convert from 1970 based time to 1900 based time */
1036       ntpns += (2208988800LL * GST_SECOND);
1037     } else {
1038       switch (rtpsession->priv->ntp_time_source) {
1039         case GST_RTP_NTP_TIME_SOURCE_NTP:
1040         case GST_RTP_NTP_TIME_SOURCE_UNIX:{
1041           GTimeVal current;
1042
1043           /* get current NTP time */
1044           g_get_current_time (&current);
1045           ntpns = GST_TIMEVAL_TO_TIME (current);
1046
1047           /* add constant to convert from 1970 based time to 1900 based time */
1048           if (rtpsession->priv->ntp_time_source == GST_RTP_NTP_TIME_SOURCE_NTP)
1049             ntpns += (2208988800LL * GST_SECOND);
1050           break;
1051         }
1052         case GST_RTP_NTP_TIME_SOURCE_RUNNING_TIME:
1053           ntpns = rt;
1054           break;
1055         case GST_RTP_NTP_TIME_SOURCE_CLOCK_TIME:
1056           ntpns = clock_time;
1057           break;
1058         default:
1059           ntpns = -1;
1060           g_assert_not_reached ();
1061           break;
1062       }
1063     }
1064
1065     gst_object_unref (clock);
1066   } else {
1067     GST_OBJECT_UNLOCK (rtpsession);
1068     rt = -1;
1069     ntpns = -1;
1070   }
1071   if (running_time)
1072     *running_time = rt;
1073   if (ntpnstime)
1074     *ntpnstime = ntpns;
1075 }
1076
1077 static void
1078 rtcp_thread (GstRtpSession * rtpsession)
1079 {
1080   GstClockID id;
1081   GstClockTime current_time;
1082   GstClockTime next_timeout;
1083   guint64 ntpnstime;
1084   GstClockTime running_time;
1085   RTPSession *session;
1086   GstClock *sysclock;
1087
1088   GST_DEBUG_OBJECT (rtpsession, "entering RTCP thread");
1089
1090   GST_RTP_SESSION_LOCK (rtpsession);
1091
1092   while (rtpsession->priv->wait_send) {
1093     GST_LOG_OBJECT (rtpsession, "waiting for getting started");
1094     GST_RTP_SESSION_WAIT (rtpsession);
1095     GST_LOG_OBJECT (rtpsession, "signaled...");
1096   }
1097
1098   sysclock = rtpsession->priv->sysclock;
1099   current_time = gst_clock_get_time (sysclock);
1100
1101   session = rtpsession->priv->session;
1102
1103   GST_DEBUG_OBJECT (rtpsession, "starting at %" GST_TIME_FORMAT,
1104       GST_TIME_ARGS (current_time));
1105   session->start_time = current_time;
1106
1107   while (!rtpsession->priv->stop_thread) {
1108     GstClockReturn res;
1109
1110     /* get initial estimate */
1111     next_timeout = rtp_session_next_timeout (session, current_time);
1112
1113     GST_DEBUG_OBJECT (rtpsession, "next check time %" GST_TIME_FORMAT,
1114         GST_TIME_ARGS (next_timeout));
1115
1116     /* leave if no more timeouts, the session ended */
1117     if (next_timeout == GST_CLOCK_TIME_NONE)
1118       break;
1119
1120     id = rtpsession->priv->id =
1121         gst_clock_new_single_shot_id (sysclock, next_timeout);
1122     GST_RTP_SESSION_UNLOCK (rtpsession);
1123
1124     res = gst_clock_id_wait (id, NULL);
1125
1126     GST_RTP_SESSION_LOCK (rtpsession);
1127     gst_clock_id_unref (id);
1128     rtpsession->priv->id = NULL;
1129
1130     if (rtpsession->priv->stop_thread)
1131       break;
1132
1133     /* update current time */
1134     current_time = gst_clock_get_time (sysclock);
1135
1136     /* get current NTP time */
1137     get_current_times (rtpsession, &running_time, &ntpnstime);
1138
1139     /* we get unlocked because we need to perform reconsideration, don't perform
1140      * the timeout but get a new reporting estimate. */
1141     GST_DEBUG_OBJECT (rtpsession, "unlocked %d, current %" GST_TIME_FORMAT,
1142         res, GST_TIME_ARGS (current_time));
1143
1144     /* perform actions, we ignore result. Release lock because it might push. */
1145     GST_RTP_SESSION_UNLOCK (rtpsession);
1146     rtp_session_on_timeout (session, current_time, ntpnstime, running_time);
1147     GST_RTP_SESSION_LOCK (rtpsession);
1148   }
1149   /* mark the thread as stopped now */
1150   rtpsession->priv->thread_stopped = TRUE;
1151   GST_RTP_SESSION_UNLOCK (rtpsession);
1152
1153   GST_DEBUG_OBJECT (rtpsession, "leaving RTCP thread");
1154 }
1155
1156 static gboolean
1157 start_rtcp_thread (GstRtpSession * rtpsession)
1158 {
1159   GError *error = NULL;
1160   gboolean res;
1161
1162   GST_DEBUG_OBJECT (rtpsession, "starting RTCP thread");
1163
1164   GST_RTP_SESSION_LOCK (rtpsession);
1165   rtpsession->priv->stop_thread = FALSE;
1166   if (rtpsession->priv->thread_stopped) {
1167     /* if the thread stopped, and we still have a handle to the thread, join it
1168      * now. We can safely join with the lock held, the thread will not take it
1169      * anymore. */
1170     if (rtpsession->priv->thread)
1171       g_thread_join (rtpsession->priv->thread);
1172     /* only create a new thread if the old one was stopped. Otherwise we can
1173      * just reuse the currently running one. */
1174     rtpsession->priv->thread = g_thread_try_new ("rtpsession-rtcp-thread",
1175         (GThreadFunc) rtcp_thread, rtpsession, &error);
1176     rtpsession->priv->thread_stopped = FALSE;
1177   }
1178   GST_RTP_SESSION_UNLOCK (rtpsession);
1179
1180   if (error != NULL) {
1181     res = FALSE;
1182     GST_DEBUG_OBJECT (rtpsession, "failed to start thread, %s", error->message);
1183     g_error_free (error);
1184   } else {
1185     res = TRUE;
1186   }
1187   return res;
1188 }
1189
1190 static void
1191 stop_rtcp_thread (GstRtpSession * rtpsession)
1192 {
1193   GST_DEBUG_OBJECT (rtpsession, "stopping RTCP thread");
1194
1195   GST_RTP_SESSION_LOCK (rtpsession);
1196   rtpsession->priv->stop_thread = TRUE;
1197   rtpsession->priv->wait_send = FALSE;
1198   GST_RTP_SESSION_SIGNAL (rtpsession);
1199   if (rtpsession->priv->id)
1200     gst_clock_id_unschedule (rtpsession->priv->id);
1201   GST_RTP_SESSION_UNLOCK (rtpsession);
1202 }
1203
1204 static void
1205 join_rtcp_thread (GstRtpSession * rtpsession)
1206 {
1207   GST_RTP_SESSION_LOCK (rtpsession);
1208   /* don't try to join when we have no thread */
1209   if (rtpsession->priv->thread != NULL) {
1210     GST_DEBUG_OBJECT (rtpsession, "joining RTCP thread");
1211     GST_RTP_SESSION_UNLOCK (rtpsession);
1212
1213     g_thread_join (rtpsession->priv->thread);
1214
1215     GST_RTP_SESSION_LOCK (rtpsession);
1216     /* after the join, take the lock and clear the thread structure. The caller
1217      * is supposed to not concurrently call start and join. */
1218     rtpsession->priv->thread = NULL;
1219   }
1220   GST_RTP_SESSION_UNLOCK (rtpsession);
1221 }
1222
1223 static GstStateChangeReturn
1224 gst_rtp_session_change_state (GstElement * element, GstStateChange transition)
1225 {
1226   GstStateChangeReturn res;
1227   GstRtpSession *rtpsession;
1228
1229   rtpsession = GST_RTP_SESSION (element);
1230
1231   switch (transition) {
1232     case GST_STATE_CHANGE_NULL_TO_READY:
1233       break;
1234     case GST_STATE_CHANGE_READY_TO_PAUSED:
1235       GST_RTP_SESSION_LOCK (rtpsession);
1236       if (rtpsession->send_rtp_src)
1237         rtpsession->priv->wait_send = TRUE;
1238       GST_RTP_SESSION_UNLOCK (rtpsession);
1239       break;
1240     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1241       break;
1242     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1243     case GST_STATE_CHANGE_PAUSED_TO_READY:
1244       /* no need to join yet, we might want to continue later. Also, the
1245        * dataflow could block downstream so that a join could just block
1246        * forever. */
1247       stop_rtcp_thread (rtpsession);
1248       break;
1249     default:
1250       break;
1251   }
1252
1253   res = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1254
1255   switch (transition) {
1256     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1257       if (!start_rtcp_thread (rtpsession))
1258         goto failed_thread;
1259       break;
1260     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1261       break;
1262     case GST_STATE_CHANGE_PAUSED_TO_READY:
1263       /* downstream is now releasing the dataflow and we can join. */
1264       join_rtcp_thread (rtpsession);
1265       break;
1266     case GST_STATE_CHANGE_READY_TO_NULL:
1267       break;
1268     default:
1269       break;
1270   }
1271   return res;
1272
1273   /* ERRORS */
1274 failed_thread:
1275   {
1276     return GST_STATE_CHANGE_FAILURE;
1277   }
1278 }
1279
1280 static gboolean
1281 return_true (gpointer key, gpointer value, gpointer user_data)
1282 {
1283   return TRUE;
1284 }
1285
1286 static void
1287 gst_rtp_session_clear_pt_map (GstRtpSession * rtpsession)
1288 {
1289   g_hash_table_foreach_remove (rtpsession->priv->ptmap, return_true, NULL);
1290 }
1291
1292 /* called when the session manager has an RTP packet or a list of packets
1293  * ready for further processing */
1294 static GstFlowReturn
1295 gst_rtp_session_process_rtp (RTPSession * sess, RTPSource * src,
1296     GstBuffer * buffer, gpointer user_data)
1297 {
1298   GstFlowReturn result;
1299   GstRtpSession *rtpsession;
1300   GstPad *rtp_src;
1301
1302   rtpsession = GST_RTP_SESSION (user_data);
1303
1304   GST_RTP_SESSION_LOCK (rtpsession);
1305   if ((rtp_src = rtpsession->recv_rtp_src))
1306     gst_object_ref (rtp_src);
1307   GST_RTP_SESSION_UNLOCK (rtpsession);
1308
1309   if (rtp_src) {
1310     GST_LOG_OBJECT (rtpsession, "pushing received RTP packet");
1311     result = gst_pad_push (rtp_src, buffer);
1312     gst_object_unref (rtp_src);
1313   } else {
1314     GST_DEBUG_OBJECT (rtpsession, "dropping received RTP packet");
1315     gst_buffer_unref (buffer);
1316     result = GST_FLOW_OK;
1317   }
1318   return result;
1319 }
1320
1321 /* called when the session manager has an RTP packet ready for further
1322  * sending */
1323 static GstFlowReturn
1324 gst_rtp_session_send_rtp (RTPSession * sess, RTPSource * src,
1325     gpointer data, gpointer user_data)
1326 {
1327   GstFlowReturn result;
1328   GstRtpSession *rtpsession;
1329   GstPad *rtp_src;
1330
1331   rtpsession = GST_RTP_SESSION (user_data);
1332
1333   GST_RTP_SESSION_LOCK (rtpsession);
1334   if ((rtp_src = rtpsession->send_rtp_src))
1335     gst_object_ref (rtp_src);
1336   if (rtpsession->priv->wait_send) {
1337     GST_LOG_OBJECT (rtpsession, "signal RTCP thread");
1338     rtpsession->priv->wait_send = FALSE;
1339     GST_RTP_SESSION_SIGNAL (rtpsession);
1340   }
1341   GST_RTP_SESSION_UNLOCK (rtpsession);
1342
1343   if (rtp_src) {
1344     if (GST_IS_BUFFER (data)) {
1345       GST_LOG_OBJECT (rtpsession, "sending RTP packet");
1346       result = gst_pad_push (rtp_src, GST_BUFFER_CAST (data));
1347     } else {
1348       GST_LOG_OBJECT (rtpsession, "sending RTP list");
1349       result = gst_pad_push_list (rtp_src, GST_BUFFER_LIST_CAST (data));
1350     }
1351     gst_object_unref (rtp_src);
1352   } else {
1353     gst_mini_object_unref (GST_MINI_OBJECT_CAST (data));
1354     result = GST_FLOW_OK;
1355   }
1356   return result;
1357 }
1358
1359 static void
1360 do_rtcp_events (GstRtpSession * rtpsession, GstPad * srcpad)
1361 {
1362   GstCaps *caps;
1363   GstSegment seg;
1364   GstEvent *event;
1365   gchar *stream_id;
1366   gboolean have_group_id;
1367   guint group_id;
1368
1369   stream_id =
1370       g_strdup_printf ("%08x%08x%08x%08x", g_random_int (), g_random_int (),
1371       g_random_int (), g_random_int ());
1372
1373   GST_RTP_SESSION_LOCK (rtpsession);
1374   if (rtpsession->recv_rtp_sink) {
1375     event =
1376         gst_pad_get_sticky_event (rtpsession->recv_rtp_sink,
1377         GST_EVENT_STREAM_START, 0);
1378     if (event) {
1379       if (gst_event_parse_group_id (event, &group_id))
1380         have_group_id = TRUE;
1381       else
1382         have_group_id = FALSE;
1383       gst_event_unref (event);
1384     } else {
1385       have_group_id = TRUE;
1386       group_id = gst_util_group_id_next ();
1387     }
1388   } else {
1389     have_group_id = TRUE;
1390     group_id = gst_util_group_id_next ();
1391   }
1392   GST_RTP_SESSION_UNLOCK (rtpsession);
1393
1394   event = gst_event_new_stream_start (stream_id);
1395   if (have_group_id)
1396     gst_event_set_group_id (event, group_id);
1397   gst_pad_push_event (srcpad, event);
1398   g_free (stream_id);
1399
1400   caps = gst_caps_new_empty_simple ("application/x-rtcp");
1401   gst_pad_set_caps (srcpad, caps);
1402   gst_caps_unref (caps);
1403
1404   gst_segment_init (&seg, GST_FORMAT_TIME);
1405   event = gst_event_new_segment (&seg);
1406   gst_pad_push_event (srcpad, event);
1407 }
1408
1409 /* called when the session manager has an RTCP packet ready for further
1410  * sending. The eos flag is set when an EOS event should be sent downstream as
1411  * well. */
1412 static GstFlowReturn
1413 gst_rtp_session_send_rtcp (RTPSession * sess, RTPSource * src,
1414     GstBuffer * buffer, gboolean eos, gpointer user_data)
1415 {
1416   GstFlowReturn result;
1417   GstRtpSession *rtpsession;
1418   GstPad *rtcp_src;
1419
1420   rtpsession = GST_RTP_SESSION (user_data);
1421
1422   GST_RTP_SESSION_LOCK (rtpsession);
1423   if (rtpsession->priv->stop_thread)
1424     goto stopping;
1425
1426   if ((rtcp_src = rtpsession->send_rtcp_src)) {
1427     gst_object_ref (rtcp_src);
1428     GST_RTP_SESSION_UNLOCK (rtpsession);
1429
1430     /* set rtcp caps on output pad */
1431     if (!gst_pad_has_current_caps (rtcp_src))
1432       do_rtcp_events (rtpsession, rtcp_src);
1433
1434     GST_LOG_OBJECT (rtpsession, "sending RTCP");
1435     result = gst_pad_push (rtcp_src, buffer);
1436
1437     /* we have to send EOS after this packet */
1438     if (eos) {
1439       GST_LOG_OBJECT (rtpsession, "sending EOS");
1440       gst_pad_push_event (rtcp_src, gst_event_new_eos ());
1441     }
1442     gst_object_unref (rtcp_src);
1443   } else {
1444     GST_RTP_SESSION_UNLOCK (rtpsession);
1445
1446     GST_DEBUG_OBJECT (rtpsession, "not sending RTCP, no output pad");
1447     gst_buffer_unref (buffer);
1448     result = GST_FLOW_OK;
1449   }
1450   return result;
1451
1452   /* ERRORS */
1453 stopping:
1454   {
1455     GST_DEBUG_OBJECT (rtpsession, "we are stopping");
1456     gst_buffer_unref (buffer);
1457     GST_RTP_SESSION_UNLOCK (rtpsession);
1458     return GST_FLOW_OK;
1459   }
1460 }
1461
1462 /* called when the session manager has an SR RTCP packet ready for handling
1463  * inter stream synchronisation */
1464 static GstFlowReturn
1465 gst_rtp_session_sync_rtcp (RTPSession * sess,
1466     GstBuffer * buffer, gpointer user_data)
1467 {
1468   GstFlowReturn result;
1469   GstRtpSession *rtpsession;
1470   GstPad *sync_src;
1471
1472   rtpsession = GST_RTP_SESSION (user_data);
1473
1474   GST_RTP_SESSION_LOCK (rtpsession);
1475   if (rtpsession->priv->stop_thread)
1476     goto stopping;
1477
1478   if ((sync_src = rtpsession->sync_src)) {
1479     gst_object_ref (sync_src);
1480     GST_RTP_SESSION_UNLOCK (rtpsession);
1481
1482     /* set rtcp caps on output pad, this happens
1483      * when we receive RTCP muxed with RTP according
1484      * to RFC5761. Otherwise we would have forwarded
1485      * the events from the recv_rtcp_sink pad already
1486      */
1487     if (!gst_pad_has_current_caps (sync_src))
1488       do_rtcp_events (rtpsession, sync_src);
1489
1490     GST_LOG_OBJECT (rtpsession, "sending Sync RTCP");
1491     result = gst_pad_push (sync_src, buffer);
1492     gst_object_unref (sync_src);
1493   } else {
1494     GST_RTP_SESSION_UNLOCK (rtpsession);
1495
1496     GST_DEBUG_OBJECT (rtpsession, "not sending Sync RTCP, no output pad");
1497     gst_buffer_unref (buffer);
1498     result = GST_FLOW_OK;
1499   }
1500   return result;
1501
1502   /* ERRORS */
1503 stopping:
1504   {
1505     GST_DEBUG_OBJECT (rtpsession, "we are stopping");
1506     gst_buffer_unref (buffer);
1507     GST_RTP_SESSION_UNLOCK (rtpsession);
1508     return GST_FLOW_OK;
1509   }
1510 }
1511
1512 static void
1513 gst_rtp_session_cache_caps (GstRtpSession * rtpsession, GstCaps * caps)
1514 {
1515   GstRtpSessionPrivate *priv;
1516   const GstStructure *s;
1517   gint payload;
1518
1519   priv = rtpsession->priv;
1520
1521   GST_DEBUG_OBJECT (rtpsession, "parsing caps");
1522
1523   s = gst_caps_get_structure (caps, 0);
1524   if (!gst_structure_get_int (s, "payload", &payload))
1525     return;
1526
1527   if (g_hash_table_lookup (priv->ptmap, GINT_TO_POINTER (payload)))
1528     return;
1529
1530   g_hash_table_insert (priv->ptmap, GINT_TO_POINTER (payload),
1531       gst_caps_ref (caps));
1532 }
1533
1534 static GstCaps *
1535 gst_rtp_session_get_caps_for_pt (GstRtpSession * rtpsession, guint payload)
1536 {
1537   GstCaps *caps = NULL;
1538   GValue args[2] = { {0}, {0} };
1539   GValue ret = { 0 };
1540
1541   GST_RTP_SESSION_LOCK (rtpsession);
1542   caps = g_hash_table_lookup (rtpsession->priv->ptmap,
1543       GINT_TO_POINTER (payload));
1544   if (caps) {
1545     gst_caps_ref (caps);
1546     goto done;
1547   }
1548
1549   /* not found in the cache, try to get it with a signal */
1550   g_value_init (&args[0], GST_TYPE_ELEMENT);
1551   g_value_set_object (&args[0], rtpsession);
1552   g_value_init (&args[1], G_TYPE_UINT);
1553   g_value_set_uint (&args[1], payload);
1554
1555   g_value_init (&ret, GST_TYPE_CAPS);
1556   g_value_set_boxed (&ret, NULL);
1557
1558   GST_RTP_SESSION_UNLOCK (rtpsession);
1559
1560   g_signal_emitv (args, gst_rtp_session_signals[SIGNAL_REQUEST_PT_MAP], 0,
1561       &ret);
1562
1563   GST_RTP_SESSION_LOCK (rtpsession);
1564
1565   g_value_unset (&args[0]);
1566   g_value_unset (&args[1]);
1567   caps = (GstCaps *) g_value_dup_boxed (&ret);
1568   g_value_unset (&ret);
1569   if (!caps)
1570     goto no_caps;
1571
1572   gst_rtp_session_cache_caps (rtpsession, caps);
1573
1574 done:
1575   GST_RTP_SESSION_UNLOCK (rtpsession);
1576
1577   return caps;
1578
1579 no_caps:
1580   {
1581     GST_DEBUG_OBJECT (rtpsession, "could not get caps");
1582     goto done;
1583   }
1584 }
1585
1586 /* called when the session manager needs the clock rate */
1587 static gint
1588 gst_rtp_session_clock_rate (RTPSession * sess, guint8 payload,
1589     gpointer user_data)
1590 {
1591   gint result = -1;
1592   GstRtpSession *rtpsession;
1593   GstCaps *caps;
1594   const GstStructure *s;
1595
1596   rtpsession = GST_RTP_SESSION_CAST (user_data);
1597
1598   caps = gst_rtp_session_get_caps_for_pt (rtpsession, payload);
1599
1600   if (!caps)
1601     goto done;
1602
1603   s = gst_caps_get_structure (caps, 0);
1604   if (!gst_structure_get_int (s, "clock-rate", &result))
1605     goto no_clock_rate;
1606
1607   gst_caps_unref (caps);
1608
1609   GST_DEBUG_OBJECT (rtpsession, "parsed clock-rate %d", result);
1610
1611 done:
1612
1613   return result;
1614
1615   /* ERRORS */
1616 no_clock_rate:
1617   {
1618     gst_caps_unref (caps);
1619     GST_DEBUG_OBJECT (rtpsession, "No clock-rate in caps!");
1620     goto done;
1621   }
1622 }
1623
1624 /* called when the session manager asks us to reconsider the timeout */
1625 static void
1626 gst_rtp_session_reconsider (RTPSession * sess, gpointer user_data)
1627 {
1628   GstRtpSession *rtpsession;
1629
1630   rtpsession = GST_RTP_SESSION_CAST (user_data);
1631
1632   GST_RTP_SESSION_LOCK (rtpsession);
1633   GST_DEBUG_OBJECT (rtpsession, "unlock timer for reconsideration");
1634   if (rtpsession->priv->id)
1635     gst_clock_id_unschedule (rtpsession->priv->id);
1636   GST_RTP_SESSION_UNLOCK (rtpsession);
1637 }
1638
1639 static gboolean
1640 gst_rtp_session_event_recv_rtp_sink (GstPad * pad, GstObject * parent,
1641     GstEvent * event)
1642 {
1643   GstRtpSession *rtpsession;
1644   gboolean ret = FALSE;
1645
1646   rtpsession = GST_RTP_SESSION (parent);
1647
1648   GST_DEBUG_OBJECT (rtpsession, "received event %s",
1649       GST_EVENT_TYPE_NAME (event));
1650
1651   switch (GST_EVENT_TYPE (event)) {
1652     case GST_EVENT_CAPS:
1653     {
1654       GstCaps *caps;
1655
1656       /* process */
1657       gst_event_parse_caps (event, &caps);
1658       gst_rtp_session_sink_setcaps (pad, rtpsession, caps);
1659       ret = gst_pad_push_event (rtpsession->recv_rtp_src, event);
1660       break;
1661     }
1662     case GST_EVENT_FLUSH_STOP:
1663       gst_segment_init (&rtpsession->recv_rtp_seg, GST_FORMAT_UNDEFINED);
1664       ret = gst_pad_push_event (rtpsession->recv_rtp_src, event);
1665       break;
1666     case GST_EVENT_SEGMENT:
1667     {
1668       GstSegment *segment, in_segment;
1669
1670       segment = &rtpsession->recv_rtp_seg;
1671
1672       /* the newsegment event is needed to convert the RTP timestamp to
1673        * running_time, which is needed to generate a mapping from RTP to NTP
1674        * timestamps in SR reports */
1675       gst_event_copy_segment (event, &in_segment);
1676       GST_DEBUG_OBJECT (rtpsession, "received segment %" GST_SEGMENT_FORMAT,
1677           &in_segment);
1678
1679       /* accept upstream */
1680       gst_segment_copy_into (&in_segment, segment);
1681
1682       /* push event forward */
1683       ret = gst_pad_push_event (rtpsession->recv_rtp_src, event);
1684       break;
1685     }
1686     case GST_EVENT_EOS:
1687     {
1688       GstPad *rtcp_src;
1689
1690       ret =
1691           gst_pad_push_event (rtpsession->recv_rtp_src, gst_event_ref (event));
1692
1693       GST_RTP_SESSION_LOCK (rtpsession);
1694       if ((rtcp_src = rtpsession->send_rtcp_src))
1695         gst_object_ref (rtcp_src);
1696       GST_RTP_SESSION_UNLOCK (rtpsession);
1697
1698       if (rtcp_src) {
1699         ret = gst_pad_push_event (rtcp_src, event);
1700         gst_object_unref (rtcp_src);
1701       } else {
1702         gst_event_unref (event);
1703         ret = TRUE;
1704       }
1705       break;
1706     }
1707     default:
1708       ret = gst_pad_push_event (rtpsession->recv_rtp_src, event);
1709       break;
1710   }
1711
1712   return ret;
1713
1714 }
1715
1716 static gboolean
1717 gst_rtp_session_request_remote_key_unit (GstRtpSession * rtpsession,
1718     guint32 ssrc, guint payload, gboolean all_headers, gint count)
1719 {
1720   GstCaps *caps;
1721
1722   caps = gst_rtp_session_get_caps_for_pt (rtpsession, payload);
1723
1724   if (caps) {
1725     const GstStructure *s = gst_caps_get_structure (caps, 0);
1726     gboolean pli;
1727     gboolean fir;
1728
1729     pli = gst_structure_has_field (s, "rtcp-fb-nack-pli");
1730     fir = gst_structure_has_field (s, "rtcp-fb-ccm-fir") && all_headers;
1731
1732     /* Google Talk uses FIR for repair, so send it even if we just want a
1733      * regular PLI */
1734     if (!pli &&
1735         gst_structure_has_field (s, "rtcp-fb-x-gstreamer-fir-as-repair"))
1736       fir = TRUE;
1737
1738     gst_caps_unref (caps);
1739
1740     if (pli || fir)
1741       return rtp_session_request_key_unit (rtpsession->priv->session, ssrc,
1742           fir, count);
1743   }
1744
1745   return FALSE;
1746 }
1747
1748 static gboolean
1749 gst_rtp_session_event_recv_rtp_src (GstPad * pad, GstObject * parent,
1750     GstEvent * event)
1751 {
1752   GstRtpSession *rtpsession;
1753   gboolean forward = TRUE;
1754   gboolean ret = TRUE;
1755   const GstStructure *s;
1756   guint32 ssrc;
1757   guint pt;
1758
1759   rtpsession = GST_RTP_SESSION (parent);
1760
1761   switch (GST_EVENT_TYPE (event)) {
1762     case GST_EVENT_CUSTOM_UPSTREAM:
1763       s = gst_event_get_structure (event);
1764       if (gst_structure_has_name (s, "GstForceKeyUnit") &&
1765           gst_structure_get_uint (s, "ssrc", &ssrc) &&
1766           gst_structure_get_uint (s, "payload", &pt)) {
1767         gboolean all_headers = FALSE;
1768         gint count = -1;
1769
1770         gst_structure_get_boolean (s, "all-headers", &all_headers);
1771         if (gst_structure_get_int (s, "count", &count) && count < 0)
1772           count += G_MAXINT;    /* Make sure count is positive if present */
1773         if (gst_rtp_session_request_remote_key_unit (rtpsession, ssrc, pt,
1774                 all_headers, count))
1775           forward = FALSE;
1776       } else if (gst_structure_has_name (s, "GstRTPRetransmissionRequest")) {
1777         GstClockTime running_time;
1778         guint seqnum, delay, deadline, max_delay, avg_rtt;
1779
1780         GST_RTP_SESSION_LOCK (rtpsession);
1781         rtpsession->priv->rtx_count++;
1782         GST_RTP_SESSION_UNLOCK (rtpsession);
1783
1784         if (!gst_structure_get_clock_time (s, "running-time", &running_time))
1785           running_time = -1;
1786         if (!gst_structure_get_uint (s, "ssrc", &ssrc))
1787           ssrc = -1;
1788         if (!gst_structure_get_uint (s, "seqnum", &seqnum))
1789           seqnum = -1;
1790         if (!gst_structure_get_uint (s, "delay", &delay))
1791           delay = 0;
1792         if (!gst_structure_get_uint (s, "deadline", &deadline))
1793           deadline = 100;
1794         if (!gst_structure_get_uint (s, "avg-rtt", &avg_rtt))
1795           avg_rtt = 40;
1796
1797         /* remaining time to receive the packet */
1798         max_delay = deadline;
1799         if (max_delay > delay)
1800           max_delay -= delay;
1801         /* estimated RTT */
1802         if (max_delay > avg_rtt)
1803           max_delay -= avg_rtt;
1804         else
1805           max_delay = 0;
1806
1807         if (rtp_session_request_nack (rtpsession->priv->session, ssrc, seqnum,
1808                 max_delay * GST_MSECOND))
1809           forward = FALSE;
1810       }
1811       break;
1812     default:
1813       break;
1814   }
1815
1816   if (forward) {
1817     GstPad *recv_rtp_sink;
1818
1819     GST_RTP_SESSION_LOCK (rtpsession);
1820     if ((recv_rtp_sink = rtpsession->recv_rtp_sink))
1821       gst_object_ref (recv_rtp_sink);
1822     GST_RTP_SESSION_UNLOCK (rtpsession);
1823
1824     if (recv_rtp_sink) {
1825       ret = gst_pad_push_event (recv_rtp_sink, event);
1826       gst_object_unref (recv_rtp_sink);
1827     } else
1828       gst_event_unref (event);
1829   } else {
1830     gst_event_unref (event);
1831   }
1832
1833   return ret;
1834 }
1835
1836
1837 static GstIterator *
1838 gst_rtp_session_iterate_internal_links (GstPad * pad, GstObject * parent)
1839 {
1840   GstRtpSession *rtpsession;
1841   GstPad *otherpad = NULL;
1842   GstIterator *it = NULL;
1843
1844   rtpsession = GST_RTP_SESSION (parent);
1845
1846   GST_RTP_SESSION_LOCK (rtpsession);
1847   if (pad == rtpsession->recv_rtp_src) {
1848     otherpad = gst_object_ref (rtpsession->recv_rtp_sink);
1849   } else if (pad == rtpsession->recv_rtp_sink) {
1850     otherpad = gst_object_ref (rtpsession->recv_rtp_src);
1851   } else if (pad == rtpsession->send_rtp_src) {
1852     otherpad = gst_object_ref (rtpsession->send_rtp_sink);
1853   } else if (pad == rtpsession->send_rtp_sink) {
1854     otherpad = gst_object_ref (rtpsession->send_rtp_src);
1855   }
1856   GST_RTP_SESSION_UNLOCK (rtpsession);
1857
1858   if (otherpad) {
1859     GValue val = { 0, };
1860
1861     g_value_init (&val, GST_TYPE_PAD);
1862     g_value_set_object (&val, otherpad);
1863     it = gst_iterator_new_single (GST_TYPE_PAD, &val);
1864     g_value_unset (&val);
1865     gst_object_unref (otherpad);
1866   } else {
1867     it = gst_iterator_new_single (GST_TYPE_PAD, NULL);
1868   }
1869
1870   return it;
1871 }
1872
1873 static gboolean
1874 gst_rtp_session_sink_setcaps (GstPad * pad, GstRtpSession * rtpsession,
1875     GstCaps * caps)
1876 {
1877   GST_RTP_SESSION_LOCK (rtpsession);
1878   gst_rtp_session_cache_caps (rtpsession, caps);
1879   GST_RTP_SESSION_UNLOCK (rtpsession);
1880
1881   return TRUE;
1882 }
1883
1884 /* receive a packet from a sender, send it to the RTP session manager and
1885  * forward the packet on the rtp_src pad
1886  */
1887 static GstFlowReturn
1888 gst_rtp_session_chain_recv_rtp (GstPad * pad, GstObject * parent,
1889     GstBuffer * buffer)
1890 {
1891   GstRtpSession *rtpsession;
1892   GstRtpSessionPrivate *priv;
1893   GstFlowReturn ret;
1894   GstClockTime current_time, running_time;
1895   GstClockTime timestamp;
1896   guint64 ntpnstime;
1897
1898   rtpsession = GST_RTP_SESSION (parent);
1899   priv = rtpsession->priv;
1900
1901   GST_LOG_OBJECT (rtpsession, "received RTP packet");
1902
1903   GST_RTP_SESSION_LOCK (rtpsession);
1904   if (rtpsession->priv->wait_send) {
1905     GST_LOG_OBJECT (rtpsession, "signal RTCP thread");
1906     rtpsession->priv->wait_send = FALSE;
1907     GST_RTP_SESSION_SIGNAL (rtpsession);
1908   }
1909   GST_RTP_SESSION_UNLOCK (rtpsession);
1910
1911   /* get NTP time when this packet was captured, this depends on the timestamp. */
1912   timestamp = GST_BUFFER_PTS (buffer);
1913   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
1914     /* convert to running time using the segment values */
1915     running_time =
1916         gst_segment_to_running_time (&rtpsession->recv_rtp_seg, GST_FORMAT_TIME,
1917         timestamp);
1918     ntpnstime = GST_CLOCK_TIME_NONE;
1919   } else {
1920     get_current_times (rtpsession, &running_time, &ntpnstime);
1921   }
1922   current_time = gst_clock_get_time (priv->sysclock);
1923
1924   ret = rtp_session_process_rtp (priv->session, buffer, current_time,
1925       running_time, ntpnstime);
1926   if (ret != GST_FLOW_OK)
1927     goto push_error;
1928
1929 done:
1930
1931   return ret;
1932
1933   /* ERRORS */
1934 push_error:
1935   {
1936     GST_DEBUG_OBJECT (rtpsession, "process returned %s",
1937         gst_flow_get_name (ret));
1938     goto done;
1939   }
1940 }
1941
1942 static gboolean
1943 gst_rtp_session_event_recv_rtcp_sink (GstPad * pad, GstObject * parent,
1944     GstEvent * event)
1945 {
1946   GstRtpSession *rtpsession;
1947   gboolean ret = FALSE;
1948
1949   rtpsession = GST_RTP_SESSION (parent);
1950
1951   GST_DEBUG_OBJECT (rtpsession, "received event %s",
1952       GST_EVENT_TYPE_NAME (event));
1953
1954   switch (GST_EVENT_TYPE (event)) {
1955     case GST_EVENT_SEGMENT:
1956       /* Make sure that the sync_src pad has caps before the segment event.
1957        * Otherwise we might get a segment event before caps from the receive
1958        * RTCP pad, and then later when receiving RTCP packets will set caps.
1959        * This will results in a sticky event misordering warning
1960        */
1961       if (!gst_pad_has_current_caps (rtpsession->sync_src)) {
1962         GstCaps *caps = gst_caps_new_empty_simple ("application/x-rtcp");
1963         gst_pad_set_caps (rtpsession->sync_src, caps);
1964         gst_caps_unref (caps);
1965       }
1966       /* fall through */
1967     default:
1968       ret = gst_pad_push_event (rtpsession->sync_src, event);
1969       break;
1970   }
1971
1972   return ret;
1973 }
1974
1975 /* Receive an RTCP packet from a sender, send it to the RTP session manager and
1976  * forward the SR packets to the sync_src pad.
1977  */
1978 static GstFlowReturn
1979 gst_rtp_session_chain_recv_rtcp (GstPad * pad, GstObject * parent,
1980     GstBuffer * buffer)
1981 {
1982   GstRtpSession *rtpsession;
1983   GstRtpSessionPrivate *priv;
1984   GstClockTime current_time;
1985   guint64 ntpnstime;
1986
1987   rtpsession = GST_RTP_SESSION (parent);
1988   priv = rtpsession->priv;
1989
1990   GST_LOG_OBJECT (rtpsession, "received RTCP packet");
1991
1992   GST_RTP_SESSION_LOCK (rtpsession);
1993   if (rtpsession->priv->wait_send) {
1994     GST_LOG_OBJECT (rtpsession, "signal RTCP thread");
1995     rtpsession->priv->wait_send = FALSE;
1996     GST_RTP_SESSION_SIGNAL (rtpsession);
1997   }
1998   GST_RTP_SESSION_UNLOCK (rtpsession);
1999
2000   current_time = gst_clock_get_time (priv->sysclock);
2001   get_current_times (rtpsession, NULL, &ntpnstime);
2002
2003   rtp_session_process_rtcp (priv->session, buffer, current_time, ntpnstime);
2004
2005   return GST_FLOW_OK;           /* always return OK */
2006 }
2007
2008 static gboolean
2009 gst_rtp_session_query_send_rtcp_src (GstPad * pad, GstObject * parent,
2010     GstQuery * query)
2011 {
2012   GstRtpSession *rtpsession;
2013   gboolean ret = FALSE;
2014
2015   rtpsession = GST_RTP_SESSION (parent);
2016
2017   GST_DEBUG_OBJECT (rtpsession, "received QUERY %s",
2018       GST_QUERY_TYPE_NAME (query));
2019
2020   switch (GST_QUERY_TYPE (query)) {
2021     case GST_QUERY_LATENCY:
2022       ret = TRUE;
2023       /* use the defaults for the latency query. */
2024       gst_query_set_latency (query, FALSE, 0, -1);
2025       break;
2026     default:
2027       /* other queries simply fail for now */
2028       break;
2029   }
2030
2031   return ret;
2032 }
2033
2034 static gboolean
2035 gst_rtp_session_event_send_rtcp_src (GstPad * pad, GstObject * parent,
2036     GstEvent * event)
2037 {
2038   GstRtpSession *rtpsession;
2039   gboolean ret = TRUE;
2040
2041   rtpsession = GST_RTP_SESSION (parent);
2042   GST_DEBUG_OBJECT (rtpsession, "received EVENT %s",
2043       GST_EVENT_TYPE_NAME (event));
2044
2045   switch (GST_EVENT_TYPE (event)) {
2046     case GST_EVENT_SEEK:
2047     case GST_EVENT_LATENCY:
2048       gst_event_unref (event);
2049       ret = TRUE;
2050       break;
2051     default:
2052       /* other events simply fail for now */
2053       gst_event_unref (event);
2054       ret = FALSE;
2055       break;
2056   }
2057
2058   return ret;
2059 }
2060
2061
2062 static gboolean
2063 gst_rtp_session_event_send_rtp_sink (GstPad * pad, GstObject * parent,
2064     GstEvent * event)
2065 {
2066   GstRtpSession *rtpsession;
2067   gboolean ret = FALSE;
2068
2069   rtpsession = GST_RTP_SESSION (parent);
2070
2071   GST_DEBUG_OBJECT (rtpsession, "received EVENT %s",
2072       GST_EVENT_TYPE_NAME (event));
2073
2074   switch (GST_EVENT_TYPE (event)) {
2075     case GST_EVENT_CAPS:
2076     {
2077       GstCaps *caps;
2078
2079       /* process */
2080       gst_event_parse_caps (event, &caps);
2081       gst_rtp_session_setcaps_send_rtp (pad, rtpsession, caps);
2082       ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
2083       break;
2084     }
2085     case GST_EVENT_FLUSH_STOP:
2086       gst_segment_init (&rtpsession->send_rtp_seg, GST_FORMAT_UNDEFINED);
2087       ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
2088       break;
2089     case GST_EVENT_SEGMENT:{
2090       GstSegment *segment, in_segment;
2091
2092       segment = &rtpsession->send_rtp_seg;
2093
2094       /* the newsegment event is needed to convert the RTP timestamp to
2095        * running_time, which is needed to generate a mapping from RTP to NTP
2096        * timestamps in SR reports */
2097       gst_event_copy_segment (event, &in_segment);
2098       GST_DEBUG_OBJECT (rtpsession, "received segment %" GST_SEGMENT_FORMAT,
2099           &in_segment);
2100
2101       /* accept upstream */
2102       gst_segment_copy_into (&in_segment, segment);
2103
2104       /* push event forward */
2105       ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
2106       break;
2107     }
2108     case GST_EVENT_EOS:{
2109       GstClockTime current_time;
2110
2111       /* push downstream FIXME, we are not supposed to leave the session just
2112        * because we stop sending. */
2113       ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
2114       current_time = gst_clock_get_time (rtpsession->priv->sysclock);
2115
2116       GST_DEBUG_OBJECT (rtpsession, "scheduling BYE message");
2117       rtp_session_mark_all_bye (rtpsession->priv->session, "End Of Stream");
2118       rtp_session_schedule_bye (rtpsession->priv->session, current_time);
2119       break;
2120     }
2121     default:{
2122       GstPad *send_rtp_src;
2123
2124       GST_RTP_SESSION_LOCK (rtpsession);
2125       if ((send_rtp_src = rtpsession->send_rtp_src))
2126         gst_object_ref (send_rtp_src);
2127       GST_RTP_SESSION_UNLOCK (rtpsession);
2128
2129       if (send_rtp_src) {
2130         ret = gst_pad_push_event (send_rtp_src, event);
2131         gst_object_unref (send_rtp_src);
2132       } else
2133         gst_event_unref (event);
2134
2135       break;
2136     }
2137   }
2138
2139   return ret;
2140 }
2141
2142 static gboolean
2143 gst_rtp_session_event_send_rtp_src (GstPad * pad, GstObject * parent,
2144     GstEvent * event)
2145 {
2146   GstRtpSession *rtpsession;
2147   gboolean ret = FALSE;
2148
2149   rtpsession = GST_RTP_SESSION (parent);
2150
2151   GST_DEBUG_OBJECT (rtpsession, "received EVENT %s",
2152       GST_EVENT_TYPE_NAME (event));
2153
2154   switch (GST_EVENT_TYPE (event)) {
2155     case GST_EVENT_LATENCY:
2156       /* save the latency, we need this to know when an RTP packet will be
2157        * rendered by the sink */
2158       gst_event_parse_latency (event, &rtpsession->priv->send_latency);
2159
2160       ret = gst_pad_event_default (pad, parent, event);
2161       break;
2162     default:
2163       ret = gst_pad_event_default (pad, parent, event);
2164       break;
2165   }
2166   return ret;
2167 }
2168
2169 static GstCaps *
2170 gst_rtp_session_getcaps_send_rtp (GstPad * pad, GstRtpSession * rtpsession,
2171     GstCaps * filter)
2172 {
2173   GstRtpSessionPrivate *priv;
2174   GstCaps *result;
2175   GstStructure *s1, *s2;
2176   guint ssrc;
2177   gboolean is_random;
2178
2179   priv = rtpsession->priv;
2180
2181   ssrc = rtp_session_suggest_ssrc (priv->session, &is_random);
2182
2183   /* we can basically accept anything but we prefer to receive packets with our
2184    * internal SSRC so that we don't have to patch it. Create a structure with
2185    * the SSRC and another one without.
2186    * Only do this if the session actually decided on an ssrc already,
2187    * otherwise we give upstream the opportunity to select an ssrc itself */
2188   if (!is_random) {
2189     s1 = gst_structure_new ("application/x-rtp", "ssrc", G_TYPE_UINT, ssrc,
2190         NULL);
2191     s2 = gst_structure_new_empty ("application/x-rtp");
2192
2193     result = gst_caps_new_full (s1, s2, NULL);
2194   } else {
2195     result = gst_caps_new_empty_simple ("application/x-rtp");
2196   }
2197
2198   if (filter) {
2199     GstCaps *caps = result;
2200
2201     result = gst_caps_intersect_full (filter, caps, GST_CAPS_INTERSECT_FIRST);
2202     gst_caps_unref (caps);
2203   }
2204
2205   GST_DEBUG_OBJECT (rtpsession, "getting caps %" GST_PTR_FORMAT, result);
2206
2207   return result;
2208 }
2209
2210 static gboolean
2211 gst_rtp_session_query_send_rtp (GstPad * pad, GstObject * parent,
2212     GstQuery * query)
2213 {
2214   gboolean res = FALSE;
2215   GstRtpSession *rtpsession;
2216
2217   rtpsession = GST_RTP_SESSION (parent);
2218
2219   switch (GST_QUERY_TYPE (query)) {
2220     case GST_QUERY_CAPS:
2221     {
2222       GstCaps *filter, *caps;
2223
2224       gst_query_parse_caps (query, &filter);
2225       caps = gst_rtp_session_getcaps_send_rtp (pad, rtpsession, filter);
2226       gst_query_set_caps_result (query, caps);
2227       gst_caps_unref (caps);
2228       res = TRUE;
2229       break;
2230     }
2231     default:
2232       res = gst_pad_query_default (pad, parent, query);
2233       break;
2234   }
2235
2236   return res;
2237 }
2238
2239 static gboolean
2240 gst_rtp_session_setcaps_send_rtp (GstPad * pad, GstRtpSession * rtpsession,
2241     GstCaps * caps)
2242 {
2243   GstRtpSessionPrivate *priv;
2244
2245   priv = rtpsession->priv;
2246
2247   rtp_session_update_send_caps (priv->session, caps);
2248
2249   return TRUE;
2250 }
2251
2252 /* Recieve an RTP packet or a list of packets to be send to the receivers,
2253  * send to RTP session manager and forward to send_rtp_src.
2254  */
2255 static GstFlowReturn
2256 gst_rtp_session_chain_send_rtp_common (GstRtpSession * rtpsession,
2257     gpointer data, gboolean is_list)
2258 {
2259   GstRtpSessionPrivate *priv;
2260   GstFlowReturn ret;
2261   GstClockTime timestamp, running_time;
2262   GstClockTime current_time;
2263
2264   priv = rtpsession->priv;
2265
2266   GST_LOG_OBJECT (rtpsession, "received RTP %s", is_list ? "list" : "packet");
2267
2268   /* get NTP time when this packet was captured, this depends on the timestamp. */
2269   if (is_list) {
2270     GstBuffer *buffer = NULL;
2271
2272     /* All groups in an list have the same timestamp.
2273      * So, just take it from the first group. */
2274     buffer = gst_buffer_list_get (GST_BUFFER_LIST_CAST (data), 0);
2275     if (buffer)
2276       timestamp = GST_BUFFER_PTS (buffer);
2277     else
2278       timestamp = -1;
2279   } else {
2280     timestamp = GST_BUFFER_PTS (GST_BUFFER_CAST (data));
2281   }
2282
2283   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
2284     /* convert to running time using the segment start value. */
2285     running_time =
2286         gst_segment_to_running_time (&rtpsession->send_rtp_seg, GST_FORMAT_TIME,
2287         timestamp);
2288     if (priv->rtcp_sync_send_time)
2289       running_time += priv->send_latency;
2290   } else {
2291     /* no timestamp. */
2292     running_time = -1;
2293   }
2294
2295   current_time = gst_clock_get_time (priv->sysclock);
2296   ret = rtp_session_send_rtp (priv->session, data, is_list, current_time,
2297       running_time);
2298   if (ret != GST_FLOW_OK)
2299     goto push_error;
2300
2301 done:
2302
2303   return ret;
2304
2305   /* ERRORS */
2306 push_error:
2307   {
2308     GST_DEBUG_OBJECT (rtpsession, "process returned %s",
2309         gst_flow_get_name (ret));
2310     goto done;
2311   }
2312 }
2313
2314 static GstFlowReturn
2315 gst_rtp_session_chain_send_rtp (GstPad * pad, GstObject * parent,
2316     GstBuffer * buffer)
2317 {
2318   GstRtpSession *rtpsession = GST_RTP_SESSION (parent);
2319
2320   return gst_rtp_session_chain_send_rtp_common (rtpsession, buffer, FALSE);
2321 }
2322
2323 static GstFlowReturn
2324 gst_rtp_session_chain_send_rtp_list (GstPad * pad, GstObject * parent,
2325     GstBufferList * list)
2326 {
2327   GstRtpSession *rtpsession = GST_RTP_SESSION (parent);
2328
2329   return gst_rtp_session_chain_send_rtp_common (rtpsession, list, TRUE);
2330 }
2331
2332 /* Create sinkpad to receive RTP packets from senders. This will also create a
2333  * srcpad for the RTP packets.
2334  */
2335 static GstPad *
2336 create_recv_rtp_sink (GstRtpSession * rtpsession)
2337 {
2338   GST_DEBUG_OBJECT (rtpsession, "creating RTP sink pad");
2339
2340   rtpsession->recv_rtp_sink =
2341       gst_pad_new_from_static_template (&rtpsession_recv_rtp_sink_template,
2342       "recv_rtp_sink");
2343   gst_pad_set_chain_function (rtpsession->recv_rtp_sink,
2344       gst_rtp_session_chain_recv_rtp);
2345   gst_pad_set_event_function (rtpsession->recv_rtp_sink,
2346       gst_rtp_session_event_recv_rtp_sink);
2347   gst_pad_set_iterate_internal_links_function (rtpsession->recv_rtp_sink,
2348       gst_rtp_session_iterate_internal_links);
2349   GST_PAD_SET_PROXY_ALLOCATION (rtpsession->recv_rtp_sink);
2350   gst_pad_set_active (rtpsession->recv_rtp_sink, TRUE);
2351   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession),
2352       rtpsession->recv_rtp_sink);
2353
2354   GST_DEBUG_OBJECT (rtpsession, "creating RTP src pad");
2355   rtpsession->recv_rtp_src =
2356       gst_pad_new_from_static_template (&rtpsession_recv_rtp_src_template,
2357       "recv_rtp_src");
2358   gst_pad_set_event_function (rtpsession->recv_rtp_src,
2359       gst_rtp_session_event_recv_rtp_src);
2360   gst_pad_set_iterate_internal_links_function (rtpsession->recv_rtp_src,
2361       gst_rtp_session_iterate_internal_links);
2362   gst_pad_use_fixed_caps (rtpsession->recv_rtp_src);
2363   gst_pad_set_active (rtpsession->recv_rtp_src, TRUE);
2364   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->recv_rtp_src);
2365
2366   return rtpsession->recv_rtp_sink;
2367 }
2368
2369 /* Remove sinkpad to receive RTP packets from senders. This will also remove
2370  * the srcpad for the RTP packets.
2371  */
2372 static void
2373 remove_recv_rtp_sink (GstRtpSession * rtpsession)
2374 {
2375   GST_DEBUG_OBJECT (rtpsession, "removing RTP sink pad");
2376
2377   /* deactivate from source to sink */
2378   gst_pad_set_active (rtpsession->recv_rtp_src, FALSE);
2379   gst_pad_set_active (rtpsession->recv_rtp_sink, FALSE);
2380
2381   /* remove pads */
2382   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2383       rtpsession->recv_rtp_sink);
2384   rtpsession->recv_rtp_sink = NULL;
2385
2386   GST_DEBUG_OBJECT (rtpsession, "removing RTP src pad");
2387   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2388       rtpsession->recv_rtp_src);
2389   rtpsession->recv_rtp_src = NULL;
2390 }
2391
2392 /* Create a sinkpad to receive RTCP messages from senders, this will also create a
2393  * sync_src pad for the SR packets.
2394  */
2395 static GstPad *
2396 create_recv_rtcp_sink (GstRtpSession * rtpsession)
2397 {
2398   GST_DEBUG_OBJECT (rtpsession, "creating RTCP sink pad");
2399
2400   rtpsession->recv_rtcp_sink =
2401       gst_pad_new_from_static_template (&rtpsession_recv_rtcp_sink_template,
2402       "recv_rtcp_sink");
2403   gst_pad_set_chain_function (rtpsession->recv_rtcp_sink,
2404       gst_rtp_session_chain_recv_rtcp);
2405   gst_pad_set_event_function (rtpsession->recv_rtcp_sink,
2406       gst_rtp_session_event_recv_rtcp_sink);
2407   gst_pad_set_iterate_internal_links_function (rtpsession->recv_rtcp_sink,
2408       gst_rtp_session_iterate_internal_links);
2409   gst_pad_set_active (rtpsession->recv_rtcp_sink, TRUE);
2410   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession),
2411       rtpsession->recv_rtcp_sink);
2412
2413   GST_DEBUG_OBJECT (rtpsession, "creating sync src pad");
2414   rtpsession->sync_src =
2415       gst_pad_new_from_static_template (&rtpsession_sync_src_template,
2416       "sync_src");
2417   gst_pad_set_iterate_internal_links_function (rtpsession->sync_src,
2418       gst_rtp_session_iterate_internal_links);
2419   gst_pad_use_fixed_caps (rtpsession->sync_src);
2420   gst_pad_set_active (rtpsession->sync_src, TRUE);
2421   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->sync_src);
2422
2423   return rtpsession->recv_rtcp_sink;
2424 }
2425
2426 static void
2427 remove_recv_rtcp_sink (GstRtpSession * rtpsession)
2428 {
2429   GST_DEBUG_OBJECT (rtpsession, "removing RTCP sink pad");
2430
2431   gst_pad_set_active (rtpsession->sync_src, FALSE);
2432   gst_pad_set_active (rtpsession->recv_rtcp_sink, FALSE);
2433
2434   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2435       rtpsession->recv_rtcp_sink);
2436   rtpsession->recv_rtcp_sink = NULL;
2437
2438   GST_DEBUG_OBJECT (rtpsession, "removing sync src pad");
2439   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->sync_src);
2440   rtpsession->sync_src = NULL;
2441 }
2442
2443 /* Create a sinkpad to receive RTP packets for receivers. This will also create a
2444  * send_rtp_src pad.
2445  */
2446 static GstPad *
2447 create_send_rtp_sink (GstRtpSession * rtpsession)
2448 {
2449   GST_DEBUG_OBJECT (rtpsession, "creating pad");
2450
2451   rtpsession->send_rtp_sink =
2452       gst_pad_new_from_static_template (&rtpsession_send_rtp_sink_template,
2453       "send_rtp_sink");
2454   gst_pad_set_chain_function (rtpsession->send_rtp_sink,
2455       gst_rtp_session_chain_send_rtp);
2456   gst_pad_set_chain_list_function (rtpsession->send_rtp_sink,
2457       gst_rtp_session_chain_send_rtp_list);
2458   gst_pad_set_query_function (rtpsession->send_rtp_sink,
2459       gst_rtp_session_query_send_rtp);
2460   gst_pad_set_event_function (rtpsession->send_rtp_sink,
2461       gst_rtp_session_event_send_rtp_sink);
2462   gst_pad_set_iterate_internal_links_function (rtpsession->send_rtp_sink,
2463       gst_rtp_session_iterate_internal_links);
2464   GST_PAD_SET_PROXY_CAPS (rtpsession->send_rtp_sink);
2465   GST_PAD_SET_PROXY_ALLOCATION (rtpsession->send_rtp_sink);
2466   gst_pad_set_active (rtpsession->send_rtp_sink, TRUE);
2467   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession),
2468       rtpsession->send_rtp_sink);
2469
2470   rtpsession->send_rtp_src =
2471       gst_pad_new_from_static_template (&rtpsession_send_rtp_src_template,
2472       "send_rtp_src");
2473   gst_pad_set_iterate_internal_links_function (rtpsession->send_rtp_src,
2474       gst_rtp_session_iterate_internal_links);
2475   gst_pad_set_event_function (rtpsession->send_rtp_src,
2476       gst_rtp_session_event_send_rtp_src);
2477   GST_PAD_SET_PROXY_CAPS (rtpsession->send_rtp_src);
2478   gst_pad_set_active (rtpsession->send_rtp_src, TRUE);
2479   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->send_rtp_src);
2480
2481   return rtpsession->send_rtp_sink;
2482 }
2483
2484 static void
2485 remove_send_rtp_sink (GstRtpSession * rtpsession)
2486 {
2487   GST_DEBUG_OBJECT (rtpsession, "removing pad");
2488
2489   gst_pad_set_active (rtpsession->send_rtp_src, FALSE);
2490   gst_pad_set_active (rtpsession->send_rtp_sink, FALSE);
2491
2492   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2493       rtpsession->send_rtp_sink);
2494   rtpsession->send_rtp_sink = NULL;
2495
2496   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2497       rtpsession->send_rtp_src);
2498   rtpsession->send_rtp_src = NULL;
2499 }
2500
2501 /* Create a srcpad with the RTCP packets to send out.
2502  * This pad will be driven by the RTP session manager when it wants to send out
2503  * RTCP packets.
2504  */
2505 static GstPad *
2506 create_send_rtcp_src (GstRtpSession * rtpsession)
2507 {
2508   GST_DEBUG_OBJECT (rtpsession, "creating pad");
2509
2510   rtpsession->send_rtcp_src =
2511       gst_pad_new_from_static_template (&rtpsession_send_rtcp_src_template,
2512       "send_rtcp_src");
2513   gst_pad_use_fixed_caps (rtpsession->send_rtcp_src);
2514   gst_pad_set_active (rtpsession->send_rtcp_src, TRUE);
2515   gst_pad_set_iterate_internal_links_function (rtpsession->send_rtcp_src,
2516       gst_rtp_session_iterate_internal_links);
2517   gst_pad_set_query_function (rtpsession->send_rtcp_src,
2518       gst_rtp_session_query_send_rtcp_src);
2519   gst_pad_set_event_function (rtpsession->send_rtcp_src,
2520       gst_rtp_session_event_send_rtcp_src);
2521   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession),
2522       rtpsession->send_rtcp_src);
2523
2524   return rtpsession->send_rtcp_src;
2525 }
2526
2527 static void
2528 remove_send_rtcp_src (GstRtpSession * rtpsession)
2529 {
2530   GST_DEBUG_OBJECT (rtpsession, "removing pad");
2531
2532   gst_pad_set_active (rtpsession->send_rtcp_src, FALSE);
2533
2534   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2535       rtpsession->send_rtcp_src);
2536   rtpsession->send_rtcp_src = NULL;
2537 }
2538
2539 static GstPad *
2540 gst_rtp_session_request_new_pad (GstElement * element,
2541     GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
2542 {
2543   GstRtpSession *rtpsession;
2544   GstElementClass *klass;
2545   GstPad *result;
2546
2547   g_return_val_if_fail (templ != NULL, NULL);
2548   g_return_val_if_fail (GST_IS_RTP_SESSION (element), NULL);
2549
2550   rtpsession = GST_RTP_SESSION (element);
2551   klass = GST_ELEMENT_GET_CLASS (element);
2552
2553   GST_DEBUG_OBJECT (element, "requesting pad %s", GST_STR_NULL (name));
2554
2555   GST_RTP_SESSION_LOCK (rtpsession);
2556
2557   /* figure out the template */
2558   if (templ == gst_element_class_get_pad_template (klass, "recv_rtp_sink")) {
2559     if (rtpsession->recv_rtp_sink != NULL)
2560       goto exists;
2561
2562     result = create_recv_rtp_sink (rtpsession);
2563   } else if (templ == gst_element_class_get_pad_template (klass,
2564           "recv_rtcp_sink")) {
2565     if (rtpsession->recv_rtcp_sink != NULL)
2566       goto exists;
2567
2568     result = create_recv_rtcp_sink (rtpsession);
2569   } else if (templ == gst_element_class_get_pad_template (klass,
2570           "send_rtp_sink")) {
2571     if (rtpsession->send_rtp_sink != NULL)
2572       goto exists;
2573
2574     result = create_send_rtp_sink (rtpsession);
2575   } else if (templ == gst_element_class_get_pad_template (klass,
2576           "send_rtcp_src")) {
2577     if (rtpsession->send_rtcp_src != NULL)
2578       goto exists;
2579
2580     result = create_send_rtcp_src (rtpsession);
2581   } else
2582     goto wrong_template;
2583
2584   GST_RTP_SESSION_UNLOCK (rtpsession);
2585
2586   return result;
2587
2588   /* ERRORS */
2589 wrong_template:
2590   {
2591     GST_RTP_SESSION_UNLOCK (rtpsession);
2592     g_warning ("rtpsession: this is not our template");
2593     return NULL;
2594   }
2595 exists:
2596   {
2597     GST_RTP_SESSION_UNLOCK (rtpsession);
2598     g_warning ("rtpsession: pad already requested");
2599     return NULL;
2600   }
2601 }
2602
2603 static void
2604 gst_rtp_session_release_pad (GstElement * element, GstPad * pad)
2605 {
2606   GstRtpSession *rtpsession;
2607
2608   g_return_if_fail (GST_IS_RTP_SESSION (element));
2609   g_return_if_fail (GST_IS_PAD (pad));
2610
2611   rtpsession = GST_RTP_SESSION (element);
2612
2613   GST_DEBUG_OBJECT (element, "releasing pad %s:%s", GST_DEBUG_PAD_NAME (pad));
2614
2615   GST_RTP_SESSION_LOCK (rtpsession);
2616
2617   if (rtpsession->recv_rtp_sink == pad) {
2618     remove_recv_rtp_sink (rtpsession);
2619   } else if (rtpsession->recv_rtcp_sink == pad) {
2620     remove_recv_rtcp_sink (rtpsession);
2621   } else if (rtpsession->send_rtp_sink == pad) {
2622     remove_send_rtp_sink (rtpsession);
2623   } else if (rtpsession->send_rtcp_src == pad) {
2624     remove_send_rtcp_src (rtpsession);
2625   } else
2626     goto wrong_pad;
2627
2628   GST_RTP_SESSION_UNLOCK (rtpsession);
2629
2630   return;
2631
2632   /* ERRORS */
2633 wrong_pad:
2634   {
2635     GST_RTP_SESSION_UNLOCK (rtpsession);
2636     g_warning ("rtpsession: asked to release an unknown pad");
2637     return;
2638   }
2639 }
2640
2641 static void
2642 gst_rtp_session_request_key_unit (RTPSession * sess,
2643     gboolean all_headers, gpointer user_data)
2644 {
2645   GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
2646   GstEvent *event;
2647   GstPad *send_rtp_sink;
2648
2649   GST_RTP_SESSION_LOCK (rtpsession);
2650   if ((send_rtp_sink = rtpsession->send_rtp_sink))
2651     gst_object_ref (send_rtp_sink);
2652   GST_RTP_SESSION_UNLOCK (rtpsession);
2653
2654   if (send_rtp_sink) {
2655     event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
2656         gst_structure_new ("GstForceKeyUnit",
2657             "all-headers", G_TYPE_BOOLEAN, all_headers, NULL));
2658     gst_pad_push_event (send_rtp_sink, event);
2659     gst_object_unref (send_rtp_sink);
2660   }
2661 }
2662
2663 static GstClockTime
2664 gst_rtp_session_request_time (RTPSession * session, gpointer user_data)
2665 {
2666   GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
2667
2668   return gst_clock_get_time (rtpsession->priv->sysclock);
2669 }
2670
2671 static void
2672 gst_rtp_session_notify_nack (RTPSession * sess, guint16 seqnum,
2673     guint16 blp, guint32 ssrc, gpointer user_data)
2674 {
2675   GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
2676   GstEvent *event;
2677   GstPad *send_rtp_sink;
2678
2679   GST_RTP_SESSION_LOCK (rtpsession);
2680   if ((send_rtp_sink = rtpsession->send_rtp_sink))
2681     gst_object_ref (send_rtp_sink);
2682   GST_RTP_SESSION_UNLOCK (rtpsession);
2683
2684   if (send_rtp_sink) {
2685     while (TRUE) {
2686       event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
2687           gst_structure_new ("GstRTPRetransmissionRequest",
2688               "seqnum", G_TYPE_UINT, (guint) seqnum,
2689               "ssrc", G_TYPE_UINT, (guint) ssrc, NULL));
2690       gst_pad_push_event (send_rtp_sink, event);
2691
2692       if (blp == 0)
2693         break;
2694
2695       seqnum++;
2696       while ((blp & 1) == 0) {
2697         seqnum++;
2698         blp >>= 1;
2699       }
2700       blp >>= 1;
2701     }
2702     gst_object_unref (send_rtp_sink);
2703   }
2704 }
2705
2706 static void
2707 gst_rtp_session_reconfigure (RTPSession * sess, gpointer user_data)
2708 {
2709   GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
2710   GstPad *send_rtp_sink;
2711
2712   GST_RTP_SESSION_LOCK (rtpsession);
2713   if ((send_rtp_sink = rtpsession->send_rtp_sink))
2714     gst_object_ref (send_rtp_sink);
2715   GST_RTP_SESSION_UNLOCK (rtpsession);
2716
2717   if (send_rtp_sink) {
2718     gst_pad_push_event (send_rtp_sink, gst_event_new_reconfigure ());
2719     gst_object_unref (send_rtp_sink);
2720   }
2721 }