gst/rtpmanager/rtpsession.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 /* FIXME 0.11: suppress warnings for deprecated API such as GValueArray
  21  * with newer GLib versions (>= 2.31.0) */
  22 #define GLIB_DISABLE_DEPRECATION_WARNINGS
  23
  24 #include <string.h>
  25
  26 #include <gst/rtp/gstrtpbuffer.h>
  27 #include <gst/rtp/gstrtcpbuffer.h>
  28
  29 #include <gst/glib-compat-private.h>
  30
  31 #include "rtpsession.h"
  32
  33 GST_DEBUG_CATEGORY_STATIC (rtp_session_debug);
  34 #define GST_CAT_DEFAULT rtp_session_debug
  35
  36 /* signals and args */
  37 enum
  38 {
  39   SIGNAL_GET_SOURCE_BY_SSRC,
  40   SIGNAL_ON_NEW_SSRC,
  41   SIGNAL_ON_SSRC_COLLISION,
  42   SIGNAL_ON_SSRC_VALIDATED,
  43   SIGNAL_ON_SSRC_ACTIVE,
  44   SIGNAL_ON_SSRC_SDES,
  45   SIGNAL_ON_BYE_SSRC,
  46   SIGNAL_ON_BYE_TIMEOUT,
  47   SIGNAL_ON_TIMEOUT,
  48   SIGNAL_ON_SENDER_TIMEOUT,
  49   SIGNAL_ON_SENDING_RTCP,
  50   SIGNAL_ON_FEEDBACK_RTCP,
  51   SIGNAL_SEND_RTCP,
  52   LAST_SIGNAL
  53 };
  54
  55 #define DEFAULT_INTERNAL_SOURCE      NULL
  56 #define DEFAULT_BANDWIDTH            RTP_STATS_BANDWIDTH
  57 #define DEFAULT_RTCP_FRACTION        (RTP_STATS_RTCP_FRACTION * RTP_STATS_BANDWIDTH)
  58 #define DEFAULT_RTCP_RR_BANDWIDTH    -1
  59 #define DEFAULT_RTCP_RS_BANDWIDTH    -1
  60 #define DEFAULT_RTCP_MTU             1400
  61 #define DEFAULT_SDES                 NULL
  62 #define DEFAULT_NUM_SOURCES          0
  63 #define DEFAULT_NUM_ACTIVE_SOURCES   0
  64 #define DEFAULT_SOURCES              NULL
  65 #define DEFAULT_RTCP_MIN_INTERVAL    (RTP_STATS_MIN_INTERVAL * GST_SECOND)
  66 #define DEFAULT_RTCP_FEEDBACK_RETENTION_WINDOW (2 * GST_SECOND)
  67 #define DEFAULT_RTCP_IMMEDIATE_FEEDBACK_THRESHOLD (3)
  68 #define DEFAULT_PROBATION            RTP_DEFAULT_PROBATION
  69
  70 enum
  71 {
  72   PROP_0,
  73   PROP_INTERNAL_SSRC,
  74   PROP_INTERNAL_SOURCE,
  75   PROP_BANDWIDTH,
  76   PROP_RTCP_FRACTION,
  77   PROP_RTCP_RR_BANDWIDTH,
  78   PROP_RTCP_RS_BANDWIDTH,
  79   PROP_RTCP_MTU,
  80   PROP_SDES,
  81   PROP_NUM_SOURCES,
  82   PROP_NUM_ACTIVE_SOURCES,
  83   PROP_SOURCES,
  84   PROP_FAVOR_NEW,
  85   PROP_RTCP_MIN_INTERVAL,
  86   PROP_RTCP_FEEDBACK_RETENTION_WINDOW,
  87   PROP_RTCP_IMMEDIATE_FEEDBACK_THRESHOLD,
  88   PROP_PROBATION,
  89   PROP_STATS,
  90   PROP_LAST
  91 };
  92
  93 /* update average packet size */
  94 #define INIT_AVG(avg, val) \
  95    (avg) = (val);
  96 #define UPDATE_AVG(avg, val)            \
  97   if ((avg) == 0)                       \
  98    (avg) = (val);                       \
  99   else                                  \
 100    (avg) = ((val) + (15 * (avg))) >> 4;
 101
 102
 103 /* GObject vmethods */
 104 static void rtp_session_finalize (GObject * object);
 105 static void rtp_session_set_property (GObject * object, guint prop_id,
 106     const GValue * value, GParamSpec * pspec);
 107 static void rtp_session_get_property (GObject * object, guint prop_id,
 108     GValue * value, GParamSpec * pspec);
 109
 110 static void rtp_session_send_rtcp (RTPSession * sess, GstClockTime max_delay);
 111
 112 static guint rtp_session_signals[LAST_SIGNAL] = { 0 };
 113
 114 G_DEFINE_TYPE (RTPSession, rtp_session, G_TYPE_OBJECT);
 115
 116 static guint32 rtp_session_create_new_ssrc (RTPSession * sess);
 117 static RTPSource *obtain_source (RTPSession * sess, guint32 ssrc,
 118     gboolean * created, RTPPacketInfo * pinfo, gboolean rtp);
 119 static RTPSource *obtain_internal_source (RTPSession * sess,
 120     guint32 ssrc, gboolean * created);
 121 static GstFlowReturn rtp_session_schedule_bye_locked (RTPSession * sess,
 122     GstClockTime current_time);
 123 static GstClockTime calculate_rtcp_interval (RTPSession * sess,
 124     gboolean deterministic, gboolean first);
 125
 126 static gboolean
 127 accumulate_trues (GSignalInvocationHint * ihint, GValue * return_accu,
 128     const GValue * handler_return, gpointer data)
 129 {
 130   if (g_value_get_boolean (handler_return))
 131     g_value_set_boolean (return_accu, TRUE);
 132
 133   return TRUE;
 134 }
 135
 136 static void
 137 rtp_session_class_init (RTPSessionClass * klass)
 138 {
 139   GObjectClass *gobject_class;
 140
 141   gobject_class = (GObjectClass *) klass;
 142
 143   gobject_class->finalize = rtp_session_finalize;
 144   gobject_class->set_property = rtp_session_set_property;
 145   gobject_class->get_property = rtp_session_get_property;
 146
 147   /**
 148    * RTPSession::get-source-by-ssrc:
 149    * @session: the object which received the signal
 150    * @ssrc: the SSRC of the RTPSource
 151    *
 152    * Request the #RTPSource object with SSRC @ssrc in @session.
 153    */
 154   rtp_session_signals[SIGNAL_GET_SOURCE_BY_SSRC] =
 155       g_signal_new ("get-source-by-ssrc", G_TYPE_FROM_CLASS (klass),
 156       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION, G_STRUCT_OFFSET (RTPSessionClass,
 157           get_source_by_ssrc), NULL, NULL, g_cclosure_marshal_generic,
 158       RTP_TYPE_SOURCE, 1, G_TYPE_UINT);
 159
 160   /**
 161    * RTPSession::on-new-ssrc:
 162    * @session: the object which received the signal
 163    * @src: the new RTPSource
 164    *
 165    * Notify of a new SSRC that entered @session.
 166    */
 167   rtp_session_signals[SIGNAL_ON_NEW_SSRC] =
 168       g_signal_new ("on-new-ssrc", G_TYPE_FROM_CLASS (klass),
 169       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_new_ssrc),
 170       NULL, NULL, g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1,
 171       RTP_TYPE_SOURCE);
 172   /**
 173    * RTPSession::on-ssrc-collision:
 174    * @session: the object which received the signal
 175    * @src: the #RTPSource that caused a collision
 176    *
 177    * Notify when we have an SSRC collision
 178    */
 179   rtp_session_signals[SIGNAL_ON_SSRC_COLLISION] =
 180       g_signal_new ("on-ssrc-collision", G_TYPE_FROM_CLASS (klass),
 181       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_ssrc_collision),
 182       NULL, NULL, g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1,
 183       RTP_TYPE_SOURCE);
 184   /**
 185    * RTPSession::on-ssrc-validated:
 186    * @session: the object which received the signal
 187    * @src: the new validated RTPSource
 188    *
 189    * Notify of a new SSRC that became validated.
 190    */
 191   rtp_session_signals[SIGNAL_ON_SSRC_VALIDATED] =
 192       g_signal_new ("on-ssrc-validated", G_TYPE_FROM_CLASS (klass),
 193       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_ssrc_validated),
 194       NULL, NULL, g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1,
 195       RTP_TYPE_SOURCE);
 196   /**
 197    * RTPSession::on-ssrc-active:
 198    * @session: the object which received the signal
 199    * @src: the active RTPSource
 200    *
 201    * Notify of a SSRC that is active, i.e., sending RTCP.
 202    */
 203   rtp_session_signals[SIGNAL_ON_SSRC_ACTIVE] =
 204       g_signal_new ("on-ssrc-active", G_TYPE_FROM_CLASS (klass),
 205       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_ssrc_active),
 206       NULL, NULL, g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1,
 207       RTP_TYPE_SOURCE);
 208   /**
 209    * RTPSession::on-ssrc-sdes:
 210    * @session: the object which received the signal
 211    * @src: the RTPSource
 212    *
 213    * Notify that a new SDES was received for SSRC.
 214    */
 215   rtp_session_signals[SIGNAL_ON_SSRC_SDES] =
 216       g_signal_new ("on-ssrc-sdes", G_TYPE_FROM_CLASS (klass),
 217       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_ssrc_sdes),
 218       NULL, NULL, g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1,
 219       RTP_TYPE_SOURCE);
 220   /**
 221    * RTPSession::on-bye-ssrc:
 222    * @session: the object which received the signal
 223    * @src: the RTPSource that went away
 224    *
 225    * Notify of an SSRC that became inactive because of a BYE packet.
 226    */
 227   rtp_session_signals[SIGNAL_ON_BYE_SSRC] =
 228       g_signal_new ("on-bye-ssrc", G_TYPE_FROM_CLASS (klass),
 229       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_bye_ssrc),
 230       NULL, NULL, g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1,
 231       RTP_TYPE_SOURCE);
 232   /**
 233    * RTPSession::on-bye-timeout:
 234    * @session: the object which received the signal
 235    * @src: the RTPSource that timed out
 236    *
 237    * Notify of an SSRC that has timed out because of BYE
 238    */
 239   rtp_session_signals[SIGNAL_ON_BYE_TIMEOUT] =
 240       g_signal_new ("on-bye-timeout", G_TYPE_FROM_CLASS (klass),
 241       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_bye_timeout),
 242       NULL, NULL, g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1,
 243       RTP_TYPE_SOURCE);
 244   /**
 245    * RTPSession::on-timeout:
 246    * @session: the object which received the signal
 247    * @src: the RTPSource that timed out
 248    *
 249    * Notify of an SSRC that has timed out
 250    */
 251   rtp_session_signals[SIGNAL_ON_TIMEOUT] =
 252       g_signal_new ("on-timeout", G_TYPE_FROM_CLASS (klass),
 253       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_timeout),
 254       NULL, NULL, g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1,
 255       RTP_TYPE_SOURCE);
 256   /**
 257    * RTPSession::on-sender-timeout:
 258    * @session: the object which received the signal
 259    * @src: the RTPSource that timed out
 260    *
 261    * Notify of an SSRC that was a sender but timed out and became a receiver.
 262    */
 263   rtp_session_signals[SIGNAL_ON_SENDER_TIMEOUT] =
 264       g_signal_new ("on-sender-timeout", G_TYPE_FROM_CLASS (klass),
 265       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_sender_timeout),
 266       NULL, NULL, g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1,
 267       RTP_TYPE_SOURCE);
 268
 269   /**
 270    * RTPSession::on-sending-rtcp
 271    * @session: the object which received the signal
 272    * @buffer: the #GstBuffer containing the RTCP packet about to be sent
 273    * @early: %TRUE if the packet is early, %FALSE if it is regular
 274    *
 275    * This signal is emitted before sending an RTCP packet, it can be used
 276    * to add extra RTCP Packets.
 277    *
 278    * Returns: %TRUE if the RTCP buffer should NOT be suppressed, %FALSE
 279    * if suppressing it is acceptable
 280    */
 281   rtp_session_signals[SIGNAL_ON_SENDING_RTCP] =
 282       g_signal_new ("on-sending-rtcp", G_TYPE_FROM_CLASS (klass),
 283       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_sending_rtcp),
 284       accumulate_trues, NULL, g_cclosure_marshal_generic, G_TYPE_BOOLEAN, 2,
 285       GST_TYPE_BUFFER | G_SIGNAL_TYPE_STATIC_SCOPE, G_TYPE_BOOLEAN);
 286
 287   /**
 288    * RTPSession::on-feedback-rtcp:
 289    * @session: the object which received the signal
 290    * @type: Type of RTCP packet, will be %GST_RTCP_TYPE_RTPFB or
 291    *  %GST_RTCP_TYPE_RTPFB
 292    * @fbtype: The type of RTCP FB packet, probably part of #GstRTCPFBType
 293    * @sender_ssrc: The SSRC of the sender
 294    * @media_ssrc: The SSRC of the media this refers to
 295    * @fci: a #GstBuffer with the FCI data from the FB packet or %NULL if
 296    * there was no FCI
 297    *
 298    * Notify that a RTCP feedback packet has been received
 299    */
 300   rtp_session_signals[SIGNAL_ON_FEEDBACK_RTCP] =
 301       g_signal_new ("on-feedback-rtcp", G_TYPE_FROM_CLASS (klass),
 302       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (RTPSessionClass, on_feedback_rtcp),
 303       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 5, G_TYPE_UINT,
 304       G_TYPE_UINT, G_TYPE_UINT, G_TYPE_UINT, GST_TYPE_BUFFER);
 305
 306   /**
 307    * RTPSession::send-rtcp:
 308    * @session: the object which received the signal
 309    * @max_delay: The maximum delay after which the feedback will not be useful
 310    *  anymore
 311    *
 312    * Requests that the #RTPSession initiate a new RTCP packet as soon as
 313    * possible within the requested delay.
 314    */
 315   rtp_session_signals[SIGNAL_SEND_RTCP] =
 316       g_signal_new ("send-rtcp", G_TYPE_FROM_CLASS (klass),
 317       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 318       G_STRUCT_OFFSET (RTPSessionClass, send_rtcp), NULL, NULL,
 319       g_cclosure_marshal_generic, G_TYPE_NONE, 1, G_TYPE_UINT64);
 320
 321   g_object_class_install_property (gobject_class, PROP_INTERNAL_SSRC,
 322       g_param_spec_uint ("internal-ssrc", "Internal SSRC",
 323           "The internal SSRC used for the session (deprecated)",
 324           0, G_MAXUINT, 0, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 325
 326   g_object_class_install_property (gobject_class, PROP_INTERNAL_SOURCE,
 327       g_param_spec_object ("internal-source", "Internal Source",
 328           "The internal source element of the session (deprecated)",
 329           RTP_TYPE_SOURCE, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 330
 331   g_object_class_install_property (gobject_class, PROP_BANDWIDTH,
 332       g_param_spec_double ("bandwidth", "Bandwidth",
 333           "The bandwidth of the session (0 for auto-discover)",
 334           0.0, G_MAXDOUBLE, DEFAULT_BANDWIDTH,
 335           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 336
 337   g_object_class_install_property (gobject_class, PROP_RTCP_FRACTION,
 338       g_param_spec_double ("rtcp-fraction", "RTCP Fraction",
 339           "The fraction of the bandwidth used for RTCP (or as a real fraction of the RTP bandwidth if < 1)",
 340           0.0, G_MAXDOUBLE, DEFAULT_RTCP_FRACTION,
 341           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 342
 343   g_object_class_install_property (gobject_class, PROP_RTCP_RR_BANDWIDTH,
 344       g_param_spec_int ("rtcp-rr-bandwidth", "RTCP RR bandwidth",
 345           "The RTCP bandwidth used for receivers in bytes per second (-1 = default)",
 346           -1, G_MAXINT, DEFAULT_RTCP_RR_BANDWIDTH,
 347           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 348
 349   g_object_class_install_property (gobject_class, PROP_RTCP_RS_BANDWIDTH,
 350       g_param_spec_int ("rtcp-rs-bandwidth", "RTCP RS bandwidth",
 351           "The RTCP bandwidth used for senders in bytes per second (-1 = default)",
 352           -1, G_MAXINT, DEFAULT_RTCP_RS_BANDWIDTH,
 353           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 354
 355   g_object_class_install_property (gobject_class, PROP_RTCP_MTU,
 356       g_param_spec_uint ("rtcp-mtu", "RTCP MTU",
 357           "The maximum size of the RTCP packets",
 358           16, G_MAXINT16, DEFAULT_RTCP_MTU,
 359           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 360
 361   g_object_class_install_property (gobject_class, PROP_SDES,
 362       g_param_spec_boxed ("sdes", "SDES",
 363           "The SDES items of this session",
 364           GST_TYPE_STRUCTURE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 365
 366   g_object_class_install_property (gobject_class, PROP_NUM_SOURCES,
 367       g_param_spec_uint ("num-sources", "Num Sources",
 368           "The number of sources in the session", 0, G_MAXUINT,
 369           DEFAULT_NUM_SOURCES, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 370
 371   g_object_class_install_property (gobject_class, PROP_NUM_ACTIVE_SOURCES,
 372       g_param_spec_uint ("num-active-sources", "Num Active Sources",
 373           "The number of active sources in the session", 0, G_MAXUINT,
 374           DEFAULT_NUM_ACTIVE_SOURCES,
 375           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 376   /**
 377    * RTPSource::sources
 378    *
 379    * Get a GValue Array of all sources in the session.
 380    *
 381    * <example>
 382    * <title>Getting the #RTPSources of a session
 383    * <programlisting>
 384    * {
 385    *   GValueArray *arr;
 386    *   GValue *val;
 387    *   guint i;
 388    *
 389    *   g_object_get (sess, "sources", &arr, NULL);
 390    *
 391    *   for (i = 0; i < arr->n_values; i++) {
 392    *     RTPSource *source;
 393    *
 394    *     val = g_value_array_get_nth (arr, i);
 395    *     source = g_value_get_object (val);
 396    *   }
 397    *   g_value_array_free (arr);
 398    * }
 399    * </programlisting>
 400    * </example>
 401    */
 402   g_object_class_install_property (gobject_class, PROP_SOURCES,
 403       g_param_spec_boxed ("sources", "Sources",
 404           "An array of all known sources in the session",
 405           G_TYPE_VALUE_ARRAY, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 406
 407   g_object_class_install_property (gobject_class, PROP_FAVOR_NEW,
 408       g_param_spec_boolean ("favor-new", "Favor new sources",
 409           "Resolve SSRC conflict in favor of new sources", FALSE,
 410           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 411
 412   g_object_class_install_property (gobject_class, PROP_RTCP_MIN_INTERVAL,
 413       g_param_spec_uint64 ("rtcp-min-interval", "Minimum RTCP interval",
 414           "Minimum interval between Regular RTCP packet (in ns)",
 415           0, G_MAXUINT64, DEFAULT_RTCP_MIN_INTERVAL,
 416           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 417
 418   g_object_class_install_property (gobject_class,
 419       PROP_RTCP_FEEDBACK_RETENTION_WINDOW,
 420       g_param_spec_uint64 ("rtcp-feedback-retention-window",
 421           "RTCP Feedback retention window",
 422           "Duration during which RTCP Feedback packets are retained (in ns)",
 423           0, G_MAXUINT64, DEFAULT_RTCP_FEEDBACK_RETENTION_WINDOW,
 424           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 425
 426   g_object_class_install_property (gobject_class,
 427       PROP_RTCP_IMMEDIATE_FEEDBACK_THRESHOLD,
 428       g_param_spec_uint ("rtcp-immediate-feedback-threshold",
 429           "RTCP Immediate Feedback threshold",
 430           "The maximum number of members of a RTP session for which immediate"
 431           " feedback is used",
 432           0, G_MAXUINT, DEFAULT_RTCP_IMMEDIATE_FEEDBACK_THRESHOLD,
 433           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 434
 435   g_object_class_install_property (gobject_class, PROP_PROBATION,
 436       g_param_spec_uint ("probation", "Number of probations",
 437           "Consecutive packet sequence numbers to accept the source",
 438           0, G_MAXUINT, DEFAULT_PROBATION,
 439           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 440
 441   /**
 442    * RTPSession::stats:
 443    *
 444    * Various session statistics. This property returns a GstStructure
 445    * with name application/x-rtp-session-stats with the following fields:
 446    *
 447    *  "rtx-drop-count"  G_TYPE_UINT   The number of retransmission events
 448    *      dropped (due to bandwidth constraints)
 449    *  "sent-nack-count" G_TYPE_UINT   Number of NACKs sent
 450    *  "recv-nack-count" G_TYPE_UINT   Number of NACKs received
 451    *
 452    * Since: 1.4
 453    */
 454   g_object_class_install_property (gobject_class, PROP_STATS,
 455       g_param_spec_boxed ("stats", "Statistics",
 456           "Various statistics", GST_TYPE_STRUCTURE,
 457           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 458
 459   klass->get_source_by_ssrc =
 460       GST_DEBUG_FUNCPTR (rtp_session_get_source_by_ssrc);
 461   klass->send_rtcp = GST_DEBUG_FUNCPTR (rtp_session_send_rtcp);
 462
 463   GST_DEBUG_CATEGORY_INIT (rtp_session_debug, "rtpsession", 0, "RTP Session");
 464 }
 465
 466 static void
 467 rtp_session_init (RTPSession * sess)
 468 {
 469   gint i;
 470   gchar *str;
 471
 472   g_mutex_init (&sess->lock);
 473   sess->key = g_random_int ();
 474   sess->mask_idx = 0;
 475   sess->mask = 0;
 476
 477   for (i = 0; i < 32; i++) {
 478     sess->ssrcs[i] =
 479         g_hash_table_new_full (NULL, NULL, NULL,
 480         (GDestroyNotify) g_object_unref);
 481   }
 482
 483   rtp_stats_init_defaults (&sess->stats);
 484   INIT_AVG (sess->stats.avg_rtcp_packet_size, 100);
 485   rtp_stats_set_min_interval (&sess->stats,
 486       (gdouble) DEFAULT_RTCP_MIN_INTERVAL / GST_SECOND);
 487
 488   sess->recalc_bandwidth = TRUE;
 489   sess->bandwidth = DEFAULT_BANDWIDTH;
 490   sess->rtcp_bandwidth = DEFAULT_RTCP_FRACTION;
 491   sess->rtcp_rr_bandwidth = DEFAULT_RTCP_RR_BANDWIDTH;
 492   sess->rtcp_rs_bandwidth = DEFAULT_RTCP_RS_BANDWIDTH;
 493
 494   /* default UDP header length */
 495   sess->header_len = 28;
 496   sess->mtu = DEFAULT_RTCP_MTU;
 497
 498   sess->probation = DEFAULT_PROBATION;
 499
 500   /* some default SDES entries */
 501   sess->sdes = gst_structure_new_empty ("application/x-rtp-source-sdes");
 502
 503   /* we do not want to leak details like the username or hostname here */
 504   str = g_strdup_printf ("user%u@host-%x", g_random_int (), g_random_int ());
 505   gst_structure_set (sess->sdes, "cname", G_TYPE_STRING, str, NULL);
 506   g_free (str);
 507
 508 #if 0
 509   /* we do not want to leak the user's real name here */
 510   str = g_strdup_printf ("Anon%u", g_random_int ());
 511   gst_structure_set (sdes, "name", G_TYPE_STRING, str, NULL);
 512   g_free (str);
 513 #endif
 514
 515   gst_structure_set (sess->sdes, "tool", G_TYPE_STRING, "GStreamer", NULL);
 516
 517   /* this is the SSRC we suggest */
 518   sess->suggested_ssrc = rtp_session_create_new_ssrc (sess);
 519
 520   sess->first_rtcp = TRUE;
 521   sess->next_rtcp_check_time = GST_CLOCK_TIME_NONE;
 522
 523   sess->allow_early = TRUE;
 524   sess->next_early_rtcp_time = GST_CLOCK_TIME_NONE;
 525   sess->rtcp_feedback_retention_window = DEFAULT_RTCP_FEEDBACK_RETENTION_WINDOW;
 526   sess->rtcp_immediate_feedback_threshold =
 527       DEFAULT_RTCP_IMMEDIATE_FEEDBACK_THRESHOLD;
 528
 529   sess->last_keyframe_request = GST_CLOCK_TIME_NONE;
 530
 531   sess->is_doing_ptp = TRUE;
 532 }
 533
 534 static void
 535 rtp_session_finalize (GObject * object)
 536 {
 537   RTPSession *sess;
 538   gint i;
 539
 540   sess = RTP_SESSION_CAST (object);
 541
 542   gst_structure_free (sess->sdes);
 543
 544   g_list_free_full (sess->conflicting_addresses,
 545       (GDestroyNotify) rtp_conflicting_address_free);
 546
 547   for (i = 0; i < 32; i++)
 548     g_hash_table_destroy (sess->ssrcs[i]);
 549
 550   g_mutex_clear (&sess->lock);
 551
 552   G_OBJECT_CLASS (rtp_session_parent_class)->finalize (object);
 553 }
 554
 555 static void
 556 copy_source (gpointer key, RTPSource * source, GValueArray * arr)
 557 {
 558   GValue value = { 0 };
 559
 560   g_value_init (&value, RTP_TYPE_SOURCE);
 561   g_value_take_object (&value, source);
 562   /* copies the value */
 563   g_value_array_append (arr, &value);
 564 }
 565
 566 static GValueArray *
 567 rtp_session_create_sources (RTPSession * sess)
 568 {
 569   GValueArray *res;
 570   guint size;
 571
 572   RTP_SESSION_LOCK (sess);
 573   /* get number of elements in the table */
 574   size = g_hash_table_size (sess->ssrcs[sess->mask_idx]);
 575   /* create the result value array */
 576   res = g_value_array_new (size);
 577
 578   /* and copy all values into the array */
 579   g_hash_table_foreach (sess->ssrcs[sess->mask_idx], (GHFunc) copy_source, res);
 580   RTP_SESSION_UNLOCK (sess);
 581
 582   return res;
 583 }
 584
 585 static GstStructure *
 586 rtp_session_create_stats (RTPSession * sess)
 587 {
 588   GstStructure *s;
 589
 590   s = gst_structure_new ("application/x-rtp-session-stats",
 591       "rtx-drop-count", G_TYPE_UINT, sess->stats.nacks_dropped,
 592       "sent-nack-count", G_TYPE_UINT, sess->stats.nacks_sent,
 593       "recv-nack-count", G_TYPE_UINT, sess->stats.nacks_received, NULL);
 594
 595   return s;
 596 }
 597
 598 static void
 599 rtp_session_set_property (GObject * object, guint prop_id,
 600     const GValue * value, GParamSpec * pspec)
 601 {
 602   RTPSession *sess;
 603
 604   sess = RTP_SESSION (object);
 605
 606   switch (prop_id) {
 607     case PROP_INTERNAL_SSRC:
 608       RTP_SESSION_LOCK (sess);
 609       sess->suggested_ssrc = g_value_get_uint (value);
 610       RTP_SESSION_UNLOCK (sess);
 611       if (sess->callbacks.reconfigure)
 612         sess->callbacks.reconfigure (sess, sess->reconfigure_user_data);
 613       break;
 614     case PROP_BANDWIDTH:
 615       RTP_SESSION_LOCK (sess);
 616       sess->bandwidth = g_value_get_double (value);
 617       sess->recalc_bandwidth = TRUE;
 618       RTP_SESSION_UNLOCK (sess);
 619       break;
 620     case PROP_RTCP_FRACTION:
 621       RTP_SESSION_LOCK (sess);
 622       sess->rtcp_bandwidth = g_value_get_double (value);
 623       sess->recalc_bandwidth = TRUE;
 624       RTP_SESSION_UNLOCK (sess);
 625       break;
 626     case PROP_RTCP_RR_BANDWIDTH:
 627       RTP_SESSION_LOCK (sess);
 628       sess->rtcp_rr_bandwidth = g_value_get_int (value);
 629       sess->recalc_bandwidth = TRUE;
 630       RTP_SESSION_UNLOCK (sess);
 631       break;
 632     case PROP_RTCP_RS_BANDWIDTH:
 633       RTP_SESSION_LOCK (sess);
 634       sess->rtcp_rs_bandwidth = g_value_get_int (value);
 635       sess->recalc_bandwidth = TRUE;
 636       RTP_SESSION_UNLOCK (sess);
 637       break;
 638     case PROP_RTCP_MTU:
 639       sess->mtu = g_value_get_uint (value);
 640       break;
 641     case PROP_SDES:
 642       rtp_session_set_sdes_struct (sess, g_value_get_boxed (value));
 643       break;
 644     case PROP_FAVOR_NEW:
 645       sess->favor_new = g_value_get_boolean (value);
 646       break;
 647     case PROP_RTCP_MIN_INTERVAL:
 648       rtp_stats_set_min_interval (&sess->stats,
 649           (gdouble) g_value_get_uint64 (value) / GST_SECOND);
 650       /* trigger reconsideration */
 651       RTP_SESSION_LOCK (sess);
 652       sess->next_rtcp_check_time = 0;
 653       RTP_SESSION_UNLOCK (sess);
 654       if (sess->callbacks.reconsider)
 655         sess->callbacks.reconsider (sess, sess->reconsider_user_data);
 656       break;
 657     case PROP_RTCP_IMMEDIATE_FEEDBACK_THRESHOLD:
 658       sess->rtcp_immediate_feedback_threshold = g_value_get_uint (value);
 659       break;
 660     case PROP_PROBATION:
 661       sess->probation = g_value_get_uint (value);
 662       break;
 663     default:
 664       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 665       break;
 666   }
 667 }
 668
 669 static void
 670 rtp_session_get_property (GObject * object, guint prop_id,
 671     GValue * value, GParamSpec * pspec)
 672 {
 673   RTPSession *sess;
 674
 675   sess = RTP_SESSION (object);
 676
 677   switch (prop_id) {
 678     case PROP_INTERNAL_SSRC:
 679       g_value_set_uint (value, rtp_session_suggest_ssrc (sess));
 680       break;
 681     case PROP_INTERNAL_SOURCE:
 682       /* FIXME, return a random source */
 683       g_value_set_object (value, NULL);
 684       break;
 685     case PROP_BANDWIDTH:
 686       g_value_set_double (value, sess->bandwidth);
 687       break;
 688     case PROP_RTCP_FRACTION:
 689       g_value_set_double (value, sess->rtcp_bandwidth);
 690       break;
 691     case PROP_RTCP_RR_BANDWIDTH:
 692       g_value_set_int (value, sess->rtcp_rr_bandwidth);
 693       break;
 694     case PROP_RTCP_RS_BANDWIDTH:
 695       g_value_set_int (value, sess->rtcp_rs_bandwidth);
 696       break;
 697     case PROP_RTCP_MTU:
 698       g_value_set_uint (value, sess->mtu);
 699       break;
 700     case PROP_SDES:
 701       g_value_take_boxed (value, rtp_session_get_sdes_struct (sess));
 702       break;
 703     case PROP_NUM_SOURCES:
 704       g_value_set_uint (value, rtp_session_get_num_sources (sess));
 705       break;
 706     case PROP_NUM_ACTIVE_SOURCES:
 707       g_value_set_uint (value, rtp_session_get_num_active_sources (sess));
 708       break;
 709     case PROP_SOURCES:
 710       g_value_take_boxed (value, rtp_session_create_sources (sess));
 711       break;
 712     case PROP_FAVOR_NEW:
 713       g_value_set_boolean (value, sess->favor_new);
 714       break;
 715     case PROP_RTCP_MIN_INTERVAL:
 716       g_value_set_uint64 (value, sess->stats.min_interval * GST_SECOND);
 717       break;
 718     case PROP_RTCP_IMMEDIATE_FEEDBACK_THRESHOLD:
 719       g_value_set_uint (value, sess->rtcp_immediate_feedback_threshold);
 720       break;
 721     case PROP_PROBATION:
 722       g_value_set_uint (value, sess->probation);
 723       break;
 724     case PROP_STATS:
 725       g_value_take_boxed (value, rtp_session_create_stats (sess));
 726       break;
 727     default:
 728       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 729       break;
 730   }
 731 }
 732
 733 static void
 734 on_new_ssrc (RTPSession * sess, RTPSource * source)
 735 {
 736   g_object_ref (source);
 737   RTP_SESSION_UNLOCK (sess);
 738   g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_NEW_SSRC], 0, source);
 739   RTP_SESSION_LOCK (sess);
 740   g_object_unref (source);
 741 }
 742
 743 static void
 744 on_ssrc_collision (RTPSession * sess, RTPSource * source)
 745 {
 746   g_object_ref (source);
 747   RTP_SESSION_UNLOCK (sess);
 748   g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_SSRC_COLLISION], 0,
 749       source);
 750   RTP_SESSION_LOCK (sess);
 751   g_object_unref (source);
 752 }
 753
 754 static void
 755 on_ssrc_validated (RTPSession * sess, RTPSource * source)
 756 {
 757   g_object_ref (source);
 758   RTP_SESSION_UNLOCK (sess);
 759   g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_SSRC_VALIDATED], 0,
 760       source);
 761   RTP_SESSION_LOCK (sess);
 762   g_object_unref (source);
 763 }
 764
 765 static void
 766 on_ssrc_active (RTPSession * sess, RTPSource * source)
 767 {
 768   g_object_ref (source);
 769   RTP_SESSION_UNLOCK (sess);
 770   g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_SSRC_ACTIVE], 0, source);
 771   RTP_SESSION_LOCK (sess);
 772   g_object_unref (source);
 773 }
 774
 775 static void
 776 on_ssrc_sdes (RTPSession * sess, RTPSource * source)
 777 {
 778   g_object_ref (source);
 779   GST_DEBUG ("SDES changed for SSRC %08x", source->ssrc);
 780   RTP_SESSION_UNLOCK (sess);
 781   g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_SSRC_SDES], 0, source);
 782   RTP_SESSION_LOCK (sess);
 783   g_object_unref (source);
 784 }
 785
 786 static void
 787 on_bye_ssrc (RTPSession * sess, RTPSource * source)
 788 {
 789   g_object_ref (source);
 790   RTP_SESSION_UNLOCK (sess);
 791   g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_BYE_SSRC], 0, source);
 792   RTP_SESSION_LOCK (sess);
 793   g_object_unref (source);
 794 }
 795
 796 static void
 797 on_bye_timeout (RTPSession * sess, RTPSource * source)
 798 {
 799   g_object_ref (source);
 800   RTP_SESSION_UNLOCK (sess);
 801   g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_BYE_TIMEOUT], 0, source);
 802   RTP_SESSION_LOCK (sess);
 803   g_object_unref (source);
 804 }
 805
 806 static void
 807 on_timeout (RTPSession * sess, RTPSource * source)
 808 {
 809   g_object_ref (source);
 810   RTP_SESSION_UNLOCK (sess);
 811   g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_TIMEOUT], 0, source);
 812   RTP_SESSION_LOCK (sess);
 813   g_object_unref (source);
 814 }
 815
 816 static void
 817 on_sender_timeout (RTPSession * sess, RTPSource * source)
 818 {
 819   g_object_ref (source);
 820   RTP_SESSION_UNLOCK (sess);
 821   g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_SENDER_TIMEOUT], 0,
 822       source);
 823   RTP_SESSION_LOCK (sess);
 824   g_object_unref (source);
 825 }
 826
 827 /**
 828  * rtp_session_new:
 829  *
 830  * Create a new session object.
 831  *
 832  * Returns: a new #RTPSession. g_object_unref() after usage.
 833  */
 834 RTPSession *
 835 rtp_session_new (void)
 836 {
 837   RTPSession *sess;
 838
 839   sess = g_object_new (RTP_TYPE_SESSION, NULL);
 840
 841   return sess;
 842 }
 843
 844 /**
 845  * rtp_session_set_callbacks:
 846  * @sess: an #RTPSession
 847  * @callbacks: callbacks to configure
 848  * @user_data: user data passed in the callbacks
 849  *
 850  * Configure a set of callbacks to be notified of actions.
 851  */
 852 void
 853 rtp_session_set_callbacks (RTPSession * sess, RTPSessionCallbacks * callbacks,
 854     gpointer user_data)
 855 {
 856   g_return_if_fail (RTP_IS_SESSION (sess));
 857
 858   if (callbacks->process_rtp) {
 859     sess->callbacks.process_rtp = callbacks->process_rtp;
 860     sess->process_rtp_user_data = user_data;
 861   }
 862   if (callbacks->send_rtp) {
 863     sess->callbacks.send_rtp = callbacks->send_rtp;
 864     sess->send_rtp_user_data = user_data;
 865   }
 866   if (callbacks->send_rtcp) {
 867     sess->callbacks.send_rtcp = callbacks->send_rtcp;
 868     sess->send_rtcp_user_data = user_data;
 869   }
 870   if (callbacks->sync_rtcp) {
 871     sess->callbacks.sync_rtcp = callbacks->sync_rtcp;
 872     sess->sync_rtcp_user_data = user_data;
 873   }
 874   if (callbacks->clock_rate) {
 875     sess->callbacks.clock_rate = callbacks->clock_rate;
 876     sess->clock_rate_user_data = user_data;
 877   }
 878   if (callbacks->reconsider) {
 879     sess->callbacks.reconsider = callbacks->reconsider;
 880     sess->reconsider_user_data = user_data;
 881   }
 882   if (callbacks->request_key_unit) {
 883     sess->callbacks.request_key_unit = callbacks->request_key_unit;
 884     sess->request_key_unit_user_data = user_data;
 885   }
 886   if (callbacks->request_time) {
 887     sess->callbacks.request_time = callbacks->request_time;
 888     sess->request_time_user_data = user_data;
 889   }
 890   if (callbacks->notify_nack) {
 891     sess->callbacks.notify_nack = callbacks->notify_nack;
 892     sess->notify_nack_user_data = user_data;
 893   }
 894   if (callbacks->reconfigure) {
 895     sess->callbacks.reconfigure = callbacks->reconfigure;
 896     sess->reconfigure_user_data = user_data;
 897   }
 898 }
 899
 900 /**
 901  * rtp_session_set_process_rtp_callback:
 902  * @sess: an #RTPSession
 903  * @callback: callback to set
 904  * @user_data: user data passed in the callback
 905  *
 906  * Configure only the process_rtp callback to be notified of the process_rtp action.
 907  */
 908 void
 909 rtp_session_set_process_rtp_callback (RTPSession * sess,
 910     RTPSessionProcessRTP callback, gpointer user_data)
 911 {
 912   g_return_if_fail (RTP_IS_SESSION (sess));
 913
 914   sess->callbacks.process_rtp = callback;
 915   sess->process_rtp_user_data = user_data;
 916 }
 917
 918 /**
 919  * rtp_session_set_send_rtp_callback:
 920  * @sess: an #RTPSession
 921  * @callback: callback to set
 922  * @user_data: user data passed in the callback
 923  *
 924  * Configure only the send_rtp callback to be notified of the send_rtp action.
 925  */
 926 void
 927 rtp_session_set_send_rtp_callback (RTPSession * sess,
 928     RTPSessionSendRTP callback, gpointer user_data)
 929 {
 930   g_return_if_fail (RTP_IS_SESSION (sess));
 931
 932   sess->callbacks.send_rtp = callback;
 933   sess->send_rtp_user_data = user_data;
 934 }
 935
 936 /**
 937  * rtp_session_set_send_rtcp_callback:
 938  * @sess: an #RTPSession
 939  * @callback: callback to set
 940  * @user_data: user data passed in the callback
 941  *
 942  * Configure only the send_rtcp callback to be notified of the send_rtcp action.
 943  */
 944 void
 945 rtp_session_set_send_rtcp_callback (RTPSession * sess,
 946     RTPSessionSendRTCP callback, gpointer user_data)
 947 {
 948   g_return_if_fail (RTP_IS_SESSION (sess));
 949
 950   sess->callbacks.send_rtcp = callback;
 951   sess->send_rtcp_user_data = user_data;
 952 }
 953
 954 /**
 955  * rtp_session_set_sync_rtcp_callback:
 956  * @sess: an #RTPSession
 957  * @callback: callback to set
 958  * @user_data: user data passed in the callback
 959  *
 960  * Configure only the sync_rtcp callback to be notified of the sync_rtcp action.
 961  */
 962 void
 963 rtp_session_set_sync_rtcp_callback (RTPSession * sess,
 964     RTPSessionSyncRTCP callback, gpointer user_data)
 965 {
 966   g_return_if_fail (RTP_IS_SESSION (sess));
 967
 968   sess->callbacks.sync_rtcp = callback;
 969   sess->sync_rtcp_user_data = user_data;
 970 }
 971
 972 /**
 973  * rtp_session_set_clock_rate_callback:
 974  * @sess: an #RTPSession
 975  * @callback: callback to set
 976  * @user_data: user data passed in the callback
 977  *
 978  * Configure only the clock_rate callback to be notified of the clock_rate action.
 979  */
 980 void
 981 rtp_session_set_clock_rate_callback (RTPSession * sess,
 982     RTPSessionClockRate callback, gpointer user_data)
 983 {
 984   g_return_if_fail (RTP_IS_SESSION (sess));
 985
 986   sess->callbacks.clock_rate = callback;
 987   sess->clock_rate_user_data = user_data;
 988 }
 989
 990 /**
 991  * rtp_session_set_reconsider_callback:
 992  * @sess: an #RTPSession
 993  * @callback: callback to set
 994  * @user_data: user data passed in the callback
 995  *
 996  * Configure only the reconsider callback to be notified of the reconsider action.
 997  */
 998 void
 999 rtp_session_set_reconsider_callback (RTPSession * sess,
1000     RTPSessionReconsider callback, gpointer user_data)
1001 {
1002   g_return_if_fail (RTP_IS_SESSION (sess));
1003
1004   sess->callbacks.reconsider = callback;
1005   sess->reconsider_user_data = user_data;
1006 }
1007
1008 /**
1009  * rtp_session_set_request_time_callback:
1010  * @sess: an #RTPSession
1011  * @callback: callback to set
1012  * @user_data: user data passed in the callback
1013  *
1014  * Configure only the request_time callback
1015  */
1016 void
1017 rtp_session_set_request_time_callback (RTPSession * sess,
1018     RTPSessionRequestTime callback, gpointer user_data)
1019 {
1020   g_return_if_fail (RTP_IS_SESSION (sess));
1021
1022   sess->callbacks.request_time = callback;
1023   sess->request_time_user_data = user_data;
1024 }
1025
1026 /**
1027  * rtp_session_set_bandwidth:
1028  * @sess: an #RTPSession
1029  * @bandwidth: the bandwidth allocated
1030  *
1031  * Set the session bandwidth in bytes per second.
1032  */
1033 void
1034 rtp_session_set_bandwidth (RTPSession * sess, gdouble bandwidth)
1035 {
1036   g_return_if_fail (RTP_IS_SESSION (sess));
1037
1038   RTP_SESSION_LOCK (sess);
1039   sess->stats.bandwidth = bandwidth;
1040   RTP_SESSION_UNLOCK (sess);
1041 }
1042
1043 /**
1044  * rtp_session_get_bandwidth:
1045  * @sess: an #RTPSession
1046  *
1047  * Get the session bandwidth.
1048  *
1049  * Returns: the session bandwidth.
1050  */
1051 gdouble
1052 rtp_session_get_bandwidth (RTPSession * sess)
1053 {
1054   gdouble result;
1055
1056   g_return_val_if_fail (RTP_IS_SESSION (sess), 0);
1057
1058   RTP_SESSION_LOCK (sess);
1059   result = sess->stats.bandwidth;
1060   RTP_SESSION_UNLOCK (sess);
1061
1062   return result;
1063 }
1064
1065 /**
1066  * rtp_session_set_rtcp_fraction:
1067  * @sess: an #RTPSession
1068  * @bandwidth: the RTCP bandwidth
1069  *
1070  * Set the bandwidth in bytes per second that should be used for RTCP
1071  * messages.
1072  */
1073 void
1074 rtp_session_set_rtcp_fraction (RTPSession * sess, gdouble bandwidth)
1075 {
1076   g_return_if_fail (RTP_IS_SESSION (sess));
1077
1078   RTP_SESSION_LOCK (sess);
1079   sess->stats.rtcp_bandwidth = bandwidth;
1080   RTP_SESSION_UNLOCK (sess);
1081 }
1082
1083 /**
1084  * rtp_session_get_rtcp_fraction:
1085  * @sess: an #RTPSession
1086  *
1087  * Get the session bandwidth used for RTCP.
1088  *
1089  * Returns: The bandwidth used for RTCP messages.
1090  */
1091 gdouble
1092 rtp_session_get_rtcp_fraction (RTPSession * sess)
1093 {
1094   gdouble result;
1095
1096   g_return_val_if_fail (RTP_IS_SESSION (sess), 0.0);
1097
1098   RTP_SESSION_LOCK (sess);
1099   result = sess->stats.rtcp_bandwidth;
1100   RTP_SESSION_UNLOCK (sess);
1101
1102   return result;
1103 }
1104
1105 /**
1106  * rtp_session_get_sdes_struct:
1107  * @sess: an #RTSPSession
1108  *
1109  * Get the SDES data as a #GstStructure
1110  *
1111  * Returns: a GstStructure with SDES items for @sess. This function returns a
1112  * copy of the SDES structure, use gst_structure_free() after usage.
1113  */
1114 GstStructure *
1115 rtp_session_get_sdes_struct (RTPSession * sess)
1116 {
1117   GstStructure *result = NULL;
1118
1119   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1120
1121   RTP_SESSION_LOCK (sess);
1122   if (sess->sdes)
1123     result = gst_structure_copy (sess->sdes);
1124   RTP_SESSION_UNLOCK (sess);
1125
1126   return result;
1127 }
1128
1129 /**
1130  * rtp_session_set_sdes_struct:
1131  * @sess: an #RTSPSession
1132  * @sdes: a #GstStructure
1133  *
1134  * Set the SDES data as a #GstStructure. This function makes a copy of @sdes.
1135  */
1136 void
1137 rtp_session_set_sdes_struct (RTPSession * sess, const GstStructure * sdes)
1138 {
1139   g_return_if_fail (sdes);
1140   g_return_if_fail (RTP_IS_SESSION (sess));
1141
1142   RTP_SESSION_LOCK (sess);
1143   if (sess->sdes)
1144     gst_structure_free (sess->sdes);
1145   sess->sdes = gst_structure_copy (sdes);
1146   RTP_SESSION_UNLOCK (sess);
1147 }
1148
1149 static GstFlowReturn
1150 source_push_rtp (RTPSource * source, gpointer data, RTPSession * session)
1151 {
1152   GstFlowReturn result = GST_FLOW_OK;
1153
1154   if (source->internal) {
1155     GST_LOG ("source %08x pushed sender RTP packet", source->ssrc);
1156
1157     RTP_SESSION_UNLOCK (session);
1158
1159     if (session->callbacks.send_rtp)
1160       result =
1161           session->callbacks.send_rtp (session, source, data,
1162           session->send_rtp_user_data);
1163     else {
1164       gst_mini_object_unref (GST_MINI_OBJECT_CAST (data));
1165     }
1166   } else {
1167     GST_LOG ("source %08x pushed receiver RTP packet", source->ssrc);
1168     RTP_SESSION_UNLOCK (session);
1169
1170     if (session->callbacks.process_rtp)
1171       result =
1172           session->callbacks.process_rtp (session, source,
1173           GST_BUFFER_CAST (data), session->process_rtp_user_data);
1174     else
1175       gst_buffer_unref (GST_BUFFER_CAST (data));
1176   }
1177   RTP_SESSION_LOCK (session);
1178
1179   return result;
1180 }
1181
1182 static gint
1183 source_clock_rate (RTPSource * source, guint8 pt, RTPSession * session)
1184 {
1185   gint result;
1186
1187   RTP_SESSION_UNLOCK (session);
1188
1189   if (session->callbacks.clock_rate)
1190     result =
1191         session->callbacks.clock_rate (session, pt,
1192         session->clock_rate_user_data);
1193   else
1194     result = -1;
1195
1196   RTP_SESSION_LOCK (session);
1197
1198   GST_DEBUG ("got clock-rate %d for pt %d", result, pt);
1199
1200   return result;
1201 }
1202
1203 static RTPSourceCallbacks callbacks = {
1204   (RTPSourcePushRTP) source_push_rtp,
1205   (RTPSourceClockRate) source_clock_rate,
1206 };
1207
1208
1209 /**
1210  * rtp_session_find_conflicting_address:
1211  * @session: The session the packet came in
1212  * @address: address to check for
1213  * @time: The time when the packet that is possibly in conflict arrived
1214  *
1215  * Checks if an address which has a conflict is already known. If it is
1216  * a known conflict, remember the time
1217  *
1218  * Returns: TRUE if it was a known conflict, FALSE otherwise
1219  */
1220 static gboolean
1221 rtp_session_find_conflicting_address (RTPSession * session,
1222     GSocketAddress * address, GstClockTime time)
1223 {
1224   return find_conflicting_address (session->conflicting_addresses, address,
1225       time);
1226 }
1227
1228 /**
1229  * rtp_session_add_conflicting_address:
1230  * @session: The session the packet came in
1231  * @address: address to remember
1232  * @time: The time when the packet that is in conflict arrived
1233  *
1234  * Adds a new conflict address
1235  */
1236 static void
1237 rtp_session_add_conflicting_address (RTPSession * sess,
1238     GSocketAddress * address, GstClockTime time)
1239 {
1240   sess->conflicting_addresses =
1241       add_conflicting_address (sess->conflicting_addresses, address, time);
1242 }
1243
1244
1245 static gboolean
1246 check_collision (RTPSession * sess, RTPSource * source,
1247     RTPPacketInfo * pinfo, gboolean rtp)
1248 {
1249   guint32 ssrc;
1250
1251   /* If we have no pinfo address, we can't do collision checking */
1252   if (!pinfo->address)
1253     return FALSE;
1254
1255   ssrc = rtp_source_get_ssrc (source);
1256
1257   if (!source->internal) {
1258     GSocketAddress *from;
1259
1260     /* This is not our local source, but lets check if two remote
1261      * source collide */
1262     if (rtp) {
1263       from = source->rtp_from;
1264     } else {
1265       from = source->rtcp_from;
1266     }
1267
1268     if (from) {
1269       if (__g_socket_address_equal (from, pinfo->address)) {
1270         /* Address is the same */
1271         return FALSE;
1272       } else {
1273         GST_LOG ("we have a third-party collision or loop ssrc:%x", ssrc);
1274         if (sess->favor_new) {
1275           if (rtp_source_find_conflicting_address (source,
1276                   pinfo->address, pinfo->current_time)) {
1277             gchar *buf1;
1278
1279             buf1 = __g_socket_address_to_string (pinfo->address);
1280             GST_LOG ("Known conflict on %x for %s, dropping packet", ssrc,
1281                 buf1);
1282             g_free (buf1);
1283
1284             return TRUE;
1285           } else {
1286             gchar *buf1, *buf2;
1287
1288             /* Current address is not a known conflict, lets assume this is
1289              * a new source. Save old address in possible conflict list
1290              */
1291             rtp_source_add_conflicting_address (source, from,
1292                 pinfo->current_time);
1293
1294             buf1 = __g_socket_address_to_string (from);
1295             buf2 = __g_socket_address_to_string (pinfo->address);
1296
1297             GST_DEBUG ("New conflict for ssrc %x, replacing %s with %s,"
1298                 " saving old as known conflict", ssrc, buf1, buf2);
1299
1300             if (rtp)
1301               rtp_source_set_rtp_from (source, pinfo->address);
1302             else
1303               rtp_source_set_rtcp_from (source, pinfo->address);
1304
1305             g_free (buf1);
1306             g_free (buf2);
1307
1308             return FALSE;
1309           }
1310         } else {
1311           /* Don't need to save old addresses, we ignore new sources */
1312           return TRUE;
1313         }
1314       }
1315     } else {
1316       /* We don't already have a from address for RTP, just set it */
1317       if (rtp)
1318         rtp_source_set_rtp_from (source, pinfo->address);
1319       else
1320         rtp_source_set_rtcp_from (source, pinfo->address);
1321       return FALSE;
1322     }
1323
1324     /* FIXME: Log 3rd party collision somehow
1325      * Maybe should be done in upper layer, only the SDES can tell us
1326      * if its a collision or a loop
1327      */
1328   } else {
1329     /* This is sending with our ssrc, is it an address we already know */
1330     if (rtp_session_find_conflicting_address (sess, pinfo->address,
1331             pinfo->current_time)) {
1332       /* Its a known conflict, its probably a loop, not a collision
1333        * lets just drop the incoming packet
1334        */
1335       GST_DEBUG ("Our packets are being looped back to us, dropping");
1336     } else {
1337       /* Its a new collision, lets change our SSRC */
1338       rtp_session_add_conflicting_address (sess, pinfo->address,
1339           pinfo->current_time);
1340
1341       GST_DEBUG ("Collision for SSRC %x", ssrc);
1342       /* mark the source BYE */
1343       rtp_source_mark_bye (source, "SSRC Collision");
1344       /* if we were suggesting this SSRC, change to something else */
1345       if (sess->suggested_ssrc == ssrc)
1346         sess->suggested_ssrc = rtp_session_create_new_ssrc (sess);
1347
1348       on_ssrc_collision (sess, source);
1349
1350       rtp_session_schedule_bye_locked (sess, pinfo->current_time);
1351     }
1352   }
1353
1354   return TRUE;
1355 }
1356
1357 typedef struct
1358 {
1359   gboolean is_doing_ptp;
1360   GSocketAddress *new_addr;
1361 } CompareAddrData;
1362
1363 /* check if the two given ip addr are the same (do not care about the port) */
1364 static gboolean
1365 ip_addr_equal (GSocketAddress * a, GSocketAddress * b)
1366 {
1367   return
1368       g_inet_address_equal (g_inet_socket_address_get_address
1369       (G_INET_SOCKET_ADDRESS (a)),
1370       g_inet_socket_address_get_address (G_INET_SOCKET_ADDRESS (b)));
1371 }
1372
1373 static void
1374 compare_rtp_source_addr (const gchar * key, RTPSource * source,
1375     CompareAddrData * data)
1376 {
1377   /* only compare ip addr of remote sources which are also not closing */
1378   if (!source->internal && !source->closing && source->rtp_from) {
1379     /* look for the first rtp source */
1380     if (!data->new_addr)
1381       data->new_addr = source->rtp_from;
1382     /* compare current ip addr with the first one */
1383     else
1384       data->is_doing_ptp &= ip_addr_equal (data->new_addr, source->rtp_from);
1385   }
1386 }
1387
1388 static void
1389 compare_rtcp_source_addr (const gchar * key, RTPSource * source,
1390     CompareAddrData * data)
1391 {
1392   /* only compare ip addr of remote sources which are also not closing */
1393   if (!source->internal && !source->closing && source->rtcp_from) {
1394     /* look for the first rtcp source */
1395     if (!data->new_addr)
1396       data->new_addr = source->rtcp_from;
1397     else
1398       /* compare current ip addr with the first one */
1399       data->is_doing_ptp &= ip_addr_equal (data->new_addr, source->rtcp_from);
1400   }
1401 }
1402
1403 /* loop over our non-internal source to know if the session
1404  * is doing point-to-point */
1405 static void
1406 session_update_ptp (RTPSession * sess)
1407 {
1408   /* to know if the session is doing point to point, the ip addr
1409    * of each non-internal (=remotes) source have to be compared
1410    * to each other.
1411    */
1412   gboolean is_doing_rtp_ptp = FALSE;
1413   gboolean is_doing_rtcp_ptp = FALSE;
1414   CompareAddrData data;
1415
1416   /* compare the first remote source's ip addr that receive rtp packets
1417    * with other remote rtp source.
1418    * it's enough because the session just needs to know if they are all
1419    * equals or not
1420    */
1421   data.is_doing_ptp = TRUE;
1422   data.new_addr = NULL;
1423   g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
1424       (GHFunc) compare_rtp_source_addr, (gpointer) & data);
1425   is_doing_rtp_ptp = data.is_doing_ptp;
1426
1427   /* same but about rtcp */
1428   data.is_doing_ptp = TRUE;
1429   data.new_addr = NULL;
1430   g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
1431       (GHFunc) compare_rtcp_source_addr, (gpointer) & data);
1432   is_doing_rtcp_ptp = data.is_doing_ptp;
1433
1434   /* the session is doing point-to-point if all rtp remote have the same
1435    * ip addr and if all rtcp remote sources have the same ip addr */
1436   sess->is_doing_ptp = is_doing_rtp_ptp && is_doing_rtcp_ptp;
1437
1438   GST_DEBUG ("doing point-to-point: %d", sess->is_doing_ptp);
1439 }
1440
1441 static void
1442 add_source (RTPSession * sess, RTPSource * src)
1443 {
1444   g_hash_table_insert (sess->ssrcs[sess->mask_idx],
1445       GINT_TO_POINTER (src->ssrc), src);
1446   /* report the new source ASAP */
1447   src->generation = sess->generation;
1448   /* we have one more source now */
1449   sess->total_sources++;
1450   if (RTP_SOURCE_IS_ACTIVE (src))
1451     sess->stats.active_sources++;
1452   if (src->internal) {
1453     sess->stats.internal_sources++;
1454     if (sess->suggested_ssrc != src->ssrc)
1455       sess->suggested_ssrc = src->ssrc;
1456   }
1457
1458   /* update point-to-point status */
1459   if (!src->internal)
1460     session_update_ptp (sess);
1461 }
1462
1463 static RTPSource *
1464 find_source (RTPSession * sess, guint32 ssrc)
1465 {
1466   return g_hash_table_lookup (sess->ssrcs[sess->mask_idx],
1467       GINT_TO_POINTER (ssrc));
1468 }
1469
1470 /* must be called with the session lock, the returned source needs to be
1471  * unreffed after usage. */
1472 static RTPSource *
1473 obtain_source (RTPSession * sess, guint32 ssrc, gboolean * created,
1474     RTPPacketInfo * pinfo, gboolean rtp)
1475 {
1476   RTPSource *source;
1477
1478   source = find_source (sess, ssrc);
1479   if (source == NULL) {
1480     /* make new Source in probation and insert */
1481     source = rtp_source_new (ssrc);
1482
1483     GST_DEBUG ("creating new source %08x %p", ssrc, source);
1484
1485     /* for RTP packets we need to set the source in probation. Receiving RTCP
1486      * packets of an SSRC, on the other hand, is a strong indication that we
1487      * are dealing with a valid source. */
1488     if (rtp)
1489       g_object_set (source, "probation", sess->probation, NULL);
1490     else
1491       g_object_set (source, "probation", 0, NULL);
1492
1493     /* store from address, if any */
1494     if (pinfo->address) {
1495       if (rtp)
1496         rtp_source_set_rtp_from (source, pinfo->address);
1497       else
1498         rtp_source_set_rtcp_from (source, pinfo->address);
1499     }
1500
1501     /* configure a callback on the source */
1502     rtp_source_set_callbacks (source, &callbacks, sess);
1503
1504     add_source (sess, source);
1505     *created = TRUE;
1506   } else {
1507     *created = FALSE;
1508     /* check for collision, this updates the address when not previously set */
1509     if (check_collision (sess, source, pinfo, rtp)) {
1510       return NULL;
1511     }
1512     /* Receiving RTCP packets of an SSRC is a strong indication that we
1513      * are dealing with a valid source. */
1514     if (!rtp)
1515       g_object_set (source, "probation", 0, NULL);
1516   }
1517   /* update last activity */
1518   source->last_activity = pinfo->current_time;
1519   if (rtp)
1520     source->last_rtp_activity = pinfo->current_time;
1521   g_object_ref (source);
1522
1523   return source;
1524 }
1525
1526 /* must be called with the session lock, the returned source needs to be
1527  * unreffed after usage. */
1528 static RTPSource *
1529 obtain_internal_source (RTPSession * sess, guint32 ssrc, gboolean * created)
1530 {
1531   RTPSource *source;
1532
1533   source = find_source (sess, ssrc);
1534   if (source == NULL) {
1535     /* make new internal Source and insert */
1536     source = rtp_source_new (ssrc);
1537
1538     GST_DEBUG ("creating new internal source %08x %p", ssrc, source);
1539
1540     source->validated = TRUE;
1541     source->internal = TRUE;
1542     source->probation = FALSE;
1543     rtp_source_set_sdes_struct (source, gst_structure_copy (sess->sdes));
1544     rtp_source_set_callbacks (source, &callbacks, sess);
1545
1546     add_source (sess, source);
1547     *created = TRUE;
1548   } else {
1549     *created = FALSE;
1550   }
1551   g_object_ref (source);
1552
1553   return source;
1554 }
1555
1556 /**
1557  * rtp_session_suggest_ssrc:
1558  * @sess: a #RTPSession
1559  *
1560  * Suggest an unused SSRC in @sess.
1561  *
1562  * Returns: a free unused SSRC
1563  */
1564 guint32
1565 rtp_session_suggest_ssrc (RTPSession * sess)
1566 {
1567   guint32 result;
1568
1569   g_return_val_if_fail (RTP_IS_SESSION (sess), 0);
1570
1571   RTP_SESSION_LOCK (sess);
1572   result = sess->suggested_ssrc;
1573   RTP_SESSION_UNLOCK (sess);
1574
1575   return result;
1576 }
1577
1578 /**
1579  * rtp_session_add_source:
1580  * @sess: a #RTPSession
1581  * @src: #RTPSource to add
1582  *
1583  * Add @src to @session.
1584  *
1585  * Returns: %TRUE on success, %FALSE if a source with the same SSRC already
1586  * existed in the session.
1587  */
1588 gboolean
1589 rtp_session_add_source (RTPSession * sess, RTPSource * src)
1590 {
1591   gboolean result = FALSE;
1592   RTPSource *find;
1593
1594   g_return_val_if_fail (RTP_IS_SESSION (sess), FALSE);
1595   g_return_val_if_fail (src != NULL, FALSE);
1596
1597   RTP_SESSION_LOCK (sess);
1598   find = find_source (sess, src->ssrc);
1599   if (find == NULL) {
1600     add_source (sess, src);
1601     result = TRUE;
1602   }
1603   RTP_SESSION_UNLOCK (sess);
1604
1605   return result;
1606 }
1607
1608 /**
1609  * rtp_session_get_num_sources:
1610  * @sess: an #RTPSession
1611  *
1612  * Get the number of sources in @sess.
1613  *
1614  * Returns: The number of sources in @sess.
1615  */
1616 guint
1617 rtp_session_get_num_sources (RTPSession * sess)
1618 {
1619   guint result;
1620
1621   g_return_val_if_fail (RTP_IS_SESSION (sess), FALSE);
1622
1623   RTP_SESSION_LOCK (sess);
1624   result = sess->total_sources;
1625   RTP_SESSION_UNLOCK (sess);
1626
1627   return result;
1628 }
1629
1630 /**
1631  * rtp_session_get_num_active_sources:
1632  * @sess: an #RTPSession
1633  *
1634  * Get the number of active sources in @sess. A source is considered active when
1635  * it has been validated and has not yet received a BYE RTCP message.
1636  *
1637  * Returns: The number of active sources in @sess.
1638  */
1639 guint
1640 rtp_session_get_num_active_sources (RTPSession * sess)
1641 {
1642   guint result;
1643
1644   g_return_val_if_fail (RTP_IS_SESSION (sess), 0);
1645
1646   RTP_SESSION_LOCK (sess);
1647   result = sess->stats.active_sources;
1648   RTP_SESSION_UNLOCK (sess);
1649
1650   return result;
1651 }
1652
1653 /**
1654  * rtp_session_get_source_by_ssrc:
1655  * @sess: an #RTPSession
1656  * @ssrc: an SSRC
1657  *
1658  * Find the source with @ssrc in @sess.
1659  *
1660  * Returns: a #RTPSource with SSRC @ssrc or NULL if the source was not found.
1661  * g_object_unref() after usage.
1662  */
1663 RTPSource *
1664 rtp_session_get_source_by_ssrc (RTPSession * sess, guint32 ssrc)
1665 {
1666   RTPSource *result;
1667
1668   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1669
1670   RTP_SESSION_LOCK (sess);
1671   result = find_source (sess, ssrc);
1672   if (result != NULL)
1673     g_object_ref (result);
1674   RTP_SESSION_UNLOCK (sess);
1675
1676   return result;
1677 }
1678
1679 /* should be called with the SESSION lock */
1680 static guint32
1681 rtp_session_create_new_ssrc (RTPSession * sess)
1682 {
1683   guint32 ssrc;
1684
1685   while (TRUE) {
1686     ssrc = g_random_int ();
1687
1688     /* see if it exists in the session, we're done if it doesn't */
1689     if (find_source (sess, ssrc) == NULL)
1690       break;
1691   }
1692   return ssrc;
1693 }
1694
1695
1696 /**
1697  * rtp_session_create_source:
1698  * @sess: an #RTPSession
1699  *
1700  * Create an #RTPSource for use in @sess. This function will create a source
1701  * with an ssrc that is currently not used by any participants in the session.
1702  *
1703  * Returns: an #RTPSource.
1704  */
1705 RTPSource *
1706 rtp_session_create_source (RTPSession * sess)
1707 {
1708   guint32 ssrc;
1709   RTPSource *source;
1710
1711   RTP_SESSION_LOCK (sess);
1712   ssrc = rtp_session_create_new_ssrc (sess);
1713   source = rtp_source_new (ssrc);
1714   rtp_source_set_callbacks (source, &callbacks, sess);
1715   /* we need an additional ref for the source in the hashtable */
1716   g_object_ref (source);
1717   add_source (sess, source);
1718   RTP_SESSION_UNLOCK (sess);
1719
1720   return source;
1721 }
1722
1723 static gboolean
1724 update_packet (GstBuffer ** buffer, guint idx, RTPPacketInfo * pinfo)
1725 {
1726   GstNetAddressMeta *meta;
1727
1728   /* get packet size including header overhead */
1729   pinfo->bytes += gst_buffer_get_size (*buffer) + pinfo->header_len;
1730   pinfo->packets++;
1731
1732   if (pinfo->rtp) {
1733     GstRTPBuffer rtp = { NULL };
1734
1735     if (!gst_rtp_buffer_map (*buffer, GST_MAP_READ, &rtp))
1736       goto invalid_packet;
1737
1738     pinfo->payload_len += gst_rtp_buffer_get_payload_len (&rtp);
1739     if (idx == 0) {
1740       gint i;
1741
1742       /* only keep info for first buffer */
1743       pinfo->ssrc = gst_rtp_buffer_get_ssrc (&rtp);
1744       pinfo->seqnum = gst_rtp_buffer_get_seq (&rtp);
1745       pinfo->pt = gst_rtp_buffer_get_payload_type (&rtp);
1746       pinfo->rtptime = gst_rtp_buffer_get_timestamp (&rtp);
1747       /* copy available csrc */
1748       pinfo->csrc_count = gst_rtp_buffer_get_csrc_count (&rtp);
1749       for (i = 0; i < pinfo->csrc_count; i++)
1750         pinfo->csrcs[i] = gst_rtp_buffer_get_csrc (&rtp, i);
1751     }
1752     gst_rtp_buffer_unmap (&rtp);
1753   }
1754
1755   if (idx == 0) {
1756     /* for netbuffer we can store the IP address to check for collisions */
1757     meta = gst_buffer_get_net_address_meta (*buffer);
1758     if (pinfo->address)
1759       g_object_unref (pinfo->address);
1760     if (meta) {
1761       pinfo->address = G_SOCKET_ADDRESS (g_object_ref (meta->addr));
1762     } else {
1763       pinfo->address = NULL;
1764     }
1765   }
1766   return TRUE;
1767
1768   /* ERRORS */
1769 invalid_packet:
1770   {
1771     GST_DEBUG ("invalid RTP packet received");
1772     return FALSE;
1773   }
1774 }
1775
1776 /* update the RTPPacketInfo structure with the current time and other bits
1777  * about the current buffer we are handling.
1778  * This function is typically called when a validated packet is received.
1779  * This function should be called with the SESSION_LOCK
1780  */
1781 static gboolean
1782 update_packet_info (RTPSession * sess, RTPPacketInfo * pinfo,
1783     gboolean send, gboolean rtp, gboolean is_list, gpointer data,
1784     GstClockTime current_time, GstClockTime running_time, guint64 ntpnstime)
1785 {
1786   gboolean res;
1787
1788   pinfo->send = send;
1789   pinfo->rtp = rtp;
1790   pinfo->is_list = is_list;
1791   pinfo->data = data;
1792   pinfo->current_time = current_time;
1793   pinfo->running_time = running_time;
1794   pinfo->ntpnstime = ntpnstime;
1795   pinfo->header_len = sess->header_len;
1796   pinfo->bytes = 0;
1797   pinfo->payload_len = 0;
1798   pinfo->packets = 0;
1799
1800   if (is_list) {
1801     GstBufferList *list = GST_BUFFER_LIST_CAST (data);
1802     res =
1803         gst_buffer_list_foreach (list, (GstBufferListFunc) update_packet,
1804         pinfo);
1805   } else {
1806     GstBuffer *buffer = GST_BUFFER_CAST (data);
1807     res = update_packet (&buffer, 0, pinfo);
1808   }
1809   return res;
1810 }
1811
1812 static void
1813 clean_packet_info (RTPPacketInfo * pinfo)
1814 {
1815   if (pinfo->address)
1816     g_object_unref (pinfo->address);
1817   if (pinfo->data) {
1818     gst_mini_object_unref (pinfo->data);
1819     pinfo->data = NULL;
1820   }
1821 }
1822
1823 static gboolean
1824 source_update_active (RTPSession * sess, RTPSource * source,
1825     gboolean prevactive)
1826 {
1827   gboolean active = RTP_SOURCE_IS_ACTIVE (source);
1828   guint32 ssrc = source->ssrc;
1829
1830   if (prevactive == active)
1831     return FALSE;
1832
1833   if (active) {
1834     sess->stats.active_sources++;
1835     GST_DEBUG ("source: %08x became active, %d active sources", ssrc,
1836         sess->stats.active_sources);
1837   } else {
1838     sess->stats.active_sources--;
1839     GST_DEBUG ("source: %08x became inactive, %d active sources", ssrc,
1840         sess->stats.active_sources);
1841   }
1842   return TRUE;
1843 }
1844
1845 static gboolean
1846 source_update_sender (RTPSession * sess, RTPSource * source,
1847     gboolean prevsender)
1848 {
1849   gboolean sender = RTP_SOURCE_IS_SENDER (source);
1850   guint32 ssrc = source->ssrc;
1851
1852   if (prevsender == sender)
1853     return FALSE;
1854
1855   if (sender) {
1856     sess->stats.sender_sources++;
1857     if (source->internal)
1858       sess->stats.internal_sender_sources++;
1859     GST_DEBUG ("source: %08x became sender, %d sender sources", ssrc,
1860         sess->stats.sender_sources);
1861   } else {
1862     sess->stats.sender_sources--;
1863     if (source->internal)
1864       sess->stats.internal_sender_sources--;
1865     GST_DEBUG ("source: %08x became non sender, %d sender sources", ssrc,
1866         sess->stats.sender_sources);
1867   }
1868   return TRUE;
1869 }
1870
1871 /**
1872  * rtp_session_process_rtp:
1873  * @sess: and #RTPSession
1874  * @buffer: an RTP buffer
1875  * @current_time: the current system time
1876  * @running_time: the running_time of @buffer
1877  *
1878  * Process an RTP buffer in the session manager. This function takes ownership
1879  * of @buffer.
1880  *
1881  * Returns: a #GstFlowReturn.
1882  */
1883 GstFlowReturn
1884 rtp_session_process_rtp (RTPSession * sess, GstBuffer * buffer,
1885     GstClockTime current_time, GstClockTime running_time, guint64 ntpnstime)
1886 {
1887   GstFlowReturn result;
1888   guint32 ssrc;
1889   RTPSource *source;
1890   gboolean created;
1891   gboolean prevsender, prevactive;
1892   RTPPacketInfo pinfo = { 0, };
1893   guint64 oldrate;
1894
1895   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1896   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1897
1898   RTP_SESSION_LOCK (sess);
1899
1900   /* update pinfo stats */
1901   if (!update_packet_info (sess, &pinfo, FALSE, TRUE, FALSE, buffer,
1902           current_time, running_time, ntpnstime)) {
1903     GST_DEBUG ("invalid RTP packet received");
1904     RTP_SESSION_UNLOCK (sess);
1905     return rtp_session_process_rtcp (sess, buffer, current_time, ntpnstime);
1906   }
1907
1908   ssrc = pinfo.ssrc;
1909
1910   source = obtain_source (sess, ssrc, &created, &pinfo, TRUE);
1911   if (!source)
1912     goto collision;
1913
1914   prevsender = RTP_SOURCE_IS_SENDER (source);
1915   prevactive = RTP_SOURCE_IS_ACTIVE (source);
1916   oldrate = source->bitrate;
1917
1918   /* let source process the packet */
1919   result = rtp_source_process_rtp (source, &pinfo);
1920
1921   /* source became active */
1922   if (source_update_active (sess, source, prevactive))
1923     on_ssrc_validated (sess, source);
1924
1925   source_update_sender (sess, source, prevsender);
1926
1927   if (oldrate != source->bitrate)
1928     sess->recalc_bandwidth = TRUE;
1929
1930   if (created)
1931     on_new_ssrc (sess, source);
1932
1933   if (source->validated) {
1934     gboolean created;
1935     gint i;
1936
1937     /* for validated sources, we add the CSRCs as well */
1938     for (i = 0; i < pinfo.csrc_count; i++) {
1939       guint32 csrc;
1940       RTPSource *csrc_src;
1941
1942       csrc = pinfo.csrcs[i];
1943
1944       /* get source */
1945       csrc_src = obtain_source (sess, csrc, &created, &pinfo, TRUE);
1946       if (!csrc_src)
1947         continue;
1948
1949       if (created) {
1950         GST_DEBUG ("created new CSRC: %08x", csrc);
1951         rtp_source_set_as_csrc (csrc_src);
1952         source_update_active (sess, csrc_src, FALSE);
1953         on_new_ssrc (sess, csrc_src);
1954       }
1955       g_object_unref (csrc_src);
1956     }
1957   }
1958   g_object_unref (source);
1959
1960   RTP_SESSION_UNLOCK (sess);
1961
1962   clean_packet_info (&pinfo);
1963
1964   return result;
1965
1966   /* ERRORS */
1967 collision:
1968   {
1969     RTP_SESSION_UNLOCK (sess);
1970     clean_packet_info (&pinfo);
1971     GST_DEBUG ("ignoring packet because its collisioning");
1972     return GST_FLOW_OK;
1973   }
1974 }
1975
1976 static void
1977 rtp_session_process_rb (RTPSession * sess, RTPSource * source,
1978     GstRTCPPacket * packet, RTPPacketInfo * pinfo)
1979 {
1980   guint count, i;
1981
1982   count = gst_rtcp_packet_get_rb_count (packet);
1983   for (i = 0; i < count; i++) {
1984     guint32 ssrc, exthighestseq, jitter, lsr, dlsr;
1985     guint8 fractionlost;
1986     gint32 packetslost;
1987     RTPSource *src;
1988
1989     gst_rtcp_packet_get_rb (packet, i, &ssrc, &fractionlost,
1990         &packetslost, &exthighestseq, &jitter, &lsr, &dlsr);
1991
1992     GST_DEBUG ("RB %d: SSRC %08x, jitter %" G_GUINT32_FORMAT, i, ssrc, jitter);
1993
1994     /* find our own source */
1995     src = find_source (sess, ssrc);
1996     if (src == NULL)
1997       continue;
1998
1999     if (src->internal && RTP_SOURCE_IS_ACTIVE (src)) {
2000       /* only deal with report blocks for our session, we update the stats of
2001        * the sender of the RTCP message. We could also compare our stats against
2002        * the other sender to see if we are better or worse. */
2003       /* FIXME, need to keep track who the RB block is from */
2004       rtp_source_process_rb (source, pinfo->ntpnstime, fractionlost,
2005           packetslost, exthighestseq, jitter, lsr, dlsr);
2006     }
2007   }
2008   on_ssrc_active (sess, source);
2009 }
2010
2011 /* A Sender report contains statistics about how the sender is doing. This
2012  * includes timing informataion such as the relation between RTP and NTP
2013  * timestamps and the number of packets/bytes it sent to us.
2014  *
2015  * In this report is also included a set of report blocks related to how this
2016  * sender is receiving data (in case we (or somebody else) is also sending stuff
2017  * to it). This info includes the packet loss, jitter and seqnum. It also
2018  * contains information to calculate the round trip time (LSR/DLSR).
2019  */
2020 static void
2021 rtp_session_process_sr (RTPSession * sess, GstRTCPPacket * packet,
2022     RTPPacketInfo * pinfo, gboolean * do_sync)
2023 {
2024   guint32 senderssrc, rtptime, packet_count, octet_count;
2025   guint64 ntptime;
2026   RTPSource *source;
2027   gboolean created, prevsender;
2028
2029   gst_rtcp_packet_sr_get_sender_info (packet, &senderssrc, &ntptime, &rtptime,
2030       &packet_count, &octet_count);
2031
2032   GST_DEBUG ("got SR packet: SSRC %08x, time %" GST_TIME_FORMAT,
2033       senderssrc, GST_TIME_ARGS (pinfo->current_time));
2034
2035   source = obtain_source (sess, senderssrc, &created, pinfo, FALSE);
2036   if (!source)
2037     return;
2038
2039   /* skip non-bye packets for sources that are marked BYE */
2040   if (sess->scheduled_bye && RTP_SOURCE_IS_MARKED_BYE (source))
2041     goto out;
2042
2043   /* don't try to do lip-sync for sources that sent a BYE */
2044   if (RTP_SOURCE_IS_MARKED_BYE (source))
2045     *do_sync = FALSE;
2046   else
2047     *do_sync = TRUE;
2048
2049   prevsender = RTP_SOURCE_IS_SENDER (source);
2050
2051   /* first update the source */
2052   rtp_source_process_sr (source, pinfo->current_time, ntptime, rtptime,
2053       packet_count, octet_count);
2054
2055   source_update_sender (sess, source, prevsender);
2056
2057   if (created)
2058     on_new_ssrc (sess, source);
2059
2060   rtp_session_process_rb (sess, source, packet, pinfo);
2061
2062 out:
2063   g_object_unref (source);
2064 }
2065
2066 /* A receiver report contains statistics about how a receiver is doing. It
2067  * includes stuff like packet loss, jitter and the seqnum it received last. It
2068  * also contains info to calculate the round trip time.
2069  *
2070  * We are only interested in how the sender of this report is doing wrt to us.
2071  */
2072 static void
2073 rtp_session_process_rr (RTPSession * sess, GstRTCPPacket * packet,
2074     RTPPacketInfo * pinfo)
2075 {
2076   guint32 senderssrc;
2077   RTPSource *source;
2078   gboolean created;
2079
2080   senderssrc = gst_rtcp_packet_rr_get_ssrc (packet);
2081
2082   GST_DEBUG ("got RR packet: SSRC %08x", senderssrc);
2083
2084   source = obtain_source (sess, senderssrc, &created, pinfo, FALSE);
2085   if (!source)
2086     return;
2087
2088   /* skip non-bye packets for sources that are marked BYE */
2089   if (sess->scheduled_bye && RTP_SOURCE_IS_MARKED_BYE (source))
2090     goto out;
2091
2092   if (created)
2093     on_new_ssrc (sess, source);
2094
2095   rtp_session_process_rb (sess, source, packet, pinfo);
2096
2097 out:
2098   g_object_unref (source);
2099 }
2100
2101 /* Get SDES items and store them in the SSRC */
2102 static void
2103 rtp_session_process_sdes (RTPSession * sess, GstRTCPPacket * packet,
2104     RTPPacketInfo * pinfo)
2105 {
2106   guint items, i, j;
2107   gboolean more_items, more_entries;
2108
2109   items = gst_rtcp_packet_sdes_get_item_count (packet);
2110   GST_DEBUG ("got SDES packet with %d items", items);
2111
2112   more_items = gst_rtcp_packet_sdes_first_item (packet);
2113   i = 0;
2114   while (more_items) {
2115     guint32 ssrc;
2116     gboolean changed, created, prevactive;
2117     RTPSource *source;
2118     GstStructure *sdes;
2119
2120     ssrc = gst_rtcp_packet_sdes_get_ssrc (packet);
2121
2122     GST_DEBUG ("item %d, SSRC %08x", i, ssrc);
2123
2124     changed = FALSE;
2125
2126     /* find src, no probation when dealing with RTCP */
2127     source = obtain_source (sess, ssrc, &created, pinfo, FALSE);
2128     if (!source)
2129       return;
2130
2131     /* skip non-bye packets for sources that are marked BYE */
2132     if (sess->scheduled_bye && RTP_SOURCE_IS_MARKED_BYE (source))
2133       goto next;
2134
2135     sdes = gst_structure_new_empty ("application/x-rtp-source-sdes");
2136
2137     more_entries = gst_rtcp_packet_sdes_first_entry (packet);
2138     j = 0;
2139     while (more_entries) {
2140       GstRTCPSDESType type;
2141       guint8 len;
2142       guint8 *data;
2143       gchar *name;
2144       gchar *value;
2145
2146       gst_rtcp_packet_sdes_get_entry (packet, &type, &len, &data);
2147
2148       GST_DEBUG ("entry %d, type %d, len %d, data %.*s", j, type, len, len,
2149           data);
2150
2151       if (type == GST_RTCP_SDES_PRIV) {
2152         name = g_strndup ((const gchar *) &data[1], data[0]);
2153         len -= data[0] + 1;
2154         data += data[0] + 1;
2155       } else {
2156         name = g_strdup (gst_rtcp_sdes_type_to_name (type));
2157       }
2158
2159       value = g_strndup ((const gchar *) data, len);
2160
2161       gst_structure_set (sdes, name, G_TYPE_STRING, value, NULL);
2162
2163       g_free (name);
2164       g_free (value);
2165
2166       more_entries = gst_rtcp_packet_sdes_next_entry (packet);
2167       j++;
2168     }
2169
2170     /* takes ownership of sdes */
2171     changed = rtp_source_set_sdes_struct (source, sdes);
2172
2173     prevactive = RTP_SOURCE_IS_ACTIVE (source);
2174     source->validated = TRUE;
2175
2176     if (created)
2177       on_new_ssrc (sess, source);
2178
2179     /* source became active */
2180     if (source_update_active (sess, source, prevactive))
2181       on_ssrc_validated (sess, source);
2182
2183     if (changed)
2184       on_ssrc_sdes (sess, source);
2185
2186   next:
2187     g_object_unref (source);
2188
2189     more_items = gst_rtcp_packet_sdes_next_item (packet);
2190     i++;
2191   }
2192 }
2193
2194 /* BYE is sent when a client leaves the session
2195  */
2196 static void
2197 rtp_session_process_bye (RTPSession * sess, GstRTCPPacket * packet,
2198     RTPPacketInfo * pinfo)
2199 {
2200   guint count, i;
2201   gchar *reason;
2202   gboolean reconsider = FALSE;
2203
2204   reason = gst_rtcp_packet_bye_get_reason (packet);
2205   GST_DEBUG ("got BYE packet (reason: %s)", GST_STR_NULL (reason));
2206
2207   count = gst_rtcp_packet_bye_get_ssrc_count (packet);
2208   for (i = 0; i < count; i++) {
2209     guint32 ssrc;
2210     RTPSource *source;
2211     gboolean created, prevactive, prevsender;
2212     guint pmembers, members;
2213
2214     ssrc = gst_rtcp_packet_bye_get_nth_ssrc (packet, i);
2215     GST_DEBUG ("SSRC: %08x", ssrc);
2216
2217     /* find src and mark bye, no probation when dealing with RTCP */
2218     source = obtain_source (sess, ssrc, &created, pinfo, FALSE);
2219     if (!source)
2220       return;
2221
2222     if (source->internal) {
2223       /* our own source, something weird with this packet */
2224       g_object_unref (source);
2225       continue;
2226     }
2227
2228     /* store time for when we need to time out this source */
2229     source->bye_time = pinfo->current_time;
2230
2231     prevactive = RTP_SOURCE_IS_ACTIVE (source);
2232     prevsender = RTP_SOURCE_IS_SENDER (source);
2233
2234     /* mark the source BYE */
2235     rtp_source_mark_bye (source, reason);
2236
2237     pmembers = sess->stats.active_sources;
2238
2239     source_update_active (sess, source, prevactive);
2240     source_update_sender (sess, source, prevsender);
2241
2242     members = sess->stats.active_sources;
2243
2244     if (!sess->scheduled_bye && members < pmembers) {
2245       /* some members went away since the previous timeout estimate.
2246        * Perform reverse reconsideration but only when we are not scheduling a
2247        * BYE ourselves. */
2248       if (sess->next_rtcp_check_time != GST_CLOCK_TIME_NONE &&
2249           pinfo->current_time < sess->next_rtcp_check_time) {
2250         GstClockTime time_remaining;
2251
2252         time_remaining = sess->next_rtcp_check_time - pinfo->current_time;
2253         sess->next_rtcp_check_time =
2254             gst_util_uint64_scale (time_remaining, members, pmembers);
2255
2256         GST_DEBUG ("reverse reconsideration %" GST_TIME_FORMAT,
2257             GST_TIME_ARGS (sess->next_rtcp_check_time));
2258
2259         sess->next_rtcp_check_time += pinfo->current_time;
2260
2261         /* mark pending reconsider. We only want to signal the reconsideration
2262          * once after we handled all the source in the bye packet */
2263         reconsider = TRUE;
2264       }
2265     }
2266
2267     if (created)
2268       on_new_ssrc (sess, source);
2269
2270     on_bye_ssrc (sess, source);
2271
2272     g_object_unref (source);
2273   }
2274   if (reconsider) {
2275     RTP_SESSION_UNLOCK (sess);
2276     /* notify app of reconsideration */
2277     if (sess->callbacks.reconsider)
2278       sess->callbacks.reconsider (sess, sess->reconsider_user_data);
2279     RTP_SESSION_LOCK (sess);
2280   }
2281   g_free (reason);
2282 }
2283
2284 static void
2285 rtp_session_process_app (RTPSession * sess, GstRTCPPacket * packet,
2286     RTPPacketInfo * pinfo)
2287 {
2288   GST_DEBUG ("received APP");
2289 }
2290
2291 static gboolean
2292 rtp_session_request_local_key_unit (RTPSession * sess, RTPSource * src,
2293     gboolean fir, GstClockTime current_time)
2294 {
2295   guint32 round_trip = 0;
2296
2297   rtp_source_get_last_rb (src, NULL, NULL, NULL, NULL, NULL, NULL, &round_trip);
2298
2299   if (sess->last_keyframe_request != GST_CLOCK_TIME_NONE && round_trip) {
2300     GstClockTime round_trip_in_ns = gst_util_uint64_scale (round_trip,
2301         GST_SECOND, 65536);
2302
2303     if (current_time - sess->last_keyframe_request < 2 * round_trip_in_ns) {
2304       GST_DEBUG ("Ignoring %s request because one was send without one "
2305           "RTT (%" GST_TIME_FORMAT " < %" GST_TIME_FORMAT ")",
2306           fir ? "FIR" : "PLI",
2307           GST_TIME_ARGS (current_time - sess->last_keyframe_request),
2308           GST_TIME_ARGS (round_trip_in_ns));;
2309       return FALSE;
2310     }
2311   }
2312
2313   sess->last_keyframe_request = current_time;
2314
2315   GST_LOG ("received %s request from %X %p(%p)", fir ? "FIR" : "PLI",
2316       rtp_source_get_ssrc (src), sess->callbacks.process_rtp,
2317       sess->callbacks.request_key_unit);
2318
2319   RTP_SESSION_UNLOCK (sess);
2320   sess->callbacks.request_key_unit (sess, fir,
2321       sess->request_key_unit_user_data);
2322   RTP_SESSION_LOCK (sess);
2323
2324   return TRUE;
2325 }
2326
2327 static void
2328 rtp_session_process_pli (RTPSession * sess, guint32 sender_ssrc,
2329     guint32 media_ssrc, GstClockTime current_time)
2330 {
2331   RTPSource *src;
2332
2333   if (!sess->callbacks.request_key_unit)
2334     return;
2335
2336   src = find_source (sess, sender_ssrc);
2337   if (src == NULL)
2338     return;
2339
2340   rtp_session_request_local_key_unit (sess, src, FALSE, current_time);
2341 }
2342
2343 static void
2344 rtp_session_process_fir (RTPSession * sess, guint32 sender_ssrc,
2345     guint8 * fci_data, guint fci_length, GstClockTime current_time)
2346 {
2347   RTPSource *src;
2348   guint32 ssrc;
2349   guint position = 0;
2350   gboolean our_request = FALSE;
2351
2352   if (!sess->callbacks.request_key_unit)
2353     return;
2354
2355   if (fci_length < 8)
2356     return;
2357
2358   src = find_source (sess, sender_ssrc);
2359
2360   /* Hack because Google fails to set the sender_ssrc correctly */
2361   if (!src && sender_ssrc == 1) {
2362     GHashTableIter iter;
2363
2364     /* we can't find the source if there are multiple */
2365     if (sess->stats.sender_sources > sess->stats.internal_sender_sources + 1)
2366       return;
2367
2368     g_hash_table_iter_init (&iter, sess->ssrcs[sess->mask_idx]);
2369     while (g_hash_table_iter_next (&iter, NULL, (gpointer *) & src)) {
2370       if (!src->internal && rtp_source_is_sender (src))
2371         break;
2372       src = NULL;
2373     }
2374   }
2375   if (!src)
2376     return;
2377
2378   for (position = 0; position < fci_length; position += 8) {
2379     guint8 *data = fci_data + position;
2380     RTPSource *own;
2381
2382     ssrc = GST_READ_UINT32_BE (data);
2383
2384     own = find_source (sess, ssrc);
2385     if (own == NULL)
2386       continue;
2387
2388     if (own->internal) {
2389       our_request = TRUE;
2390       break;
2391     }
2392   }
2393   if (!our_request)
2394     return;
2395
2396   rtp_session_request_local_key_unit (sess, src, TRUE, current_time);
2397 }
2398
2399 static void
2400 rtp_session_process_nack (RTPSession * sess, guint32 sender_ssrc,
2401     guint32 media_ssrc, guint8 * fci_data, guint fci_length,
2402     GstClockTime current_time)
2403 {
2404   sess->stats.nacks_received++;
2405
2406   if (!sess->callbacks.notify_nack)
2407     return;
2408
2409   while (fci_length > 0) {
2410     guint16 seqnum, blp;
2411
2412     seqnum = GST_READ_UINT16_BE (fci_data);
2413     blp = GST_READ_UINT16_BE (fci_data + 2);
2414
2415     GST_DEBUG ("NACK #%u, blp %04x, SSRC 0x%08x", seqnum, blp, media_ssrc);
2416
2417     RTP_SESSION_UNLOCK (sess);
2418     sess->callbacks.notify_nack (sess, seqnum, blp, media_ssrc,
2419         sess->notify_nack_user_data);
2420     RTP_SESSION_LOCK (sess);
2421
2422     fci_data += 4;
2423     fci_length -= 4;
2424   }
2425 }
2426
2427 static void
2428 rtp_session_process_feedback (RTPSession * sess, GstRTCPPacket * packet,
2429     RTPPacketInfo * pinfo, GstClockTime current_time)
2430 {
2431   GstRTCPType type = gst_rtcp_packet_get_type (packet);
2432   GstRTCPFBType fbtype = gst_rtcp_packet_fb_get_type (packet);
2433   guint32 sender_ssrc = gst_rtcp_packet_fb_get_sender_ssrc (packet);
2434   guint32 media_ssrc = gst_rtcp_packet_fb_get_media_ssrc (packet);
2435   guint8 *fci_data = gst_rtcp_packet_fb_get_fci (packet);
2436   guint fci_length = 4 * gst_rtcp_packet_fb_get_fci_length (packet);
2437   RTPSource *src;
2438
2439   src = find_source (sess, media_ssrc);
2440
2441   /* skip non-bye packets for sources that are marked BYE */
2442   if (sess->scheduled_bye && src && RTP_SOURCE_IS_MARKED_BYE (src))
2443     return;
2444
2445   GST_DEBUG ("received feedback %d:%d from %08X about %08X with FCI of "
2446       "length %d", type, fbtype, sender_ssrc, media_ssrc, fci_length);
2447
2448   if (g_signal_has_handler_pending (sess,
2449           rtp_session_signals[SIGNAL_ON_FEEDBACK_RTCP], 0, TRUE)) {
2450     GstBuffer *fci_buffer = NULL;
2451
2452     if (fci_length > 0) {
2453       fci_buffer = gst_buffer_copy_region (packet->rtcp->buffer,
2454           GST_BUFFER_COPY_MEMORY, fci_data - packet->rtcp->map.data,
2455           fci_length);
2456       GST_BUFFER_TIMESTAMP (fci_buffer) = pinfo->running_time;
2457     }
2458
2459     RTP_SESSION_UNLOCK (sess);
2460     g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_FEEDBACK_RTCP], 0,
2461         type, fbtype, sender_ssrc, media_ssrc, fci_buffer);
2462     RTP_SESSION_LOCK (sess);
2463
2464     if (fci_buffer)
2465       gst_buffer_unref (fci_buffer);
2466   }
2467
2468   if (src && sess->rtcp_feedback_retention_window) {
2469     rtp_source_retain_rtcp_packet (src, packet, pinfo->running_time);
2470   }
2471
2472   if ((src && src->internal) ||
2473       /* PSFB FIR puts the media ssrc inside the FCI */
2474       (type == GST_RTCP_TYPE_PSFB && fbtype == GST_RTCP_PSFB_TYPE_FIR)) {
2475     switch (type) {
2476       case GST_RTCP_TYPE_PSFB:
2477         switch (fbtype) {
2478           case GST_RTCP_PSFB_TYPE_PLI:
2479             rtp_session_process_pli (sess, sender_ssrc, media_ssrc,
2480                 current_time);
2481             break;
2482           case GST_RTCP_PSFB_TYPE_FIR:
2483             rtp_session_process_fir (sess, sender_ssrc, fci_data, fci_length,
2484                 current_time);
2485             break;
2486           default:
2487             break;
2488         }
2489         break;
2490       case GST_RTCP_TYPE_RTPFB:
2491         switch (fbtype) {
2492           case GST_RTCP_RTPFB_TYPE_NACK:
2493             rtp_session_process_nack (sess, sender_ssrc, media_ssrc,
2494                 fci_data, fci_length, current_time);
2495             break;
2496           default:
2497             break;
2498         }
2499       default:
2500         break;
2501     }
2502   }
2503 }
2504
2505 /**
2506  * rtp_session_process_rtcp:
2507  * @sess: and #RTPSession
2508  * @buffer: an RTCP buffer
2509  * @current_time: the current system time
2510  * @ntpnstime: the current NTP time in nanoseconds
2511  *
2512  * Process an RTCP buffer in the session manager. This function takes ownership
2513  * of @buffer.
2514  *
2515  * Returns: a #GstFlowReturn.
2516  */
2517 GstFlowReturn
2518 rtp_session_process_rtcp (RTPSession * sess, GstBuffer * buffer,
2519     GstClockTime current_time, guint64 ntpnstime)
2520 {
2521   GstRTCPPacket packet;
2522   gboolean more, is_bye = FALSE, do_sync = FALSE;
2523   RTPPacketInfo pinfo = { 0, };
2524   GstFlowReturn result = GST_FLOW_OK;
2525   GstRTCPBuffer rtcp = { NULL, };
2526
2527   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
2528   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
2529
2530   if (!gst_rtcp_buffer_validate (buffer))
2531     goto invalid_packet;
2532
2533   GST_DEBUG ("received RTCP packet");
2534
2535   RTP_SESSION_LOCK (sess);
2536   /* update pinfo stats */
2537   update_packet_info (sess, &pinfo, FALSE, FALSE, FALSE, buffer, current_time,
2538       -1, ntpnstime);
2539
2540   /* start processing the compound packet */
2541   gst_rtcp_buffer_map (buffer, GST_MAP_READ, &rtcp);
2542   more = gst_rtcp_buffer_get_first_packet (&rtcp, &packet);
2543   while (more) {
2544     GstRTCPType type;
2545
2546     type = gst_rtcp_packet_get_type (&packet);
2547
2548     switch (type) {
2549       case GST_RTCP_TYPE_SR:
2550         rtp_session_process_sr (sess, &packet, &pinfo, &do_sync);
2551         break;
2552       case GST_RTCP_TYPE_RR:
2553         rtp_session_process_rr (sess, &packet, &pinfo);
2554         break;
2555       case GST_RTCP_TYPE_SDES:
2556         rtp_session_process_sdes (sess, &packet, &pinfo);
2557         break;
2558       case GST_RTCP_TYPE_BYE:
2559         is_bye = TRUE;
2560         /* don't try to attempt lip-sync anymore for streams with a BYE */
2561         do_sync = FALSE;
2562         rtp_session_process_bye (sess, &packet, &pinfo);
2563         break;
2564       case GST_RTCP_TYPE_APP:
2565         rtp_session_process_app (sess, &packet, &pinfo);
2566         break;
2567       case GST_RTCP_TYPE_RTPFB:
2568       case GST_RTCP_TYPE_PSFB:
2569         rtp_session_process_feedback (sess, &packet, &pinfo, current_time);
2570         break;
2571       default:
2572         GST_WARNING ("got unknown RTCP packet");
2573         break;
2574     }
2575     more = gst_rtcp_packet_move_to_next (&packet);
2576   }
2577
2578   gst_rtcp_buffer_unmap (&rtcp);
2579
2580   /* if we are scheduling a BYE, we only want to count bye packets, else we
2581    * count everything */
2582   if (sess->scheduled_bye && is_bye) {
2583     sess->bye_stats.bye_members++;
2584     UPDATE_AVG (sess->bye_stats.avg_rtcp_packet_size, pinfo.bytes);
2585   }
2586
2587   /* keep track of average packet size */
2588   UPDATE_AVG (sess->stats.avg_rtcp_packet_size, pinfo.bytes);
2589
2590   GST_DEBUG ("%p, received RTCP packet, avg size %u, %u", &sess->stats,
2591       sess->stats.avg_rtcp_packet_size, pinfo.bytes);
2592   RTP_SESSION_UNLOCK (sess);
2593
2594   pinfo.data = NULL;
2595   clean_packet_info (&pinfo);
2596
2597   /* notify caller of sr packets in the callback */
2598   if (do_sync && sess->callbacks.sync_rtcp) {
2599     result = sess->callbacks.sync_rtcp (sess, buffer,
2600         sess->sync_rtcp_user_data);
2601   } else
2602     gst_buffer_unref (buffer);
2603
2604   return result;
2605
2606   /* ERRORS */
2607 invalid_packet:
2608   {
2609     GST_DEBUG ("invalid RTCP packet received");
2610     gst_buffer_unref (buffer);
2611     return GST_FLOW_OK;
2612   }
2613 }
2614
2615 /**
2616  * rtp_session_update_send_caps:
2617  * @sess: an #RTPSession
2618  * @caps: a #GstCaps
2619  *
2620  * Update the caps of the sender in the rtp session.
2621  */
2622 void
2623 rtp_session_update_send_caps (RTPSession * sess, GstCaps * caps)
2624 {
2625   GstStructure *s;
2626   guint ssrc;
2627
2628   g_return_if_fail (RTP_IS_SESSION (sess));
2629   g_return_if_fail (GST_IS_CAPS (caps));
2630
2631   GST_LOG ("received caps %" GST_PTR_FORMAT, caps);
2632
2633   s = gst_caps_get_structure (caps, 0);
2634
2635   if (gst_structure_get_uint (s, "ssrc", &ssrc)) {
2636     RTPSource *source;
2637     gboolean created;
2638
2639     RTP_SESSION_LOCK (sess);
2640     source = obtain_internal_source (sess, ssrc, &created);
2641     if (source) {
2642       rtp_source_update_caps (source, caps);
2643       g_object_unref (source);
2644     }
2645     RTP_SESSION_UNLOCK (sess);
2646   }
2647 }
2648
2649 /**
2650  * rtp_session_send_rtp:
2651  * @sess: an #RTPSession
2652  * @data: pointer to either an RTP buffer or a list of RTP buffers
2653  * @is_list: TRUE when @data is a buffer list
2654  * @current_time: the current system time
2655  * @running_time: the running time of @data
2656  *
2657  * Send the RTP buffer in the session manager. This function takes ownership of
2658  * @buffer.
2659  *
2660  * Returns: a #GstFlowReturn.
2661  */
2662 GstFlowReturn
2663 rtp_session_send_rtp (RTPSession * sess, gpointer data, gboolean is_list,
2664     GstClockTime current_time, GstClockTime running_time)
2665 {
2666   GstFlowReturn result;
2667   RTPSource *source;
2668   gboolean prevsender;
2669   guint64 oldrate;
2670   RTPPacketInfo pinfo = { 0, };
2671   gboolean created;
2672
2673   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
2674   g_return_val_if_fail (is_list || GST_IS_BUFFER (data), GST_FLOW_ERROR);
2675
2676   GST_LOG ("received RTP %s for sending", is_list ? "list" : "packet");
2677
2678   RTP_SESSION_LOCK (sess);
2679   if (!update_packet_info (sess, &pinfo, TRUE, TRUE, is_list, data,
2680           current_time, running_time, -1))
2681     goto invalid_packet;
2682
2683   source = obtain_internal_source (sess, pinfo.ssrc, &created);
2684
2685   /* update last activity */
2686   source->last_rtp_activity = current_time;
2687
2688   prevsender = RTP_SOURCE_IS_SENDER (source);
2689   oldrate = source->bitrate;
2690
2691   /* we use our own source to send */
2692   result = rtp_source_send_rtp (source, &pinfo);
2693
2694   source_update_sender (sess, source, prevsender);
2695
2696   if (oldrate != source->bitrate)
2697     sess->recalc_bandwidth = TRUE;
2698   RTP_SESSION_UNLOCK (sess);
2699
2700   g_object_unref (source);
2701   clean_packet_info (&pinfo);
2702
2703   return result;
2704
2705 invalid_packet:
2706   {
2707     gst_mini_object_unref (GST_MINI_OBJECT_CAST (data));
2708     RTP_SESSION_UNLOCK (sess);
2709     GST_DEBUG ("invalid RTP packet received");
2710     return GST_FLOW_OK;
2711   }
2712 }
2713
2714 static void
2715 add_bitrates (gpointer key, RTPSource * source, gdouble * bandwidth)
2716 {
2717   *bandwidth += source->bitrate;
2718 }
2719
2720 /* must be called with session lock */
2721 static GstClockTime
2722 calculate_rtcp_interval (RTPSession * sess, gboolean deterministic,
2723     gboolean first)
2724 {
2725   GstClockTime result;
2726   RTPSessionStats *stats;
2727
2728   /* recalculate bandwidth when it changed */
2729   if (sess->recalc_bandwidth) {
2730     gdouble bandwidth;
2731
2732     if (sess->bandwidth > 0)
2733       bandwidth = sess->bandwidth;
2734     else {
2735       /* If it is <= 0, then try to estimate the actual bandwidth */
2736       bandwidth = 0;
2737
2738       g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
2739           (GHFunc) add_bitrates, &bandwidth);
2740       bandwidth /= 8.0;
2741     }
2742     if (bandwidth < 8000)
2743       bandwidth = RTP_STATS_BANDWIDTH;
2744
2745     rtp_stats_set_bandwidths (&sess->stats, bandwidth,
2746         sess->rtcp_bandwidth, sess->rtcp_rs_bandwidth, sess->rtcp_rr_bandwidth);
2747
2748     sess->recalc_bandwidth = FALSE;
2749   }
2750
2751   if (sess->scheduled_bye) {
2752     stats = &sess->bye_stats;
2753     result = rtp_stats_calculate_bye_interval (stats);
2754   } else {
2755     stats = &sess->stats;
2756     result = rtp_stats_calculate_rtcp_interval (stats,
2757         stats->internal_sender_sources > 0, first);
2758   }
2759
2760   GST_DEBUG ("next deterministic interval: %" GST_TIME_FORMAT ", first %d",
2761       GST_TIME_ARGS (result), first);
2762
2763   if (!deterministic && result != GST_CLOCK_TIME_NONE)
2764     result = rtp_stats_add_rtcp_jitter (stats, result);
2765
2766   GST_DEBUG ("next interval: %" GST_TIME_FORMAT, GST_TIME_ARGS (result));
2767
2768   return result;
2769 }
2770
2771 static void
2772 source_mark_bye (const gchar * key, RTPSource * source, const gchar * reason)
2773 {
2774   if (source->internal)
2775     rtp_source_mark_bye (source, reason);
2776 }
2777
2778 /**
2779  * rtp_session_mark_all_bye:
2780  * @sess: an #RTPSession
2781  * @reason: a reason
2782  *
2783  * Mark all internal sources of the session as BYE with @reason.
2784  */
2785 void
2786 rtp_session_mark_all_bye (RTPSession * sess, const gchar * reason)
2787 {
2788   g_return_if_fail (RTP_IS_SESSION (sess));
2789
2790   RTP_SESSION_LOCK (sess);
2791   g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
2792       (GHFunc) source_mark_bye, (gpointer) reason);
2793   RTP_SESSION_UNLOCK (sess);
2794 }
2795
2796 /* Stop the current @sess and schedule a BYE message for the other members.
2797  * One must have the session lock to call this function
2798  */
2799 static GstFlowReturn
2800 rtp_session_schedule_bye_locked (RTPSession * sess, GstClockTime current_time)
2801 {
2802   GstFlowReturn result = GST_FLOW_OK;
2803   GstClockTime interval;
2804
2805   /* nothing to do it we already scheduled bye */
2806   if (sess->scheduled_bye)
2807     goto done;
2808
2809   /* we schedule BYE now */
2810   sess->scheduled_bye = TRUE;
2811   /* at least one member wants to send a BYE */
2812   memcpy (&sess->bye_stats, &sess->stats, sizeof (RTPSessionStats));
2813   INIT_AVG (sess->bye_stats.avg_rtcp_packet_size, 100);
2814   sess->bye_stats.bye_members = 1;
2815   sess->first_rtcp = TRUE;
2816   sess->allow_early = TRUE;
2817
2818   /* reschedule transmission */
2819   sess->last_rtcp_send_time = current_time;
2820   interval = calculate_rtcp_interval (sess, FALSE, TRUE);
2821
2822   if (interval != GST_CLOCK_TIME_NONE)
2823     sess->next_rtcp_check_time = current_time + interval;
2824   else
2825     sess->next_rtcp_check_time = GST_CLOCK_TIME_NONE;
2826
2827   GST_DEBUG ("Schedule BYE for %" GST_TIME_FORMAT ", %" GST_TIME_FORMAT,
2828       GST_TIME_ARGS (interval), GST_TIME_ARGS (sess->next_rtcp_check_time));
2829
2830   RTP_SESSION_UNLOCK (sess);
2831   /* notify app of reconsideration */
2832   if (sess->callbacks.reconsider)
2833     sess->callbacks.reconsider (sess, sess->reconsider_user_data);
2834   RTP_SESSION_LOCK (sess);
2835 done:
2836
2837   return result;
2838 }
2839
2840 /**
2841  * rtp_session_schedule_bye:
2842  * @sess: an #RTPSession
2843  * @current_time: the current system time
2844  *
2845  * Schedule a BYE message for all sources marked as BYE in @sess.
2846  *
2847  * Returns: a #GstFlowReturn.
2848  */
2849 GstFlowReturn
2850 rtp_session_schedule_bye (RTPSession * sess, GstClockTime current_time)
2851 {
2852   GstFlowReturn result = GST_FLOW_OK;
2853
2854   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
2855
2856   RTP_SESSION_LOCK (sess);
2857   result = rtp_session_schedule_bye_locked (sess, current_time);
2858   RTP_SESSION_UNLOCK (sess);
2859
2860   return result;
2861 }
2862
2863 /**
2864  * rtp_session_next_timeout:
2865  * @sess: an #RTPSession
2866  * @current_time: the current system time
2867  *
2868  * Get the next time we should perform session maintenance tasks.
2869  *
2870  * Returns: a time when rtp_session_on_timeout() should be called with the
2871  * current system time.
2872  */
2873 GstClockTime
2874 rtp_session_next_timeout (RTPSession * sess, GstClockTime current_time)
2875 {
2876   GstClockTime result, interval = 0;
2877
2878   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_CLOCK_TIME_NONE);
2879
2880   RTP_SESSION_LOCK (sess);
2881
2882   if (GST_CLOCK_TIME_IS_VALID (sess->next_early_rtcp_time)) {
2883     GST_DEBUG ("have early rtcp time");
2884     result = sess->next_early_rtcp_time;
2885     goto early_exit;
2886   }
2887
2888   result = sess->next_rtcp_check_time;
2889
2890   GST_DEBUG ("current time: %" GST_TIME_FORMAT
2891       ", next time: %" GST_TIME_FORMAT,
2892       GST_TIME_ARGS (current_time), GST_TIME_ARGS (result));
2893
2894   if (result == GST_CLOCK_TIME_NONE || result < current_time) {
2895     GST_DEBUG ("take current time as base");
2896     /* our previous check time expired, start counting from the current time
2897      * again. */
2898     result = current_time;
2899   }
2900
2901   if (sess->scheduled_bye) {
2902     if (sess->bye_stats.active_sources >= 50) {
2903       GST_DEBUG ("reconsider BYE, more than 50 sources");
2904       /* reconsider BYE if members >= 50 */
2905       interval = calculate_rtcp_interval (sess, FALSE, TRUE);
2906     }
2907   } else {
2908     if (sess->first_rtcp) {
2909       GST_DEBUG ("first RTCP packet");
2910       /* we are called for the first time */
2911       interval = calculate_rtcp_interval (sess, FALSE, TRUE);
2912     } else if (sess->next_rtcp_check_time < current_time) {
2913       GST_DEBUG ("old check time expired, getting new timeout");
2914       /* get a new timeout when we need to */
2915       interval = calculate_rtcp_interval (sess, FALSE, FALSE);
2916     }
2917   }
2918
2919   if (interval != GST_CLOCK_TIME_NONE)
2920     result += interval;
2921   else
2922     result = GST_CLOCK_TIME_NONE;
2923
2924   sess->next_rtcp_check_time = result;
2925
2926 early_exit:
2927
2928   GST_DEBUG ("current time: %" GST_TIME_FORMAT
2929       ", next time: %" GST_TIME_FORMAT,
2930       GST_TIME_ARGS (current_time), GST_TIME_ARGS (result));
2931   RTP_SESSION_UNLOCK (sess);
2932
2933   return result;
2934 }
2935
2936 typedef struct
2937 {
2938   RTPSource *source;
2939   gboolean is_bye;
2940   GstBuffer *buffer;
2941 } ReportOutput;
2942
2943 typedef struct
2944 {
2945   GstRTCPBuffer rtcpbuf;
2946   RTPSession *sess;
2947   RTPSource *source;
2948   guint num_to_report;
2949   gboolean have_fir;
2950   gboolean have_pli;
2951   gboolean have_nack;
2952   GstBuffer *rtcp;
2953   GstClockTime current_time;
2954   guint64 ntpnstime;
2955   GstClockTime running_time;
2956   GstClockTime interval;
2957   GstRTCPPacket packet;
2958   gboolean has_sdes;
2959   gboolean is_early;
2960   gboolean may_suppress;
2961   GQueue output;
2962   guint nacked_seqnums;
2963 } ReportData;
2964
2965 static void
2966 session_start_rtcp (RTPSession * sess, ReportData * data)
2967 {
2968   GstRTCPPacket *packet = &data->packet;
2969   RTPSource *own = data->source;
2970   GstRTCPBuffer *rtcp = &data->rtcpbuf;
2971
2972   data->rtcp = gst_rtcp_buffer_new (sess->mtu);
2973   data->has_sdes = FALSE;
2974
2975   gst_rtcp_buffer_map (data->rtcp, GST_MAP_READWRITE, rtcp);
2976
2977   if (RTP_SOURCE_IS_SENDER (own)) {
2978     guint64 ntptime;
2979     guint32 rtptime;
2980     guint32 packet_count, octet_count;
2981
2982     /* we are a sender, create SR */
2983     GST_DEBUG ("create SR for SSRC %08x", own->ssrc);
2984     gst_rtcp_buffer_add_packet (rtcp, GST_RTCP_TYPE_SR, packet);
2985
2986     /* get latest stats */
2987     rtp_source_get_new_sr (own, data->ntpnstime, data->running_time,
2988         &ntptime, &rtptime, &packet_count, &octet_count);
2989     /* store stats */
2990     rtp_source_process_sr (own, data->current_time, ntptime, rtptime,
2991         packet_count, octet_count);
2992
2993     /* fill in sender report info */
2994     gst_rtcp_packet_sr_set_sender_info (packet, own->ssrc,
2995         ntptime, rtptime, packet_count, octet_count);
2996   } else {
2997     /* we are only receiver, create RR */
2998     GST_DEBUG ("create RR for SSRC %08x", own->ssrc);
2999     gst_rtcp_buffer_add_packet (rtcp, GST_RTCP_TYPE_RR, packet);
3000     gst_rtcp_packet_rr_set_ssrc (packet, own->ssrc);
3001   }
3002 }
3003
3004 /* construct a Sender or Receiver Report */
3005 static void
3006 session_report_blocks (const gchar * key, RTPSource * source, ReportData * data)
3007 {
3008   RTPSession *sess = data->sess;
3009   GstRTCPPacket *packet = &data->packet;
3010   guint8 fractionlost;
3011   gint32 packetslost;
3012   guint32 exthighestseq, jitter;
3013   guint32 lsr, dlsr;
3014
3015   /* don't report for sources in future generations */
3016   if (((gint16) (source->generation - sess->generation)) > 0) {
3017     GST_DEBUG ("source %08x generation %u > %u", source->ssrc,
3018         source->generation, sess->generation);
3019     return;
3020   }
3021
3022   if (g_hash_table_contains (source->reported_in_sr_of,
3023           GUINT_TO_POINTER (data->source->ssrc))) {
3024     GST_DEBUG ("source %08x already reported in this generation", source->ssrc);
3025     return;
3026   }
3027
3028   if (gst_rtcp_packet_get_rb_count (packet) == GST_RTCP_MAX_RB_COUNT) {
3029     GST_DEBUG ("max RB count reached");
3030     return;
3031   }
3032
3033   /* only report about other sender */
3034   if (source == data->source)
3035     goto reported;
3036
3037   if (!RTP_SOURCE_IS_SENDER (source)) {
3038     GST_DEBUG ("source %08x not sender", source->ssrc);
3039     goto reported;
3040   }
3041
3042   GST_DEBUG ("create RB for SSRC %08x", source->ssrc);
3043
3044   /* get new stats */
3045   rtp_source_get_new_rb (source, data->current_time, &fractionlost,
3046       &packetslost, &exthighestseq, &jitter, &lsr, &dlsr);
3047
3048   /* store last generated RR packet */
3049   source->last_rr.is_valid = TRUE;
3050   source->last_rr.fractionlost = fractionlost;
3051   source->last_rr.packetslost = packetslost;
3052   source->last_rr.exthighestseq = exthighestseq;
3053   source->last_rr.jitter = jitter;
3054   source->last_rr.lsr = lsr;
3055   source->last_rr.dlsr = dlsr;
3056
3057   /* packet is not yet filled, add report block for this source. */
3058   gst_rtcp_packet_add_rb (packet, source->ssrc, fractionlost, packetslost,
3059       exthighestseq, jitter, lsr, dlsr);
3060
3061 reported:
3062   g_hash_table_add (source->reported_in_sr_of,
3063       GUINT_TO_POINTER (data->source->ssrc));
3064 }
3065
3066 /* construct FIR */
3067 static void
3068 session_add_fir (const gchar * key, RTPSource * source, ReportData * data)
3069 {
3070   GstRTCPPacket *packet = &data->packet;
3071   guint16 len;
3072   guint8 *fci_data;
3073
3074   if (!source->send_fir)
3075     return;
3076
3077   len = gst_rtcp_packet_fb_get_fci_length (packet);
3078   if (!gst_rtcp_packet_fb_set_fci_length (packet, len + 2))
3079     /* exit because the packet is full, will put next request in a
3080      * further packet */
3081     return;
3082
3083   fci_data = gst_rtcp_packet_fb_get_fci (packet) + (len * 4);
3084
3085   GST_WRITE_UINT32_BE (fci_data, source->ssrc);
3086   fci_data += 4;
3087   fci_data[0] = source->current_send_fir_seqnum;
3088   fci_data[1] = fci_data[2] = fci_data[3] = 0;
3089
3090   source->send_fir = FALSE;
3091 }
3092
3093 static void
3094 session_fir (RTPSession * sess, ReportData * data)
3095 {
3096   GstRTCPBuffer *rtcp = &data->rtcpbuf;
3097   GstRTCPPacket *packet = &data->packet;
3098
3099   if (!gst_rtcp_buffer_add_packet (rtcp, GST_RTCP_TYPE_PSFB, packet))
3100     return;
3101
3102   gst_rtcp_packet_fb_set_type (packet, GST_RTCP_PSFB_TYPE_FIR);
3103   gst_rtcp_packet_fb_set_sender_ssrc (packet, data->source->ssrc);
3104   gst_rtcp_packet_fb_set_media_ssrc (packet, 0);
3105
3106   g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
3107       (GHFunc) session_add_fir, data);
3108
3109   if (gst_rtcp_packet_fb_get_fci_length (packet) == 0)
3110     gst_rtcp_packet_remove (packet);
3111   else
3112     data->may_suppress = FALSE;
3113 }
3114
3115 static gboolean
3116 has_pli_compare_func (gconstpointer a, gconstpointer ignored)
3117 {
3118   GstRTCPPacket packet;
3119   GstRTCPBuffer rtcp = { NULL, };
3120   gboolean ret = FALSE;
3121
3122   gst_rtcp_buffer_map ((GstBuffer *) a, GST_MAP_READ, &rtcp);
3123
3124   if (gst_rtcp_buffer_get_first_packet (&rtcp, &packet)) {
3125     if (gst_rtcp_packet_get_type (&packet) == GST_RTCP_TYPE_PSFB &&
3126         gst_rtcp_packet_fb_get_type (&packet) == GST_RTCP_PSFB_TYPE_PLI)
3127       ret = TRUE;
3128   }
3129
3130   gst_rtcp_buffer_unmap (&rtcp);
3131
3132   return ret;
3133 }
3134
3135 /* construct PLI */
3136 static void
3137 session_pli (const gchar * key, RTPSource * source, ReportData * data)
3138 {
3139   GstRTCPBuffer *rtcp = &data->rtcpbuf;
3140   GstRTCPPacket *packet = &data->packet;
3141
3142   if (!source->send_pli)
3143     return;
3144
3145   if (rtp_source_has_retained (source, has_pli_compare_func, NULL))
3146     return;
3147
3148   if (!gst_rtcp_buffer_add_packet (rtcp, GST_RTCP_TYPE_PSFB, packet))
3149     /* exit because the packet is full, will put next request in a
3150      * further packet */
3151     return;
3152
3153   gst_rtcp_packet_fb_set_type (packet, GST_RTCP_PSFB_TYPE_PLI);
3154   gst_rtcp_packet_fb_set_sender_ssrc (packet, data->source->ssrc);
3155   gst_rtcp_packet_fb_set_media_ssrc (packet, source->ssrc);
3156
3157   source->send_pli = FALSE;
3158   data->may_suppress = FALSE;
3159 }
3160
3161 /* construct NACK */
3162 static void
3163 session_nack (const gchar * key, RTPSource * source, ReportData * data)
3164 {
3165   GstRTCPBuffer *rtcp = &data->rtcpbuf;
3166   GstRTCPPacket *packet = &data->packet;
3167   guint32 *nacks;
3168   guint n_nacks, i;
3169   guint8 *fci_data;
3170
3171   if (!source->send_nack)
3172     return;
3173
3174   if (!gst_rtcp_buffer_add_packet (rtcp, GST_RTCP_TYPE_RTPFB, packet))
3175     /* exit because the packet is full, will put next request in a
3176      * further packet */
3177     return;
3178
3179   gst_rtcp_packet_fb_set_type (packet, GST_RTCP_RTPFB_TYPE_NACK);
3180   gst_rtcp_packet_fb_set_sender_ssrc (packet, data->source->ssrc);
3181   gst_rtcp_packet_fb_set_media_ssrc (packet, source->ssrc);
3182
3183   nacks = rtp_source_get_nacks (source, &n_nacks);
3184   GST_DEBUG ("%u NACKs", n_nacks);
3185   if (!gst_rtcp_packet_fb_set_fci_length (packet, n_nacks))
3186     return;
3187
3188   fci_data = gst_rtcp_packet_fb_get_fci (packet);
3189   for (i = 0; i < n_nacks; i++) {
3190     GST_WRITE_UINT32_BE (fci_data, nacks[i]);
3191     fci_data += 4;
3192     data->nacked_seqnums++;
3193   }
3194
3195   rtp_source_clear_nacks (source);
3196   data->may_suppress = FALSE;
3197 }
3198
3199 /* perform cleanup of sources that timed out */
3200 static void
3201 session_cleanup (const gchar * key, RTPSource * source, ReportData * data)
3202 {
3203   gboolean remove = FALSE;
3204   gboolean byetimeout = FALSE;
3205   gboolean sendertimeout = FALSE;
3206   gboolean is_sender, is_active;
3207   RTPSession *sess = data->sess;
3208   GstClockTime interval, binterval;
3209   GstClockTime btime;
3210
3211   GST_DEBUG ("look at %08x, generation %u", source->ssrc, source->generation);
3212
3213   /* check for outdated collisions */
3214   if (source->internal) {
3215     GST_DEBUG ("Timing out collisions for %x", source->ssrc);
3216     rtp_source_timeout (source, data->current_time,
3217         data->running_time - sess->rtcp_feedback_retention_window);
3218   }
3219
3220   /* nothing else to do when without RTCP */
3221   if (data->interval == GST_CLOCK_TIME_NONE)
3222     return;
3223
3224   is_sender = RTP_SOURCE_IS_SENDER (source);
3225   is_active = RTP_SOURCE_IS_ACTIVE (source);
3226
3227   /* our own rtcp interval may have been forced low by secondary configuration,
3228    * while sender side may still operate with higher interval,
3229    * so do not just take our interval to decide on timing out sender,
3230    * but take (if data->interval <= 5 * GST_SECOND):
3231    *   interval = CLAMP (sender_interval, data->interval, 5 * GST_SECOND)
3232    * where sender_interval is difference between last 2 received RTCP reports
3233    */
3234   if (data->interval >= 5 * GST_SECOND || source->internal) {
3235     binterval = data->interval;
3236   } else {
3237     GST_LOG ("prev_rtcp %" GST_TIME_FORMAT ", last_rtcp %" GST_TIME_FORMAT,
3238         GST_TIME_ARGS (source->stats.prev_rtcptime),
3239         GST_TIME_ARGS (source->stats.last_rtcptime));
3240     /* if not received enough yet, fallback to larger default */
3241     if (source->stats.last_rtcptime > source->stats.prev_rtcptime)
3242       binterval = source->stats.last_rtcptime - source->stats.prev_rtcptime;
3243     else
3244       binterval = 5 * GST_SECOND;
3245     binterval = CLAMP (binterval, data->interval, 5 * GST_SECOND);
3246   }
3247   GST_LOG ("timeout base interval %" GST_TIME_FORMAT,
3248       GST_TIME_ARGS (binterval));
3249
3250   if (!source->internal && source->marked_bye) {
3251     /* if we received a BYE from the source, remove the source after some
3252      * time. */
3253     if (data->current_time > source->bye_time &&
3254         data->current_time - source->bye_time > sess->stats.bye_timeout) {
3255       GST_DEBUG ("removing BYE source %08x", source->ssrc);
3256       remove = TRUE;
3257       byetimeout = TRUE;
3258     }
3259   }
3260
3261   if (source->internal && source->sent_bye) {
3262     GST_DEBUG ("removing internal source that has sent BYE %08x", source->ssrc);
3263     remove = TRUE;
3264   }
3265
3266   /* sources that were inactive for more than 5 times the deterministic reporting
3267    * interval get timed out. the min timeout is 5 seconds. */
3268   /* mind old time that might pre-date last time going to PLAYING */
3269   btime = MAX (source->last_activity, sess->start_time);
3270   if (data->current_time > btime) {
3271     interval = MAX (binterval * 5, 5 * GST_SECOND);
3272     if (data->current_time - btime > interval) {
3273       GST_DEBUG ("removing timeout source %08x, last %" GST_TIME_FORMAT,
3274           source->ssrc, GST_TIME_ARGS (btime));
3275       if (source->internal) {
3276         /* this is an internal source that is not using our suggested ssrc.
3277          * since there must be another source using this ssrc, we can remove
3278          * this one instead of making it a receiver forever */
3279         if (source->ssrc != sess->suggested_ssrc) {
3280           rtp_source_mark_bye (source, "timed out");
3281           /* do not schedule bye here, since we are inside the RTCP timeout
3282            * processing and scheduling bye will interfere with SR/RR sending */
3283         }
3284       } else {
3285         remove = TRUE;
3286       }
3287     }
3288   }
3289
3290   /* senders that did not send for a long time become a receiver, this also
3291    * holds for our own sources. */
3292   if (is_sender) {
3293     /* mind old time that might pre-date last time going to PLAYING */
3294     btime = MAX (source->last_rtp_activity, sess->start_time);
3295     if (data->current_time > btime) {
3296       interval = MAX (binterval * 2, 5 * GST_SECOND);
3297       if (data->current_time - btime > interval) {
3298         if (source->internal && source->sent_bye) {
3299           /* an internal source is BYE and stopped sending RTP, remove */
3300           GST_DEBUG ("internal BYE source %08x timed out, last %"
3301               GST_TIME_FORMAT, source->ssrc, GST_TIME_ARGS (btime));
3302           remove = TRUE;
3303         } else {
3304           GST_DEBUG ("sender source %08x timed out and became receiver, last %"
3305               GST_TIME_FORMAT, source->ssrc, GST_TIME_ARGS (btime));
3306           sendertimeout = TRUE;
3307         }
3308       }
3309     }
3310   }
3311
3312   if (remove) {
3313     sess->total_sources--;
3314     if (is_sender) {
3315       sess->stats.sender_sources--;
3316       if (source->internal)
3317         sess->stats.internal_sender_sources--;
3318     }
3319     if (is_active)
3320       sess->stats.active_sources--;
3321
3322     if (source->internal)
3323       sess->stats.internal_sources--;
3324
3325     if (byetimeout)
3326       on_bye_timeout (sess, source);
3327     else
3328       on_timeout (sess, source);
3329   } else {
3330     if (sendertimeout) {
3331       source->is_sender = FALSE;
3332       sess->stats.sender_sources--;
3333       if (source->internal)
3334         sess->stats.internal_sender_sources--;
3335
3336       on_sender_timeout (sess, source);
3337     }
3338     /* count how many source to report in this generation */
3339     if (((gint16) (source->generation - sess->generation)) <= 0)
3340       data->num_to_report++;
3341   }
3342   source->closing = remove;
3343 }
3344
3345 static void
3346 session_sdes (RTPSession * sess, ReportData * data)
3347 {
3348   GstRTCPPacket *packet = &data->packet;
3349   const GstStructure *sdes;
3350   gint i, n_fields;
3351   GstRTCPBuffer *rtcp = &data->rtcpbuf;
3352
3353   /* add SDES packet */
3354   gst_rtcp_buffer_add_packet (rtcp, GST_RTCP_TYPE_SDES, packet);
3355
3356   gst_rtcp_packet_sdes_add_item (packet, data->source->ssrc);
3357
3358   sdes = rtp_source_get_sdes_struct (data->source);
3359
3360   /* add all fields in the structure, the order is not important. */
3361   n_fields = gst_structure_n_fields (sdes);
3362   for (i = 0; i < n_fields; ++i) {
3363     const gchar *field;
3364     const gchar *value;
3365     GstRTCPSDESType type;
3366
3367     field = gst_structure_nth_field_name (sdes, i);
3368     if (field == NULL)
3369       continue;
3370     value = gst_structure_get_string (sdes, field);
3371     if (value == NULL)
3372       continue;
3373     type = gst_rtcp_sdes_name_to_type (field);
3374
3375     /* Early packets are minimal and only include the CNAME */
3376     if (data->is_early && type != GST_RTCP_SDES_CNAME)
3377       continue;
3378
3379     if (type > GST_RTCP_SDES_END && type < GST_RTCP_SDES_PRIV) {
3380       gst_rtcp_packet_sdes_add_entry (packet, type, strlen (value),
3381           (const guint8 *) value);
3382     } else if (type == GST_RTCP_SDES_PRIV) {
3383       gsize prefix_len;
3384       gsize value_len;
3385       gsize data_len;
3386       guint8 data[256];
3387
3388       /* don't accept entries that are too big */
3389       prefix_len = strlen (field);
3390       if (prefix_len > 255)
3391         continue;
3392       value_len = strlen (value);
3393       if (value_len > 255)
3394         continue;
3395       data_len = 1 + prefix_len + value_len;
3396       if (data_len > 255)
3397         continue;
3398
3399       data[0] = prefix_len;
3400       memcpy (&data[1], field, prefix_len);
3401       memcpy (&data[1 + prefix_len], value, value_len);
3402
3403       gst_rtcp_packet_sdes_add_entry (packet, type, data_len, data);
3404     }
3405   }
3406
3407   data->has_sdes = TRUE;
3408 }
3409
3410 /* schedule a BYE packet */
3411 static void
3412 make_source_bye (RTPSession * sess, RTPSource * source, ReportData * data)
3413 {
3414   GstRTCPPacket *packet = &data->packet;
3415   GstRTCPBuffer *rtcp = &data->rtcpbuf;
3416
3417   /* add SDES */
3418   session_sdes (sess, data);
3419   /* add a BYE packet */
3420   gst_rtcp_buffer_add_packet (rtcp, GST_RTCP_TYPE_BYE, packet);
3421   gst_rtcp_packet_bye_add_ssrc (packet, source->ssrc);
3422   if (source->bye_reason)
3423     gst_rtcp_packet_bye_set_reason (packet, source->bye_reason);
3424
3425   /* we have a BYE packet now */
3426   source->sent_bye = TRUE;
3427 }
3428
3429 static gboolean
3430 is_rtcp_time (RTPSession * sess, GstClockTime current_time, ReportData * data)
3431 {
3432   GstClockTime new_send_time, elapsed;
3433   GstClockTime interval;
3434   RTPSessionStats *stats;
3435
3436   if (sess->scheduled_bye)
3437     stats = &sess->bye_stats;
3438   else
3439     stats = &sess->stats;
3440
3441   if (GST_CLOCK_TIME_IS_VALID (sess->next_early_rtcp_time))
3442     data->is_early = TRUE;
3443   else
3444     data->is_early = FALSE;
3445
3446   if (data->is_early && sess->next_early_rtcp_time < current_time) {
3447     GST_DEBUG ("early feedback %" GST_TIME_FORMAT " < now %"
3448         GST_TIME_FORMAT, GST_TIME_ARGS (sess->next_early_rtcp_time),
3449         GST_TIME_ARGS (current_time));
3450     goto early;
3451   }
3452
3453   /* no need to check yet */
3454   if (sess->next_rtcp_check_time == GST_CLOCK_TIME_NONE ||
3455       sess->next_rtcp_check_time > current_time) {
3456     GST_DEBUG ("no check time yet, next %" GST_TIME_FORMAT " > now %"
3457         GST_TIME_FORMAT, GST_TIME_ARGS (sess->next_rtcp_check_time),
3458         GST_TIME_ARGS (current_time));
3459     return FALSE;
3460   }
3461
3462 early:
3463   /* get elapsed time since we last reported */
3464   elapsed = current_time - sess->last_rtcp_send_time;
3465
3466   /* take interval and add jitter */
3467   interval = data->interval;
3468   if (interval != GST_CLOCK_TIME_NONE)
3469     interval = rtp_stats_add_rtcp_jitter (stats, interval);
3470
3471   /* perform forward reconsideration */
3472   if (interval != GST_CLOCK_TIME_NONE) {
3473     GST_DEBUG ("forward reconsideration %" GST_TIME_FORMAT ", elapsed %"
3474         GST_TIME_FORMAT, GST_TIME_ARGS (interval), GST_TIME_ARGS (elapsed));
3475     new_send_time = interval + sess->last_rtcp_send_time;
3476   } else {
3477     new_send_time = sess->last_rtcp_send_time;
3478   }
3479
3480   if (!data->is_early) {
3481     /* check if reconsideration */
3482     if (new_send_time == GST_CLOCK_TIME_NONE || current_time < new_send_time) {
3483       GST_DEBUG ("reconsider RTCP for %" GST_TIME_FORMAT,
3484           GST_TIME_ARGS (new_send_time));
3485       /* store new check time */
3486       sess->next_rtcp_check_time = new_send_time;
3487       return FALSE;
3488     }
3489     sess->next_rtcp_check_time = current_time + interval;
3490   } else if (interval != GST_CLOCK_TIME_NONE) {
3491     /* Apply the rules from RFC 4585 section 3.5.3 */
3492     if (stats->min_interval != 0 && !sess->first_rtcp) {
3493       GstClockTime T_rr_current_interval =
3494           g_random_double_range (0.5, 1.5) * stats->min_interval;
3495
3496       /* This will caused the RTCP to be suppressed if no FB packets are added */
3497       if (sess->last_rtcp_send_time + T_rr_current_interval > new_send_time) {
3498         GST_DEBUG ("RTCP packet could be suppressed min: %" GST_TIME_FORMAT
3499             " last: %" GST_TIME_FORMAT
3500             " + T_rr_current_interval: %" GST_TIME_FORMAT
3501             " >  new_send_time: %" GST_TIME_FORMAT,
3502             GST_TIME_ARGS (stats->min_interval),
3503             GST_TIME_ARGS (sess->last_rtcp_send_time),
3504             GST_TIME_ARGS (T_rr_current_interval),
3505             GST_TIME_ARGS (new_send_time));
3506         data->may_suppress = TRUE;
3507       }
3508     }
3509   }
3510
3511   GST_DEBUG ("can send RTCP now, next interval %" GST_TIME_FORMAT,
3512       GST_TIME_ARGS (new_send_time));
3513
3514   return TRUE;
3515 }
3516
3517 static void
3518 clone_ssrcs_hashtable (gchar * key, RTPSource * source, GHashTable * hash_table)
3519 {
3520   g_hash_table_insert (hash_table, key, g_object_ref (source));
3521 }
3522
3523 static gboolean
3524 remove_closing_sources (const gchar * key, RTPSource * source,
3525     ReportData * data)
3526 {
3527   if (source->closing)
3528     return TRUE;
3529
3530   if (source->send_fir)
3531     data->have_fir = TRUE;
3532   if (source->send_pli)
3533     data->have_pli = TRUE;
3534   if (source->send_nack)
3535     data->have_nack = TRUE;
3536
3537   return FALSE;
3538 }
3539
3540 static void
3541 generate_rtcp (const gchar * key, RTPSource * source, ReportData * data)
3542 {
3543   RTPSession *sess = data->sess;
3544   gboolean is_bye = FALSE;
3545   ReportOutput *output;
3546
3547   /* only generate RTCP for active internal sources */
3548   if (!source->internal || source->sent_bye)
3549     return;
3550
3551   /* ignore other sources when we do the timeout after a scheduled BYE */
3552   if (sess->scheduled_bye && !source->marked_bye)
3553     return;
3554
3555   data->source = source;
3556
3557   /* open packet */
3558   session_start_rtcp (sess, data);
3559
3560   if (source->marked_bye) {
3561     /* send BYE */
3562     make_source_bye (sess, source, data);
3563     is_bye = TRUE;
3564   } else if (!data->is_early) {
3565     /* loop over all known sources and add report blocks. If we are early, we
3566      * just make a minimal RTCP packet and skip this step */
3567     g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
3568         (GHFunc) session_report_blocks, data);
3569   }
3570   if (!data->has_sdes)
3571     session_sdes (sess, data);
3572
3573   if (data->have_fir)
3574     session_fir (sess, data);
3575
3576   if (data->have_pli)
3577     g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
3578         (GHFunc) session_pli, data);
3579
3580   if (data->have_nack)
3581     g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
3582         (GHFunc) session_nack, data);
3583
3584   gst_rtcp_buffer_unmap (&data->rtcpbuf);
3585
3586   output = g_slice_new (ReportOutput);
3587   output->source = g_object_ref (source);
3588   output->is_bye = is_bye;
3589   output->buffer = data->rtcp;
3590   /* queue the RTCP packet to push later */
3591   g_queue_push_tail (&data->output, output);
3592 }
3593
3594 static void
3595 update_generation (const gchar * key, RTPSource * source, ReportData * data)
3596 {
3597   RTPSession *sess = data->sess;
3598
3599   if (g_hash_table_size (source->reported_in_sr_of) >=
3600       sess->stats.internal_sources) {
3601     /* source is reported, move to next generation */
3602     source->generation = sess->generation + 1;
3603     g_hash_table_remove_all (source->reported_in_sr_of);
3604
3605     GST_LOG ("reported source %x, new generation: %d", source->ssrc,
3606         source->generation);
3607
3608     /* if we reported all sources in this generation, move to next */
3609     if (--data->num_to_report == 0) {
3610       sess->generation++;
3611       GST_DEBUG ("all reported, generation now %u", sess->generation);
3612     }
3613   }
3614 }
3615
3616 /**
3617  * rtp_session_on_timeout:
3618  * @sess: an #RTPSession
3619  * @current_time: the current system time
3620  * @ntpnstime: the current NTP time in nanoseconds
3621  * @running_time: the current running_time of the pipeline
3622  *
3623  * Perform maintenance actions after the timeout obtained with
3624  * rtp_session_next_timeout() expired.
3625  *
3626  * This function will perform timeouts of receivers and senders, send a BYE
3627  * packet or generate RTCP packets with current session stats.
3628  *
3629  * This function can call the #RTPSessionSendRTCP callback, possibly multiple
3630  * times, for each packet that should be processed.
3631  *
3632  * Returns: a #GstFlowReturn.
3633  */
3634 GstFlowReturn
3635 rtp_session_on_timeout (RTPSession * sess, GstClockTime current_time,
3636     guint64 ntpnstime, GstClockTime running_time)
3637 {
3638   GstFlowReturn result = GST_FLOW_OK;
3639   ReportData data = { GST_RTCP_BUFFER_INIT };
3640   GHashTable *table_copy;
3641   ReportOutput *output;
3642
3643   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
3644
3645   GST_DEBUG ("reporting at %" GST_TIME_FORMAT ", NTP time %" GST_TIME_FORMAT
3646       ", running-time %" GST_TIME_FORMAT, GST_TIME_ARGS (current_time),
3647       GST_TIME_ARGS (ntpnstime), GST_TIME_ARGS (running_time));
3648
3649   data.sess = sess;
3650   data.current_time = current_time;
3651   data.ntpnstime = ntpnstime;
3652   data.running_time = running_time;
3653   data.num_to_report = 0;
3654   data.may_suppress = FALSE;
3655   data.nacked_seqnums = 0;
3656   g_queue_init (&data.output);
3657
3658   RTP_SESSION_LOCK (sess);
3659   /* get a new interval, we need this for various cleanups etc */
3660   data.interval = calculate_rtcp_interval (sess, TRUE, sess->first_rtcp);
3661
3662   GST_DEBUG ("interval %" GST_TIME_FORMAT, GST_TIME_ARGS (data.interval));
3663
3664   /* we need an internal source now */
3665   if (sess->stats.internal_sources == 0) {
3666     RTPSource *source;
3667     gboolean created;
3668
3669     source = obtain_internal_source (sess, sess->suggested_ssrc, &created);
3670     g_object_unref (source);
3671   }
3672
3673   sess->conflicting_addresses =
3674       timeout_conflicting_addresses (sess->conflicting_addresses, current_time);
3675
3676   /* Make a local copy of the hashtable. We need to do this because the
3677    * cleanup stage below releases the session lock. */
3678   table_copy = g_hash_table_new_full (NULL, NULL, NULL,
3679       (GDestroyNotify) g_object_unref);
3680   g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
3681       (GHFunc) clone_ssrcs_hashtable, table_copy);
3682
3683   /* Clean up the session, mark the source for removing, this might release the
3684    * session lock. */
3685   g_hash_table_foreach (table_copy, (GHFunc) session_cleanup, &data);
3686   g_hash_table_destroy (table_copy);
3687
3688   /* Now remove the marked sources */
3689   g_hash_table_foreach_remove (sess->ssrcs[sess->mask_idx],
3690       (GHRFunc) remove_closing_sources, &data);
3691
3692   /* update point-to-point status */
3693   session_update_ptp (sess);
3694
3695   /* see if we need to generate SR or RR packets */
3696   if (!is_rtcp_time (sess, current_time, &data))
3697     goto done;
3698
3699   GST_DEBUG ("doing RTCP generation %u for %u sources, early %d",
3700       sess->generation, data.num_to_report, data.is_early);
3701
3702   /* generate RTCP for all internal sources */
3703   g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
3704       (GHFunc) generate_rtcp, &data);
3705
3706   /* update the generation for all the sources that have been reported */
3707   g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
3708       (GHFunc) update_generation, &data);
3709
3710   /* we keep track of the last report time in order to timeout inactive
3711    * receivers or senders */
3712   if (!data.is_early && !data.may_suppress)
3713     sess->last_rtcp_send_time = data.current_time;
3714   sess->first_rtcp = FALSE;
3715   sess->next_early_rtcp_time = GST_CLOCK_TIME_NONE;
3716   sess->scheduled_bye = FALSE;
3717
3718 done:
3719   RTP_SESSION_UNLOCK (sess);
3720
3721   /* push out the RTCP packets */
3722   while ((output = g_queue_pop_head (&data.output))) {
3723     gboolean do_not_suppress;
3724     GstBuffer *buffer = output->buffer;
3725     RTPSource *source = output->source;
3726
3727     /* Give the user a change to add its own packet */
3728     g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_SENDING_RTCP], 0,
3729         buffer, data.is_early, &do_not_suppress);
3730
3731     if (sess->callbacks.send_rtcp && (do_not_suppress || !data.may_suppress)) {
3732       guint packet_size;
3733
3734       packet_size = gst_buffer_get_size (buffer) + sess->header_len;
3735
3736       UPDATE_AVG (sess->stats.avg_rtcp_packet_size, packet_size);
3737       GST_DEBUG ("%p, sending RTCP packet, avg size %u, %u", &sess->stats,
3738           sess->stats.avg_rtcp_packet_size, packet_size);
3739       result =
3740           sess->callbacks.send_rtcp (sess, source, buffer, output->is_bye,
3741           sess->send_rtcp_user_data);
3742       sess->stats.nacks_sent += data.nacked_seqnums;
3743     } else {
3744       GST_DEBUG ("freeing packet callback: %p"
3745           " do_not_suppress: %d may_suppress: %d",
3746           sess->callbacks.send_rtcp, do_not_suppress, data.may_suppress);
3747       sess->stats.nacks_dropped += data.nacked_seqnums;
3748       gst_buffer_unref (buffer);
3749     }
3750     g_object_unref (source);
3751     g_slice_free (ReportOutput, output);
3752   }
3753   return result;
3754 }
3755
3756 /**
3757  * rtp_session_request_early_rtcp:
3758  * @sess: an #RTPSession
3759  * @current_time: the current system time
3760  * @max_delay: maximum delay
3761  *
3762  * Request transmission of early RTCP
3763  */
3764 void
3765 rtp_session_request_early_rtcp (RTPSession * sess, GstClockTime current_time,
3766     GstClockTime max_delay)
3767 {
3768   GstClockTime T_dither_max;
3769
3770   /* Implements the algorithm described in RFC 4585 section 3.5.2 */
3771
3772   RTP_SESSION_LOCK (sess);
3773
3774   /* Check if already requested */
3775   /*  RFC 4585 section 3.5.2 step 2 */
3776   if (GST_CLOCK_TIME_IS_VALID (sess->next_early_rtcp_time)) {
3777     GST_LOG_OBJECT (sess, "already have next early rtcp time");
3778     goto dont_send;
3779   }
3780
3781   if (!GST_CLOCK_TIME_IS_VALID (sess->next_rtcp_check_time)) {
3782     GST_LOG_OBJECT (sess, "no next RTCP check time");
3783     goto dont_send;
3784   }
3785
3786   /* Ignore the request a scheduled packet will be in time anyway */
3787   if (current_time + max_delay > sess->next_rtcp_check_time) {
3788     GST_LOG_OBJECT (sess, "next scheduled time is soon %" GST_TIME_FORMAT " + %"
3789         GST_TIME_FORMAT " > %" GST_TIME_FORMAT,
3790         GST_TIME_ARGS (current_time),
3791         GST_TIME_ARGS (max_delay), GST_TIME_ARGS (sess->next_rtcp_check_time));
3792     goto dont_send;
3793   }
3794
3795   /*  RFC 4585 section 3.5.2 step 2b */
3796   /* If the total sources is <=2, then there is only us and one peer */
3797   /* When there is one auxiliary stream the session can still do point
3798    * to point.
3799    */
3800   if (sess->is_doing_ptp) {
3801     T_dither_max = 0;
3802   } else {
3803     /* Divide by 2 because l = 0.5 */
3804     T_dither_max = sess->next_rtcp_check_time - sess->last_rtcp_send_time;
3805     T_dither_max /= 2;
3806   }
3807
3808   /*  RFC 4585 section 3.5.2 step 3 */
3809   if (current_time + T_dither_max > sess->next_rtcp_check_time) {
3810     GST_LOG_OBJECT (sess, "don't send because of dither");
3811     goto dont_send;
3812   }
3813
3814   /*  RFC 4585 section 3.5.2 step 4
3815    * Don't send if allow_early is FALSE, but not if we are in
3816    * immediate mode, meaning we are part of a group of at most the
3817    * application-specific threshold.
3818    */
3819   if (sess->total_sources > sess->rtcp_immediate_feedback_threshold &&
3820       sess->allow_early == FALSE) {
3821     GST_LOG_OBJECT (sess, "can't allow early feedback");
3822     goto dont_send;
3823   }
3824
3825   if (T_dither_max) {
3826     /* Schedule an early transmission later */
3827     sess->next_early_rtcp_time = g_random_double () * T_dither_max +
3828         current_time;
3829   } else {
3830     /* If no dithering, schedule it for NOW */
3831     sess->next_early_rtcp_time = current_time;
3832   }
3833
3834   GST_LOG_OBJECT (sess, "next early RTCP time %" GST_TIME_FORMAT,
3835       GST_TIME_ARGS (sess->next_early_rtcp_time));
3836   RTP_SESSION_UNLOCK (sess);
3837
3838   /* notify app of need to send packet early
3839    * and therefore of timeout change */
3840   if (sess->callbacks.reconsider)
3841     sess->callbacks.reconsider (sess, sess->reconsider_user_data);
3842
3843   return;
3844
3845 dont_send:
3846
3847   RTP_SESSION_UNLOCK (sess);
3848 }
3849
3850 static void
3851 rtp_session_send_rtcp (RTPSession * sess, GstClockTime max_delay)
3852 {
3853   GstClockTime now;
3854
3855   if (!sess->callbacks.send_rtcp)
3856     return;
3857
3858   now = sess->callbacks.request_time (sess, sess->request_time_user_data);
3859
3860   rtp_session_request_early_rtcp (sess, now, max_delay);
3861 }
3862
3863 gboolean
3864 rtp_session_request_key_unit (RTPSession * sess, guint32 ssrc,
3865     gboolean fir, gint count)
3866 {
3867   RTPSource *src;
3868
3869   RTP_SESSION_LOCK (sess);
3870   src = find_source (sess, ssrc);
3871   if (src == NULL)
3872     goto no_source;
3873
3874   if (fir) {
3875     src->send_pli = FALSE;
3876     src->send_fir = TRUE;
3877
3878     if (count == -1 || count != src->last_fir_count)
3879       src->current_send_fir_seqnum++;
3880     src->last_fir_count = count;
3881   } else if (!src->send_fir) {
3882     src->send_pli = TRUE;
3883   }
3884   RTP_SESSION_UNLOCK (sess);
3885
3886   rtp_session_send_rtcp (sess, 200 * GST_MSECOND);
3887
3888   return TRUE;
3889
3890   /* ERRORS */
3891 no_source:
3892   {
3893     RTP_SESSION_UNLOCK (sess);
3894     return FALSE;
3895   }
3896 }
3897
3898 /**
3899  * rtp_session_request_nack:
3900  * @sess: a #RTPSession
3901  * @ssrc: the SSRC
3902  * @seqnum: the missing seqnum
3903  * @max_delay: max delay to request NACK
3904  *
3905  * Request scheduling of a NACK feedback packet for @seqnum in @ssrc.
3906  *
3907  * Returns: %TRUE if the NACK feedback could be scheduled
3908  */
3909 gboolean
3910 rtp_session_request_nack (RTPSession * sess, guint32 ssrc, guint16 seqnum,
3911     GstClockTime max_delay)
3912 {
3913   RTPSource *source;
3914
3915   RTP_SESSION_LOCK (sess);
3916   source = find_source (sess, ssrc);
3917   if (source == NULL)
3918     goto no_source;
3919
3920   GST_DEBUG ("request NACK for %08x, #%u", ssrc, seqnum);
3921   rtp_source_register_nack (source, seqnum);
3922   RTP_SESSION_UNLOCK (sess);
3923
3924   rtp_session_send_rtcp (sess, max_delay);
3925
3926   return TRUE;
3927
3928   /* ERRORS */
3929 no_source:
3930   {
3931     RTP_SESSION_UNLOCK (sess);
3932     return FALSE;
3933   }
3934 }