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