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