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