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