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 all_sources_bye, 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     /* Forward send an EOS on the RTCP sink if we received an EOS on the
1448      * send_rtp_sink. We don't need to check the recv_rtp_sink since in this
1449      * case the EOS event would already have been sent */
1450     if (all_sources_bye && rtpsession->send_rtp_sink &&
1451         GST_PAD_IS_EOS (rtpsession->send_rtp_sink)) {
1452       GstEvent *event;
1453
1454       GST_LOG_OBJECT (rtpsession, "sending EOS");
1455
1456       event = gst_event_new_eos ();
1457       gst_event_set_seqnum (event, rtpsession->recv_rtcp_segment_seqnum);
1458       gst_pad_push_event (rtcp_src, event);
1459     }
1460     gst_object_unref (rtcp_src);
1461   } else {
1462     GST_RTP_SESSION_UNLOCK (rtpsession);
1463
1464     GST_DEBUG_OBJECT (rtpsession, "not sending RTCP, no output pad");
1465     gst_buffer_unref (buffer);
1466     result = GST_FLOW_OK;
1467   }
1468   return result;
1469
1470   /* ERRORS */
1471 stopping:
1472   {
1473     GST_DEBUG_OBJECT (rtpsession, "we are stopping");
1474     gst_buffer_unref (buffer);
1475     GST_RTP_SESSION_UNLOCK (rtpsession);
1476     return GST_FLOW_OK;
1477   }
1478 }
1479
1480 /* called when the session manager has an SR RTCP packet ready for handling
1481  * inter stream synchronisation */
1482 static GstFlowReturn
1483 gst_rtp_session_sync_rtcp (RTPSession * sess,
1484     GstBuffer * buffer, gpointer user_data)
1485 {
1486   GstFlowReturn result;
1487   GstRtpSession *rtpsession;
1488   GstPad *sync_src;
1489
1490   rtpsession = GST_RTP_SESSION (user_data);
1491
1492   GST_RTP_SESSION_LOCK (rtpsession);
1493   if (rtpsession->priv->stop_thread)
1494     goto stopping;
1495
1496   if ((sync_src = rtpsession->sync_src)) {
1497     gst_object_ref (sync_src);
1498     GST_RTP_SESSION_UNLOCK (rtpsession);
1499
1500     /* set rtcp caps on output pad, this happens
1501      * when we receive RTCP muxed with RTP according
1502      * to RFC5761. Otherwise we would have forwarded
1503      * the events from the recv_rtcp_sink pad already
1504      */
1505     if (!gst_pad_has_current_caps (sync_src))
1506       do_rtcp_events (rtpsession, sync_src);
1507
1508     GST_LOG_OBJECT (rtpsession, "sending Sync RTCP");
1509     result = gst_pad_push (sync_src, buffer);
1510     gst_object_unref (sync_src);
1511   } else {
1512     GST_RTP_SESSION_UNLOCK (rtpsession);
1513
1514     GST_DEBUG_OBJECT (rtpsession, "not sending Sync RTCP, no output pad");
1515     gst_buffer_unref (buffer);
1516     result = GST_FLOW_OK;
1517   }
1518   return result;
1519
1520   /* ERRORS */
1521 stopping:
1522   {
1523     GST_DEBUG_OBJECT (rtpsession, "we are stopping");
1524     gst_buffer_unref (buffer);
1525     GST_RTP_SESSION_UNLOCK (rtpsession);
1526     return GST_FLOW_OK;
1527   }
1528 }
1529
1530 static void
1531 gst_rtp_session_cache_caps (GstRtpSession * rtpsession, GstCaps * caps)
1532 {
1533   GstRtpSessionPrivate *priv;
1534   const GstStructure *s;
1535   gint payload;
1536
1537   priv = rtpsession->priv;
1538
1539   GST_DEBUG_OBJECT (rtpsession, "parsing caps");
1540
1541   s = gst_caps_get_structure (caps, 0);
1542   if (!gst_structure_get_int (s, "payload", &payload))
1543     return;
1544
1545   if (g_hash_table_lookup (priv->ptmap, GINT_TO_POINTER (payload)))
1546     return;
1547
1548   g_hash_table_insert (priv->ptmap, GINT_TO_POINTER (payload),
1549       gst_caps_ref (caps));
1550 }
1551
1552 static GstCaps *
1553 gst_rtp_session_get_caps_for_pt (GstRtpSession * rtpsession, guint payload)
1554 {
1555   GstCaps *caps = NULL;
1556   GValue args[2] = { {0}, {0} };
1557   GValue ret = { 0 };
1558
1559   GST_RTP_SESSION_LOCK (rtpsession);
1560   caps = g_hash_table_lookup (rtpsession->priv->ptmap,
1561       GINT_TO_POINTER (payload));
1562   if (caps) {
1563     gst_caps_ref (caps);
1564     goto done;
1565   }
1566
1567   /* not found in the cache, try to get it with a signal */
1568   g_value_init (&args[0], GST_TYPE_ELEMENT);
1569   g_value_set_object (&args[0], rtpsession);
1570   g_value_init (&args[1], G_TYPE_UINT);
1571   g_value_set_uint (&args[1], payload);
1572
1573   g_value_init (&ret, GST_TYPE_CAPS);
1574   g_value_set_boxed (&ret, NULL);
1575
1576   GST_RTP_SESSION_UNLOCK (rtpsession);
1577
1578   g_signal_emitv (args, gst_rtp_session_signals[SIGNAL_REQUEST_PT_MAP], 0,
1579       &ret);
1580
1581   GST_RTP_SESSION_LOCK (rtpsession);
1582
1583   g_value_unset (&args[0]);
1584   g_value_unset (&args[1]);
1585   caps = (GstCaps *) g_value_dup_boxed (&ret);
1586   g_value_unset (&ret);
1587   if (!caps)
1588     goto no_caps;
1589
1590   gst_rtp_session_cache_caps (rtpsession, caps);
1591
1592 done:
1593   GST_RTP_SESSION_UNLOCK (rtpsession);
1594
1595   return caps;
1596
1597 no_caps:
1598   {
1599     GST_DEBUG_OBJECT (rtpsession, "could not get caps");
1600     goto done;
1601   }
1602 }
1603
1604 /* called when the session manager needs the clock rate */
1605 static gint
1606 gst_rtp_session_clock_rate (RTPSession * sess, guint8 payload,
1607     gpointer user_data)
1608 {
1609   gint result = -1;
1610   GstRtpSession *rtpsession;
1611   GstCaps *caps;
1612   const GstStructure *s;
1613
1614   rtpsession = GST_RTP_SESSION_CAST (user_data);
1615
1616   caps = gst_rtp_session_get_caps_for_pt (rtpsession, payload);
1617
1618   if (!caps)
1619     goto done;
1620
1621   s = gst_caps_get_structure (caps, 0);
1622   if (!gst_structure_get_int (s, "clock-rate", &result))
1623     goto no_clock_rate;
1624
1625   gst_caps_unref (caps);
1626
1627   GST_DEBUG_OBJECT (rtpsession, "parsed clock-rate %d", result);
1628
1629 done:
1630
1631   return result;
1632
1633   /* ERRORS */
1634 no_clock_rate:
1635   {
1636     gst_caps_unref (caps);
1637     GST_DEBUG_OBJECT (rtpsession, "No clock-rate in caps!");
1638     goto done;
1639   }
1640 }
1641
1642 /* called when the session manager asks us to reconsider the timeout */
1643 static void
1644 gst_rtp_session_reconsider (RTPSession * sess, gpointer user_data)
1645 {
1646   GstRtpSession *rtpsession;
1647
1648   rtpsession = GST_RTP_SESSION_CAST (user_data);
1649
1650   GST_RTP_SESSION_LOCK (rtpsession);
1651   GST_DEBUG_OBJECT (rtpsession, "unlock timer for reconsideration");
1652   if (rtpsession->priv->id)
1653     gst_clock_id_unschedule (rtpsession->priv->id);
1654   GST_RTP_SESSION_UNLOCK (rtpsession);
1655 }
1656
1657 static gboolean
1658 gst_rtp_session_event_recv_rtp_sink (GstPad * pad, GstObject * parent,
1659     GstEvent * event)
1660 {
1661   GstRtpSession *rtpsession;
1662   gboolean ret = FALSE;
1663
1664   rtpsession = GST_RTP_SESSION (parent);
1665
1666   GST_DEBUG_OBJECT (rtpsession, "received event %s",
1667       GST_EVENT_TYPE_NAME (event));
1668
1669   switch (GST_EVENT_TYPE (event)) {
1670     case GST_EVENT_CAPS:
1671     {
1672       GstCaps *caps;
1673
1674       /* process */
1675       gst_event_parse_caps (event, &caps);
1676       gst_rtp_session_sink_setcaps (pad, rtpsession, caps);
1677       ret = gst_pad_push_event (rtpsession->recv_rtp_src, event);
1678       break;
1679     }
1680     case GST_EVENT_FLUSH_STOP:
1681       gst_segment_init (&rtpsession->recv_rtp_seg, GST_FORMAT_UNDEFINED);
1682       rtpsession->recv_rtcp_segment_seqnum = GST_SEQNUM_INVALID;
1683       ret = gst_pad_push_event (rtpsession->recv_rtp_src, event);
1684       break;
1685     case GST_EVENT_SEGMENT:
1686     {
1687       GstSegment *segment, in_segment;
1688
1689       segment = &rtpsession->recv_rtp_seg;
1690
1691       /* the newsegment event is needed to convert the RTP timestamp to
1692        * running_time, which is needed to generate a mapping from RTP to NTP
1693        * timestamps in SR reports */
1694       gst_event_copy_segment (event, &in_segment);
1695       GST_DEBUG_OBJECT (rtpsession, "received segment %" GST_SEGMENT_FORMAT,
1696           &in_segment);
1697
1698       /* accept upstream */
1699       gst_segment_copy_into (&in_segment, segment);
1700
1701       /* push event forward */
1702       ret = gst_pad_push_event (rtpsession->recv_rtp_src, event);
1703       break;
1704     }
1705     case GST_EVENT_EOS:
1706     {
1707       GstPad *rtcp_src;
1708
1709       ret =
1710           gst_pad_push_event (rtpsession->recv_rtp_src, gst_event_ref (event));
1711
1712       GST_RTP_SESSION_LOCK (rtpsession);
1713       if ((rtcp_src = rtpsession->send_rtcp_src))
1714         gst_object_ref (rtcp_src);
1715       GST_RTP_SESSION_UNLOCK (rtpsession);
1716
1717       gst_event_unref (event);
1718
1719       if (rtcp_src) {
1720         event = gst_event_new_eos ();
1721         if (rtpsession->recv_rtcp_segment_seqnum != GST_SEQNUM_INVALID)
1722           gst_event_set_seqnum (event, rtpsession->recv_rtcp_segment_seqnum);
1723         ret = gst_pad_push_event (rtcp_src, event);
1724         gst_object_unref (rtcp_src);
1725       } else {
1726         ret = TRUE;
1727       }
1728       break;
1729     }
1730     default:
1731       ret = gst_pad_push_event (rtpsession->recv_rtp_src, event);
1732       break;
1733   }
1734
1735   return ret;
1736
1737 }
1738
1739 static gboolean
1740 gst_rtp_session_request_remote_key_unit (GstRtpSession * rtpsession,
1741     guint32 ssrc, guint payload, gboolean all_headers, gint count)
1742 {
1743   GstCaps *caps;
1744
1745   caps = gst_rtp_session_get_caps_for_pt (rtpsession, payload);
1746
1747   if (caps) {
1748     const GstStructure *s = gst_caps_get_structure (caps, 0);
1749     gboolean pli;
1750     gboolean fir;
1751
1752     pli = gst_structure_has_field (s, "rtcp-fb-nack-pli");
1753     fir = gst_structure_has_field (s, "rtcp-fb-ccm-fir") && all_headers;
1754
1755     /* Google Talk uses FIR for repair, so send it even if we just want a
1756      * regular PLI */
1757     if (!pli &&
1758         gst_structure_has_field (s, "rtcp-fb-x-gstreamer-fir-as-repair"))
1759       fir = TRUE;
1760
1761     gst_caps_unref (caps);
1762
1763     if (pli || fir)
1764       return rtp_session_request_key_unit (rtpsession->priv->session, ssrc,
1765           fir, count);
1766   }
1767
1768   return FALSE;
1769 }
1770
1771 static gboolean
1772 gst_rtp_session_event_recv_rtp_src (GstPad * pad, GstObject * parent,
1773     GstEvent * event)
1774 {
1775   GstRtpSession *rtpsession;
1776   gboolean forward = TRUE;
1777   gboolean ret = TRUE;
1778   const GstStructure *s;
1779   guint32 ssrc;
1780   guint pt;
1781
1782   rtpsession = GST_RTP_SESSION (parent);
1783
1784   switch (GST_EVENT_TYPE (event)) {
1785     case GST_EVENT_CUSTOM_UPSTREAM:
1786       s = gst_event_get_structure (event);
1787       if (gst_structure_has_name (s, "GstForceKeyUnit") &&
1788           gst_structure_get_uint (s, "ssrc", &ssrc) &&
1789           gst_structure_get_uint (s, "payload", &pt)) {
1790         gboolean all_headers = FALSE;
1791         gint count = -1;
1792
1793         gst_structure_get_boolean (s, "all-headers", &all_headers);
1794         if (gst_structure_get_int (s, "count", &count) && count < 0)
1795           count += G_MAXINT;    /* Make sure count is positive if present */
1796         if (gst_rtp_session_request_remote_key_unit (rtpsession, ssrc, pt,
1797                 all_headers, count))
1798           forward = FALSE;
1799       } else if (gst_structure_has_name (s, "GstRTPRetransmissionRequest")) {
1800         GstClockTime running_time;
1801         guint seqnum, delay, deadline, max_delay, avg_rtt;
1802
1803         GST_RTP_SESSION_LOCK (rtpsession);
1804         rtpsession->priv->rtx_count++;
1805         GST_RTP_SESSION_UNLOCK (rtpsession);
1806
1807         if (!gst_structure_get_clock_time (s, "running-time", &running_time))
1808           running_time = -1;
1809         if (!gst_structure_get_uint (s, "ssrc", &ssrc))
1810           ssrc = -1;
1811         if (!gst_structure_get_uint (s, "seqnum", &seqnum))
1812           seqnum = -1;
1813         if (!gst_structure_get_uint (s, "delay", &delay))
1814           delay = 0;
1815         if (!gst_structure_get_uint (s, "deadline", &deadline))
1816           deadline = 100;
1817         if (!gst_structure_get_uint (s, "avg-rtt", &avg_rtt))
1818           avg_rtt = 40;
1819
1820         /* remaining time to receive the packet */
1821         max_delay = deadline;
1822         if (max_delay > delay)
1823           max_delay -= delay;
1824         /* estimated RTT */
1825         if (max_delay > avg_rtt)
1826           max_delay -= avg_rtt;
1827         else
1828           max_delay = 0;
1829
1830         if (rtp_session_request_nack (rtpsession->priv->session, ssrc, seqnum,
1831                 max_delay * GST_MSECOND))
1832           forward = FALSE;
1833       }
1834       break;
1835     default:
1836       break;
1837   }
1838
1839   if (forward) {
1840     GstPad *recv_rtp_sink;
1841
1842     GST_RTP_SESSION_LOCK (rtpsession);
1843     if ((recv_rtp_sink = rtpsession->recv_rtp_sink))
1844       gst_object_ref (recv_rtp_sink);
1845     GST_RTP_SESSION_UNLOCK (rtpsession);
1846
1847     if (recv_rtp_sink) {
1848       ret = gst_pad_push_event (recv_rtp_sink, event);
1849       gst_object_unref (recv_rtp_sink);
1850     } else
1851       gst_event_unref (event);
1852   } else {
1853     gst_event_unref (event);
1854   }
1855
1856   return ret;
1857 }
1858
1859
1860 static GstIterator *
1861 gst_rtp_session_iterate_internal_links (GstPad * pad, GstObject * parent)
1862 {
1863   GstRtpSession *rtpsession;
1864   GstPad *otherpad = NULL;
1865   GstIterator *it = NULL;
1866
1867   rtpsession = GST_RTP_SESSION (parent);
1868
1869   GST_RTP_SESSION_LOCK (rtpsession);
1870   if (pad == rtpsession->recv_rtp_src) {
1871     otherpad = gst_object_ref (rtpsession->recv_rtp_sink);
1872   } else if (pad == rtpsession->recv_rtp_sink) {
1873     otherpad = gst_object_ref (rtpsession->recv_rtp_src);
1874   } else if (pad == rtpsession->send_rtp_src) {
1875     otherpad = gst_object_ref (rtpsession->send_rtp_sink);
1876   } else if (pad == rtpsession->send_rtp_sink) {
1877     otherpad = gst_object_ref (rtpsession->send_rtp_src);
1878   }
1879   GST_RTP_SESSION_UNLOCK (rtpsession);
1880
1881   if (otherpad) {
1882     GValue val = { 0, };
1883
1884     g_value_init (&val, GST_TYPE_PAD);
1885     g_value_set_object (&val, otherpad);
1886     it = gst_iterator_new_single (GST_TYPE_PAD, &val);
1887     g_value_unset (&val);
1888     gst_object_unref (otherpad);
1889   } else {
1890     it = gst_iterator_new_single (GST_TYPE_PAD, NULL);
1891   }
1892
1893   return it;
1894 }
1895
1896 static gboolean
1897 gst_rtp_session_sink_setcaps (GstPad * pad, GstRtpSession * rtpsession,
1898     GstCaps * caps)
1899 {
1900   GST_RTP_SESSION_LOCK (rtpsession);
1901   gst_rtp_session_cache_caps (rtpsession, caps);
1902   GST_RTP_SESSION_UNLOCK (rtpsession);
1903
1904   return TRUE;
1905 }
1906
1907 /* receive a packet from a sender, send it to the RTP session manager and
1908  * forward the packet on the rtp_src pad
1909  */
1910 static GstFlowReturn
1911 gst_rtp_session_chain_recv_rtp (GstPad * pad, GstObject * parent,
1912     GstBuffer * buffer)
1913 {
1914   GstRtpSession *rtpsession;
1915   GstRtpSessionPrivate *priv;
1916   GstFlowReturn ret;
1917   GstClockTime current_time, running_time;
1918   GstClockTime timestamp;
1919   guint64 ntpnstime;
1920
1921   rtpsession = GST_RTP_SESSION (parent);
1922   priv = rtpsession->priv;
1923
1924   GST_LOG_OBJECT (rtpsession, "received RTP packet");
1925
1926   GST_RTP_SESSION_LOCK (rtpsession);
1927   signal_waiting_rtcp_thread_unlocked (rtpsession);
1928   GST_RTP_SESSION_UNLOCK (rtpsession);
1929
1930   /* get NTP time when this packet was captured, this depends on the timestamp. */
1931   timestamp = GST_BUFFER_PTS (buffer);
1932   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
1933     /* convert to running time using the segment values */
1934     running_time =
1935         gst_segment_to_running_time (&rtpsession->recv_rtp_seg, GST_FORMAT_TIME,
1936         timestamp);
1937     ntpnstime = GST_CLOCK_TIME_NONE;
1938   } else {
1939     get_current_times (rtpsession, &running_time, &ntpnstime);
1940   }
1941   current_time = gst_clock_get_time (priv->sysclock);
1942
1943   ret = rtp_session_process_rtp (priv->session, buffer, current_time,
1944       running_time, ntpnstime);
1945   if (ret != GST_FLOW_OK)
1946     goto push_error;
1947
1948 done:
1949
1950   return ret;
1951
1952   /* ERRORS */
1953 push_error:
1954   {
1955     GST_DEBUG_OBJECT (rtpsession, "process returned %s",
1956         gst_flow_get_name (ret));
1957     goto done;
1958   }
1959 }
1960
1961 static gboolean
1962 gst_rtp_session_event_recv_rtcp_sink (GstPad * pad, GstObject * parent,
1963     GstEvent * event)
1964 {
1965   GstRtpSession *rtpsession;
1966   gboolean ret = FALSE;
1967
1968   rtpsession = GST_RTP_SESSION (parent);
1969
1970   GST_DEBUG_OBJECT (rtpsession, "received event %s",
1971       GST_EVENT_TYPE_NAME (event));
1972
1973   switch (GST_EVENT_TYPE (event)) {
1974     case GST_EVENT_SEGMENT:
1975       /* Make sure that the sync_src pad has caps before the segment event.
1976        * Otherwise we might get a segment event before caps from the receive
1977        * RTCP pad, and then later when receiving RTCP packets will set caps.
1978        * This will results in a sticky event misordering warning
1979        */
1980       if (!gst_pad_has_current_caps (rtpsession->sync_src)) {
1981         GstCaps *caps = gst_caps_new_empty_simple ("application/x-rtcp");
1982         gst_pad_set_caps (rtpsession->sync_src, caps);
1983         gst_caps_unref (caps);
1984       }
1985       /* fall through */
1986     default:
1987       ret = gst_pad_push_event (rtpsession->sync_src, event);
1988       break;
1989   }
1990
1991   return ret;
1992 }
1993
1994 /* Receive an RTCP packet from a sender, send it to the RTP session manager and
1995  * forward the SR packets to the sync_src pad.
1996  */
1997 static GstFlowReturn
1998 gst_rtp_session_chain_recv_rtcp (GstPad * pad, GstObject * parent,
1999     GstBuffer * buffer)
2000 {
2001   GstRtpSession *rtpsession;
2002   GstRtpSessionPrivate *priv;
2003   GstClockTime current_time;
2004   GstClockTime running_time;
2005   guint64 ntpnstime;
2006
2007   rtpsession = GST_RTP_SESSION (parent);
2008   priv = rtpsession->priv;
2009
2010   GST_LOG_OBJECT (rtpsession, "received RTCP packet");
2011
2012   GST_RTP_SESSION_LOCK (rtpsession);
2013   signal_waiting_rtcp_thread_unlocked (rtpsession);
2014   GST_RTP_SESSION_UNLOCK (rtpsession);
2015
2016   current_time = gst_clock_get_time (priv->sysclock);
2017   get_current_times (rtpsession, &running_time, &ntpnstime);
2018
2019   rtp_session_process_rtcp (priv->session, buffer, current_time, running_time,
2020       ntpnstime);
2021
2022   return GST_FLOW_OK;           /* always return OK */
2023 }
2024
2025 static gboolean
2026 gst_rtp_session_query_send_rtcp_src (GstPad * pad, GstObject * parent,
2027     GstQuery * query)
2028 {
2029   GstRtpSession *rtpsession;
2030   gboolean ret = FALSE;
2031
2032   rtpsession = GST_RTP_SESSION (parent);
2033
2034   GST_DEBUG_OBJECT (rtpsession, "received QUERY %s",
2035       GST_QUERY_TYPE_NAME (query));
2036
2037   switch (GST_QUERY_TYPE (query)) {
2038     case GST_QUERY_LATENCY:
2039       ret = TRUE;
2040       /* use the defaults for the latency query. */
2041       gst_query_set_latency (query, FALSE, 0, -1);
2042       break;
2043     default:
2044       /* other queries simply fail for now */
2045       break;
2046   }
2047
2048   return ret;
2049 }
2050
2051 static gboolean
2052 gst_rtp_session_event_send_rtcp_src (GstPad * pad, GstObject * parent,
2053     GstEvent * event)
2054 {
2055   GstRtpSession *rtpsession;
2056   gboolean ret = TRUE;
2057
2058   rtpsession = GST_RTP_SESSION (parent);
2059   GST_DEBUG_OBJECT (rtpsession, "received EVENT %s",
2060       GST_EVENT_TYPE_NAME (event));
2061
2062   switch (GST_EVENT_TYPE (event)) {
2063     case GST_EVENT_SEEK:
2064     case GST_EVENT_LATENCY:
2065       gst_event_unref (event);
2066       ret = TRUE;
2067       break;
2068     default:
2069       /* other events simply fail for now */
2070       gst_event_unref (event);
2071       ret = FALSE;
2072       break;
2073   }
2074
2075   return ret;
2076 }
2077
2078
2079 static gboolean
2080 gst_rtp_session_event_send_rtp_sink (GstPad * pad, GstObject * parent,
2081     GstEvent * event)
2082 {
2083   GstRtpSession *rtpsession;
2084   gboolean ret = FALSE;
2085
2086   rtpsession = GST_RTP_SESSION (parent);
2087
2088   GST_DEBUG_OBJECT (rtpsession, "received EVENT %s",
2089       GST_EVENT_TYPE_NAME (event));
2090
2091   switch (GST_EVENT_TYPE (event)) {
2092     case GST_EVENT_CAPS:
2093     {
2094       GstCaps *caps;
2095
2096       /* process */
2097       gst_event_parse_caps (event, &caps);
2098       gst_rtp_session_setcaps_send_rtp (pad, rtpsession, caps);
2099       ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
2100       break;
2101     }
2102     case GST_EVENT_FLUSH_STOP:
2103       gst_segment_init (&rtpsession->send_rtp_seg, GST_FORMAT_UNDEFINED);
2104       ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
2105       break;
2106     case GST_EVENT_SEGMENT:{
2107       GstSegment *segment, in_segment;
2108
2109       segment = &rtpsession->send_rtp_seg;
2110
2111       /* the newsegment event is needed to convert the RTP timestamp to
2112        * running_time, which is needed to generate a mapping from RTP to NTP
2113        * timestamps in SR reports */
2114       gst_event_copy_segment (event, &in_segment);
2115       GST_DEBUG_OBJECT (rtpsession, "received segment %" GST_SEGMENT_FORMAT,
2116           &in_segment);
2117
2118       /* accept upstream */
2119       gst_segment_copy_into (&in_segment, segment);
2120
2121       /* push event forward */
2122       ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
2123       break;
2124     }
2125     case GST_EVENT_EOS:{
2126       GstClockTime current_time;
2127
2128       /* push downstream FIXME, we are not supposed to leave the session just
2129        * because we stop sending. */
2130       ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
2131       current_time = gst_clock_get_time (rtpsession->priv->sysclock);
2132
2133       GST_DEBUG_OBJECT (rtpsession, "scheduling BYE message");
2134       rtp_session_mark_all_bye (rtpsession->priv->session, "End Of Stream");
2135       rtp_session_schedule_bye (rtpsession->priv->session, current_time);
2136       break;
2137     }
2138     default:{
2139       GstPad *send_rtp_src;
2140
2141       GST_RTP_SESSION_LOCK (rtpsession);
2142       if ((send_rtp_src = rtpsession->send_rtp_src))
2143         gst_object_ref (send_rtp_src);
2144       GST_RTP_SESSION_UNLOCK (rtpsession);
2145
2146       if (send_rtp_src) {
2147         ret = gst_pad_push_event (send_rtp_src, event);
2148         gst_object_unref (send_rtp_src);
2149       } else
2150         gst_event_unref (event);
2151
2152       break;
2153     }
2154   }
2155
2156   return ret;
2157 }
2158
2159 static gboolean
2160 gst_rtp_session_event_send_rtp_src (GstPad * pad, GstObject * parent,
2161     GstEvent * event)
2162 {
2163   GstRtpSession *rtpsession;
2164   gboolean ret = FALSE;
2165
2166   rtpsession = GST_RTP_SESSION (parent);
2167
2168   GST_DEBUG_OBJECT (rtpsession, "received EVENT %s",
2169       GST_EVENT_TYPE_NAME (event));
2170
2171   switch (GST_EVENT_TYPE (event)) {
2172     case GST_EVENT_LATENCY:
2173       /* save the latency, we need this to know when an RTP packet will be
2174        * rendered by the sink */
2175       gst_event_parse_latency (event, &rtpsession->priv->send_latency);
2176
2177       ret = gst_pad_event_default (pad, parent, event);
2178       break;
2179     default:
2180       ret = gst_pad_event_default (pad, parent, event);
2181       break;
2182   }
2183   return ret;
2184 }
2185
2186 static GstCaps *
2187 gst_rtp_session_getcaps_send_rtp (GstPad * pad, GstRtpSession * rtpsession,
2188     GstCaps * filter)
2189 {
2190   GstRtpSessionPrivate *priv;
2191   GstCaps *result;
2192   GstStructure *s1, *s2;
2193   guint ssrc;
2194   gboolean is_random;
2195
2196   priv = rtpsession->priv;
2197
2198   ssrc = rtp_session_suggest_ssrc (priv->session, &is_random);
2199
2200   /* we can basically accept anything but we prefer to receive packets with our
2201    * internal SSRC so that we don't have to patch it. Create a structure with
2202    * the SSRC and another one without.
2203    * Only do this if the session actually decided on an ssrc already,
2204    * otherwise we give upstream the opportunity to select an ssrc itself */
2205   if (!is_random) {
2206     s1 = gst_structure_new ("application/x-rtp", "ssrc", G_TYPE_UINT, ssrc,
2207         NULL);
2208     s2 = gst_structure_new_empty ("application/x-rtp");
2209
2210     result = gst_caps_new_full (s1, s2, NULL);
2211   } else {
2212     result = gst_caps_new_empty_simple ("application/x-rtp");
2213   }
2214
2215   if (filter) {
2216     GstCaps *caps = result;
2217
2218     result = gst_caps_intersect_full (filter, caps, GST_CAPS_INTERSECT_FIRST);
2219     gst_caps_unref (caps);
2220   }
2221
2222   GST_DEBUG_OBJECT (rtpsession, "getting caps %" GST_PTR_FORMAT, result);
2223
2224   return result;
2225 }
2226
2227 static gboolean
2228 gst_rtp_session_query_send_rtp (GstPad * pad, GstObject * parent,
2229     GstQuery * query)
2230 {
2231   gboolean res = FALSE;
2232   GstRtpSession *rtpsession;
2233
2234   rtpsession = GST_RTP_SESSION (parent);
2235
2236   switch (GST_QUERY_TYPE (query)) {
2237     case GST_QUERY_CAPS:
2238     {
2239       GstCaps *filter, *caps;
2240
2241       gst_query_parse_caps (query, &filter);
2242       caps = gst_rtp_session_getcaps_send_rtp (pad, rtpsession, filter);
2243       gst_query_set_caps_result (query, caps);
2244       gst_caps_unref (caps);
2245       res = TRUE;
2246       break;
2247     }
2248     default:
2249       res = gst_pad_query_default (pad, parent, query);
2250       break;
2251   }
2252
2253   return res;
2254 }
2255
2256 static gboolean
2257 gst_rtp_session_setcaps_send_rtp (GstPad * pad, GstRtpSession * rtpsession,
2258     GstCaps * caps)
2259 {
2260   GstRtpSessionPrivate *priv;
2261
2262   priv = rtpsession->priv;
2263
2264   rtp_session_update_send_caps (priv->session, caps);
2265
2266   return TRUE;
2267 }
2268
2269 /* Recieve an RTP packet or a list of packets to be send to the receivers,
2270  * send to RTP session manager and forward to send_rtp_src.
2271  */
2272 static GstFlowReturn
2273 gst_rtp_session_chain_send_rtp_common (GstRtpSession * rtpsession,
2274     gpointer data, gboolean is_list)
2275 {
2276   GstRtpSessionPrivate *priv;
2277   GstFlowReturn ret;
2278   GstClockTime timestamp, running_time;
2279   GstClockTime current_time;
2280
2281   priv = rtpsession->priv;
2282
2283   GST_LOG_OBJECT (rtpsession, "received RTP %s", is_list ? "list" : "packet");
2284
2285   /* get NTP time when this packet was captured, this depends on the timestamp. */
2286   if (is_list) {
2287     GstBuffer *buffer = NULL;
2288
2289     /* All groups in an list have the same timestamp.
2290      * So, just take it from the first group. */
2291     buffer = gst_buffer_list_get (GST_BUFFER_LIST_CAST (data), 0);
2292     if (buffer)
2293       timestamp = GST_BUFFER_PTS (buffer);
2294     else
2295       timestamp = -1;
2296   } else {
2297     timestamp = GST_BUFFER_PTS (GST_BUFFER_CAST (data));
2298   }
2299
2300   if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
2301     /* convert to running time using the segment start value. */
2302     running_time =
2303         gst_segment_to_running_time (&rtpsession->send_rtp_seg, GST_FORMAT_TIME,
2304         timestamp);
2305     if (priv->rtcp_sync_send_time)
2306       running_time += priv->send_latency;
2307   } else {
2308     /* no timestamp. */
2309     running_time = -1;
2310   }
2311
2312   current_time = gst_clock_get_time (priv->sysclock);
2313   ret = rtp_session_send_rtp (priv->session, data, is_list, current_time,
2314       running_time);
2315   if (ret != GST_FLOW_OK)
2316     goto push_error;
2317
2318 done:
2319
2320   return ret;
2321
2322   /* ERRORS */
2323 push_error:
2324   {
2325     GST_DEBUG_OBJECT (rtpsession, "process returned %s",
2326         gst_flow_get_name (ret));
2327     goto done;
2328   }
2329 }
2330
2331 static GstFlowReturn
2332 gst_rtp_session_chain_send_rtp (GstPad * pad, GstObject * parent,
2333     GstBuffer * buffer)
2334 {
2335   GstRtpSession *rtpsession = GST_RTP_SESSION (parent);
2336
2337   return gst_rtp_session_chain_send_rtp_common (rtpsession, buffer, FALSE);
2338 }
2339
2340 static GstFlowReturn
2341 gst_rtp_session_chain_send_rtp_list (GstPad * pad, GstObject * parent,
2342     GstBufferList * list)
2343 {
2344   GstRtpSession *rtpsession = GST_RTP_SESSION (parent);
2345
2346   return gst_rtp_session_chain_send_rtp_common (rtpsession, list, TRUE);
2347 }
2348
2349 /* Create sinkpad to receive RTP packets from senders. This will also create a
2350  * srcpad for the RTP packets.
2351  */
2352 static GstPad *
2353 create_recv_rtp_sink (GstRtpSession * rtpsession)
2354 {
2355   GST_DEBUG_OBJECT (rtpsession, "creating RTP sink pad");
2356
2357   rtpsession->recv_rtp_sink =
2358       gst_pad_new_from_static_template (&rtpsession_recv_rtp_sink_template,
2359       "recv_rtp_sink");
2360   gst_pad_set_chain_function (rtpsession->recv_rtp_sink,
2361       gst_rtp_session_chain_recv_rtp);
2362   gst_pad_set_event_function (rtpsession->recv_rtp_sink,
2363       gst_rtp_session_event_recv_rtp_sink);
2364   gst_pad_set_iterate_internal_links_function (rtpsession->recv_rtp_sink,
2365       gst_rtp_session_iterate_internal_links);
2366   GST_PAD_SET_PROXY_ALLOCATION (rtpsession->recv_rtp_sink);
2367   gst_pad_set_active (rtpsession->recv_rtp_sink, TRUE);
2368   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession),
2369       rtpsession->recv_rtp_sink);
2370
2371   GST_DEBUG_OBJECT (rtpsession, "creating RTP src pad");
2372   rtpsession->recv_rtp_src =
2373       gst_pad_new_from_static_template (&rtpsession_recv_rtp_src_template,
2374       "recv_rtp_src");
2375   gst_pad_set_event_function (rtpsession->recv_rtp_src,
2376       gst_rtp_session_event_recv_rtp_src);
2377   gst_pad_set_iterate_internal_links_function (rtpsession->recv_rtp_src,
2378       gst_rtp_session_iterate_internal_links);
2379   gst_pad_use_fixed_caps (rtpsession->recv_rtp_src);
2380   gst_pad_set_active (rtpsession->recv_rtp_src, TRUE);
2381   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->recv_rtp_src);
2382
2383   return rtpsession->recv_rtp_sink;
2384 }
2385
2386 /* Remove sinkpad to receive RTP packets from senders. This will also remove
2387  * the srcpad for the RTP packets.
2388  */
2389 static void
2390 remove_recv_rtp_sink (GstRtpSession * rtpsession)
2391 {
2392   GST_DEBUG_OBJECT (rtpsession, "removing RTP sink pad");
2393
2394   /* deactivate from source to sink */
2395   gst_pad_set_active (rtpsession->recv_rtp_src, FALSE);
2396   gst_pad_set_active (rtpsession->recv_rtp_sink, FALSE);
2397
2398   /* remove pads */
2399   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2400       rtpsession->recv_rtp_sink);
2401   rtpsession->recv_rtp_sink = NULL;
2402
2403   GST_DEBUG_OBJECT (rtpsession, "removing RTP src pad");
2404   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2405       rtpsession->recv_rtp_src);
2406   rtpsession->recv_rtp_src = NULL;
2407 }
2408
2409 /* Create a sinkpad to receive RTCP messages from senders, this will also create a
2410  * sync_src pad for the SR packets.
2411  */
2412 static GstPad *
2413 create_recv_rtcp_sink (GstRtpSession * rtpsession)
2414 {
2415   GST_DEBUG_OBJECT (rtpsession, "creating RTCP sink pad");
2416
2417   rtpsession->recv_rtcp_sink =
2418       gst_pad_new_from_static_template (&rtpsession_recv_rtcp_sink_template,
2419       "recv_rtcp_sink");
2420   gst_pad_set_chain_function (rtpsession->recv_rtcp_sink,
2421       gst_rtp_session_chain_recv_rtcp);
2422   gst_pad_set_event_function (rtpsession->recv_rtcp_sink,
2423       gst_rtp_session_event_recv_rtcp_sink);
2424   gst_pad_set_iterate_internal_links_function (rtpsession->recv_rtcp_sink,
2425       gst_rtp_session_iterate_internal_links);
2426   gst_pad_set_active (rtpsession->recv_rtcp_sink, TRUE);
2427   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession),
2428       rtpsession->recv_rtcp_sink);
2429
2430   GST_DEBUG_OBJECT (rtpsession, "creating sync src pad");
2431   rtpsession->sync_src =
2432       gst_pad_new_from_static_template (&rtpsession_sync_src_template,
2433       "sync_src");
2434   gst_pad_set_iterate_internal_links_function (rtpsession->sync_src,
2435       gst_rtp_session_iterate_internal_links);
2436   gst_pad_use_fixed_caps (rtpsession->sync_src);
2437   gst_pad_set_active (rtpsession->sync_src, TRUE);
2438   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->sync_src);
2439
2440   return rtpsession->recv_rtcp_sink;
2441 }
2442
2443 static void
2444 remove_recv_rtcp_sink (GstRtpSession * rtpsession)
2445 {
2446   GST_DEBUG_OBJECT (rtpsession, "removing RTCP sink pad");
2447
2448   gst_pad_set_active (rtpsession->sync_src, FALSE);
2449   gst_pad_set_active (rtpsession->recv_rtcp_sink, FALSE);
2450
2451   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2452       rtpsession->recv_rtcp_sink);
2453   rtpsession->recv_rtcp_sink = NULL;
2454
2455   GST_DEBUG_OBJECT (rtpsession, "removing sync src pad");
2456   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->sync_src);
2457   rtpsession->sync_src = NULL;
2458 }
2459
2460 /* Create a sinkpad to receive RTP packets for receivers. This will also create a
2461  * send_rtp_src pad.
2462  */
2463 static GstPad *
2464 create_send_rtp_sink (GstRtpSession * rtpsession)
2465 {
2466   GST_DEBUG_OBJECT (rtpsession, "creating pad");
2467
2468   rtpsession->send_rtp_sink =
2469       gst_pad_new_from_static_template (&rtpsession_send_rtp_sink_template,
2470       "send_rtp_sink");
2471   gst_pad_set_chain_function (rtpsession->send_rtp_sink,
2472       gst_rtp_session_chain_send_rtp);
2473   gst_pad_set_chain_list_function (rtpsession->send_rtp_sink,
2474       gst_rtp_session_chain_send_rtp_list);
2475   gst_pad_set_query_function (rtpsession->send_rtp_sink,
2476       gst_rtp_session_query_send_rtp);
2477   gst_pad_set_event_function (rtpsession->send_rtp_sink,
2478       gst_rtp_session_event_send_rtp_sink);
2479   gst_pad_set_iterate_internal_links_function (rtpsession->send_rtp_sink,
2480       gst_rtp_session_iterate_internal_links);
2481   GST_PAD_SET_PROXY_CAPS (rtpsession->send_rtp_sink);
2482   GST_PAD_SET_PROXY_ALLOCATION (rtpsession->send_rtp_sink);
2483   gst_pad_set_active (rtpsession->send_rtp_sink, TRUE);
2484   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession),
2485       rtpsession->send_rtp_sink);
2486
2487   rtpsession->send_rtp_src =
2488       gst_pad_new_from_static_template (&rtpsession_send_rtp_src_template,
2489       "send_rtp_src");
2490   gst_pad_set_iterate_internal_links_function (rtpsession->send_rtp_src,
2491       gst_rtp_session_iterate_internal_links);
2492   gst_pad_set_event_function (rtpsession->send_rtp_src,
2493       gst_rtp_session_event_send_rtp_src);
2494   GST_PAD_SET_PROXY_CAPS (rtpsession->send_rtp_src);
2495   gst_pad_set_active (rtpsession->send_rtp_src, TRUE);
2496   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession), rtpsession->send_rtp_src);
2497
2498   return rtpsession->send_rtp_sink;
2499 }
2500
2501 static void
2502 remove_send_rtp_sink (GstRtpSession * rtpsession)
2503 {
2504   GST_DEBUG_OBJECT (rtpsession, "removing pad");
2505
2506   gst_pad_set_active (rtpsession->send_rtp_src, FALSE);
2507   gst_pad_set_active (rtpsession->send_rtp_sink, FALSE);
2508
2509   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2510       rtpsession->send_rtp_sink);
2511   rtpsession->send_rtp_sink = NULL;
2512
2513   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2514       rtpsession->send_rtp_src);
2515   rtpsession->send_rtp_src = NULL;
2516 }
2517
2518 /* Create a srcpad with the RTCP packets to send out.
2519  * This pad will be driven by the RTP session manager when it wants to send out
2520  * RTCP packets.
2521  */
2522 static GstPad *
2523 create_send_rtcp_src (GstRtpSession * rtpsession)
2524 {
2525   GST_DEBUG_OBJECT (rtpsession, "creating pad");
2526
2527   rtpsession->send_rtcp_src =
2528       gst_pad_new_from_static_template (&rtpsession_send_rtcp_src_template,
2529       "send_rtcp_src");
2530   gst_pad_use_fixed_caps (rtpsession->send_rtcp_src);
2531   gst_pad_set_active (rtpsession->send_rtcp_src, TRUE);
2532   gst_pad_set_iterate_internal_links_function (rtpsession->send_rtcp_src,
2533       gst_rtp_session_iterate_internal_links);
2534   gst_pad_set_query_function (rtpsession->send_rtcp_src,
2535       gst_rtp_session_query_send_rtcp_src);
2536   gst_pad_set_event_function (rtpsession->send_rtcp_src,
2537       gst_rtp_session_event_send_rtcp_src);
2538   gst_element_add_pad (GST_ELEMENT_CAST (rtpsession),
2539       rtpsession->send_rtcp_src);
2540
2541   return rtpsession->send_rtcp_src;
2542 }
2543
2544 static void
2545 remove_send_rtcp_src (GstRtpSession * rtpsession)
2546 {
2547   GST_DEBUG_OBJECT (rtpsession, "removing pad");
2548
2549   gst_pad_set_active (rtpsession->send_rtcp_src, FALSE);
2550
2551   gst_element_remove_pad (GST_ELEMENT_CAST (rtpsession),
2552       rtpsession->send_rtcp_src);
2553   rtpsession->send_rtcp_src = NULL;
2554 }
2555
2556 static GstPad *
2557 gst_rtp_session_request_new_pad (GstElement * element,
2558     GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
2559 {
2560   GstRtpSession *rtpsession;
2561   GstElementClass *klass;
2562   GstPad *result;
2563
2564   g_return_val_if_fail (templ != NULL, NULL);
2565   g_return_val_if_fail (GST_IS_RTP_SESSION (element), NULL);
2566
2567   rtpsession = GST_RTP_SESSION (element);
2568   klass = GST_ELEMENT_GET_CLASS (element);
2569
2570   GST_DEBUG_OBJECT (element, "requesting pad %s", GST_STR_NULL (name));
2571
2572   GST_RTP_SESSION_LOCK (rtpsession);
2573
2574   /* figure out the template */
2575   if (templ == gst_element_class_get_pad_template (klass, "recv_rtp_sink")) {
2576     if (rtpsession->recv_rtp_sink != NULL)
2577       goto exists;
2578
2579     result = create_recv_rtp_sink (rtpsession);
2580   } else if (templ == gst_element_class_get_pad_template (klass,
2581           "recv_rtcp_sink")) {
2582     if (rtpsession->recv_rtcp_sink != NULL)
2583       goto exists;
2584
2585     result = create_recv_rtcp_sink (rtpsession);
2586   } else if (templ == gst_element_class_get_pad_template (klass,
2587           "send_rtp_sink")) {
2588     if (rtpsession->send_rtp_sink != NULL)
2589       goto exists;
2590
2591     result = create_send_rtp_sink (rtpsession);
2592   } else if (templ == gst_element_class_get_pad_template (klass,
2593           "send_rtcp_src")) {
2594     if (rtpsession->send_rtcp_src != NULL)
2595       goto exists;
2596
2597     result = create_send_rtcp_src (rtpsession);
2598   } else
2599     goto wrong_template;
2600
2601   GST_RTP_SESSION_UNLOCK (rtpsession);
2602
2603   return result;
2604
2605   /* ERRORS */
2606 wrong_template:
2607   {
2608     GST_RTP_SESSION_UNLOCK (rtpsession);
2609     g_warning ("rtpsession: this is not our template");
2610     return NULL;
2611   }
2612 exists:
2613   {
2614     GST_RTP_SESSION_UNLOCK (rtpsession);
2615     g_warning ("rtpsession: pad already requested");
2616     return NULL;
2617   }
2618 }
2619
2620 static void
2621 gst_rtp_session_release_pad (GstElement * element, GstPad * pad)
2622 {
2623   GstRtpSession *rtpsession;
2624
2625   g_return_if_fail (GST_IS_RTP_SESSION (element));
2626   g_return_if_fail (GST_IS_PAD (pad));
2627
2628   rtpsession = GST_RTP_SESSION (element);
2629
2630   GST_DEBUG_OBJECT (element, "releasing pad %s:%s", GST_DEBUG_PAD_NAME (pad));
2631
2632   GST_RTP_SESSION_LOCK (rtpsession);
2633
2634   if (rtpsession->recv_rtp_sink == pad) {
2635     remove_recv_rtp_sink (rtpsession);
2636   } else if (rtpsession->recv_rtcp_sink == pad) {
2637     remove_recv_rtcp_sink (rtpsession);
2638   } else if (rtpsession->send_rtp_sink == pad) {
2639     remove_send_rtp_sink (rtpsession);
2640   } else if (rtpsession->send_rtcp_src == pad) {
2641     remove_send_rtcp_src (rtpsession);
2642   } else
2643     goto wrong_pad;
2644
2645   GST_RTP_SESSION_UNLOCK (rtpsession);
2646
2647   return;
2648
2649   /* ERRORS */
2650 wrong_pad:
2651   {
2652     GST_RTP_SESSION_UNLOCK (rtpsession);
2653     g_warning ("rtpsession: asked to release an unknown pad");
2654     return;
2655   }
2656 }
2657
2658 static void
2659 gst_rtp_session_request_key_unit (RTPSession * sess,
2660     guint32 ssrc, gboolean all_headers, gpointer user_data)
2661 {
2662   GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
2663   GstEvent *event;
2664   GstPad *send_rtp_sink;
2665
2666   GST_RTP_SESSION_LOCK (rtpsession);
2667   if ((send_rtp_sink = rtpsession->send_rtp_sink))
2668     gst_object_ref (send_rtp_sink);
2669   GST_RTP_SESSION_UNLOCK (rtpsession);
2670
2671   if (send_rtp_sink) {
2672     event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
2673         gst_structure_new ("GstForceKeyUnit", "ssrc", G_TYPE_UINT, ssrc,
2674             "all-headers", G_TYPE_BOOLEAN, all_headers, NULL));
2675     gst_pad_push_event (send_rtp_sink, event);
2676     gst_object_unref (send_rtp_sink);
2677   }
2678 }
2679
2680 static GstClockTime
2681 gst_rtp_session_request_time (RTPSession * session, gpointer user_data)
2682 {
2683   GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
2684
2685   return gst_clock_get_time (rtpsession->priv->sysclock);
2686 }
2687
2688 static void
2689 gst_rtp_session_notify_nack (RTPSession * sess, guint16 seqnum,
2690     guint16 blp, guint32 ssrc, gpointer user_data)
2691 {
2692   GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
2693   GstEvent *event;
2694   GstPad *send_rtp_sink;
2695
2696   GST_RTP_SESSION_LOCK (rtpsession);
2697   if ((send_rtp_sink = rtpsession->send_rtp_sink))
2698     gst_object_ref (send_rtp_sink);
2699   GST_RTP_SESSION_UNLOCK (rtpsession);
2700
2701   if (send_rtp_sink) {
2702     while (TRUE) {
2703       event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
2704           gst_structure_new ("GstRTPRetransmissionRequest",
2705               "seqnum", G_TYPE_UINT, (guint) seqnum,
2706               "ssrc", G_TYPE_UINT, (guint) ssrc, NULL));
2707       gst_pad_push_event (send_rtp_sink, event);
2708
2709       if (blp == 0)
2710         break;
2711
2712       seqnum++;
2713       while ((blp & 1) == 0) {
2714         seqnum++;
2715         blp >>= 1;
2716       }
2717       blp >>= 1;
2718     }
2719     gst_object_unref (send_rtp_sink);
2720   }
2721 }
2722
2723 static void
2724 gst_rtp_session_reconfigure (RTPSession * sess, gpointer user_data)
2725 {
2726   GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
2727   GstPad *send_rtp_sink;
2728
2729   GST_RTP_SESSION_LOCK (rtpsession);
2730   if ((send_rtp_sink = rtpsession->send_rtp_sink))
2731     gst_object_ref (send_rtp_sink);
2732   GST_RTP_SESSION_UNLOCK (rtpsession);
2733
2734   if (send_rtp_sink) {
2735     gst_pad_push_event (send_rtp_sink, gst_event_new_reconfigure ());
2736     gst_object_unref (send_rtp_sink);
2737   }
2738 }
2739
2740 static void
2741 gst_rtp_session_notify_early_rtcp (RTPSession * sess, gpointer user_data)
2742 {
2743   GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
2744
2745   GST_DEBUG_OBJECT (rtpsession, "Notified of early RTCP");
2746   /* with an early RTCP request, we might have to start the RTCP thread */
2747   GST_RTP_SESSION_LOCK (rtpsession);
2748   signal_waiting_rtcp_thread_unlocked (rtpsession);
2749   GST_RTP_SESSION_UNLOCK (rtpsession);
2750 }