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