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   RTP_SESSION_UNLOCK (session);
 848
 849   if (session->callbacks.clock_rate)
 850     result =
 851         session->callbacks.clock_rate (session, pt,
 852         session->clock_rate_user_data);
 853   else
 854     result = -1;
 855
 856   RTP_SESSION_LOCK (session);
 857
 858   GST_DEBUG ("got clock-rate %d for pt %d", result, pt);
 859
 860   return result;
 861 }
 862
 863 static RTPSourceCallbacks callbacks = {
 864   (RTPSourcePushRTP) source_push_rtp,
 865   (RTPSourceClockRate) source_clock_rate,
 866 };
 867
 868 /**
 869  * find_add_conflicting_addresses:
 870  * @sess: The session to check in
 871  * @arrival: The arrival stats for the buffer
 872  *
 873  * Checks if an address which has a conflict is already known,
 874  *  otherwise remembers it to prevent loops.
 875  *
 876  * Returns: TRUE if it was a known conflict, FALSE otherwise
 877  */
 878
 879 static gboolean
 880 find_add_conflicting_addresses (RTPSession * sess, RTPArrivalStats * arrival)
 881 {
 882   GList *item;
 883   RTPConflictingAddress *new_conflict;
 884
 885   for (item = g_list_first (sess->conflicting_addresses);
 886       item; item = g_list_next (item)) {
 887     RTPConflictingAddress *known_conflict = item->data;
 888
 889     if (gst_netaddress_equal (&arrival->address, &known_conflict->address)) {
 890       known_conflict->time = arrival->time;
 891       return TRUE;
 892     }
 893   }
 894
 895   new_conflict = g_new0 (RTPConflictingAddress, 1);
 896
 897   memcpy (&new_conflict->address, &arrival->address, sizeof (GstNetAddress));
 898   new_conflict->time = arrival->time;
 899
 900   sess->conflicting_addresses = g_list_prepend (sess->conflicting_addresses,
 901       new_conflict);
 902
 903   return FALSE;
 904 }
 905
 906 static gboolean
 907 check_collision (RTPSession * sess, RTPSource * source,
 908     RTPArrivalStats * arrival, gboolean rtp)
 909 {
 910   /* If we have not arrival address, we can't do collision checking */
 911   if (!arrival->have_address) {
 912     return FALSE;
 913   }
 914
 915   if (sess->source != source) {
 916     /* This is not our local source, but lets check if two remote
 917      * source collide
 918      */
 919
 920     if (rtp) {
 921       if (source->have_rtp_from) {
 922         if (gst_netaddress_equal (&source->rtp_from, &arrival->address))
 923           /* Address is the same */
 924           return FALSE;
 925       } else {
 926         /* We don't already have a from address for RTP, just set it */
 927         rtp_source_set_rtp_from (source, &arrival->address);
 928         return FALSE;
 929       }
 930     } else {
 931       if (source->have_rtcp_from) {
 932         if (gst_netaddress_equal (&source->rtcp_from, &arrival->address))
 933           /* Address is the same */
 934           return FALSE;
 935       } else {
 936         /* We don't already have a from address for RTCP, just set it */
 937         rtp_source_set_rtcp_from (source, &arrival->address);
 938         return FALSE;
 939       }
 940     }
 941
 942     /* In this case, we have third-party collision or loop */
 943
 944     /* FIXME: Log 3rd party collision somehow
 945      * Maybe should be done in upper layer, only the SDES can tell us
 946      * if its a collision or a loop
 947      */
 948   } else {
 949     /* This is sending with our ssrc, is it an address we already know */
 950
 951     if (find_add_conflicting_addresses (sess, arrival)) {
 952       /* Its a known conflict, its probably a loop, not a collision
 953        * lets just drop the incoming packet
 954        */
 955       GST_DEBUG ("Our packets are being looped back to us, dropping");
 956     } else {
 957       /* Its a new collision, lets change our SSRC */
 958
 959       GST_DEBUG ("Collision for SSRC %x", rtp_source_get_ssrc (source));
 960       on_ssrc_collision (sess, source);
 961
 962       rtp_session_send_bye_locked (sess, "SSRC Collision");
 963
 964       sess->change_ssrc = TRUE;
 965     }
 966   }
 967
 968   return TRUE;
 969 }
 970
 971
 972 /* must be called with the session lock */
 973 static RTPSource *
 974 obtain_source (RTPSession * sess, guint32 ssrc, gboolean * created,
 975     RTPArrivalStats * arrival, gboolean rtp)
 976 {
 977   RTPSource *source;
 978
 979   source =
 980       g_hash_table_lookup (sess->ssrcs[sess->mask_idx], GINT_TO_POINTER (ssrc));
 981   if (source == NULL) {
 982     /* make new Source in probation and insert */
 983     source = rtp_source_new (ssrc);
 984
 985     /* for RTP packets we need to set the source in probation. Receiving RTCP
 986      * packets of an SSRC, on the other hand, is a strong indication that we
 987      * are dealing with a valid source. */
 988     if (rtp)
 989       source->probation = RTP_DEFAULT_PROBATION;
 990     else
 991       source->probation = 0;
 992
 993     /* store from address, if any */
 994     if (arrival->have_address) {
 995       if (rtp)
 996         rtp_source_set_rtp_from (source, &arrival->address);
 997       else
 998         rtp_source_set_rtcp_from (source, &arrival->address);
 999     }
1000
1001     /* configure a callback on the source */
1002     rtp_source_set_callbacks (source, &callbacks, sess);
1003
1004     g_hash_table_insert (sess->ssrcs[sess->mask_idx], GINT_TO_POINTER (ssrc),
1005         source);
1006
1007     /* we have one more source now */
1008     sess->total_sources++;
1009     *created = TRUE;
1010   } else {
1011     *created = FALSE;
1012     /* check for collision, this updates the address when not previously set */
1013     if (check_collision (sess, source, arrival, rtp)) {
1014       return NULL;
1015     }
1016   }
1017   /* update last activity */
1018   source->last_activity = arrival->time;
1019   if (rtp)
1020     source->last_rtp_activity = arrival->time;
1021
1022   return source;
1023 }
1024
1025 /**
1026  * rtp_session_get_internal_source:
1027  * @sess: a #RTPSession
1028  *
1029  * Get the internal #RTPSource of @session.
1030  *
1031  * Returns: The internal #RTPSource. g_object_unref() after usage.
1032  */
1033 RTPSource *
1034 rtp_session_get_internal_source (RTPSession * sess)
1035 {
1036   RTPSource *result;
1037
1038   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1039
1040   result = g_object_ref (sess->source);
1041
1042   return result;
1043 }
1044
1045 /**
1046  * rtp_session_add_source:
1047  * @sess: a #RTPSession
1048  * @src: #RTPSource to add
1049  *
1050  * Add @src to @session.
1051  *
1052  * Returns: %TRUE on success, %FALSE if a source with the same SSRC already
1053  * existed in the session.
1054  */
1055 gboolean
1056 rtp_session_add_source (RTPSession * sess, RTPSource * src)
1057 {
1058   gboolean result = FALSE;
1059   RTPSource *find;
1060
1061   g_return_val_if_fail (RTP_IS_SESSION (sess), FALSE);
1062   g_return_val_if_fail (src != NULL, FALSE);
1063
1064   RTP_SESSION_LOCK (sess);
1065   find =
1066       g_hash_table_lookup (sess->ssrcs[sess->mask_idx],
1067       GINT_TO_POINTER (src->ssrc));
1068   if (find == NULL) {
1069     g_hash_table_insert (sess->ssrcs[sess->mask_idx],
1070         GINT_TO_POINTER (src->ssrc), src);
1071     /* we have one more source now */
1072     sess->total_sources++;
1073     result = TRUE;
1074   }
1075   RTP_SESSION_UNLOCK (sess);
1076
1077   return result;
1078 }
1079
1080 /**
1081  * rtp_session_get_num_sources:
1082  * @sess: an #RTPSession
1083  *
1084  * Get the number of sources in @sess.
1085  *
1086  * Returns: The number of sources in @sess.
1087  */
1088 guint
1089 rtp_session_get_num_sources (RTPSession * sess)
1090 {
1091   guint result;
1092
1093   g_return_val_if_fail (RTP_IS_SESSION (sess), FALSE);
1094
1095   RTP_SESSION_LOCK (sess);
1096   result = sess->total_sources;
1097   RTP_SESSION_UNLOCK (sess);
1098
1099   return result;
1100 }
1101
1102 /**
1103  * rtp_session_get_num_active_sources:
1104  * @sess: an #RTPSession
1105  *
1106  * Get the number of active sources in @sess. A source is considered active when
1107  * it has been validated and has not yet received a BYE RTCP message.
1108  *
1109  * Returns: The number of active sources in @sess.
1110  */
1111 guint
1112 rtp_session_get_num_active_sources (RTPSession * sess)
1113 {
1114   guint result;
1115
1116   g_return_val_if_fail (RTP_IS_SESSION (sess), 0);
1117
1118   RTP_SESSION_LOCK (sess);
1119   result = sess->stats.active_sources;
1120   RTP_SESSION_UNLOCK (sess);
1121
1122   return result;
1123 }
1124
1125 /**
1126  * rtp_session_get_source_by_ssrc:
1127  * @sess: an #RTPSession
1128  * @ssrc: an SSRC
1129  *
1130  * Find the source with @ssrc in @sess.
1131  *
1132  * Returns: a #RTPSource with SSRC @ssrc or NULL if the source was not found.
1133  * g_object_unref() after usage.
1134  */
1135 RTPSource *
1136 rtp_session_get_source_by_ssrc (RTPSession * sess, guint32 ssrc)
1137 {
1138   RTPSource *result;
1139
1140   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1141
1142   RTP_SESSION_LOCK (sess);
1143   result =
1144       g_hash_table_lookup (sess->ssrcs[sess->mask_idx], GINT_TO_POINTER (ssrc));
1145   if (result)
1146     g_object_ref (result);
1147   RTP_SESSION_UNLOCK (sess);
1148
1149   return result;
1150 }
1151
1152 /**
1153  * rtp_session_get_source_by_cname:
1154  * @sess: a #RTPSession
1155  * @cname: an CNAME
1156  *
1157  * Find the source with @cname in @sess.
1158  *
1159  * Returns: a #RTPSource with CNAME @cname or NULL if the source was not found.
1160  * g_object_unref() after usage.
1161  */
1162 RTPSource *
1163 rtp_session_get_source_by_cname (RTPSession * sess, const gchar * cname)
1164 {
1165   RTPSource *result;
1166
1167   g_return_val_if_fail (RTP_IS_SESSION (sess), NULL);
1168   g_return_val_if_fail (cname != NULL, NULL);
1169
1170   RTP_SESSION_LOCK (sess);
1171   result = g_hash_table_lookup (sess->cnames, cname);
1172   if (result)
1173     g_object_ref (result);
1174   RTP_SESSION_UNLOCK (sess);
1175
1176   return result;
1177 }
1178
1179 static guint32
1180 rtp_session_create_new_ssrc (RTPSession * sess)
1181 {
1182   guint32 ssrc;
1183
1184   while (TRUE) {
1185     ssrc = g_random_int ();
1186
1187     /* see if it exists in the session, we're done if it doesn't */
1188     if (g_hash_table_lookup (sess->ssrcs[sess->mask_idx],
1189             GINT_TO_POINTER (ssrc)) == NULL)
1190       break;
1191   }
1192
1193   return ssrc;
1194 }
1195
1196
1197 /**
1198  * rtp_session_create_source:
1199  * @sess: an #RTPSession
1200  *
1201  * Create an #RTPSource for use in @sess. This function will create a source
1202  * with an ssrc that is currently not used by any participants in the session.
1203  *
1204  * Returns: an #RTPSource.
1205  */
1206 RTPSource *
1207 rtp_session_create_source (RTPSession * sess)
1208 {
1209   guint32 ssrc;
1210   RTPSource *source;
1211
1212   RTP_SESSION_LOCK (sess);
1213   ssrc = rtp_session_create_new_ssrc (sess);
1214   source = rtp_source_new (ssrc);
1215   g_object_ref (source);
1216   rtp_source_set_callbacks (source, &callbacks, sess);
1217   g_hash_table_insert (sess->ssrcs[sess->mask_idx], GINT_TO_POINTER (ssrc),
1218       source);
1219   /* we have one more source now */
1220   sess->total_sources++;
1221   RTP_SESSION_UNLOCK (sess);
1222
1223   return source;
1224 }
1225
1226 /* update the RTPArrivalStats structure with the current time and other bits
1227  * about the current buffer we are handling.
1228  * This function is typically called when a validated packet is received.
1229  * This function should be called with the SESSION_LOCK
1230  */
1231 static void
1232 update_arrival_stats (RTPSession * sess, RTPArrivalStats * arrival,
1233     gboolean rtp, GstBuffer * buffer, guint64 ntpnstime)
1234 {
1235   GTimeVal current;
1236
1237   /* get time of arrival */
1238   g_get_current_time (&current);
1239   arrival->time = GST_TIMEVAL_TO_TIME (current);
1240   arrival->timestamp = GST_BUFFER_TIMESTAMP (buffer);
1241   arrival->ntpnstime = ntpnstime;
1242
1243   /* get packet size including header overhead */
1244   arrival->bytes = GST_BUFFER_SIZE (buffer) + sess->header_len;
1245
1246   if (rtp) {
1247     arrival->payload_len = gst_rtp_buffer_get_payload_len (buffer);
1248   } else {
1249     arrival->payload_len = 0;
1250   }
1251
1252   /* for netbuffer we can store the IP address to check for collisions */
1253   arrival->have_address = GST_IS_NETBUFFER (buffer);
1254   if (arrival->have_address) {
1255     GstNetBuffer *netbuf = (GstNetBuffer *) buffer;
1256
1257     memcpy (&arrival->address, &netbuf->from, sizeof (GstNetAddress));
1258   }
1259 }
1260
1261 /**
1262  * rtp_session_process_rtp:
1263  * @sess: and #RTPSession
1264  * @buffer: an RTP buffer
1265  * @ntpnstime: the NTP arrival time in nanoseconds
1266  *
1267  * Process an RTP buffer in the session manager. This function takes ownership
1268  * of @buffer.
1269  *
1270  * Returns: a #GstFlowReturn.
1271  */
1272 GstFlowReturn
1273 rtp_session_process_rtp (RTPSession * sess, GstBuffer * buffer,
1274     guint64 ntpnstime)
1275 {
1276   GstFlowReturn result;
1277   guint32 ssrc;
1278   RTPSource *source;
1279   gboolean created;
1280   gboolean prevsender, prevactive;
1281   RTPArrivalStats arrival;
1282
1283   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1284   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1285
1286   if (!gst_rtp_buffer_validate (buffer))
1287     goto invalid_packet;
1288
1289   RTP_SESSION_LOCK (sess);
1290   /* update arrival stats */
1291   update_arrival_stats (sess, &arrival, TRUE, buffer, ntpnstime);
1292
1293   /* ignore more RTP packets when we left the session */
1294   if (sess->source->received_bye)
1295     goto ignore;
1296
1297   /* get SSRC and look up in session database */
1298   ssrc = gst_rtp_buffer_get_ssrc (buffer);
1299   source = obtain_source (sess, ssrc, &created, &arrival, TRUE);
1300
1301   if (!source)
1302     goto collision;
1303
1304   prevsender = RTP_SOURCE_IS_SENDER (source);
1305   prevactive = RTP_SOURCE_IS_ACTIVE (source);
1306
1307   /* we need to ref so that we can process the CSRCs later */
1308   gst_buffer_ref (buffer);
1309
1310   /* let source process the packet */
1311   result = rtp_source_process_rtp (source, buffer, &arrival);
1312
1313   /* source became active */
1314   if (prevactive != RTP_SOURCE_IS_ACTIVE (source)) {
1315     sess->stats.active_sources++;
1316     GST_DEBUG ("source: %08x became active, %d active sources", ssrc,
1317         sess->stats.active_sources);
1318     on_ssrc_validated (sess, source);
1319   }
1320   if (prevsender != RTP_SOURCE_IS_SENDER (source)) {
1321     sess->stats.sender_sources++;
1322     GST_DEBUG ("source: %08x became sender, %d sender sources", ssrc,
1323         sess->stats.sender_sources);
1324   }
1325
1326   if (created)
1327     on_new_ssrc (sess, source);
1328
1329   if (source->validated) {
1330     guint8 i, count;
1331     gboolean created;
1332
1333     /* for validated sources, we add the CSRCs as well */
1334     count = gst_rtp_buffer_get_csrc_count (buffer);
1335
1336     for (i = 0; i < count; i++) {
1337       guint32 csrc;
1338       RTPSource *csrc_src;
1339
1340       csrc = gst_rtp_buffer_get_csrc (buffer, i);
1341
1342       /* get source */
1343       csrc_src = obtain_source (sess, csrc, &created, &arrival, TRUE);
1344
1345       if (created) {
1346         GST_DEBUG ("created new CSRC: %08x", csrc);
1347         rtp_source_set_as_csrc (csrc_src);
1348         if (RTP_SOURCE_IS_ACTIVE (csrc_src))
1349           sess->stats.active_sources++;
1350         on_new_ssrc (sess, source);
1351       }
1352     }
1353   }
1354   gst_buffer_unref (buffer);
1355
1356   RTP_SESSION_UNLOCK (sess);
1357
1358   return result;
1359
1360   /* ERRORS */
1361 invalid_packet:
1362   {
1363     gst_buffer_unref (buffer);
1364     GST_DEBUG ("invalid RTP packet received");
1365     return GST_FLOW_OK;
1366   }
1367 ignore:
1368   {
1369     gst_buffer_unref (buffer);
1370     RTP_SESSION_UNLOCK (sess);
1371     GST_DEBUG ("ignoring RTP packet because we are leaving");
1372     return GST_FLOW_OK;
1373   }
1374 collision:
1375   {
1376     gst_buffer_unref (buffer);
1377     RTP_SESSION_UNLOCK (sess);
1378     GST_DEBUG ("ignoring packet because its collisioning");
1379     return GST_FLOW_OK;
1380   }
1381 }
1382
1383 static void
1384 rtp_session_process_rb (RTPSession * sess, RTPSource * source,
1385     GstRTCPPacket * packet, RTPArrivalStats * arrival)
1386 {
1387   guint count, i;
1388
1389   count = gst_rtcp_packet_get_rb_count (packet);
1390   for (i = 0; i < count; i++) {
1391     guint32 ssrc, exthighestseq, jitter, lsr, dlsr;
1392     guint8 fractionlost;
1393     gint32 packetslost;
1394
1395     gst_rtcp_packet_get_rb (packet, i, &ssrc, &fractionlost,
1396         &packetslost, &exthighestseq, &jitter, &lsr, &dlsr);
1397
1398     GST_DEBUG ("RB %d: SSRC %08x, jitter %" G_GUINT32_FORMAT, i, ssrc, jitter);
1399
1400     if (ssrc == sess->source->ssrc) {
1401       /* only deal with report blocks for our session, we update the stats of
1402        * the sender of the RTCP message. We could also compare our stats against
1403        * the other sender to see if we are better or worse. */
1404       rtp_source_process_rb (source, arrival->time, fractionlost, packetslost,
1405           exthighestseq, jitter, lsr, dlsr);
1406
1407       on_ssrc_active (sess, source);
1408     }
1409   }
1410 }
1411
1412 /* A Sender report contains statistics about how the sender is doing. This
1413  * includes timing informataion such as the relation between RTP and NTP
1414  * timestamps and the number of packets/bytes it sent to us.
1415  *
1416  * In this report is also included a set of report blocks related to how this
1417  * sender is receiving data (in case we (or somebody else) is also sending stuff
1418  * to it). This info includes the packet loss, jitter and seqnum. It also
1419  * contains information to calculate the round trip time (LSR/DLSR).
1420  */
1421 static void
1422 rtp_session_process_sr (RTPSession * sess, GstRTCPPacket * packet,
1423     RTPArrivalStats * arrival)
1424 {
1425   guint32 senderssrc, rtptime, packet_count, octet_count;
1426   guint64 ntptime;
1427   RTPSource *source;
1428   gboolean created, prevsender;
1429
1430   gst_rtcp_packet_sr_get_sender_info (packet, &senderssrc, &ntptime, &rtptime,
1431       &packet_count, &octet_count);
1432
1433   GST_DEBUG ("got SR packet: SSRC %08x, time %" GST_TIME_FORMAT,
1434       senderssrc, GST_TIME_ARGS (arrival->time));
1435
1436   source = obtain_source (sess, senderssrc, &created, arrival, FALSE);
1437
1438   if (!source)
1439     return;
1440
1441   GST_BUFFER_OFFSET (packet->buffer) = source->clock_base;
1442   GST_BUFFER_OFFSET_END (packet->buffer) = source->clock_base_time;
1443
1444   prevsender = RTP_SOURCE_IS_SENDER (source);
1445
1446   /* first update the source */
1447   rtp_source_process_sr (source, arrival->time, ntptime, rtptime, packet_count,
1448       octet_count);
1449
1450   if (prevsender != RTP_SOURCE_IS_SENDER (source)) {
1451     sess->stats.sender_sources++;
1452     GST_DEBUG ("source: %08x became sender, %d sender sources", senderssrc,
1453         sess->stats.sender_sources);
1454   }
1455
1456   if (created)
1457     on_new_ssrc (sess, source);
1458
1459   rtp_session_process_rb (sess, source, packet, arrival);
1460 }
1461
1462 /* A receiver report contains statistics about how a receiver is doing. It
1463  * includes stuff like packet loss, jitter and the seqnum it received last. It
1464  * also contains info to calculate the round trip time.
1465  *
1466  * We are only interested in how the sender of this report is doing wrt to us.
1467  */
1468 static void
1469 rtp_session_process_rr (RTPSession * sess, GstRTCPPacket * packet,
1470     RTPArrivalStats * arrival)
1471 {
1472   guint32 senderssrc;
1473   RTPSource *source;
1474   gboolean created;
1475
1476   senderssrc = gst_rtcp_packet_rr_get_ssrc (packet);
1477
1478   GST_DEBUG ("got RR packet: SSRC %08x", senderssrc);
1479
1480   source = obtain_source (sess, senderssrc, &created, arrival, FALSE);
1481
1482   if (!source)
1483     return;
1484
1485   if (created)
1486     on_new_ssrc (sess, source);
1487
1488   rtp_session_process_rb (sess, source, packet, arrival);
1489 }
1490
1491 /* Get SDES items and store them in the SSRC */
1492 static void
1493 rtp_session_process_sdes (RTPSession * sess, GstRTCPPacket * packet,
1494     RTPArrivalStats * arrival)
1495 {
1496   guint items, i, j;
1497   gboolean more_items, more_entries;
1498
1499   items = gst_rtcp_packet_sdes_get_item_count (packet);
1500   GST_DEBUG ("got SDES packet with %d items", items);
1501
1502   more_items = gst_rtcp_packet_sdes_first_item (packet);
1503   i = 0;
1504   while (more_items) {
1505     guint32 ssrc;
1506     gboolean changed, created;
1507     RTPSource *source;
1508
1509     ssrc = gst_rtcp_packet_sdes_get_ssrc (packet);
1510
1511     GST_DEBUG ("item %d, SSRC %08x", i, ssrc);
1512
1513     /* find src, no probation when dealing with RTCP */
1514     source = obtain_source (sess, ssrc, &created, arrival, FALSE);
1515     changed = FALSE;
1516
1517     if (!source)
1518       return;
1519
1520     more_entries = gst_rtcp_packet_sdes_first_entry (packet);
1521     j = 0;
1522     while (more_entries) {
1523       GstRTCPSDESType type;
1524       guint8 len;
1525       guint8 *data;
1526
1527       gst_rtcp_packet_sdes_get_entry (packet, &type, &len, &data);
1528
1529       GST_DEBUG ("entry %d, type %d, len %d, data %.*s", j, type, len, len,
1530           data);
1531
1532       changed |= rtp_source_set_sdes (source, type, data, len);
1533
1534       more_entries = gst_rtcp_packet_sdes_next_entry (packet);
1535       j++;
1536     }
1537
1538     if (created)
1539       on_new_ssrc (sess, source);
1540     if (changed)
1541       on_ssrc_sdes (sess, source);
1542
1543     more_items = gst_rtcp_packet_sdes_next_item (packet);
1544     i++;
1545   }
1546 }
1547
1548 /* BYE is sent when a client leaves the session
1549  */
1550 static void
1551 rtp_session_process_bye (RTPSession * sess, GstRTCPPacket * packet,
1552     RTPArrivalStats * arrival)
1553 {
1554   guint count, i;
1555   gchar *reason;
1556
1557   reason = gst_rtcp_packet_bye_get_reason (packet);
1558   GST_DEBUG ("got BYE packet (reason: %s)", GST_STR_NULL (reason));
1559
1560   count = gst_rtcp_packet_bye_get_ssrc_count (packet);
1561   for (i = 0; i < count; i++) {
1562     guint32 ssrc;
1563     RTPSource *source;
1564     gboolean created, prevactive, prevsender;
1565     guint pmembers, members;
1566
1567     ssrc = gst_rtcp_packet_bye_get_nth_ssrc (packet, i);
1568     GST_DEBUG ("SSRC: %08x", ssrc);
1569
1570     /* find src and mark bye, no probation when dealing with RTCP */
1571     source = obtain_source (sess, ssrc, &created, arrival, FALSE);
1572
1573     if (!source)
1574       return;
1575
1576     /* store time for when we need to time out this source */
1577     source->bye_time = arrival->time;
1578
1579     prevactive = RTP_SOURCE_IS_ACTIVE (source);
1580     prevsender = RTP_SOURCE_IS_SENDER (source);
1581
1582     /* let the source handle the rest */
1583     rtp_source_process_bye (source, reason);
1584
1585     pmembers = sess->stats.active_sources;
1586
1587     if (prevactive && !RTP_SOURCE_IS_ACTIVE (source)) {
1588       sess->stats.active_sources--;
1589       GST_DEBUG ("source: %08x became inactive, %d active sources", ssrc,
1590           sess->stats.active_sources);
1591     }
1592     if (prevsender && !RTP_SOURCE_IS_SENDER (source)) {
1593       sess->stats.sender_sources--;
1594       GST_DEBUG ("source: %08x became non sender, %d sender sources", ssrc,
1595           sess->stats.sender_sources);
1596     }
1597     members = sess->stats.active_sources;
1598
1599     if (!sess->source->received_bye && members < pmembers) {
1600       /* some members went away since the previous timeout estimate.
1601        * Perform reverse reconsideration but only when we are not scheduling a
1602        * BYE ourselves. */
1603       if (arrival->time < sess->next_rtcp_check_time) {
1604         GstClockTime time_remaining;
1605
1606         time_remaining = sess->next_rtcp_check_time - arrival->time;
1607         sess->next_rtcp_check_time =
1608             gst_util_uint64_scale (time_remaining, members, pmembers);
1609
1610         GST_DEBUG ("reverse reconsideration %" GST_TIME_FORMAT,
1611             GST_TIME_ARGS (sess->next_rtcp_check_time));
1612
1613         sess->next_rtcp_check_time += arrival->time;
1614
1615         RTP_SESSION_UNLOCK (sess);
1616         /* notify app of reconsideration */
1617         if (sess->callbacks.reconsider)
1618           sess->callbacks.reconsider (sess, sess->reconsider_user_data);
1619         RTP_SESSION_LOCK (sess);
1620       }
1621     }
1622
1623     if (created)
1624       on_new_ssrc (sess, source);
1625
1626     on_bye_ssrc (sess, source);
1627   }
1628   g_free (reason);
1629 }
1630
1631 static void
1632 rtp_session_process_app (RTPSession * sess, GstRTCPPacket * packet,
1633     RTPArrivalStats * arrival)
1634 {
1635   GST_DEBUG ("received APP");
1636 }
1637
1638 /**
1639  * rtp_session_process_rtcp:
1640  * @sess: and #RTPSession
1641  * @buffer: an RTCP buffer
1642  *
1643  * Process an RTCP buffer in the session manager. This function takes ownership
1644  * of @buffer.
1645  *
1646  * Returns: a #GstFlowReturn.
1647  */
1648 GstFlowReturn
1649 rtp_session_process_rtcp (RTPSession * sess, GstBuffer * buffer)
1650 {
1651   GstRTCPPacket packet;
1652   gboolean more, is_bye = FALSE, is_sr = FALSE;
1653   RTPArrivalStats arrival;
1654   GstFlowReturn result = GST_FLOW_OK;
1655
1656   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1657   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1658
1659   if (!gst_rtcp_buffer_validate (buffer))
1660     goto invalid_packet;
1661
1662   GST_DEBUG ("received RTCP packet");
1663
1664   RTP_SESSION_LOCK (sess);
1665   /* update arrival stats */
1666   update_arrival_stats (sess, &arrival, FALSE, buffer, -1);
1667
1668   if (sess->sent_bye)
1669     goto ignore;
1670
1671   /* start processing the compound packet */
1672   more = gst_rtcp_buffer_get_first_packet (buffer, &packet);
1673   while (more) {
1674     GstRTCPType type;
1675
1676     type = gst_rtcp_packet_get_type (&packet);
1677
1678     /* when we are leaving the session, we should ignore all non-BYE messages */
1679     if (sess->source->received_bye && type != GST_RTCP_TYPE_BYE) {
1680       GST_DEBUG ("ignoring non-BYE RTCP packet because we are leaving");
1681       goto next;
1682     }
1683
1684     switch (type) {
1685       case GST_RTCP_TYPE_SR:
1686         rtp_session_process_sr (sess, &packet, &arrival);
1687         is_sr = TRUE;
1688         break;
1689       case GST_RTCP_TYPE_RR:
1690         rtp_session_process_rr (sess, &packet, &arrival);
1691         break;
1692       case GST_RTCP_TYPE_SDES:
1693         rtp_session_process_sdes (sess, &packet, &arrival);
1694         break;
1695       case GST_RTCP_TYPE_BYE:
1696         is_bye = TRUE;
1697         rtp_session_process_bye (sess, &packet, &arrival);
1698         break;
1699       case GST_RTCP_TYPE_APP:
1700         rtp_session_process_app (sess, &packet, &arrival);
1701         break;
1702       default:
1703         GST_WARNING ("got unknown RTCP packet");
1704         break;
1705     }
1706   next:
1707     more = gst_rtcp_packet_move_to_next (&packet);
1708   }
1709
1710   /* if we are scheduling a BYE, we only want to count bye packets, else we
1711    * count everything */
1712   if (sess->source->received_bye) {
1713     if (is_bye) {
1714       sess->stats.bye_members++;
1715       UPDATE_AVG (sess->stats.avg_rtcp_packet_size, arrival.bytes);
1716     }
1717   } else {
1718     /* keep track of average packet size */
1719     UPDATE_AVG (sess->stats.avg_rtcp_packet_size, arrival.bytes);
1720   }
1721   RTP_SESSION_UNLOCK (sess);
1722
1723   /* notify caller of sr packets in the callback */
1724   if (is_sr && sess->callbacks.sync_rtcp)
1725     result = sess->callbacks.sync_rtcp (sess, sess->source, buffer,
1726         sess->sync_rtcp_user_data);
1727   else
1728     gst_buffer_unref (buffer);
1729
1730   return result;
1731
1732   /* ERRORS */
1733 invalid_packet:
1734   {
1735     GST_DEBUG ("invalid RTCP packet received");
1736     gst_buffer_unref (buffer);
1737     return GST_FLOW_OK;
1738   }
1739 ignore:
1740   {
1741     gst_buffer_unref (buffer);
1742     RTP_SESSION_UNLOCK (sess);
1743     GST_DEBUG ("ignoring RTP packet because we left");
1744     return GST_FLOW_OK;
1745   }
1746 }
1747
1748 /**
1749  * rtp_session_send_rtp:
1750  * @sess: an #RTPSession
1751  * @buffer: an RTP buffer
1752  * @ntpnstime: the NTP time in nanoseconds of when this buffer was captured.
1753  *
1754  * Send the RTP buffer in the session manager. This function takes ownership of
1755  * @buffer.
1756  *
1757  * Returns: a #GstFlowReturn.
1758  */
1759 GstFlowReturn
1760 rtp_session_send_rtp (RTPSession * sess, GstBuffer * buffer, guint64 ntpnstime)
1761 {
1762   GstFlowReturn result;
1763   RTPSource *source;
1764   gboolean prevsender;
1765   GTimeVal current;
1766
1767   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1768   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1769
1770   if (!gst_rtp_buffer_validate (buffer))
1771     goto invalid_packet;
1772
1773   GST_DEBUG ("received RTP packet for sending");
1774
1775   RTP_SESSION_LOCK (sess);
1776   source = sess->source;
1777
1778   /* update last activity */
1779   g_get_current_time (&current);
1780   source->last_rtp_activity = GST_TIMEVAL_TO_TIME (current);
1781
1782   prevsender = RTP_SOURCE_IS_SENDER (source);
1783
1784   /* we use our own source to send */
1785   result = rtp_source_send_rtp (source, buffer, ntpnstime);
1786
1787   if (RTP_SOURCE_IS_SENDER (source) && !prevsender)
1788     sess->stats.sender_sources++;
1789   RTP_SESSION_UNLOCK (sess);
1790
1791   return result;
1792
1793   /* ERRORS */
1794 invalid_packet:
1795   {
1796     gst_buffer_unref (buffer);
1797     GST_DEBUG ("invalid RTP packet received");
1798     return GST_FLOW_OK;
1799   }
1800 }
1801
1802 static GstClockTime
1803 calculate_rtcp_interval (RTPSession * sess, gboolean deterministic,
1804     gboolean first)
1805 {
1806   GstClockTime result;
1807
1808   if (sess->source->received_bye) {
1809     result = rtp_stats_calculate_bye_interval (&sess->stats);
1810   } else {
1811     result = rtp_stats_calculate_rtcp_interval (&sess->stats,
1812         RTP_SOURCE_IS_SENDER (sess->source), first);
1813   }
1814
1815   GST_DEBUG ("next deterministic interval: %" GST_TIME_FORMAT ", first %d",
1816       GST_TIME_ARGS (result), first);
1817
1818   if (!deterministic)
1819     result = rtp_stats_add_rtcp_jitter (&sess->stats, result);
1820
1821   GST_DEBUG ("next interval: %" GST_TIME_FORMAT, GST_TIME_ARGS (result));
1822
1823   return result;
1824 }
1825
1826 /**
1827  * rtp_session_send_bye_locked:
1828  * @sess: an #RTPSession
1829  * @reason: a reason or NULL
1830  *
1831  * Stop the current @sess and schedule a BYE message for the other members.
1832  *
1833  * One must have the session lock to call this function
1834  *
1835  * Returns: a #GstFlowReturn.
1836  */
1837 static GstFlowReturn
1838 rtp_session_send_bye_locked (RTPSession * sess, const gchar * reason)
1839 {
1840   GstFlowReturn result = GST_FLOW_OK;
1841   RTPSource *source;
1842   GstClockTime current, interval;
1843   GTimeVal curtv;
1844
1845   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1846
1847   source = sess->source;
1848
1849   /* ignore more BYEs */
1850   if (source->received_bye)
1851     goto done;
1852
1853   /* we have BYE now */
1854   source->received_bye = TRUE;
1855   /* at least one member wants to send a BYE */
1856   g_free (sess->bye_reason);
1857   sess->bye_reason = g_strdup (reason);
1858   sess->stats.avg_rtcp_packet_size = 100;
1859   sess->stats.bye_members = 1;
1860   sess->first_rtcp = TRUE;
1861   sess->sent_bye = FALSE;
1862
1863   /* get current time */
1864   g_get_current_time (&curtv);
1865   current = GST_TIMEVAL_TO_TIME (curtv);
1866
1867   /* reschedule transmission */
1868   sess->last_rtcp_send_time = current;
1869   interval = calculate_rtcp_interval (sess, FALSE, TRUE);
1870   sess->next_rtcp_check_time = current + interval;
1871
1872   GST_DEBUG ("Schedule BYE for %" GST_TIME_FORMAT ", %" GST_TIME_FORMAT,
1873       GST_TIME_ARGS (interval), GST_TIME_ARGS (sess->next_rtcp_check_time));
1874
1875   RTP_SESSION_UNLOCK (sess);
1876   /* notify app of reconsideration */
1877   if (sess->callbacks.reconsider)
1878     sess->callbacks.reconsider (sess, sess->reconsider_user_data);
1879   RTP_SESSION_LOCK (sess);
1880 done:
1881
1882   return result;
1883 }
1884
1885 /**
1886  * rtp_session_send_bye:
1887  * @sess: an #RTPSession
1888  * @reason: a reason or NULL
1889  *
1890  * Stop the current @sess and schedule a BYE message for the other members.
1891  *
1892  * One must have the session lock to call this function
1893  *
1894  * Returns: a #GstFlowReturn.
1895  */
1896 GstFlowReturn
1897 rtp_session_send_bye (RTPSession * sess, const gchar * reason)
1898 {
1899   GstFlowReturn result = GST_FLOW_OK;
1900
1901   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1902
1903   RTP_SESSION_LOCK (sess);
1904   result = rtp_session_send_bye_locked (sess, reason);
1905   RTP_SESSION_UNLOCK (sess);
1906
1907   return result;
1908 }
1909
1910 /**
1911  * rtp_session_next_timeout:
1912  * @sess: an #RTPSession
1913  * @time: the current system time
1914  *
1915  * Get the next time we should perform session maintenance tasks.
1916  *
1917  * Returns: a time when rtp_session_on_timeout() should be called with the
1918  * current system time.
1919  */
1920 GstClockTime
1921 rtp_session_next_timeout (RTPSession * sess, GstClockTime time)
1922 {
1923   GstClockTime result;
1924
1925   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1926
1927   RTP_SESSION_LOCK (sess);
1928
1929   result = sess->next_rtcp_check_time;
1930
1931   GST_DEBUG ("current time: %" GST_TIME_FORMAT ", next :%" GST_TIME_FORMAT,
1932       GST_TIME_ARGS (time), GST_TIME_ARGS (result));
1933
1934   if (result < time) {
1935     GST_DEBUG ("take current time as base");
1936     /* our previous check time expired, start counting from the current time
1937      * again. */
1938     result = time;
1939   }
1940
1941   if (sess->source->received_bye) {
1942     if (sess->sent_bye) {
1943       GST_DEBUG ("we sent BYE already");
1944       result = GST_CLOCK_TIME_NONE;
1945     } else if (sess->stats.active_sources >= 50) {
1946       GST_DEBUG ("reconsider BYE, more than 50 sources");
1947       /* reconsider BYE if members >= 50 */
1948       result += calculate_rtcp_interval (sess, FALSE, TRUE);
1949     }
1950   } else {
1951     if (sess->first_rtcp) {
1952       GST_DEBUG ("first RTCP packet");
1953       /* we are called for the first time */
1954       result += calculate_rtcp_interval (sess, FALSE, TRUE);
1955     } else if (sess->next_rtcp_check_time < time) {
1956       GST_DEBUG ("old check time expired, getting new timeout");
1957       /* get a new timeout when we need to */
1958       result += calculate_rtcp_interval (sess, FALSE, FALSE);
1959     }
1960   }
1961   sess->next_rtcp_check_time = result;
1962
1963   GST_DEBUG ("next timeout: %" GST_TIME_FORMAT, GST_TIME_ARGS (result));
1964   RTP_SESSION_UNLOCK (sess);
1965
1966   return result;
1967 }
1968
1969 typedef struct
1970 {
1971   RTPSession *sess;
1972   GstBuffer *rtcp;
1973   GstClockTime time;
1974   guint64 ntpnstime;
1975   GstClockTime interval;
1976   GstRTCPPacket packet;
1977   gboolean is_bye;
1978   gboolean has_sdes;
1979 } ReportData;
1980
1981 static void
1982 session_start_rtcp (RTPSession * sess, ReportData * data)
1983 {
1984   GstRTCPPacket *packet = &data->packet;
1985   RTPSource *own = sess->source;
1986
1987   data->rtcp = gst_rtcp_buffer_new (sess->mtu);
1988
1989   if (RTP_SOURCE_IS_SENDER (own)) {
1990     guint64 ntptime;
1991     guint32 rtptime;
1992     guint32 packet_count, octet_count;
1993
1994     /* we are a sender, create SR */
1995     GST_DEBUG ("create SR for SSRC %08x", own->ssrc);
1996     gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_SR, packet);
1997
1998     /* get latest stats */
1999     rtp_source_get_new_sr (own, data->ntpnstime, &ntptime, &rtptime,
2000         &packet_count, &octet_count);
2001     /* store stats */
2002     rtp_source_process_sr (own, data->ntpnstime, ntptime, rtptime, packet_count,
2003         octet_count);
2004
2005     /* fill in sender report info */
2006     gst_rtcp_packet_sr_set_sender_info (packet, own->ssrc,
2007         ntptime, rtptime, packet_count, octet_count);
2008   } else {
2009     /* we are only receiver, create RR */
2010     GST_DEBUG ("create RR for SSRC %08x", own->ssrc);
2011     gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_RR, packet);
2012     gst_rtcp_packet_rr_set_ssrc (packet, own->ssrc);
2013   }
2014 }
2015
2016 /* construct a Sender or Receiver Report */
2017 static void
2018 session_report_blocks (const gchar * key, RTPSource * source, ReportData * data)
2019 {
2020   RTPSession *sess = data->sess;
2021   GstRTCPPacket *packet = &data->packet;
2022
2023   /* create a new buffer if needed */
2024   if (data->rtcp == NULL) {
2025     session_start_rtcp (sess, data);
2026   }
2027   if (gst_rtcp_packet_get_rb_count (packet) < GST_RTCP_MAX_RB_COUNT) {
2028     /* only report about other sender sources */
2029     if (source != sess->source && RTP_SOURCE_IS_SENDER (source)) {
2030       guint8 fractionlost;
2031       gint32 packetslost;
2032       guint32 exthighestseq, jitter;
2033       guint32 lsr, dlsr;
2034
2035       /* get new stats */
2036       rtp_source_get_new_rb (source, data->time, &fractionlost, &packetslost,
2037           &exthighestseq, &jitter, &lsr, &dlsr);
2038
2039       /* packet is not yet filled, add report block for this source. */
2040       gst_rtcp_packet_add_rb (packet, source->ssrc, fractionlost, packetslost,
2041           exthighestseq, jitter, lsr, dlsr);
2042     }
2043   }
2044 }
2045
2046 /* perform cleanup of sources that timed out */
2047 static gboolean
2048 session_cleanup (const gchar * key, RTPSource * source, ReportData * data)
2049 {
2050   gboolean remove = FALSE;
2051   gboolean byetimeout = FALSE;
2052   gboolean is_sender, is_active;
2053   RTPSession *sess = data->sess;
2054   GstClockTime interval;
2055
2056   is_sender = RTP_SOURCE_IS_SENDER (source);
2057   is_active = RTP_SOURCE_IS_ACTIVE (source);
2058
2059   /* check for our own source, we don't want to delete our own source. */
2060   if (!(source == sess->source)) {
2061     if (source->received_bye) {
2062       /* if we received a BYE from the source, remove the source after some
2063        * time. */
2064       if (data->time > source->bye_time &&
2065           data->time - source->bye_time > sess->stats.bye_timeout) {
2066         GST_DEBUG ("removing BYE source %08x", source->ssrc);
2067         remove = TRUE;
2068         byetimeout = TRUE;
2069       }
2070     }
2071     /* sources that were inactive for more than 5 times the deterministic reporting
2072      * interval get timed out. the min timeout is 5 seconds. */
2073     if (data->time > source->last_activity) {
2074       interval = MAX (data->interval * 5, 5 * GST_SECOND);
2075       if (data->time - source->last_activity > interval) {
2076         GST_DEBUG ("removing timeout source %08x, last %" GST_TIME_FORMAT,
2077             source->ssrc, GST_TIME_ARGS (source->last_activity));
2078         remove = TRUE;
2079       }
2080     }
2081   }
2082
2083   /* senders that did not send for a long time become a receiver, this also
2084    * holds for our own source. */
2085   if (is_sender) {
2086     if (data->time > source->last_rtp_activity) {
2087       interval = MAX (data->interval * 2, 5 * GST_SECOND);
2088       if (data->time - source->last_rtp_activity > interval) {
2089         GST_DEBUG ("sender source %08x timed out and became receiver, last %"
2090             GST_TIME_FORMAT, source->ssrc,
2091             GST_TIME_ARGS (source->last_rtp_activity));
2092         source->is_sender = FALSE;
2093         sess->stats.sender_sources--;
2094       }
2095     }
2096   }
2097
2098   if (remove) {
2099     sess->total_sources--;
2100     if (is_sender)
2101       sess->stats.sender_sources--;
2102     if (is_active)
2103       sess->stats.active_sources--;
2104
2105     if (byetimeout)
2106       on_bye_timeout (sess, source);
2107     else
2108       on_timeout (sess, source);
2109   }
2110   return remove;
2111 }
2112
2113 static void
2114 session_sdes (RTPSession * sess, ReportData * data)
2115 {
2116   GstRTCPPacket *packet = &data->packet;
2117   guint8 *sdes_data;
2118   guint sdes_len;
2119
2120   /* add SDES packet */
2121   gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_SDES, packet);
2122
2123   gst_rtcp_packet_sdes_add_item (packet, sess->source->ssrc);
2124
2125   rtp_source_get_sdes (sess->source, GST_RTCP_SDES_CNAME, &sdes_data,
2126       &sdes_len);
2127   gst_rtcp_packet_sdes_add_entry (packet, GST_RTCP_SDES_CNAME, sdes_len,
2128       sdes_data);
2129
2130   /* other SDES items must only be added at regular intervals and only when the
2131    * user requests to since it might be a privacy problem */
2132 #if 0
2133   gst_rtcp_packet_sdes_add_entry (&packet, GST_RTCP_SDES_NAME,
2134       strlen (sess->name), (guint8 *) sess->name);
2135   gst_rtcp_packet_sdes_add_entry (&packet, GST_RTCP_SDES_TOOL,
2136       strlen (sess->tool), (guint8 *) sess->tool);
2137 #endif
2138
2139   data->has_sdes = TRUE;
2140 }
2141
2142 /* schedule a BYE packet */
2143 static void
2144 session_bye (RTPSession * sess, ReportData * data)
2145 {
2146   GstRTCPPacket *packet = &data->packet;
2147
2148   /* open packet */
2149   session_start_rtcp (sess, data);
2150
2151   /* add SDES */
2152   session_sdes (sess, data);
2153
2154   /* add a BYE packet */
2155   gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_BYE, packet);
2156   gst_rtcp_packet_bye_add_ssrc (packet, sess->source->ssrc);
2157   if (sess->bye_reason)
2158     gst_rtcp_packet_bye_set_reason (packet, sess->bye_reason);
2159
2160   /* we have a BYE packet now */
2161   data->is_bye = TRUE;
2162 }
2163
2164 static gboolean
2165 is_rtcp_time (RTPSession * sess, GstClockTime time, ReportData * data)
2166 {
2167   GstClockTime new_send_time, elapsed;
2168   gboolean result;
2169
2170   /* no need to check yet */
2171   if (sess->next_rtcp_check_time > time) {
2172     GST_DEBUG ("no check time yet, next %" GST_TIME_FORMAT " > now %"
2173         GST_TIME_FORMAT, GST_TIME_ARGS (sess->next_rtcp_check_time),
2174         GST_TIME_ARGS (time));
2175     return FALSE;
2176   }
2177
2178   /* get elapsed time since we last reported */
2179   elapsed = time - sess->last_rtcp_send_time;
2180
2181   /* perform forward reconsideration */
2182   new_send_time = rtp_stats_add_rtcp_jitter (&sess->stats, data->interval);
2183
2184   GST_DEBUG ("forward reconsideration %" GST_TIME_FORMAT ", elapsed %"
2185       GST_TIME_FORMAT, GST_TIME_ARGS (new_send_time), GST_TIME_ARGS (elapsed));
2186
2187   new_send_time += sess->last_rtcp_send_time;
2188
2189   /* check if reconsideration */
2190   if (time < new_send_time) {
2191     GST_DEBUG ("reconsider RTCP for %" GST_TIME_FORMAT,
2192         GST_TIME_ARGS (new_send_time));
2193     result = FALSE;
2194     /* store new check time */
2195     sess->next_rtcp_check_time = new_send_time;
2196   } else {
2197     result = TRUE;
2198     new_send_time = calculate_rtcp_interval (sess, FALSE, FALSE);
2199
2200     GST_DEBUG ("can send RTCP now, next interval %" GST_TIME_FORMAT,
2201         GST_TIME_ARGS (new_send_time));
2202     sess->next_rtcp_check_time = time + new_send_time;
2203   }
2204   return result;
2205 }
2206
2207 /**
2208  * rtp_session_on_timeout:
2209  * @sess: an #RTPSession
2210  * @time: the current system time
2211  * @ntpnstime: the current NTP time in nanoseconds
2212  *
2213  * Perform maintenance actions after the timeout obtained with
2214  * rtp_session_next_timeout() expired.
2215  *
2216  * This function will perform timeouts of receivers and senders, send a BYE
2217  * packet or generate RTCP packets with current session stats.
2218  *
2219  * This function can call the #RTPSessionSendRTCP callback, possibly multiple
2220  * times, for each packet that should be processed.
2221  *
2222  * Returns: a #GstFlowReturn.
2223  */
2224 GstFlowReturn
2225 rtp_session_on_timeout (RTPSession * sess, GstClockTime time, guint64 ntpnstime)
2226 {
2227   GstFlowReturn result = GST_FLOW_OK;
2228   GList *item;
2229   ReportData data;
2230
2231   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
2232
2233   GST_DEBUG ("reporting at %" GST_TIME_FORMAT ", NTP time %" GST_TIME_FORMAT,
2234       GST_TIME_ARGS (time), GST_TIME_ARGS (ntpnstime));
2235
2236   data.sess = sess;
2237   data.rtcp = NULL;
2238   data.time = time;
2239   data.ntpnstime = ntpnstime;
2240   data.is_bye = FALSE;
2241   data.has_sdes = FALSE;
2242
2243   RTP_SESSION_LOCK (sess);
2244   /* get a new interval, we need this for various cleanups etc */
2245   data.interval = calculate_rtcp_interval (sess, TRUE, sess->first_rtcp);
2246
2247   /* first perform cleanups */
2248   g_hash_table_foreach_remove (sess->ssrcs[sess->mask_idx],
2249       (GHRFunc) session_cleanup, &data);
2250
2251   /* see if we need to generate SR or RR packets */
2252   if (is_rtcp_time (sess, time, &data)) {
2253     if (sess->source->received_bye) {
2254       /* generate BYE instead */
2255       session_bye (sess, &data);
2256       sess->sent_bye = TRUE;
2257     } else {
2258       /* loop over all known sources and do something */
2259       g_hash_table_foreach (sess->ssrcs[sess->mask_idx],
2260           (GHFunc) session_report_blocks, &data);
2261     }
2262   }
2263
2264   if (data.rtcp) {
2265     guint size;
2266
2267     /* we keep track of the last report time in order to timeout inactive
2268      * receivers or senders */
2269     sess->last_rtcp_send_time = data.time;
2270     sess->first_rtcp = FALSE;
2271
2272     /* add SDES for this source when not already added */
2273     if (!data.has_sdes)
2274       session_sdes (sess, &data);
2275
2276     /* update average RTCP size before sending */
2277     size = GST_BUFFER_SIZE (data.rtcp) + sess->header_len;
2278     UPDATE_AVG (sess->stats.avg_rtcp_packet_size, size);
2279   }
2280
2281   /* check for outdated collisions */
2282   item = g_list_first (sess->conflicting_addresses);
2283   while (item) {
2284     RTPConflictingAddress *known_conflict = item->data;
2285     GList *next_item = g_list_next (item);
2286
2287     if (known_conflict->time < time - (data.interval *
2288             RTCP_INTERVAL_COLLISION_TIMEOUT)) {
2289       sess->conflicting_addresses =
2290           g_list_delete_link (sess->conflicting_addresses, item);
2291       g_free (known_conflict);
2292     }
2293     item = next_item;
2294   }
2295
2296   if (sess->change_ssrc) {
2297     g_hash_table_steal (sess->ssrcs[sess->mask_idx],
2298         GINT_TO_POINTER (sess->source->ssrc));
2299
2300     sess->source->ssrc = rtp_session_create_new_ssrc (sess);
2301     rtp_source_reset (sess->source);
2302
2303     g_hash_table_insert (sess->ssrcs[sess->mask_idx],
2304         GINT_TO_POINTER (sess->source->ssrc), sess->source);
2305
2306     g_free (sess->bye_reason);
2307     sess->bye_reason = NULL;
2308     sess->sent_bye = FALSE;
2309     sess->change_ssrc = FALSE;
2310   }
2311   RTP_SESSION_UNLOCK (sess);
2312
2313   /* push out the RTCP packet */
2314   if (data.rtcp) {
2315     /* close the RTCP packet */
2316     gst_rtcp_buffer_end (data.rtcp);
2317
2318     if (sess->callbacks.send_rtcp)
2319       result = sess->callbacks.send_rtcp (sess, sess->source, data.rtcp,
2320           sess->send_rtcp_user_data);
2321     else
2322       gst_buffer_unref (data.rtcp);
2323   }
2324
2325   return result;
2326 }