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 }
 773
 774 /**
 775  * rtp_session_set_process_rtp_callback:
 776  * @sess: an #RTPSession
 777  * @callback: callback to set
 778  * @user_data: user data passed in the callback
 779  *
 780  * Configure only the process_rtp callback to be notified of the process_rtp action.
 781  */
 782 void
 783 rtp_session_set_process_rtp_callback (RTPSession * sess,
 784     RTPSessionProcessRTP callback, gpointer user_data)
 785 {
 786   g_return_if_fail (RTP_IS_SESSION (sess));
 787
 788   sess->callbacks.process_rtp = callback;
 789   sess->process_rtp_user_data = user_data;
 790 }
 791
 792 /**
 793  * rtp_session_set_send_rtp_callback:
 794  * @sess: an #RTPSession
 795  * @callback: callback to set
 796  * @user_data: user data passed in the callback
 797  *
 798  * Configure only the send_rtp callback to be notified of the send_rtp action.
 799  */
 800 void
 801 rtp_session_set_send_rtp_callback (RTPSession * sess,
 802     RTPSessionSendRTP callback, gpointer user_data)
 803 {
 804   g_return_if_fail (RTP_IS_SESSION (sess));
 805
 806   sess->callbacks.send_rtp = callback;
 807   sess->send_rtp_user_data = user_data;
 808 }
 809
 810 /**
 811  * rtp_session_set_send_rtcp_callback:
 812  * @sess: an #RTPSession
 813  * @callback: callback to set
 814  * @user_data: user data passed in the callback
 815  *
 816  * Configure only the send_rtcp callback to be notified of the send_rtcp action.
 817  */
 818 void
 819 rtp_session_set_send_rtcp_callback (RTPSession * sess,
 820     RTPSessionSendRTCP callback, gpointer user_data)
 821 {
 822   g_return_if_fail (RTP_IS_SESSION (sess));
 823
 824   sess->callbacks.send_rtcp = callback;
 825   sess->send_rtcp_user_data = user_data;
 826 }
 827
 828 /**
 829  * rtp_session_set_sync_rtcp_callback:
 830  * @sess: an #RTPSession
 831  * @callback: callback to set
 832  * @user_data: user data passed in the callback
 833  *
 834  * Configure only the sync_rtcp callback to be notified of the sync_rtcp action.
 835  */
 836 void
 837 rtp_session_set_sync_rtcp_callback (RTPSession * sess,
 838     RTPSessionSyncRTCP callback, gpointer user_data)
 839 {
 840   g_return_if_fail (RTP_IS_SESSION (sess));
 841
 842   sess->callbacks.sync_rtcp = callback;
 843   sess->sync_rtcp_user_data = user_data;
 844 }
 845
 846 /**
 847  * rtp_session_set_clock_rate_callback:
 848  * @sess: an #RTPSession
 849  * @callback: callback to set
 850  * @user_data: user data passed in the callback
 851  *
 852  * Configure only the clock_rate callback to be notified of the clock_rate action.
 853  */
 854 void
 855 rtp_session_set_clock_rate_callback (RTPSession * sess,
 856     RTPSessionClockRate callback, gpointer user_data)
 857 {
 858   g_return_if_fail (RTP_IS_SESSION (sess));
 859
 860   sess->callbacks.clock_rate = callback;
 861   sess->clock_rate_user_data = user_data;
 862 }
 863
 864 /**
 865  * rtp_session_set_reconsider_callback:
 866  * @sess: an #RTPSession
 867  * @callback: callback to set
 868  * @user_data: user data passed in the callback
 869  *
 870  * Configure only the reconsider callback to be notified of the reconsider action.
 871  */
 872 void
 873 rtp_session_set_reconsider_callback (RTPSession * sess,
 874     RTPSessionReconsider callback, gpointer user_data)
 875 {
 876   g_return_if_fail (RTP_IS_SESSION (sess));
 877
 878   sess->callbacks.reconsider = callback;
 879   sess->reconsider_user_data = user_data;
 880 }
 881
 882 /**
 883  * rtp_session_set_bandwidth:
 884  * @sess: an #RTPSession
 885  * @bandwidth: the bandwidth allocated
 886  *
 887  * Set the session bandwidth in bytes per second.
 888  */
 889 void
 890 rtp_session_set_bandwidth (RTPSession * sess, gdouble bandwidth)
 891 {
 892   g_return_if_fail (RTP_IS_SESSION (sess));
 893
 894   RTP_SESSION_LOCK (sess);
 895   sess->stats.bandwidth = bandwidth;
 896   RTP_SESSION_UNLOCK (sess);
 897 }
 898
 899 /**
 900  * rtp_session_get_bandwidth:
 901  * @sess: an #RTPSession
 902  *
 903  * Get the session bandwidth.
 904  *
 905  * Returns: the session bandwidth.
 906  */
 907 gdouble
 908 rtp_session_get_bandwidth (RTPSession * sess)
 909 {
 910   gdouble result;
 911
 912   g_return_val_if_fail (RTP_IS_SESSION (sess), 0);
 913
 914   RTP_SESSION_LOCK (sess);
 915   result = sess->stats.bandwidth;
 916   RTP_SESSION_UNLOCK (sess);
 917
 918   return result;
 919 }
 920
 921 /**
 922  * rtp_session_set_rtcp_fraction:
 923  * @sess: an #RTPSession
 924  * @bandwidth: the RTCP bandwidth
 925  *
 926  * Set the bandwidth in bytes per second that should be used for RTCP
 927  * messages.
 928  */
 929 void
 930 rtp_session_set_rtcp_fraction (RTPSession * sess, gdouble bandwidth)
 931 {
 932   g_return_if_fail (RTP_IS_SESSION (sess));
 933
 934   RTP_SESSION_LOCK (sess);
 935   sess->stats.rtcp_bandwidth = bandwidth;
 936   RTP_SESSION_UNLOCK (sess);
 937 }
 938
 939 /**
 940  * rtp_session_get_rtcp_fraction:
 941  * @sess: an #RTPSession
 942  *
 943  * Get the session bandwidth used for RTCP.
 944  *
 945  * Returns: The bandwidth used for RTCP messages.
 946  */
 947 gdouble
 948 rtp_session_get_rtcp_fraction (RTPSession * sess)
 949 {
 950   gdouble result;
 951
 952   g_return_val_if_fail (RTP_IS_SESSION (sess), 0.0);
 953
 954   RTP_SESSION_LOCK (sess);
 955   result = sess->stats.rtcp_bandwidth;
 956   RTP_SESSION_UNLOCK (sess);
 957
 958   return result;
 959 }
 960
 961 /**
 962  * rtp_session_set_sdes_string:
 963  * @sess: an #RTPSession
 964  * @type: the type of the SDES item
 965  * @item: a null-terminated string to set.
 966  *
 967  * Store an SDES item of @type in @sess.
 968  *
 969  * Returns: %FALSE if the data was unchanged @type is invalid.
 970  */
 971 gboolean
 972 rtp_session_set_sdes_string (RTPSession * sess, GstRTCPSDESType type,
 973     const gchar * item)
 974 {
 975   gboolean result;
 976
 977   g_return_val_if_fail (RTP_IS_SESSION (sess), FALSE);
 978
 979   RTP_SESSION_LOCK (sess);
 980   result = rtp_source_set_sdes_string (sess->source, type, item);
 981   RTP_SESSION_UNLOCK (sess);
 982
 983   return result;
 984 }
 985
 986 /**
 987  * rtp_session_get_sdes_string:
 988  * @sess: an #RTPSession
 989  * @type: the type of the SDES item
 990  *
 991  * Get the SDES item of @type from @sess.
 992  *
 993  * Returns: a null-terminated copy of the SDES item or NULL when @type was not
 994  * valid. g_free() after usage.
 995  */
 996 gchar *
 997 rtp_session_get_sdes_string (RTPSession * sess, GstRTCPSDESType type)
 998 {
 999   gchar *result;
1000
1001   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1002
1003   RTP_SESSION_LOCK (sess);
1004   result = rtp_source_get_sdes_string (sess->source, type);
1005   RTP_SESSION_UNLOCK (sess);
1006
1007   return result;
1008 }
1009
1010 /**
1011  * rtp_session_get_sdes_struct:
1012  * @sess: an #RTSPSession
1013  *
1014  * Get the SDES data as a #GstStructure
1015  *
1016  * Returns: a GstStructure with SDES items for @sess. This function returns a
1017  * copy of the SDES structure, use gst_structure_free() after usage.
1018  */
1019 GstStructure *
1020 rtp_session_get_sdes_struct (RTPSession * sess)
1021 {
1022   const GstStructure *sdes;
1023   GstStructure *result = NULL;
1024
1025   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1026
1027   RTP_SESSION_LOCK (sess);
1028   sdes = rtp_source_get_sdes_struct (sess->source);
1029   if (sdes)
1030     result = gst_structure_copy (sdes);
1031   RTP_SESSION_UNLOCK (sess);
1032
1033   return result;
1034 }
1035
1036 /**
1037  * rtp_session_set_sdes_struct:
1038  * @sess: an #RTSPSession
1039  * @sdes: a #GstStructure
1040  *
1041  * Set the SDES data as a #GstStructure. This function makes a copy of @sdes.
1042  */
1043 void
1044 rtp_session_set_sdes_struct (RTPSession * sess, const GstStructure * sdes)
1045 {
1046   g_return_if_fail (sdes);
1047   g_return_if_fail (RTP_IS_SESSION (sess));
1048
1049   RTP_SESSION_LOCK (sess);
1050   rtp_source_set_sdes_struct (sess->source, gst_structure_copy (sdes));
1051   RTP_SESSION_UNLOCK (sess);
1052 }
1053
1054 static GstFlowReturn
1055 source_push_rtp (RTPSource * source, gpointer data, RTPSession * session)
1056 {
1057   GstFlowReturn result = GST_FLOW_OK;
1058
1059   if (source == session->source) {
1060     GST_LOG ("source %08x pushed sender RTP packet", source->ssrc);
1061
1062     RTP_SESSION_UNLOCK (session);
1063
1064     if (session->callbacks.send_rtp)
1065       result =
1066           session->callbacks.send_rtp (session, source, data,
1067           session->send_rtp_user_data);
1068     else {
1069       gst_mini_object_unref (GST_MINI_OBJECT_CAST (data));
1070     }
1071   } else {
1072     GST_LOG ("source %08x pushed receiver RTP packet", source->ssrc);
1073     RTP_SESSION_UNLOCK (session);
1074
1075     if (session->callbacks.process_rtp)
1076       result =
1077           session->callbacks.process_rtp (session, source,
1078           GST_BUFFER_CAST (data), session->process_rtp_user_data);
1079     else
1080       gst_buffer_unref (GST_BUFFER_CAST (data));
1081   }
1082   RTP_SESSION_LOCK (session);
1083
1084   return result;
1085 }
1086
1087 static gint
1088 source_clock_rate (RTPSource * source, guint8 pt, RTPSession * session)
1089 {
1090   gint result;
1091
1092   RTP_SESSION_UNLOCK (session);
1093
1094   if (session->callbacks.clock_rate)
1095     result =
1096         session->callbacks.clock_rate (session, pt,
1097         session->clock_rate_user_data);
1098   else
1099     result = -1;
1100
1101   RTP_SESSION_LOCK (session);
1102
1103   GST_DEBUG ("got clock-rate %d for pt %d", result, pt);
1104
1105   return result;
1106 }
1107
1108 static RTPSourceCallbacks callbacks = {
1109   (RTPSourcePushRTP) source_push_rtp,
1110   (RTPSourceClockRate) source_clock_rate,
1111 };
1112
1113 static gboolean
1114 check_collision (RTPSession * sess, RTPSource * source,
1115     RTPArrivalStats * arrival, gboolean rtp)
1116 {
1117   /* If we have no arrival address, we can't do collision checking */
1118   if (!arrival->have_address)
1119     return FALSE;
1120
1121   if (sess->source != source) {
1122     GstNetAddress *from;
1123     gboolean have_from;
1124
1125     /* This is not our local source, but lets check if two remote
1126      * source collide
1127      */
1128
1129     if (rtp) {
1130       from = &source->rtp_from;
1131       have_from = source->have_rtp_from;
1132     } else {
1133       from = &source->rtcp_from;
1134       have_from = source->have_rtcp_from;
1135     }
1136
1137     if (have_from) {
1138       if (gst_netaddress_equal (from, &arrival->address)) {
1139         /* Address is the same */
1140         return FALSE;
1141       } else {
1142         GST_LOG ("we have a third-party collision or loop ssrc:%x",
1143             rtp_source_get_ssrc (source));
1144         if (sess->favor_new) {
1145           if (rtp_source_find_conflicting_address (source,
1146                   &arrival->address, arrival->current_time)) {
1147             gchar buf1[40];
1148             gst_netaddress_to_string (&arrival->address, buf1, 40);
1149             GST_LOG ("Known conflict on %x for %s, dropping packet",
1150                 rtp_source_get_ssrc (source), buf1);
1151             return TRUE;
1152           } else {
1153             gchar buf1[40], buf2[40];
1154
1155             /* Current address is not a known conflict, lets assume this is
1156              * a new source. Save old address in possible conflict list
1157              */
1158             rtp_source_add_conflicting_address (source, from,
1159                 arrival->current_time);
1160
1161             gst_netaddress_to_string (from, buf1, 40);
1162             gst_netaddress_to_string (&arrival->address, buf2, 40);
1163             GST_DEBUG ("New conflict for ssrc %x, replacing %s with %s,"
1164                 " saving old as known conflict",
1165                 rtp_source_get_ssrc (source), buf1, buf2);
1166
1167             if (rtp)
1168               rtp_source_set_rtp_from (source, &arrival->address);
1169             else
1170               rtp_source_set_rtcp_from (source, &arrival->address);
1171             return FALSE;
1172           }
1173         } else {
1174           /* Don't need to save old addresses, we ignore new sources */
1175           return TRUE;
1176         }
1177       }
1178     } else {
1179       /* We don't already have a from address for RTP, just set it */
1180       if (rtp)
1181         rtp_source_set_rtp_from (source, &arrival->address);
1182       else
1183         rtp_source_set_rtcp_from (source, &arrival->address);
1184       return FALSE;
1185     }
1186
1187     /* FIXME: Log 3rd party collision somehow
1188      * Maybe should be done in upper layer, only the SDES can tell us
1189      * if its a collision or a loop
1190      */
1191
1192     /* If the source has been inactive for some time, we assume that it has
1193      * simply changed its transport source address. Hence, there is no true
1194      * third-party collision - only a simulated one. */
1195     if (arrival->current_time > source->last_activity) {
1196       GstClockTime inactivity_period =
1197           arrival->current_time - source->last_activity;
1198       if (inactivity_period > 1 * GST_SECOND) {
1199         /* Use new network address */
1200         if (rtp) {
1201           g_assert (source->have_rtp_from);
1202           rtp_source_set_rtp_from (source, &arrival->address);
1203         } else {
1204           g_assert (source->have_rtcp_from);
1205           rtp_source_set_rtcp_from (source, &arrival->address);
1206         }
1207         return FALSE;
1208       }
1209     }
1210   } else {
1211     /* This is sending with our ssrc, is it an address we already know */
1212
1213     if (rtp_source_find_conflicting_address (source, &arrival->address,
1214             arrival->current_time)) {
1215       /* Its a known conflict, its probably a loop, not a collision
1216        * lets just drop the incoming packet
1217        */
1218       GST_DEBUG ("Our packets are being looped back to us, dropping");
1219     } else {
1220       /* Its a new collision, lets change our SSRC */
1221
1222       rtp_source_add_conflicting_address (source, &arrival->address,
1223           arrival->current_time);
1224
1225       GST_DEBUG ("Collision for SSRC %x", rtp_source_get_ssrc (source));
1226       on_ssrc_collision (sess, source);
1227
1228       rtp_session_schedule_bye_locked (sess, "SSRC Collision",
1229           arrival->current_time);
1230
1231       sess->change_ssrc = TRUE;
1232     }
1233   }
1234
1235   return TRUE;
1236 }
1237
1238
1239 /* must be called with the session lock, the returned source needs to be
1240  * unreffed after usage. */
1241 static RTPSource *
1242 obtain_source (RTPSession * sess, guint32 ssrc, gboolean * created,
1243     RTPArrivalStats * arrival, gboolean rtp)
1244 {
1245   RTPSource *source;
1246
1247   source =
1248       g_hash_table_lookup (sess->ssrcs[sess->mask_idx], GINT_TO_POINTER (ssrc));
1249   if (source == NULL) {
1250     /* make new Source in probation and insert */
1251     source = rtp_source_new (ssrc);
1252
1253     /* for RTP packets we need to set the source in probation. Receiving RTCP
1254      * packets of an SSRC, on the other hand, is a strong indication that we
1255      * are dealing with a valid source. */
1256     if (rtp)
1257       source->probation = RTP_DEFAULT_PROBATION;
1258     else
1259       source->probation = 0;
1260
1261     /* store from address, if any */
1262     if (arrival->have_address) {
1263       if (rtp)
1264         rtp_source_set_rtp_from (source, &arrival->address);
1265       else
1266         rtp_source_set_rtcp_from (source, &arrival->address);
1267     }
1268
1269     /* configure a callback on the source */
1270     rtp_source_set_callbacks (source, &callbacks, sess);
1271
1272     g_hash_table_insert (sess->ssrcs[sess->mask_idx], GINT_TO_POINTER (ssrc),
1273         source);
1274
1275     /* we have one more source now */
1276     sess->total_sources++;
1277     *created = TRUE;
1278   } else {
1279     *created = FALSE;
1280     /* check for collision, this updates the address when not previously set */
1281     if (check_collision (sess, source, arrival, rtp)) {
1282       return NULL;
1283     }
1284   }
1285   /* update last activity */
1286   source->last_activity = arrival->current_time;
1287   if (rtp)
1288     source->last_rtp_activity = arrival->current_time;
1289   g_object_ref (source);
1290
1291   return source;
1292 }
1293
1294 /**
1295  * rtp_session_get_internal_source:
1296  * @sess: a #RTPSession
1297  *
1298  * Get the internal #RTPSource of @sess.
1299  *
1300  * Returns: The internal #RTPSource. g_object_unref() after usage.
1301  */
1302 RTPSource *
1303 rtp_session_get_internal_source (RTPSession * sess)
1304 {
1305   RTPSource *result;
1306
1307   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1308
1309   result = g_object_ref (sess->source);
1310
1311   return result;
1312 }
1313
1314 /**
1315  * rtp_session_set_internal_ssrc:
1316  * @sess: a #RTPSession
1317  * @ssrc: an SSRC
1318  *
1319  * Set the SSRC of @sess to @ssrc.
1320  */
1321 void
1322 rtp_session_set_internal_ssrc (RTPSession * sess, guint32 ssrc)
1323 {
1324   RTP_SESSION_LOCK (sess);
1325   if (ssrc != sess->source->ssrc) {
1326     g_hash_table_steal (sess->ssrcs[sess->mask_idx],
1327         GINT_TO_POINTER (sess->source->ssrc));
1328
1329     GST_DEBUG ("setting internal SSRC to %08x", ssrc);
1330     /* After this call, any receiver of the old SSRC either in RTP or RTCP
1331      * packets will timeout on the old SSRC, we could potentially schedule a
1332      * BYE RTCP for the old SSRC... */
1333     sess->source->ssrc = ssrc;
1334     rtp_source_reset (sess->source);
1335
1336     /* rehash with the new SSRC */
1337     g_hash_table_insert (sess->ssrcs[sess->mask_idx],
1338         GINT_TO_POINTER (sess->source->ssrc), sess->source);
1339   }
1340   RTP_SESSION_UNLOCK (sess);
1341
1342   g_object_notify (G_OBJECT (sess), "internal-ssrc");
1343 }
1344
1345 /**
1346  * rtp_session_get_internal_ssrc:
1347  * @sess: a #RTPSession
1348  *
1349  * Get the internal SSRC of @sess.
1350  *
1351  * Returns: The SSRC of the session.
1352  */
1353 guint32
1354 rtp_session_get_internal_ssrc (RTPSession * sess)
1355 {
1356   guint32 ssrc;
1357
1358   RTP_SESSION_LOCK (sess);
1359   ssrc = sess->source->ssrc;
1360   RTP_SESSION_UNLOCK (sess);
1361
1362   return ssrc;
1363 }
1364
1365 /**
1366  * rtp_session_add_source:
1367  * @sess: a #RTPSession
1368  * @src: #RTPSource to add
1369  *
1370  * Add @src to @session.
1371  *
1372  * Returns: %TRUE on success, %FALSE if a source with the same SSRC already
1373  * existed in the session.
1374  */
1375 gboolean
1376 rtp_session_add_source (RTPSession * sess, RTPSource * src)
1377 {
1378   gboolean result = FALSE;
1379   RTPSource *find;
1380
1381   g_return_val_if_fail (RTP_IS_SESSION (sess), FALSE);
1382   g_return_val_if_fail (src != NULL, FALSE);
1383
1384   RTP_SESSION_LOCK (sess);
1385   find =
1386       g_hash_table_lookup (sess->ssrcs[sess->mask_idx],
1387       GINT_TO_POINTER (src->ssrc));
1388   if (find == NULL) {
1389     g_hash_table_insert (sess->ssrcs[sess->mask_idx],
1390         GINT_TO_POINTER (src->ssrc), src);
1391     /* we have one more source now */
1392     sess->total_sources++;
1393     result = TRUE;
1394   }
1395   RTP_SESSION_UNLOCK (sess);
1396
1397   return result;
1398 }
1399
1400 /**
1401  * rtp_session_get_num_sources:
1402  * @sess: an #RTPSession
1403  *
1404  * Get the number of sources in @sess.
1405  *
1406  * Returns: The number of sources in @sess.
1407  */
1408 guint
1409 rtp_session_get_num_sources (RTPSession * sess)
1410 {
1411   guint result;
1412
1413   g_return_val_if_fail (RTP_IS_SESSION (sess), FALSE);
1414
1415   RTP_SESSION_LOCK (sess);
1416   result = sess->total_sources;
1417   RTP_SESSION_UNLOCK (sess);
1418
1419   return result;
1420 }
1421
1422 /**
1423  * rtp_session_get_num_active_sources:
1424  * @sess: an #RTPSession
1425  *
1426  * Get the number of active sources in @sess. A source is considered active when
1427  * it has been validated and has not yet received a BYE RTCP message.
1428  *
1429  * Returns: The number of active sources in @sess.
1430  */
1431 guint
1432 rtp_session_get_num_active_sources (RTPSession * sess)
1433 {
1434   guint result;
1435
1436   g_return_val_if_fail (RTP_IS_SESSION (sess), 0);
1437
1438   RTP_SESSION_LOCK (sess);
1439   result = sess->stats.active_sources;
1440   RTP_SESSION_UNLOCK (sess);
1441
1442   return result;
1443 }
1444
1445 /**
1446  * rtp_session_get_source_by_ssrc:
1447  * @sess: an #RTPSession
1448  * @ssrc: an SSRC
1449  *
1450  * Find the source with @ssrc in @sess.
1451  *
1452  * Returns: a #RTPSource with SSRC @ssrc or NULL if the source was not found.
1453  * g_object_unref() after usage.
1454  */
1455 RTPSource *
1456 rtp_session_get_source_by_ssrc (RTPSession * sess, guint32 ssrc)
1457 {
1458   RTPSource *result;
1459
1460   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1461
1462   RTP_SESSION_LOCK (sess);
1463   result =
1464       g_hash_table_lookup (sess->ssrcs[sess->mask_idx], GINT_TO_POINTER (ssrc));
1465   if (result)
1466     g_object_ref (result);
1467   RTP_SESSION_UNLOCK (sess);
1468
1469   return result;
1470 }
1471
1472 /**
1473  * rtp_session_get_source_by_cname:
1474  * @sess: a #RTPSession
1475  * @cname: an CNAME
1476  *
1477  * Find the source with @cname in @sess.
1478  *
1479  * Returns: a #RTPSource with CNAME @cname or NULL if the source was not found.
1480  * g_object_unref() after usage.
1481  */
1482 RTPSource *
1483 rtp_session_get_source_by_cname (RTPSession * sess, const gchar * cname)
1484 {
1485   RTPSource *result;
1486
1487   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1488   g_return_val_if_fail (cname != NULL, NULL);
1489
1490   RTP_SESSION_LOCK (sess);
1491   result = g_hash_table_lookup (sess->cnames, cname);
1492   if (result)
1493     g_object_ref (result);
1494   RTP_SESSION_UNLOCK (sess);
1495
1496   return result;
1497 }
1498
1499 /* should be called with the SESSION lock */
1500 static guint32
1501 rtp_session_create_new_ssrc (RTPSession * sess)
1502 {
1503   guint32 ssrc;
1504
1505   while (TRUE) {
1506     ssrc = g_random_int ();
1507
1508     /* see if it exists in the session, we're done if it doesn't */
1509     if (g_hash_table_lookup (sess->ssrcs[sess->mask_idx],
1510             GINT_TO_POINTER (ssrc)) == NULL)
1511       break;
1512   }
1513   return ssrc;
1514 }
1515
1516
1517 /**
1518  * rtp_session_create_source:
1519  * @sess: an #RTPSession
1520  *
1521  * Create an #RTPSource for use in @sess. This function will create a source
1522  * with an ssrc that is currently not used by any participants in the session.
1523  *
1524  * Returns: an #RTPSource.
1525  */
1526 RTPSource *
1527 rtp_session_create_source (RTPSession * sess)
1528 {
1529   guint32 ssrc;
1530   RTPSource *source;
1531
1532   RTP_SESSION_LOCK (sess);
1533   ssrc = rtp_session_create_new_ssrc (sess);
1534   source = rtp_source_new (ssrc);
1535   rtp_source_set_callbacks (source, &callbacks, sess);
1536   /* we need an additional ref for the source in the hashtable */
1537   g_object_ref (source);
1538   g_hash_table_insert (sess->ssrcs[sess->mask_idx], GINT_TO_POINTER (ssrc),
1539       source);
1540   /* we have one more source now */
1541   sess->total_sources++;
1542   RTP_SESSION_UNLOCK (sess);
1543
1544   return source;
1545 }
1546
1547 /* update the RTPArrivalStats structure with the current time and other bits
1548  * about the current buffer we are handling.
1549  * This function is typically called when a validated packet is received.
1550  * This function should be called with the SESSION_LOCK
1551  */
1552 static void
1553 update_arrival_stats (RTPSession * sess, RTPArrivalStats * arrival,
1554     gboolean rtp, GstBuffer * buffer, GstClockTime current_time,
1555     GstClockTime running_time)
1556 {
1557   /* get time of arrival */
1558   arrival->current_time = current_time;
1559   arrival->running_time = running_time;
1560
1561   /* get packet size including header overhead */
1562   arrival->bytes = GST_BUFFER_SIZE (buffer) + sess->header_len;
1563
1564   if (rtp) {
1565     arrival->payload_len = gst_rtp_buffer_get_payload_len (buffer);
1566   } else {
1567     arrival->payload_len = 0;
1568   }
1569
1570   /* for netbuffer we can store the IP address to check for collisions */
1571   arrival->have_address = GST_IS_NETBUFFER (buffer);
1572   if (arrival->have_address) {
1573     GstNetBuffer *netbuf = (GstNetBuffer *) buffer;
1574
1575     memcpy (&arrival->address, &netbuf->from, sizeof (GstNetAddress));
1576   }
1577 }
1578
1579 /**
1580  * rtp_session_process_rtp:
1581  * @sess: and #RTPSession
1582  * @buffer: an RTP buffer
1583  * @current_time: the current system time
1584  * @running_time: the running_time of @buffer
1585  *
1586  * Process an RTP buffer in the session manager. This function takes ownership
1587  * of @buffer.
1588  *
1589  * Returns: a #GstFlowReturn.
1590  */
1591 GstFlowReturn
1592 rtp_session_process_rtp (RTPSession * sess, GstBuffer * buffer,
1593     GstClockTime current_time, GstClockTime running_time)
1594 {
1595   GstFlowReturn result;
1596   guint32 ssrc;
1597   RTPSource *source;
1598   gboolean created;
1599   gboolean prevsender, prevactive;
1600   RTPArrivalStats arrival;
1601   guint32 csrcs[16];
1602   guint8 i, count;
1603   guint64 oldrate;
1604
1605   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1606   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1607
1608   if (!gst_rtp_buffer_validate (buffer))
1609     goto invalid_packet;
1610
1611   RTP_SESSION_LOCK (sess);
1612   /* update arrival stats */
1613   update_arrival_stats (sess, &arrival, TRUE, buffer, current_time,
1614       running_time);
1615
1616   /* ignore more RTP packets when we left the session */
1617   if (sess->source->received_bye)
1618     goto ignore;
1619
1620   /* get SSRC and look up in session database */
1621   ssrc = gst_rtp_buffer_get_ssrc (buffer);
1622   source = obtain_source (sess, ssrc, &created, &arrival, TRUE);
1623   if (!source)
1624     goto collision;
1625
1626   prevsender = RTP_SOURCE_IS_SENDER (source);
1627   prevactive = RTP_SOURCE_IS_ACTIVE (source);
1628   oldrate = source->bitrate;
1629
1630   /* copy available csrc for later */
1631   count = gst_rtp_buffer_get_csrc_count (buffer);
1632   /* make sure to not overflow our array. An RTP buffer can maximally contain
1633    * 16 CSRCs */
1634   count = MIN (count, 16);
1635
1636   for (i = 0; i < count; i++)
1637     csrcs[i] = gst_rtp_buffer_get_csrc (buffer, i);
1638
1639   /* let source process the packet */
1640   result = rtp_source_process_rtp (source, buffer, &arrival);
1641
1642   /* source became active */
1643   if (prevactive != RTP_SOURCE_IS_ACTIVE (source)) {
1644     sess->stats.active_sources++;
1645     GST_DEBUG ("source: %08x became active, %d active sources", ssrc,
1646         sess->stats.active_sources);
1647     on_ssrc_validated (sess, source);
1648   }
1649   if (prevsender != RTP_SOURCE_IS_SENDER (source)) {
1650     sess->stats.sender_sources++;
1651     GST_DEBUG ("source: %08x became sender, %d sender sources", ssrc,
1652         sess->stats.sender_sources);
1653   }
1654   if (oldrate != source->bitrate)
1655     sess->recalc_bandwidth = TRUE;
1656
1657   if (created)
1658     on_new_ssrc (sess, source);
1659
1660   if (source->validated) {
1661     gboolean created;
1662
1663     /* for validated sources, we add the CSRCs as well */
1664     for (i = 0; i < count; i++) {
1665       guint32 csrc;
1666       RTPSource *csrc_src;
1667
1668       csrc = csrcs[i];
1669
1670       /* get source */
1671       csrc_src = obtain_source (sess, csrc, &created, &arrival, TRUE);
1672       if (!csrc_src)
1673         continue;
1674
1675       if (created) {
1676         GST_DEBUG ("created new CSRC: %08x", csrc);
1677         rtp_source_set_as_csrc (csrc_src);
1678         if (RTP_SOURCE_IS_ACTIVE (csrc_src))
1679           sess->stats.active_sources++;
1680         on_new_ssrc (sess, csrc_src);
1681       }
1682       g_object_unref (csrc_src);
1683     }
1684   }
1685   g_object_unref (source);
1686
1687   RTP_SESSION_UNLOCK (sess);
1688
1689   return result;
1690
1691   /* ERRORS */
1692 invalid_packet:
1693   {
1694     gst_buffer_unref (buffer);
1695     GST_DEBUG ("invalid RTP packet received");
1696     return GST_FLOW_OK;
1697   }
1698 ignore:
1699   {
1700     gst_buffer_unref (buffer);
1701     RTP_SESSION_UNLOCK (sess);
1702     GST_DEBUG ("ignoring RTP packet because we are leaving");
1703     return GST_FLOW_OK;
1704   }
1705 collision:
1706   {
1707     gst_buffer_unref (buffer);
1708     RTP_SESSION_UNLOCK (sess);
1709     GST_DEBUG ("ignoring packet because its collisioning");
1710     return GST_FLOW_OK;
1711   }
1712 }
1713
1714 static void
1715 rtp_session_process_rb (RTPSession * sess, RTPSource * source,
1716     GstRTCPPacket * packet, RTPArrivalStats * arrival)
1717 {
1718   guint count, i;
1719
1720   count = gst_rtcp_packet_get_rb_count (packet);
1721   for (i = 0; i < count; i++) {
1722     guint32 ssrc, exthighestseq, jitter, lsr, dlsr;
1723     guint8 fractionlost;
1724     gint32 packetslost;
1725
1726     gst_rtcp_packet_get_rb (packet, i, &ssrc, &fractionlost,
1727         &packetslost, &exthighestseq, &jitter, &lsr, &dlsr);
1728
1729     GST_DEBUG ("RB %d: SSRC %08x, jitter %" G_GUINT32_FORMAT, i, ssrc, jitter);
1730
1731     if (ssrc == sess->source->ssrc) {
1732       /* only deal with report blocks for our session, we update the stats of
1733        * the sender of the RTCP message. We could also compare our stats against
1734        * the other sender to see if we are better or worse. */
1735       rtp_source_process_rb (source, arrival->current_time, fractionlost,
1736           packetslost, exthighestseq, jitter, lsr, dlsr);
1737     }
1738   }
1739   on_ssrc_active (sess, source);
1740 }
1741
1742 /* A Sender report contains statistics about how the sender is doing. This
1743  * includes timing informataion such as the relation between RTP and NTP
1744  * timestamps and the number of packets/bytes it sent to us.
1745  *
1746  * In this report is also included a set of report blocks related to how this
1747  * sender is receiving data (in case we (or somebody else) is also sending stuff
1748  * to it). This info includes the packet loss, jitter and seqnum. It also
1749  * contains information to calculate the round trip time (LSR/DLSR).
1750  */
1751 static void
1752 rtp_session_process_sr (RTPSession * sess, GstRTCPPacket * packet,
1753     RTPArrivalStats * arrival, gboolean * do_sync)
1754 {
1755   guint32 senderssrc, rtptime, packet_count, octet_count;
1756   guint64 ntptime;
1757   RTPSource *source;
1758   gboolean created, prevsender;
1759
1760   gst_rtcp_packet_sr_get_sender_info (packet, &senderssrc, &ntptime, &rtptime,
1761       &packet_count, &octet_count);
1762
1763   GST_DEBUG ("got SR packet: SSRC %08x, time %" GST_TIME_FORMAT,
1764       senderssrc, GST_TIME_ARGS (arrival->current_time));
1765
1766   source = obtain_source (sess, senderssrc, &created, arrival, FALSE);
1767   if (!source)
1768     return;
1769
1770   /* don't try to do lip-sync for sources that sent a BYE */
1771   if (rtp_source_received_bye (source))
1772     *do_sync = FALSE;
1773   else
1774     *do_sync = TRUE;
1775
1776   prevsender = RTP_SOURCE_IS_SENDER (source);
1777
1778   /* first update the source */
1779   rtp_source_process_sr (source, arrival->current_time, ntptime, rtptime,
1780       packet_count, octet_count);
1781
1782   if (prevsender != RTP_SOURCE_IS_SENDER (source)) {
1783     sess->stats.sender_sources++;
1784     GST_DEBUG ("source: %08x became sender, %d sender sources", senderssrc,
1785         sess->stats.sender_sources);
1786   }
1787
1788   if (created)
1789     on_new_ssrc (sess, source);
1790
1791   rtp_session_process_rb (sess, source, packet, arrival);
1792   g_object_unref (source);
1793 }
1794
1795 /* A receiver report contains statistics about how a receiver is doing. It
1796  * includes stuff like packet loss, jitter and the seqnum it received last. It
1797  * also contains info to calculate the round trip time.
1798  *
1799  * We are only interested in how the sender of this report is doing wrt to us.
1800  */
1801 static void
1802 rtp_session_process_rr (RTPSession * sess, GstRTCPPacket * packet,
1803     RTPArrivalStats * arrival)
1804 {
1805   guint32 senderssrc;
1806   RTPSource *source;
1807   gboolean created;
1808
1809   senderssrc = gst_rtcp_packet_rr_get_ssrc (packet);
1810
1811   GST_DEBUG ("got RR packet: SSRC %08x", senderssrc);
1812
1813   source = obtain_source (sess, senderssrc, &created, arrival, FALSE);
1814   if (!source)
1815     return;
1816
1817   if (created)
1818     on_new_ssrc (sess, source);
1819
1820   rtp_session_process_rb (sess, source, packet, arrival);
1821   g_object_unref (source);
1822 }
1823
1824 /* Get SDES items and store them in the SSRC */
1825 static void
1826 rtp_session_process_sdes (RTPSession * sess, GstRTCPPacket * packet,
1827     RTPArrivalStats * arrival)
1828 {
1829   guint items, i, j;
1830   gboolean more_items, more_entries;
1831
1832   items = gst_rtcp_packet_sdes_get_item_count (packet);
1833   GST_DEBUG ("got SDES packet with %d items", items);
1834
1835   more_items = gst_rtcp_packet_sdes_first_item (packet);
1836   i = 0;
1837   while (more_items) {
1838     guint32 ssrc;
1839     gboolean changed, created, validated;
1840     RTPSource *source;
1841     GstStructure *sdes;
1842
1843     ssrc = gst_rtcp_packet_sdes_get_ssrc (packet);
1844
1845     GST_DEBUG ("item %d, SSRC %08x", i, ssrc);
1846
1847     changed = FALSE;
1848
1849     /* find src, no probation when dealing with RTCP */
1850     source = obtain_source (sess, ssrc, &created, arrival, FALSE);
1851     if (!source)
1852       return;
1853
1854     sdes = gst_structure_new ("application/x-rtp-source-sdes", NULL);
1855
1856     more_entries = gst_rtcp_packet_sdes_first_entry (packet);
1857     j = 0;
1858     while (more_entries) {
1859       GstRTCPSDESType type;
1860       guint8 len;
1861       guint8 *data;
1862       gchar *name;
1863       gchar *value;
1864
1865       gst_rtcp_packet_sdes_get_entry (packet, &type, &len, &data);
1866
1867       GST_DEBUG ("entry %d, type %d, len %d, data %.*s", j, type, len, len,
1868           data);
1869
1870       if (type == GST_RTCP_SDES_PRIV) {
1871         name = g_strndup ((const gchar *) &data[1], data[0]);
1872         len -= data[0] + 1;
1873         data += data[0] + 1;
1874       } else {
1875         name = g_strdup (gst_rtcp_sdes_type_to_name (type));
1876       }
1877
1878       value = g_strndup ((const gchar *) data, len);
1879
1880       gst_structure_set (sdes, name, G_TYPE_STRING, value, NULL);
1881
1882       g_free (name);
1883       g_free (value);
1884
1885       more_entries = gst_rtcp_packet_sdes_next_entry (packet);
1886       j++;
1887     }
1888
1889     /* takes ownership of sdes */
1890     changed = rtp_source_set_sdes_struct (source, sdes);
1891
1892     validated = !RTP_SOURCE_IS_ACTIVE (source);
1893     source->validated = TRUE;
1894
1895     if (created)
1896       on_new_ssrc (sess, source);
1897     if (validated)
1898       on_ssrc_validated (sess, source);
1899     if (changed)
1900       on_ssrc_sdes (sess, source);
1901
1902     g_object_unref (source);
1903
1904     more_items = gst_rtcp_packet_sdes_next_item (packet);
1905     i++;
1906   }
1907 }
1908
1909 /* BYE is sent when a client leaves the session
1910  */
1911 static void
1912 rtp_session_process_bye (RTPSession * sess, GstRTCPPacket * packet,
1913     RTPArrivalStats * arrival)
1914 {
1915   guint count, i;
1916   gchar *reason;
1917   gboolean reconsider = FALSE;
1918
1919   reason = gst_rtcp_packet_bye_get_reason (packet);
1920   GST_DEBUG ("got BYE packet (reason: %s)", GST_STR_NULL (reason));
1921
1922   count = gst_rtcp_packet_bye_get_ssrc_count (packet);
1923   for (i = 0; i < count; i++) {
1924     guint32 ssrc;
1925     RTPSource *source;
1926     gboolean created, prevactive, prevsender;
1927     guint pmembers, members;
1928
1929     ssrc = gst_rtcp_packet_bye_get_nth_ssrc (packet, i);
1930     GST_DEBUG ("SSRC: %08x", ssrc);
1931
1932     /* find src and mark bye, no probation when dealing with RTCP */
1933     source = obtain_source (sess, ssrc, &created, arrival, FALSE);
1934     if (!source)
1935       return;
1936
1937     /* store time for when we need to time out this source */
1938     source->bye_time = arrival->current_time;
1939
1940     prevactive = RTP_SOURCE_IS_ACTIVE (source);
1941     prevsender = RTP_SOURCE_IS_SENDER (source);
1942
1943     /* let the source handle the rest */
1944     rtp_source_process_bye (source, reason);
1945
1946     pmembers = sess->stats.active_sources;
1947
1948     if (prevactive && !RTP_SOURCE_IS_ACTIVE (source)) {
1949       sess->stats.active_sources--;
1950       GST_DEBUG ("source: %08x became inactive, %d active sources", ssrc,
1951           sess->stats.active_sources);
1952     }
1953     if (prevsender && !RTP_SOURCE_IS_SENDER (source)) {
1954       sess->stats.sender_sources--;
1955       GST_DEBUG ("source: %08x became non sender, %d sender sources", ssrc,
1956           sess->stats.sender_sources);
1957     }
1958     members = sess->stats.active_sources;
1959
1960     if (!sess->source->received_bye && members < pmembers) {
1961       /* some members went away since the previous timeout estimate.
1962        * Perform reverse reconsideration but only when we are not scheduling a
1963        * BYE ourselves. */
1964       if (arrival->current_time < sess->next_rtcp_check_time) {
1965         GstClockTime time_remaining;
1966
1967         time_remaining = sess->next_rtcp_check_time - arrival->current_time;
1968         sess->next_rtcp_check_time =
1969             gst_util_uint64_scale (time_remaining, members, pmembers);
1970
1971         GST_DEBUG ("reverse reconsideration %" GST_TIME_FORMAT,
1972             GST_TIME_ARGS (sess->next_rtcp_check_time));
1973
1974         sess->next_rtcp_check_time += arrival->current_time;
1975
1976         /* mark pending reconsider. We only want to signal the reconsideration
1977          * once after we handled all the source in the bye packet */
1978         reconsider = TRUE;
1979       }
1980     }
1981
1982     if (created)
1983       on_new_ssrc (sess, source);
1984
1985     on_bye_ssrc (sess, source);
1986
1987     g_object_unref (source);
1988   }
1989   if (reconsider) {
1990     RTP_SESSION_UNLOCK (sess);
1991     /* notify app of reconsideration */
1992     if (sess->callbacks.reconsider)
1993       sess->callbacks.reconsider (sess, sess->reconsider_user_data);
1994     RTP_SESSION_LOCK (sess);
1995   }
1996   g_free (reason);
1997 }
1998
1999 static void
2000 rtp_session_process_app (RTPSession * sess, GstRTCPPacket * packet,
2001     RTPArrivalStats * arrival)
2002 {
2003   GST_DEBUG ("received APP");
2004 }
2005
2006
2007 static void
2008 rtp_session_process_feedback (RTPSession * sess, GstRTCPPacket * packet,
2009     RTPArrivalStats * arrival)
2010 {
2011   GstRTCPType type = gst_rtcp_packet_get_type (packet);
2012   GstRTCPFBType fbtype = gst_rtcp_packet_fb_get_type (packet);
2013   guint32 sender_ssrc = gst_rtcp_packet_fb_get_sender_ssrc (packet);
2014   guint32 media_ssrc = gst_rtcp_packet_fb_get_media_ssrc (packet);
2015   guint length = 4 * (gst_rtcp_packet_get_length (packet) - 2);
2016
2017   GST_DEBUG ("received feedback %d:%d from %08X about %08X"
2018       " with FCI of length %d", type, fbtype, sender_ssrc, media_ssrc, length);
2019
2020   if (g_signal_has_handler_pending (sess,
2021           rtp_session_signals[SIGNAL_ON_FEEDBACK_RTCP], 0, TRUE)) {
2022     GstBuffer *fci = NULL;
2023
2024     if (length) {
2025       fci = gst_buffer_create_sub (packet->buffer, packet->offset + 72, length);
2026       GST_BUFFER_TIMESTAMP (fci) = arrival->running_time;
2027     }
2028
2029     RTP_SESSION_UNLOCK (sess);
2030     g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_FEEDBACK_RTCP], 0,
2031         type, fbtype, sender_ssrc, media_ssrc, fci);
2032     RTP_SESSION_LOCK (sess);
2033
2034     if (fci)
2035       gst_buffer_unref (fci);
2036   }
2037
2038   if (sess->rtcp_feedback_retention_window) {
2039     RTPSource *src = g_hash_table_lookup (sess->ssrcs[sess->mask_idx],
2040         GINT_TO_POINTER (media_ssrc));
2041
2042     if (src)
2043       rtp_source_retain_rtcp_packet (src, packet, arrival->running_time);
2044   }
2045 }
2046
2047 /**
2048  * rtp_session_process_rtcp:
2049  * @sess: and #RTPSession
2050  * @buffer: an RTCP buffer
2051  * @current_time: the current system time
2052  *
2053  * Process an RTCP buffer in the session manager. This function takes ownership
2054  * of @buffer.
2055  *
2056  * Returns: a #GstFlowReturn.
2057  */
2058 GstFlowReturn
2059 rtp_session_process_rtcp (RTPSession * sess, GstBuffer * buffer,
2060     GstClockTime current_time)
2061 {
2062   GstRTCPPacket packet;
2063   gboolean more, is_bye = FALSE, do_sync = FALSE;
2064   RTPArrivalStats arrival;
2065   GstFlowReturn result = GST_FLOW_OK;
2066
2067   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
2068   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
2069
2070   if (!gst_rtcp_buffer_validate (buffer))
2071     goto invalid_packet;
2072
2073   GST_DEBUG ("received RTCP packet");
2074
2075   RTP_SESSION_LOCK (sess);
2076   /* update arrival stats */
2077   update_arrival_stats (sess, &arrival, FALSE, buffer, current_time, -1);
2078
2079   if (sess->sent_bye)
2080     goto ignore;
2081
2082   /* start processing the compound packet */
2083   more = gst_rtcp_buffer_get_first_packet (buffer, &packet);
2084   while (more) {
2085     GstRTCPType type;
2086
2087     type = gst_rtcp_packet_get_type (&packet);
2088
2089     /* when we are leaving the session, we should ignore all non-BYE messages */
2090     if (sess->source->received_bye && type != GST_RTCP_TYPE_BYE) {
2091       GST_DEBUG ("ignoring non-BYE RTCP packet because we are leaving");
2092       goto next;
2093     }
2094
2095     switch (type) {
2096       case GST_RTCP_TYPE_SR:
2097         rtp_session_process_sr (sess, &packet, &arrival, &do_sync);
2098         break;
2099       case GST_RTCP_TYPE_RR:
2100         rtp_session_process_rr (sess, &packet, &arrival);
2101         break;
2102       case GST_RTCP_TYPE_SDES:
2103         rtp_session_process_sdes (sess, &packet, &arrival);
2104         break;
2105       case GST_RTCP_TYPE_BYE:
2106         is_bye = TRUE;
2107         /* don't try to attempt lip-sync anymore for streams with a BYE */
2108         do_sync = FALSE;
2109         rtp_session_process_bye (sess, &packet, &arrival);
2110         break;
2111       case GST_RTCP_TYPE_APP:
2112         rtp_session_process_app (sess, &packet, &arrival);
2113         break;
2114       case GST_RTCP_TYPE_RTPFB:
2115       case GST_RTCP_TYPE_PSFB:
2116         rtp_session_process_feedback (sess, &packet, &arrival);
2117         break;
2118       default:
2119         GST_WARNING ("got unknown RTCP packet");
2120         break;
2121     }
2122   next:
2123     more = gst_rtcp_packet_move_to_next (&packet);
2124   }
2125
2126   /* if we are scheduling a BYE, we only want to count bye packets, else we
2127    * count everything */
2128   if (sess->source->received_bye) {
2129     if (is_bye) {
2130       sess->stats.bye_members++;
2131       UPDATE_AVG (sess->stats.avg_rtcp_packet_size, arrival.bytes);
2132     }
2133   } else {
2134     /* keep track of average packet size */
2135     UPDATE_AVG (sess->stats.avg_rtcp_packet_size, arrival.bytes);
2136   }
2137   GST_DEBUG ("%p, received RTCP packet, avg size %u, %u", &sess->stats,
2138       sess->stats.avg_rtcp_packet_size, arrival.bytes);
2139   RTP_SESSION_UNLOCK (sess);
2140
2141   /* notify caller of sr packets in the callback */
2142   if (do_sync && sess->callbacks.sync_rtcp) {
2143     /* make writable, we might want to change the buffer */
2144     buffer = gst_buffer_make_metadata_writable (buffer);
2145
2146     result = sess->callbacks.sync_rtcp (sess, sess->source, buffer,
2147         sess->sync_rtcp_user_data);
2148   } else
2149     gst_buffer_unref (buffer);
2150
2151   return result;
2152
2153   /* ERRORS */
2154 invalid_packet:
2155   {
2156     GST_DEBUG ("invalid RTCP packet received");
2157     gst_buffer_unref (buffer);
2158     return GST_FLOW_OK;
2159   }
2160 ignore:
2161   {
2162     gst_buffer_unref (buffer);
2163     RTP_SESSION_UNLOCK (sess);
2164     GST_DEBUG ("ignoring RTP packet because we left");
2165     return GST_FLOW_OK;
2166   }
2167 }
2168
2169 /**
2170  * rtp_session_send_rtp:
2171  * @sess: an #RTPSession
2172  * @data: pointer to either an RTP buffer or a list of RTP buffers
2173  * @is_list: TRUE when @data is a buffer list
2174  * @current_time: the current system time
2175  * @running_time: the running time of @data
2176  *
2177  * Send the RTP buffer in the session manager. This function takes ownership of
2178  * @buffer.
2179  *
2180  * Returns: a #GstFlowReturn.
2181  */
2182 GstFlowReturn
2183 rtp_session_send_rtp (RTPSession * sess, gpointer data, gboolean is_list,
2184     GstClockTime current_time, GstClockTime running_time)
2185 {
2186   GstFlowReturn result;
2187   RTPSource *source;
2188   gboolean prevsender;
2189   gboolean valid_packet;
2190   guint64 oldrate;
2191
2192   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
2193   g_return_val_if_fail (is_list || GST_IS_BUFFER (data), GST_FLOW_ERROR);
2194
2195   if (is_list) {
2196     valid_packet = gst_rtp_buffer_list_validate (GST_BUFFER_LIST_CAST (data));
2197   } else {
2198     valid_packet = gst_rtp_buffer_validate (GST_BUFFER_CAST (data));
2199   }
2200
2201   if (!valid_packet)
2202     goto invalid_packet;
2203
2204   GST_LOG ("received RTP %s for sending", is_list ? "list" : "packet");
2205
2206   RTP_SESSION_LOCK (sess);
2207   source = sess->source;
2208
2209   /* update last activity */
2210   source->last_rtp_activity = current_time;
2211
2212   prevsender = RTP_SOURCE_IS_SENDER (source);
2213   oldrate = source->bitrate;
2214
2215   /* we use our own source to send */
2216   result = rtp_source_send_rtp (source, data, is_list, running_time);
2217
2218   if (RTP_SOURCE_IS_SENDER (source) && !prevsender)
2219     sess->stats.sender_sources++;
2220   if (oldrate != source->bitrate)
2221     sess->recalc_bandwidth = TRUE;
2222   RTP_SESSION_UNLOCK (sess);
2223
2224   return result;
2225
2226   /* ERRORS */
2227 invalid_packet:
2228   {
2229     gst_mini_object_unref (GST_MINI_OBJECT_CAST (data));
2230     GST_DEBUG ("invalid RTP packet received");
2231     return GST_FLOW_OK;
2232   }
2233 }
2234
2235 static void
2236 add_bitrates (gpointer key, RTPSource * source, gdouble * bandwidth)
2237 {
2238   *bandwidth += source->bitrate;
2239 }
2240
2241 static GstClockTime
2242 calculate_rtcp_interval (RTPSession * sess, gboolean deterministic,
2243     gboolean first)
2244 {
2245   GstClockTime result;
2246
2247   /* recalculate bandwidth when it changed */
2248   if (sess->recalc_bandwidth) {
2249     gdouble bandwidth;
2250
2251     if (sess->bandwidth > 0)
2252       bandwidth = sess->bandwidth;
2253     else {
2254       /* If it is <= 0, then try to estimate the actual bandwidth */
2255       bandwidth = sess->source->bitrate;
2256
2257       g_hash_table_foreach (sess->cnames, (GHFunc) add_bitrates, &bandwidth);
2258       bandwidth /= 8.0;
2259     }
2260     if (bandwidth == 0)
2261       bandwidth = RTP_STATS_BANDWIDTH;
2262
2263     rtp_stats_set_bandwidths (&sess->stats, bandwidth,
2264         sess->rtcp_bandwidth, sess->rtcp_rs_bandwidth, sess->rtcp_rr_bandwidth);
2265
2266     sess->recalc_bandwidth = FALSE;
2267   }
2268
2269   if (sess->source->received_bye) {
2270     result = rtp_stats_calculate_bye_interval (&sess->stats);
2271   } else {
2272     result = rtp_stats_calculate_rtcp_interval (&sess->stats,
2273         RTP_SOURCE_IS_SENDER (sess->source), first);
2274   }
2275
2276   GST_DEBUG ("next deterministic interval: %" GST_TIME_FORMAT ", first %d",
2277       GST_TIME_ARGS (result), first);
2278
2279   if (!deterministic && result != GST_CLOCK_TIME_NONE)
2280     result = rtp_stats_add_rtcp_jitter (&sess->stats, result);
2281
2282   GST_DEBUG ("next interval: %" GST_TIME_FORMAT, GST_TIME_ARGS (result));
2283
2284   return result;
2285 }
2286
2287 /* Stop the current @sess and schedule a BYE message for the other members.
2288  * One must have the session lock to call this function
2289  */
2290 static GstFlowReturn
2291 rtp_session_schedule_bye_locked (RTPSession * sess, const gchar * reason,
2292     GstClockTime current_time)
2293 {
2294   GstFlowReturn result = GST_FLOW_OK;
2295   RTPSource *source;
2296   GstClockTime interval;
2297
2298   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
2299
2300   source = sess->source;
2301
2302   /* ignore more BYEs */
2303   if (source->received_bye)
2304     goto done;
2305
2306   /* we have BYE now */
2307   source->received_bye = TRUE;
2308   /* at least one member wants to send a BYE */
2309   g_free (sess->bye_reason);
2310   sess->bye_reason = g_strdup (reason);
2311   INIT_AVG (sess->stats.avg_rtcp_packet_size, 100);
2312   sess->stats.bye_members = 1;
2313   sess->first_rtcp = TRUE;
2314   sess->sent_bye = FALSE;
2315   sess->allow_early = TRUE;
2316
2317   /* reschedule transmission */
2318   sess->last_rtcp_send_time = current_time;
2319   interval = calculate_rtcp_interval (sess, FALSE, TRUE);
2320   sess->next_rtcp_check_time = current_time + interval;
2321
2322   GST_DEBUG ("Schedule BYE for %" GST_TIME_FORMAT ", %" GST_TIME_FORMAT,
2323       GST_TIME_ARGS (interval), GST_TIME_ARGS (sess->next_rtcp_check_time));
2324
2325   RTP_SESSION_UNLOCK (sess);
2326   /* notify app of reconsideration */
2327   if (sess->callbacks.reconsider)
2328     sess->callbacks.reconsider (sess, sess->reconsider_user_data);
2329   RTP_SESSION_LOCK (sess);
2330 done:
2331
2332   return result;
2333 }
2334
2335 /**
2336  * rtp_session_schedule_bye:
2337  * @sess: an #RTPSession
2338  * @reason: a reason or NULL
2339  * @current_time: the current system time
2340  *
2341  * Stop the current @sess and schedule a BYE message for the other members.
2342  *
2343  * Returns: a #GstFlowReturn.
2344  */
2345 GstFlowReturn
2346 rtp_session_schedule_bye (RTPSession * sess, const gchar * reason,
2347     GstClockTime current_time)
2348 {
2349   GstFlowReturn result = GST_FLOW_OK;
2350
2351   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
2352
2353   RTP_SESSION_LOCK (sess);
2354   result = rtp_session_schedule_bye_locked (sess, reason, current_time);
2355   RTP_SESSION_UNLOCK (sess);
2356
2357   return result;
2358 }
2359
2360 /**
2361  * rtp_session_next_timeout:
2362  * @sess: an #RTPSession
2363  * @current_time: the current system time
2364  *
2365  * Get the next time we should perform session maintenance tasks.
2366  *
2367  * Returns: a time when rtp_session_on_timeout() should be called with the
2368  * current system time.
2369  */
2370 GstClockTime
2371 rtp_session_next_timeout (RTPSession * sess, GstClockTime current_time)
2372 {
2373   GstClockTime result, interval = 0;
2374
2375   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_CLOCK_TIME_NONE);
2376
2377   RTP_SESSION_LOCK (sess);
2378
2379   if (GST_CLOCK_TIME_IS_VALID (sess->next_early_rtcp_time)) {
2380     result = sess->next_early_rtcp_time;
2381     goto early_exit;
2382   }
2383
2384   result = sess->next_rtcp_check_time;
2385
2386   GST_DEBUG ("current time: %" GST_TIME_FORMAT ", next :%" GST_TIME_FORMAT,
2387       GST_TIME_ARGS (current_time), GST_TIME_ARGS (result));
2388
2389   if (result < current_time) {
2390     GST_DEBUG ("take current time as base");
2391     /* our previous check time expired, start counting from the current time
2392      * again. */
2393     result = current_time;
2394   }
2395
2396   if (sess->source->received_bye) {
2397     if (sess->sent_bye) {
2398       GST_DEBUG ("we sent BYE already");
2399       interval = GST_CLOCK_TIME_NONE;
2400     } else if (sess->stats.active_sources >= 50) {
2401       GST_DEBUG ("reconsider BYE, more than 50 sources");
2402       /* reconsider BYE if members >= 50 */
2403       interval = calculate_rtcp_interval (sess, FALSE, TRUE);
2404     }
2405   } else {
2406     if (sess->first_rtcp) {
2407       GST_DEBUG ("first RTCP packet");
2408       /* we are called for the first time */
2409       interval = calculate_rtcp_interval (sess, FALSE, TRUE);
2410     } else if (sess->next_rtcp_check_time < current_time) {
2411       GST_DEBUG ("old check time expired, getting new timeout");
2412       /* get a new timeout when we need to */
2413       interval = calculate_rtcp_interval (sess, FALSE, FALSE);
2414     }
2415   }
2416
2417   if (interval != GST_CLOCK_TIME_NONE)
2418     result += interval;
2419   else
2420     result = GST_CLOCK_TIME_NONE;
2421
2422   sess->next_rtcp_check_time = result;
2423
2424 early_exit:
2425
2426   GST_DEBUG ("current time: %" GST_TIME_FORMAT
2427       ", next time: %" GST_TIME_FORMAT,
2428       GST_TIME_ARGS (current_time), GST_TIME_ARGS (result));
2429   RTP_SESSION_UNLOCK (sess);
2430
2431   return result;
2432 }
2433
2434 typedef struct
2435 {
2436   RTPSession *sess;
2437   GstBuffer *rtcp;
2438   GstClockTime current_time;
2439   guint64 ntpnstime;
2440   GstClockTime running_time;
2441   GstClockTime interval;
2442   GstRTCPPacket packet;
2443   gboolean is_bye;
2444   gboolean has_sdes;
2445   gboolean is_early;
2446   gboolean may_suppress;
2447 } ReportData;
2448
2449 static void
2450 session_start_rtcp (RTPSession * sess, ReportData * data)
2451 {
2452   GstRTCPPacket *packet = &data->packet;
2453   RTPSource *own = sess->source;
2454
2455   data->rtcp = gst_rtcp_buffer_new (sess->mtu);
2456
2457   if (RTP_SOURCE_IS_SENDER (own)) {
2458     guint64 ntptime;
2459     guint32 rtptime;
2460     guint32 packet_count, octet_count;
2461
2462     /* we are a sender, create SR */
2463     GST_DEBUG ("create SR for SSRC %08x", own->ssrc);
2464     gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_SR, packet);
2465
2466     /* get latest stats */
2467     rtp_source_get_new_sr (own, data->ntpnstime, data->running_time,
2468         &ntptime, &rtptime, &packet_count, &octet_count);
2469     /* store stats */
2470     rtp_source_process_sr (own, data->current_time, ntptime, rtptime,
2471         packet_count, octet_count);
2472
2473     /* fill in sender report info */
2474     gst_rtcp_packet_sr_set_sender_info (packet, own->ssrc,
2475         ntptime, rtptime, packet_count, octet_count);
2476   } else {
2477     /* we are only receiver, create RR */
2478     GST_DEBUG ("create RR for SSRC %08x", own->ssrc);
2479     gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_RR, packet);
2480     gst_rtcp_packet_rr_set_ssrc (packet, own->ssrc);
2481   }
2482 }
2483
2484 /* construct a Sender or Receiver Report */
2485 static void
2486 session_report_blocks (const gchar * key, RTPSource * source, ReportData * data)
2487 {
2488   RTPSession *sess = data->sess;
2489   GstRTCPPacket *packet = &data->packet;
2490
2491   /* create a new buffer if needed */
2492   if (data->rtcp == NULL) {
2493     session_start_rtcp (sess, data);
2494   } else if (data->is_early) {
2495     /* Put a single RR or SR in minimal compound packets */
2496     return;
2497   }
2498   if (gst_rtcp_packet_get_rb_count (packet) < GST_RTCP_MAX_RB_COUNT) {
2499     /* only report about other sender sources */
2500     if (source != sess->source && RTP_SOURCE_IS_SENDER (source)) {
2501       guint8 fractionlost;
2502       gint32 packetslost;
2503       guint32 exthighestseq, jitter;
2504       guint32 lsr, dlsr;
2505
2506       /* get new stats */
2507       rtp_source_get_new_rb (source, data->current_time, &fractionlost,
2508           &packetslost, &exthighestseq, &jitter, &lsr, &dlsr);
2509
2510       /* store last generated RR packet */
2511       source->last_rr.is_valid = TRUE;
2512       source->last_rr.fractionlost = fractionlost;
2513       source->last_rr.packetslost = packetslost;
2514       source->last_rr.exthighestseq = exthighestseq;
2515       source->last_rr.jitter = jitter;
2516       source->last_rr.lsr = lsr;
2517       source->last_rr.dlsr = dlsr;
2518
2519       /* packet is not yet filled, add report block for this source. */
2520       gst_rtcp_packet_add_rb (packet, source->ssrc, fractionlost, packetslost,
2521           exthighestseq, jitter, lsr, dlsr);
2522     }
2523   }
2524 }
2525
2526 /* perform cleanup of sources that timed out */
2527 static void
2528 session_cleanup (const gchar * key, RTPSource * source, ReportData * data)
2529 {
2530   gboolean remove = FALSE;
2531   gboolean byetimeout = FALSE;
2532   gboolean sendertimeout = FALSE;
2533   gboolean is_sender, is_active;
2534   RTPSession *sess = data->sess;
2535   GstClockTime interval;
2536
2537   is_sender = RTP_SOURCE_IS_SENDER (source);
2538   is_active = RTP_SOURCE_IS_ACTIVE (source);
2539
2540   /* check for our own source, we don't want to delete our own source. */
2541   if (!(source == sess->source)) {
2542     if (source->received_bye) {
2543       /* if we received a BYE from the source, remove the source after some
2544        * time. */
2545       if (data->current_time > source->bye_time &&
2546           data->current_time - source->bye_time > sess->stats.bye_timeout) {
2547         GST_DEBUG ("removing BYE source %08x", source->ssrc);
2548         remove = TRUE;
2549         byetimeout = TRUE;
2550       }
2551     }
2552     /* sources that were inactive for more than 5 times the deterministic reporting
2553      * interval get timed out. the min timeout is 5 seconds. */
2554     if (data->current_time > source->last_activity) {
2555       interval = MAX (data->interval * 5, 5 * GST_SECOND);
2556       if (data->current_time - source->last_activity > interval) {
2557         GST_DEBUG ("removing timeout source %08x, last %" GST_TIME_FORMAT,
2558             source->ssrc, GST_TIME_ARGS (source->last_activity));
2559         remove = TRUE;
2560       }
2561     }
2562   }
2563
2564   /* senders that did not send for a long time become a receiver, this also
2565    * holds for our own source. */
2566   if (is_sender) {
2567     if (data->current_time > source->last_rtp_activity) {
2568       interval = MAX (data->interval * 2, 5 * GST_SECOND);
2569       if (data->current_time - source->last_rtp_activity > interval) {
2570         GST_DEBUG ("sender source %08x timed out and became receiver, last %"
2571             GST_TIME_FORMAT, source->ssrc,
2572             GST_TIME_ARGS (source->last_rtp_activity));
2573         source->is_sender = FALSE;
2574         sess->stats.sender_sources--;
2575         sendertimeout = TRUE;
2576       }
2577     }
2578   }
2579
2580   if (remove) {
2581     sess->total_sources--;
2582     if (is_sender)
2583       sess->stats.sender_sources--;
2584     if (is_active)
2585       sess->stats.active_sources--;
2586
2587     if (byetimeout)
2588       on_bye_timeout (sess, source);
2589     else
2590       on_timeout (sess, source);
2591   } else {
2592     if (sendertimeout)
2593       on_sender_timeout (sess, source);
2594   }
2595
2596   source->closing = remove;
2597 }
2598
2599 static void
2600 session_sdes (RTPSession * sess, ReportData * data)
2601 {
2602   GstRTCPPacket *packet = &data->packet;
2603   const GstStructure *sdes;
2604   gint i, n_fields;
2605
2606   /* add SDES packet */
2607   gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_SDES, packet);
2608
2609   gst_rtcp_packet_sdes_add_item (packet, sess->source->ssrc);
2610
2611   sdes = rtp_source_get_sdes_struct (sess->source);
2612
2613   /* add all fields in the structure, the order is not important. */
2614   n_fields = gst_structure_n_fields (sdes);
2615   for (i = 0; i < n_fields; ++i) {
2616     const gchar *field;
2617     const gchar *value;
2618     GstRTCPSDESType type;
2619
2620     field = gst_structure_nth_field_name (sdes, i);
2621     if (field == NULL)
2622       continue;
2623     value = gst_structure_get_string (sdes, field);
2624     if (value == NULL)
2625       continue;
2626     type = gst_rtcp_sdes_name_to_type (field);
2627
2628     /* Early packets are minimal and only include the CNAME */
2629     if (data->is_early && type != GST_RTCP_SDES_CNAME)
2630       continue;
2631
2632     if (type > GST_RTCP_SDES_END && type < GST_RTCP_SDES_PRIV) {
2633       gst_rtcp_packet_sdes_add_entry (packet, type, strlen (value),
2634           (const guint8 *) value);
2635     } else if (type == GST_RTCP_SDES_PRIV) {
2636       gsize prefix_len;
2637       gsize value_len;
2638       gsize data_len;
2639       guint8 data[256];
2640
2641       /* don't accept entries that are too big */
2642       prefix_len = strlen (field);
2643       if (prefix_len > 255)
2644         continue;
2645       value_len = strlen (value);
2646       if (value_len > 255)
2647         continue;
2648       data_len = 1 + prefix_len + value_len;
2649       if (data_len > 255)
2650         continue;
2651
2652       data[0] = prefix_len;
2653       memcpy (&data[1], field, prefix_len);
2654       memcpy (&data[1 + prefix_len], value, value_len);
2655
2656       gst_rtcp_packet_sdes_add_entry (packet, type, data_len, data);
2657     }
2658   }
2659
2660   data->has_sdes = TRUE;
2661 }
2662
2663 /* schedule a BYE packet */
2664 static void
2665 session_bye (RTPSession * sess, ReportData * data)
2666 {
2667   GstRTCPPacket *packet = &data->packet;
2668
2669   /* open packet */
2670   session_start_rtcp (sess, data);
2671
2672   /* add SDES */
2673   session_sdes (sess, data);
2674
2675   /* add a BYE packet */
2676   gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_BYE, packet);
2677   gst_rtcp_packet_bye_add_ssrc (packet, sess->source->ssrc);
2678   if (sess->bye_reason)
2679     gst_rtcp_packet_bye_set_reason (packet, sess->bye_reason);
2680
2681   /* we have a BYE packet now */
2682   data->is_bye = TRUE;
2683 }
2684
2685 static gboolean
2686 is_rtcp_time (RTPSession * sess, GstClockTime current_time, ReportData * data)
2687 {
2688   GstClockTime new_send_time, elapsed;
2689
2690   if (data->is_early && sess->next_early_rtcp_time < current_time)
2691     goto early;
2692
2693   /* no need to check yet */
2694   if (sess->next_rtcp_check_time > current_time) {
2695     GST_DEBUG ("no check time yet, next %" GST_TIME_FORMAT " > now %"
2696         GST_TIME_FORMAT, GST_TIME_ARGS (sess->next_rtcp_check_time),
2697         GST_TIME_ARGS (current_time));
2698     return FALSE;
2699   }
2700
2701   /* get elapsed time since we last reported */
2702   elapsed = current_time - sess->last_rtcp_send_time;
2703
2704   /* perform forward reconsideration */
2705   new_send_time = rtp_stats_add_rtcp_jitter (&sess->stats, data->interval);
2706
2707   GST_DEBUG ("forward reconsideration %" GST_TIME_FORMAT ", elapsed %"
2708       GST_TIME_FORMAT, GST_TIME_ARGS (new_send_time), GST_TIME_ARGS (elapsed));
2709
2710   new_send_time += sess->last_rtcp_send_time;
2711
2712   /* check if reconsideration */
2713   if (current_time < new_send_time) {
2714     GST_DEBUG ("reconsider RTCP for %" GST_TIME_FORMAT,
2715         GST_TIME_ARGS (new_send_time));
2716     /* store new check time */
2717     sess->next_rtcp_check_time = new_send_time;
2718     return FALSE;
2719   }
2720
2721 early:
2722
2723   new_send_time = calculate_rtcp_interval (sess, FALSE, FALSE);
2724
2725   GST_DEBUG ("can send RTCP now, next interval %" GST_TIME_FORMAT,
2726       GST_TIME_ARGS (new_send_time));
2727   sess->next_rtcp_check_time = current_time + new_send_time;
2728
2729   /* Apply the rules from RFC 4585 section 3.5.3 */
2730   if (sess->stats.min_interval != 0 && !sess->first_rtcp) {
2731     GstClockTimeDiff T_rr_current_interval = g_random_double_range (0.5, 1.5) *
2732         sess->stats.min_interval;
2733
2734     /* This will caused the RTCP to be suppressed if no FB packets are added */
2735     if (sess->last_rtcp_send_time + T_rr_current_interval >
2736         sess->next_rtcp_check_time) {
2737       GST_DEBUG ("RTCP packet could be suppressed min: %" GST_TIME_FORMAT
2738           " last: %" GST_TIME_FORMAT
2739           " + T_rr_current_interval: %" GST_TIME_FORMAT
2740           " >  sess->next_rtcp_check_time: %" GST_TIME_FORMAT,
2741           GST_TIME_ARGS (sess->stats.min_interval),
2742           GST_TIME_ARGS (sess->last_rtcp_send_time),
2743           GST_TIME_ARGS (T_rr_current_interval),
2744           GST_TIME_ARGS (sess->next_rtcp_check_time));
2745       data->may_suppress = TRUE;
2746     }
2747   }
2748
2749   return TRUE;
2750 }
2751
2752 static void
2753 clone_ssrcs_hashtable (gchar * key, RTPSource * source, GHashTable * hash_table)
2754 {
2755   g_hash_table_insert (hash_table, key, g_object_ref (source));
2756 }
2757
2758 static gboolean
2759 remove_closing_sources (const gchar * key, RTPSource * source, gpointer * data)
2760 {
2761   return source->closing;
2762 }
2763
2764 /**
2765  * rtp_session_on_timeout:
2766  * @sess: an #RTPSession
2767  * @current_time: the current system time
2768  * @ntpnstime: the current NTP time in nanoseconds
2769  * @running_time: the current running_time of the pipeline
2770  *
2771  * Perform maintenance actions after the timeout obtained with
2772  * rtp_session_next_timeout() expired.
2773  *
2774  * This function will perform timeouts of receivers and senders, send a BYE
2775  * packet or generate RTCP packets with current session stats.
2776  *
2777  * This function can call the #RTPSessionSendRTCP callback, possibly multiple
2778  * times, for each packet that should be processed.
2779  *
2780  * Returns: a #GstFlowReturn.
2781  */
2782 GstFlowReturn
2783 rtp_session_on_timeout (RTPSession * sess, GstClockTime current_time,
2784     guint64 ntpnstime, GstClockTime running_time)
2785 {
2786   GstFlowReturn result = GST_FLOW_OK;
2787   ReportData data;
2788   RTPSource *own;
2789   GHashTable *table_copy;
2790   gboolean notify = FALSE;
2791
2792   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
2793
2794   GST_DEBUG ("reporting at %" GST_TIME_FORMAT ", NTP time %" GST_TIME_FORMAT,
2795       GST_TIME_ARGS (current_time), GST_TIME_ARGS (ntpnstime));
2796
2797   data.sess = sess;
2798   data.rtcp = NULL;
2799   data.current_time = current_time;
2800   data.ntpnstime = ntpnstime;
2801   data.is_bye = FALSE;
2802   data.has_sdes = FALSE;
2803   data.may_suppress = FALSE;
2804   data.running_time = running_time;
2805
2806   own = sess->source;
2807
2808   RTP_SESSION_LOCK (sess);
2809   /* get a new interval, we need this for various cleanups etc */
2810   data.interval = calculate_rtcp_interval (sess, TRUE, sess->first_rtcp);
2811
2812   /* Make a local copy of the hashtable. We need to do this because the
2813    * cleanup stage below releases the session lock. */
2814   table_copy = g_hash_table_new_full (NULL, NULL, NULL,
2815       (GDestroyNotify) g_object_unref);
2816   g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
2817       (GHFunc) clone_ssrcs_hashtable, table_copy);
2818
2819   /* Clean up the session, mark the source for removing, this might release the
2820    * session lock. */
2821   g_hash_table_foreach (table_copy, (GHFunc) session_cleanup, &data);
2822   g_hash_table_destroy (table_copy);
2823
2824   /* Now remove the marked sources */
2825   g_hash_table_foreach_remove (sess->ssrcs[sess->mask_idx],
2826       (GHRFunc) remove_closing_sources, NULL);
2827
2828   if (GST_CLOCK_TIME_IS_VALID (sess->next_early_rtcp_time))
2829     data.is_early = TRUE;
2830   else
2831     data.is_early = FALSE;
2832
2833   /* see if we need to generate SR or RR packets */
2834   if (is_rtcp_time (sess, current_time, &data)) {
2835     if (own->received_bye) {
2836       /* generate BYE instead */
2837       GST_DEBUG ("generating BYE message");
2838       session_bye (sess, &data);
2839       sess->sent_bye = TRUE;
2840     } else {
2841       /* loop over all known sources and do something */
2842       g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
2843           (GHFunc) session_report_blocks, &data);
2844     }
2845   }
2846
2847   if (data.rtcp) {
2848     /* we keep track of the last report time in order to timeout inactive
2849      * receivers or senders */
2850     if (!data.is_early && !data.may_suppress)
2851       sess->last_rtcp_send_time = data.current_time;
2852     sess->first_rtcp = FALSE;
2853     sess->next_early_rtcp_time = GST_CLOCK_TIME_NONE;
2854
2855     /* add SDES for this source when not already added */
2856     if (!data.has_sdes)
2857       session_sdes (sess, &data);
2858   }
2859
2860   /* check for outdated collisions */
2861   GST_DEBUG ("Timing out collisions");
2862   rtp_source_timeout (sess->source, current_time,
2863       data.interval * RTCP_INTERVAL_COLLISION_TIMEOUT,
2864       running_time - sess->rtcp_feedback_retention_window);
2865
2866   if (sess->change_ssrc) {
2867     GST_DEBUG ("need to change our SSRC (%08x)", own->ssrc);
2868     g_hash_table_steal (sess->ssrcs[sess->mask_idx],
2869         GINT_TO_POINTER (own->ssrc));
2870
2871     own->ssrc = rtp_session_create_new_ssrc (sess);
2872     rtp_source_reset (own);
2873
2874     g_hash_table_insert (sess->ssrcs[sess->mask_idx],
2875         GINT_TO_POINTER (own->ssrc), own);
2876
2877     g_free (sess->bye_reason);
2878     sess->bye_reason = NULL;
2879     sess->sent_bye = FALSE;
2880     sess->change_ssrc = FALSE;
2881     notify = TRUE;
2882     GST_DEBUG ("changed our SSRC to %08x", own->ssrc);
2883   }
2884
2885   sess->allow_early = TRUE;
2886
2887   RTP_SESSION_UNLOCK (sess);
2888
2889   if (notify)
2890     g_object_notify (G_OBJECT (sess), "internal-ssrc");
2891
2892   /* push out the RTCP packet */
2893   if (data.rtcp) {
2894     gboolean do_not_suppress;
2895
2896     /* Give the user a change to add its own packet */
2897     g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_SENDING_RTCP], 0,
2898         data.rtcp, data.is_early, &do_not_suppress);
2899
2900     if (sess->callbacks.send_rtcp && (do_not_suppress || !data.may_suppress)) {
2901       guint packet_size;
2902
2903       /* close the RTCP packet */
2904       gst_rtcp_buffer_end (data.rtcp);
2905
2906       packet_size = GST_BUFFER_SIZE (data.rtcp) + sess->header_len;
2907
2908       UPDATE_AVG (sess->stats.avg_rtcp_packet_size, packet_size);
2909       GST_DEBUG ("%p, sending RTCP packet, avg size %u, %u", &sess->stats,
2910           sess->stats.avg_rtcp_packet_size, packet_size);
2911       result =
2912           sess->callbacks.send_rtcp (sess, own, data.rtcp, sess->sent_bye,
2913           sess->send_rtcp_user_data);
2914     } else {
2915       GST_DEBUG ("freeing packet callback: %p"
2916           " do_not_suppress: %d may_suppress: %d",
2917           sess->callbacks.send_rtcp, do_not_suppress, data.may_suppress);
2918       gst_buffer_unref (data.rtcp);
2919     }
2920   }
2921
2922   return result;
2923 }
2924
2925 void
2926 rtp_session_request_early_rtcp (RTPSession * sess, GstClockTime current_time,
2927     GstClockTimeDiff max_delay)
2928 {
2929   GstClockTime T_dither_max;
2930
2931   /* Implements the algorithm described in RFC 4585 section 3.5.2 */
2932
2933   RTP_SESSION_LOCK (sess);
2934
2935   /* Check if already requested */
2936   /*  RFC 4585 section 3.5.2 step 2 */
2937   if (GST_CLOCK_TIME_IS_VALID (sess->next_early_rtcp_time))
2938     goto dont_send;
2939
2940   /* Ignore the request a scheduled packet will be in time anyway */
2941   if (current_time + max_delay > sess->next_rtcp_check_time)
2942     goto dont_send;
2943
2944   /*  RFC 4585 section 3.5.2 step 2b */
2945   /* If the total sources is <=2, then there is only us and one peer */
2946   if (sess->total_sources <= 2) {
2947     T_dither_max = 0;
2948   } else {
2949     /* Divide by 2 because l = 0.5 */
2950     T_dither_max = sess->next_rtcp_check_time - sess->last_rtcp_send_time;
2951     T_dither_max /= 2;
2952   }
2953
2954   /*  RFC 4585 section 3.5.2 step 3 */
2955   if (current_time + T_dither_max > sess->next_rtcp_check_time)
2956     goto dont_send;
2957
2958   /*  RFC 4585 section 3.5.2 step 4 */
2959   if (sess->allow_early == FALSE)
2960     goto dont_send;
2961
2962   if (T_dither_max) {
2963     /* Schedule an early transmission later */
2964     sess->next_early_rtcp_time = g_random_double () * T_dither_max +
2965         current_time;
2966   } else {
2967     /* If no dithering, schedule it for NOW */
2968     sess->next_early_rtcp_time = current_time;
2969   }
2970
2971   RTP_SESSION_UNLOCK (sess);
2972
2973   /* notify app of need to send packet early
2974    * and therefore of timeout change */
2975   if (sess->callbacks.reconsider)
2976     sess->callbacks.reconsider (sess, sess->reconsider_user_data);
2977
2978   return;
2979
2980 dont_send:
2981
2982   RTP_SESSION_UNLOCK (sess);
2983
2984 }
2985
2986 void
2987 rtp_session_request_key_unit (RTPSession * sess, guint32 ssrc, gboolean fir)
2988 {
2989   guint i;
2990
2991   if (fir)
2992     return;
2993
2994   for (i = 0; i < sess->rtcp_pli_requests->len; i++)
2995     if (ssrc == g_array_index (sess->rtcp_pli_requests, guint32, i))
2996       return;
2997
2998   g_array_append_val (sess->rtcp_pli_requests, ssrc);
2999 }
3000
3001 static gboolean
3002 has_pli_compare_func (gconstpointer a, gconstpointer ignored)
3003 {
3004   GstRTCPPacket packet;
3005
3006   packet.buffer = (GstBuffer *) a;
3007   packet.offset = 0;
3008
3009   if (gst_rtcp_packet_get_type (&packet) == GST_RTCP_TYPE_PSFB &&
3010       gst_rtcp_packet_fb_get_type (&packet) == GST_RTCP_PSFB_TYPE_PLI)
3011     return TRUE;
3012   else
3013     return FALSE;
3014 }
3015
3016 static gboolean
3017 rtp_session_on_sending_rtcp (RTPSession * sess, GstBuffer * buffer,
3018     gboolean early)
3019 {
3020   gboolean ret = FALSE;
3021
3022   RTP_SESSION_LOCK (sess);
3023
3024   while (sess->rtcp_pli_requests->len) {
3025     GstRTCPPacket rtcppacket;
3026     guint media_ssrc = g_array_index (sess->rtcp_pli_requests, guint32, 0);
3027     RTPSource *media_src = g_hash_table_lookup (sess->ssrcs[sess->mask_idx],
3028         GUINT_TO_POINTER (media_ssrc));
3029
3030     if (media_src && !rtp_source_has_retained (media_src,
3031             has_pli_compare_func, NULL)) {
3032       if (gst_rtcp_buffer_add_packet (buffer, GST_RTCP_TYPE_PSFB, &rtcppacket)) {
3033         gst_rtcp_packet_fb_set_type (&rtcppacket, GST_RTCP_PSFB_TYPE_PLI);
3034         gst_rtcp_packet_fb_set_sender_ssrc (&rtcppacket,
3035             rtp_source_get_ssrc (sess->source));
3036         gst_rtcp_packet_fb_set_media_ssrc (&rtcppacket, media_ssrc);
3037         ret = TRUE;
3038       } else {
3039         /* Break because the packet is full, will put next request in a
3040          * further packet
3041          */
3042         break;
3043       }
3044     }
3045
3046     g_array_remove_index (sess->rtcp_pli_requests, 0);
3047   }
3048
3049   RTP_SESSION_UNLOCK (sess);
3050
3051   return ret;
3052 }