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