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, GstClockTime current_time);
 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", arrival->time);
 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, GstClockTime current_time,
1234     guint64 ntpnstime)
1235 {
1236   /* get time of arrival */
1237   arrival->time = current_time;
1238   arrival->timestamp = GST_BUFFER_TIMESTAMP (buffer);
1239   arrival->ntpnstime = ntpnstime;
1240
1241   /* get packet size including header overhead */
1242   arrival->bytes = GST_BUFFER_SIZE (buffer) + sess->header_len;
1243
1244   if (rtp) {
1245     arrival->payload_len = gst_rtp_buffer_get_payload_len (buffer);
1246   } else {
1247     arrival->payload_len = 0;
1248   }
1249
1250   /* for netbuffer we can store the IP address to check for collisions */
1251   arrival->have_address = GST_IS_NETBUFFER (buffer);
1252   if (arrival->have_address) {
1253     GstNetBuffer *netbuf = (GstNetBuffer *) buffer;
1254
1255     memcpy (&arrival->address, &netbuf->from, sizeof (GstNetAddress));
1256   }
1257 }
1258
1259 /**
1260  * rtp_session_process_rtp:
1261  * @sess: and #RTPSession
1262  * @buffer: an RTP buffer
1263  * @current_time: the current system time
1264  * @ntpnstime: the NTP arrival time in nanoseconds
1265  *
1266  * Process an RTP buffer in the session manager. This function takes ownership
1267  * of @buffer.
1268  *
1269  * Returns: a #GstFlowReturn.
1270  */
1271 GstFlowReturn
1272 rtp_session_process_rtp (RTPSession * sess, GstBuffer * buffer,
1273     GstClockTime current_time, guint64 ntpnstime)
1274 {
1275   GstFlowReturn result;
1276   guint32 ssrc;
1277   RTPSource *source;
1278   gboolean created;
1279   gboolean prevsender, prevactive;
1280   RTPArrivalStats arrival;
1281
1282   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1283   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1284
1285   if (!gst_rtp_buffer_validate (buffer))
1286     goto invalid_packet;
1287
1288   RTP_SESSION_LOCK (sess);
1289   /* update arrival stats */
1290   update_arrival_stats (sess, &arrival, TRUE, buffer, current_time, ntpnstime);
1291
1292   /* ignore more RTP packets when we left the session */
1293   if (sess->source->received_bye)
1294     goto ignore;
1295
1296   /* get SSRC and look up in session database */
1297   ssrc = gst_rtp_buffer_get_ssrc (buffer);
1298   source = obtain_source (sess, ssrc, &created, &arrival, TRUE);
1299
1300   if (!source)
1301     goto collision;
1302
1303   prevsender = RTP_SOURCE_IS_SENDER (source);
1304   prevactive = RTP_SOURCE_IS_ACTIVE (source);
1305
1306   /* we need to ref so that we can process the CSRCs later */
1307   gst_buffer_ref (buffer);
1308
1309   /* let source process the packet */
1310   result = rtp_source_process_rtp (source, buffer, &arrival);
1311
1312   /* source became active */
1313   if (prevactive != RTP_SOURCE_IS_ACTIVE (source)) {
1314     sess->stats.active_sources++;
1315     GST_DEBUG ("source: %08x became active, %d active sources", ssrc,
1316         sess->stats.active_sources);
1317     on_ssrc_validated (sess, source);
1318   }
1319   if (prevsender != RTP_SOURCE_IS_SENDER (source)) {
1320     sess->stats.sender_sources++;
1321     GST_DEBUG ("source: %08x became sender, %d sender sources", ssrc,
1322         sess->stats.sender_sources);
1323   }
1324
1325   if (created)
1326     on_new_ssrc (sess, source);
1327
1328   if (source->validated) {
1329     guint8 i, count;
1330     gboolean created;
1331
1332     /* for validated sources, we add the CSRCs as well */
1333     count = gst_rtp_buffer_get_csrc_count (buffer);
1334
1335     for (i = 0; i < count; i++) {
1336       guint32 csrc;
1337       RTPSource *csrc_src;
1338
1339       csrc = gst_rtp_buffer_get_csrc (buffer, i);
1340
1341       /* get source */
1342       csrc_src = obtain_source (sess, csrc, &created, &arrival, TRUE);
1343
1344       if (created) {
1345         GST_DEBUG ("created new CSRC: %08x", csrc);
1346         rtp_source_set_as_csrc (csrc_src);
1347         if (RTP_SOURCE_IS_ACTIVE (csrc_src))
1348           sess->stats.active_sources++;
1349         on_new_ssrc (sess, source);
1350       }
1351     }
1352   }
1353   gst_buffer_unref (buffer);
1354
1355   RTP_SESSION_UNLOCK (sess);
1356
1357   return result;
1358
1359   /* ERRORS */
1360 invalid_packet:
1361   {
1362     gst_buffer_unref (buffer);
1363     GST_DEBUG ("invalid RTP packet received");
1364     return GST_FLOW_OK;
1365   }
1366 ignore:
1367   {
1368     gst_buffer_unref (buffer);
1369     RTP_SESSION_UNLOCK (sess);
1370     GST_DEBUG ("ignoring RTP packet because we are leaving");
1371     return GST_FLOW_OK;
1372   }
1373 collision:
1374   {
1375     gst_buffer_unref (buffer);
1376     RTP_SESSION_UNLOCK (sess);
1377     GST_DEBUG ("ignoring packet because its collisioning");
1378     return GST_FLOW_OK;
1379   }
1380 }
1381
1382 static void
1383 rtp_session_process_rb (RTPSession * sess, RTPSource * source,
1384     GstRTCPPacket * packet, RTPArrivalStats * arrival)
1385 {
1386   guint count, i;
1387
1388   count = gst_rtcp_packet_get_rb_count (packet);
1389   for (i = 0; i < count; i++) {
1390     guint32 ssrc, exthighestseq, jitter, lsr, dlsr;
1391     guint8 fractionlost;
1392     gint32 packetslost;
1393
1394     gst_rtcp_packet_get_rb (packet, i, &ssrc, &fractionlost,
1395         &packetslost, &exthighestseq, &jitter, &lsr, &dlsr);
1396
1397     GST_DEBUG ("RB %d: SSRC %08x, jitter %" G_GUINT32_FORMAT, i, ssrc, jitter);
1398
1399     if (ssrc == sess->source->ssrc) {
1400       /* only deal with report blocks for our session, we update the stats of
1401        * the sender of the RTCP message. We could also compare our stats against
1402        * the other sender to see if we are better or worse. */
1403       rtp_source_process_rb (source, arrival->time, fractionlost, packetslost,
1404           exthighestseq, jitter, lsr, dlsr);
1405
1406       on_ssrc_active (sess, source);
1407     }
1408   }
1409 }
1410
1411 /* A Sender report contains statistics about how the sender is doing. This
1412  * includes timing informataion such as the relation between RTP and NTP
1413  * timestamps and the number of packets/bytes it sent to us.
1414  *
1415  * In this report is also included a set of report blocks related to how this
1416  * sender is receiving data (in case we (or somebody else) is also sending stuff
1417  * to it). This info includes the packet loss, jitter and seqnum. It also
1418  * contains information to calculate the round trip time (LSR/DLSR).
1419  */
1420 static void
1421 rtp_session_process_sr (RTPSession * sess, GstRTCPPacket * packet,
1422     RTPArrivalStats * arrival)
1423 {
1424   guint32 senderssrc, rtptime, packet_count, octet_count;
1425   guint64 ntptime;
1426   RTPSource *source;
1427   gboolean created, prevsender;
1428
1429   gst_rtcp_packet_sr_get_sender_info (packet, &senderssrc, &ntptime, &rtptime,
1430       &packet_count, &octet_count);
1431
1432   GST_DEBUG ("got SR packet: SSRC %08x, time %" GST_TIME_FORMAT,
1433       senderssrc, GST_TIME_ARGS (arrival->time));
1434
1435   source = obtain_source (sess, senderssrc, &created, arrival, FALSE);
1436
1437   if (!source)
1438     return;
1439
1440   GST_BUFFER_OFFSET (packet->buffer) = source->clock_base;
1441   GST_BUFFER_OFFSET_END (packet->buffer) = source->clock_base_time;
1442
1443   prevsender = RTP_SOURCE_IS_SENDER (source);
1444
1445   /* first update the source */
1446   rtp_source_process_sr (source, arrival->time, ntptime, rtptime, packet_count,
1447       octet_count);
1448
1449   if (prevsender != RTP_SOURCE_IS_SENDER (source)) {
1450     sess->stats.sender_sources++;
1451     GST_DEBUG ("source: %08x became sender, %d sender sources", senderssrc,
1452         sess->stats.sender_sources);
1453   }
1454
1455   if (created)
1456     on_new_ssrc (sess, source);
1457
1458   rtp_session_process_rb (sess, source, packet, arrival);
1459 }
1460
1461 /* A receiver report contains statistics about how a receiver is doing. It
1462  * includes stuff like packet loss, jitter and the seqnum it received last. It
1463  * also contains info to calculate the round trip time.
1464  *
1465  * We are only interested in how the sender of this report is doing wrt to us.
1466  */
1467 static void
1468 rtp_session_process_rr (RTPSession * sess, GstRTCPPacket * packet,
1469     RTPArrivalStats * arrival)
1470 {
1471   guint32 senderssrc;
1472   RTPSource *source;
1473   gboolean created;
1474
1475   senderssrc = gst_rtcp_packet_rr_get_ssrc (packet);
1476
1477   GST_DEBUG ("got RR packet: SSRC %08x", senderssrc);
1478
1479   source = obtain_source (sess, senderssrc, &created, arrival, FALSE);
1480
1481   if (!source)
1482     return;
1483
1484   if (created)
1485     on_new_ssrc (sess, source);
1486
1487   rtp_session_process_rb (sess, source, packet, arrival);
1488 }
1489
1490 /* Get SDES items and store them in the SSRC */
1491 static void
1492 rtp_session_process_sdes (RTPSession * sess, GstRTCPPacket * packet,
1493     RTPArrivalStats * arrival)
1494 {
1495   guint items, i, j;
1496   gboolean more_items, more_entries;
1497
1498   items = gst_rtcp_packet_sdes_get_item_count (packet);
1499   GST_DEBUG ("got SDES packet with %d items", items);
1500
1501   more_items = gst_rtcp_packet_sdes_first_item (packet);
1502   i = 0;
1503   while (more_items) {
1504     guint32 ssrc;
1505     gboolean changed, created;
1506     RTPSource *source;
1507
1508     ssrc = gst_rtcp_packet_sdes_get_ssrc (packet);
1509
1510     GST_DEBUG ("item %d, SSRC %08x", i, ssrc);
1511
1512     /* find src, no probation when dealing with RTCP */
1513     source = obtain_source (sess, ssrc, &created, arrival, FALSE);
1514     changed = FALSE;
1515
1516     if (!source)
1517       return;
1518
1519     more_entries = gst_rtcp_packet_sdes_first_entry (packet);
1520     j = 0;
1521     while (more_entries) {
1522       GstRTCPSDESType type;
1523       guint8 len;
1524       guint8 *data;
1525
1526       gst_rtcp_packet_sdes_get_entry (packet, &type, &len, &data);
1527
1528       GST_DEBUG ("entry %d, type %d, len %d, data %.*s", j, type, len, len,
1529           data);
1530
1531       changed |= rtp_source_set_sdes (source, type, data, len);
1532
1533       more_entries = gst_rtcp_packet_sdes_next_entry (packet);
1534       j++;
1535     }
1536
1537     if (created)
1538       on_new_ssrc (sess, source);
1539     if (changed)
1540       on_ssrc_sdes (sess, source);
1541
1542     more_items = gst_rtcp_packet_sdes_next_item (packet);
1543     i++;
1544   }
1545 }
1546
1547 /* BYE is sent when a client leaves the session
1548  */
1549 static void
1550 rtp_session_process_bye (RTPSession * sess, GstRTCPPacket * packet,
1551     RTPArrivalStats * arrival)
1552 {
1553   guint count, i;
1554   gchar *reason;
1555
1556   reason = gst_rtcp_packet_bye_get_reason (packet);
1557   GST_DEBUG ("got BYE packet (reason: %s)", GST_STR_NULL (reason));
1558
1559   count = gst_rtcp_packet_bye_get_ssrc_count (packet);
1560   for (i = 0; i < count; i++) {
1561     guint32 ssrc;
1562     RTPSource *source;
1563     gboolean created, prevactive, prevsender;
1564     guint pmembers, members;
1565
1566     ssrc = gst_rtcp_packet_bye_get_nth_ssrc (packet, i);
1567     GST_DEBUG ("SSRC: %08x", ssrc);
1568
1569     /* find src and mark bye, no probation when dealing with RTCP */
1570     source = obtain_source (sess, ssrc, &created, arrival, FALSE);
1571
1572     if (!source)
1573       return;
1574
1575     /* store time for when we need to time out this source */
1576     source->bye_time = arrival->time;
1577
1578     prevactive = RTP_SOURCE_IS_ACTIVE (source);
1579     prevsender = RTP_SOURCE_IS_SENDER (source);
1580
1581     /* let the source handle the rest */
1582     rtp_source_process_bye (source, reason);
1583
1584     pmembers = sess->stats.active_sources;
1585
1586     if (prevactive && !RTP_SOURCE_IS_ACTIVE (source)) {
1587       sess->stats.active_sources--;
1588       GST_DEBUG ("source: %08x became inactive, %d active sources", ssrc,
1589           sess->stats.active_sources);
1590     }
1591     if (prevsender && !RTP_SOURCE_IS_SENDER (source)) {
1592       sess->stats.sender_sources--;
1593       GST_DEBUG ("source: %08x became non sender, %d sender sources", ssrc,
1594           sess->stats.sender_sources);
1595     }
1596     members = sess->stats.active_sources;
1597
1598     if (!sess->source->received_bye && members < pmembers) {
1599       /* some members went away since the previous timeout estimate.
1600        * Perform reverse reconsideration but only when we are not scheduling a
1601        * BYE ourselves. */
1602       if (arrival->time < sess->next_rtcp_check_time) {
1603         GstClockTime time_remaining;
1604
1605         time_remaining = sess->next_rtcp_check_time - arrival->time;
1606         sess->next_rtcp_check_time =
1607             gst_util_uint64_scale (time_remaining, members, pmembers);
1608
1609         GST_DEBUG ("reverse reconsideration %" GST_TIME_FORMAT,
1610             GST_TIME_ARGS (sess->next_rtcp_check_time));
1611
1612         sess->next_rtcp_check_time += arrival->time;
1613
1614         RTP_SESSION_UNLOCK (sess);
1615         /* notify app of reconsideration */
1616         if (sess->callbacks.reconsider)
1617           sess->callbacks.reconsider (sess, sess->reconsider_user_data);
1618         RTP_SESSION_LOCK (sess);
1619       }
1620     }
1621
1622     if (created)
1623       on_new_ssrc (sess, source);
1624
1625     on_bye_ssrc (sess, source);
1626   }
1627   g_free (reason);
1628 }
1629
1630 static void
1631 rtp_session_process_app (RTPSession * sess, GstRTCPPacket * packet,
1632     RTPArrivalStats * arrival)
1633 {
1634   GST_DEBUG ("received APP");
1635 }
1636
1637 /**
1638  * rtp_session_process_rtcp:
1639  * @sess: and #RTPSession
1640  * @buffer: an RTCP buffer
1641  * @current_time: the current system time
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     GstClockTime current_time)
1651 {
1652   GstRTCPPacket packet;
1653   gboolean more, is_bye = FALSE, is_sr = FALSE;
1654   RTPArrivalStats arrival;
1655   GstFlowReturn result = GST_FLOW_OK;
1656
1657   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1658   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1659
1660   if (!gst_rtcp_buffer_validate (buffer))
1661     goto invalid_packet;
1662
1663   GST_DEBUG ("received RTCP packet");
1664
1665   RTP_SESSION_LOCK (sess);
1666   /* update arrival stats */
1667   update_arrival_stats (sess, &arrival, FALSE, buffer, current_time, -1);
1668
1669   if (sess->sent_bye)
1670     goto ignore;
1671
1672   /* start processing the compound packet */
1673   more = gst_rtcp_buffer_get_first_packet (buffer, &packet);
1674   while (more) {
1675     GstRTCPType type;
1676
1677     type = gst_rtcp_packet_get_type (&packet);
1678
1679     /* when we are leaving the session, we should ignore all non-BYE messages */
1680     if (sess->source->received_bye && type != GST_RTCP_TYPE_BYE) {
1681       GST_DEBUG ("ignoring non-BYE RTCP packet because we are leaving");
1682       goto next;
1683     }
1684
1685     switch (type) {
1686       case GST_RTCP_TYPE_SR:
1687         rtp_session_process_sr (sess, &packet, &arrival);
1688         is_sr = TRUE;
1689         break;
1690       case GST_RTCP_TYPE_RR:
1691         rtp_session_process_rr (sess, &packet, &arrival);
1692         break;
1693       case GST_RTCP_TYPE_SDES:
1694         rtp_session_process_sdes (sess, &packet, &arrival);
1695         break;
1696       case GST_RTCP_TYPE_BYE:
1697         is_bye = TRUE;
1698         rtp_session_process_bye (sess, &packet, &arrival);
1699         break;
1700       case GST_RTCP_TYPE_APP:
1701         rtp_session_process_app (sess, &packet, &arrival);
1702         break;
1703       default:
1704         GST_WARNING ("got unknown RTCP packet");
1705         break;
1706     }
1707   next:
1708     more = gst_rtcp_packet_move_to_next (&packet);
1709   }
1710
1711   /* if we are scheduling a BYE, we only want to count bye packets, else we
1712    * count everything */
1713   if (sess->source->received_bye) {
1714     if (is_bye) {
1715       sess->stats.bye_members++;
1716       UPDATE_AVG (sess->stats.avg_rtcp_packet_size, arrival.bytes);
1717     }
1718   } else {
1719     /* keep track of average packet size */
1720     UPDATE_AVG (sess->stats.avg_rtcp_packet_size, arrival.bytes);
1721   }
1722   RTP_SESSION_UNLOCK (sess);
1723
1724   /* notify caller of sr packets in the callback */
1725   if (is_sr && sess->callbacks.sync_rtcp)
1726     result = sess->callbacks.sync_rtcp (sess, sess->source, buffer,
1727         sess->sync_rtcp_user_data);
1728   else
1729     gst_buffer_unref (buffer);
1730
1731   return result;
1732
1733   /* ERRORS */
1734 invalid_packet:
1735   {
1736     GST_DEBUG ("invalid RTCP packet received");
1737     gst_buffer_unref (buffer);
1738     return GST_FLOW_OK;
1739   }
1740 ignore:
1741   {
1742     gst_buffer_unref (buffer);
1743     RTP_SESSION_UNLOCK (sess);
1744     GST_DEBUG ("ignoring RTP packet because we left");
1745     return GST_FLOW_OK;
1746   }
1747 }
1748
1749 /**
1750  * rtp_session_send_rtp:
1751  * @sess: an #RTPSession
1752  * @buffer: an RTP buffer
1753  * @current_time: the current system time
1754  * @ntpnstime: the NTP time in nanoseconds of when this buffer was captured.
1755  *
1756  * Send the RTP buffer in the session manager. This function takes ownership of
1757  * @buffer.
1758  *
1759  * Returns: a #GstFlowReturn.
1760  */
1761 GstFlowReturn
1762 rtp_session_send_rtp (RTPSession * sess, GstBuffer * buffer,
1763     GstClockTime current_time, guint64 ntpnstime)
1764 {
1765   GstFlowReturn result;
1766   RTPSource *source;
1767   gboolean prevsender;
1768
1769   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1770   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1771
1772   if (!gst_rtp_buffer_validate (buffer))
1773     goto invalid_packet;
1774
1775   GST_DEBUG ("received RTP packet for sending");
1776
1777   RTP_SESSION_LOCK (sess);
1778   source = sess->source;
1779
1780   /* update last activity */
1781   source->last_rtp_activity = current_time;
1782
1783   prevsender = RTP_SOURCE_IS_SENDER (source);
1784
1785   /* we use our own source to send */
1786   result = rtp_source_send_rtp (source, buffer, ntpnstime);
1787
1788   if (RTP_SOURCE_IS_SENDER (source) && !prevsender)
1789     sess->stats.sender_sources++;
1790   RTP_SESSION_UNLOCK (sess);
1791
1792   return result;
1793
1794   /* ERRORS */
1795 invalid_packet:
1796   {
1797     gst_buffer_unref (buffer);
1798     GST_DEBUG ("invalid RTP packet received");
1799     return GST_FLOW_OK;
1800   }
1801 }
1802
1803 static GstClockTime
1804 calculate_rtcp_interval (RTPSession * sess, gboolean deterministic,
1805     gboolean first)
1806 {
1807   GstClockTime result;
1808
1809   if (sess->source->received_bye) {
1810     result = rtp_stats_calculate_bye_interval (&sess->stats);
1811   } else {
1812     result = rtp_stats_calculate_rtcp_interval (&sess->stats,
1813         RTP_SOURCE_IS_SENDER (sess->source), first);
1814   }
1815
1816   GST_DEBUG ("next deterministic interval: %" GST_TIME_FORMAT ", first %d",
1817       GST_TIME_ARGS (result), first);
1818
1819   if (!deterministic)
1820     result = rtp_stats_add_rtcp_jitter (&sess->stats, result);
1821
1822   GST_DEBUG ("next interval: %" GST_TIME_FORMAT, GST_TIME_ARGS (result));
1823
1824   return result;
1825 }
1826
1827 /**
1828  * rtp_session_send_bye_locked:
1829  * @sess: an #RTPSession
1830  * @reason: a reason or NULL
1831  *
1832  * Stop the current @sess and schedule a BYE message for the other members.
1833  *
1834  * One must have the session lock to call this function
1835  *
1836  * Returns: a #GstFlowReturn.
1837  */
1838 static GstFlowReturn
1839 rtp_session_send_bye_locked (RTPSession * sess, const gchar * reason,
1840     GstClockTime current_time)
1841 {
1842   GstFlowReturn result = GST_FLOW_OK;
1843   RTPSource *source;
1844   GstClockTime interval;
1845
1846   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1847
1848   source = sess->source;
1849
1850   /* ignore more BYEs */
1851   if (source->received_bye)
1852     goto done;
1853
1854   /* we have BYE now */
1855   source->received_bye = TRUE;
1856   /* at least one member wants to send a BYE */
1857   g_free (sess->bye_reason);
1858   sess->bye_reason = g_strdup (reason);
1859   sess->stats.avg_rtcp_packet_size = 100;
1860   sess->stats.bye_members = 1;
1861   sess->first_rtcp = TRUE;
1862   sess->sent_bye = FALSE;
1863
1864   /* reschedule transmission */
1865   sess->last_rtcp_send_time = current_time;
1866   interval = calculate_rtcp_interval (sess, FALSE, TRUE);
1867   sess->next_rtcp_check_time = current_time + interval;
1868
1869   GST_DEBUG ("Schedule BYE for %" GST_TIME_FORMAT ", %" GST_TIME_FORMAT,
1870       GST_TIME_ARGS (interval), GST_TIME_ARGS (sess->next_rtcp_check_time));
1871
1872   RTP_SESSION_UNLOCK (sess);
1873   /* notify app of reconsideration */
1874   if (sess->callbacks.reconsider)
1875     sess->callbacks.reconsider (sess, sess->reconsider_user_data);
1876   RTP_SESSION_LOCK (sess);
1877 done:
1878
1879   return result;
1880 }
1881
1882 /**
1883  * rtp_session_send_bye:
1884  * @sess: an #RTPSession
1885  * @reason: a reason or NULL
1886  * @current_time: the current system time
1887  *
1888  * Stop the current @sess and schedule a BYE message for the other members.
1889  *
1890  * One must have the session lock to call this function
1891  *
1892  * Returns: a #GstFlowReturn.
1893  */
1894 GstFlowReturn
1895 rtp_session_send_bye (RTPSession * sess, const gchar * reason,
1896     GstClockTime current_time)
1897 {
1898   GstFlowReturn result = GST_FLOW_OK;
1899
1900   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1901
1902   RTP_SESSION_LOCK (sess);
1903   result = rtp_session_send_bye_locked (sess, reason, current_time);
1904   RTP_SESSION_UNLOCK (sess);
1905
1906   return result;
1907 }
1908
1909 /**
1910  * rtp_session_next_timeout:
1911  * @sess: an #RTPSession
1912  * @current_time: the current system time
1913  *
1914  * Get the next time we should perform session maintenance tasks.
1915  *
1916  * Returns: a time when rtp_session_on_timeout() should be called with the
1917  * current system time.
1918  */
1919 GstClockTime
1920 rtp_session_next_timeout (RTPSession * sess, GstClockTime current_time)
1921 {
1922   GstClockTime result;
1923
1924   g_return_val_if_fail (RTP_IS_SESSION (sess), GST_FLOW_ERROR);
1925
1926   RTP_SESSION_LOCK (sess);
1927
1928   result = sess->next_rtcp_check_time;
1929
1930   GST_DEBUG ("current time: %" GST_TIME_FORMAT ", next :%" GST_TIME_FORMAT,
1931       GST_TIME_ARGS (current_time), GST_TIME_ARGS (result));
1932
1933   if (result < current_time) {
1934     GST_DEBUG ("take current time as base");
1935     /* our previous check time expired, start counting from the current time
1936      * again. */
1937     result = current_time;
1938   }
1939
1940   if (sess->source->received_bye) {
1941     if (sess->sent_bye) {
1942       GST_DEBUG ("we sent BYE already");
1943       result = GST_CLOCK_TIME_NONE;
1944     } else if (sess->stats.active_sources >= 50) {
1945       GST_DEBUG ("reconsider BYE, more than 50 sources");
1946       /* reconsider BYE if members >= 50 */
1947       result += calculate_rtcp_interval (sess, FALSE, TRUE);
1948     }
1949   } else {
1950     if (sess->first_rtcp) {
1951       GST_DEBUG ("first RTCP packet");
1952       /* we are called for the first time */
1953       result += calculate_rtcp_interval (sess, FALSE, TRUE);
1954     } else if (sess->next_rtcp_check_time < current_time) {
1955       GST_DEBUG ("old check time expired, getting new timeout");
1956       /* get a new timeout when we need to */
1957       result += calculate_rtcp_interval (sess, FALSE, FALSE);
1958     }
1959   }
1960   sess->next_rtcp_check_time = result;
1961
1962   GST_DEBUG ("next timeout: %" GST_TIME_FORMAT, GST_TIME_ARGS (result));
1963   RTP_SESSION_UNLOCK (sess);
1964
1965   return result;
1966 }
1967
1968 typedef struct
1969 {
1970   RTPSession *sess;
1971   GstBuffer *rtcp;
1972   GstClockTime current_time;
1973   guint64 ntpnstime;
1974   GstClockTime interval;
1975   GstRTCPPacket packet;
1976   gboolean is_bye;
1977   gboolean has_sdes;
1978 } ReportData;
1979
1980 static void
1981 session_start_rtcp (RTPSession * sess, ReportData * data)
1982 {
1983   GstRTCPPacket *packet = &data->packet;
1984   RTPSource *own = sess->source;
1985
1986   data->rtcp = gst_rtcp_buffer_new (sess->mtu);
1987
1988   if (RTP_SOURCE_IS_SENDER (own)) {
1989     guint64 ntptime;
1990     guint32 rtptime;
1991     guint32 packet_count, octet_count;
1992
1993     /* we are a sender, create SR */
1994     GST_DEBUG ("create SR for SSRC %08x", own->ssrc);
1995     gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_SR, packet);
1996
1997     /* get latest stats */
1998     rtp_source_get_new_sr (own, data->ntpnstime, &ntptime, &rtptime,
1999         &packet_count, &octet_count);
2000     /* store stats */
2001     rtp_source_process_sr (own, data->ntpnstime, ntptime, rtptime, packet_count,
2002         octet_count);
2003
2004     /* fill in sender report info */
2005     gst_rtcp_packet_sr_set_sender_info (packet, own->ssrc,
2006         ntptime, rtptime, packet_count, octet_count);
2007   } else {
2008     /* we are only receiver, create RR */
2009     GST_DEBUG ("create RR for SSRC %08x", own->ssrc);
2010     gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_RR, packet);
2011     gst_rtcp_packet_rr_set_ssrc (packet, own->ssrc);
2012   }
2013 }
2014
2015 /* construct a Sender or Receiver Report */
2016 static void
2017 session_report_blocks (const gchar * key, RTPSource * source, ReportData * data)
2018 {
2019   RTPSession *sess = data->sess;
2020   GstRTCPPacket *packet = &data->packet;
2021
2022   /* create a new buffer if needed */
2023   if (data->rtcp == NULL) {
2024     session_start_rtcp (sess, data);
2025   }
2026   if (gst_rtcp_packet_get_rb_count (packet) < GST_RTCP_MAX_RB_COUNT) {
2027     /* only report about other sender sources */
2028     if (source != sess->source && RTP_SOURCE_IS_SENDER (source)) {
2029       guint8 fractionlost;
2030       gint32 packetslost;
2031       guint32 exthighestseq, jitter;
2032       guint32 lsr, dlsr;
2033
2034       /* get new stats */
2035       rtp_source_get_new_rb (source, data->current_time, &fractionlost,
2036           &packetslost, &exthighestseq, &jitter, &lsr, &dlsr);
2037
2038       /* packet is not yet filled, add report block for this source. */
2039       gst_rtcp_packet_add_rb (packet, source->ssrc, fractionlost, packetslost,
2040           exthighestseq, jitter, lsr, dlsr);
2041     }
2042   }
2043 }
2044
2045 /* perform cleanup of sources that timed out */
2046 static gboolean
2047 session_cleanup (const gchar * key, RTPSource * source, ReportData * data)
2048 {
2049   gboolean remove = FALSE;
2050   gboolean byetimeout = FALSE;
2051   gboolean is_sender, is_active;
2052   RTPSession *sess = data->sess;
2053   GstClockTime interval;
2054
2055   is_sender = RTP_SOURCE_IS_SENDER (source);
2056   is_active = RTP_SOURCE_IS_ACTIVE (source);
2057
2058   /* check for our own source, we don't want to delete our own source. */
2059   if (!(source == sess->source)) {
2060     if (source->received_bye) {
2061       /* if we received a BYE from the source, remove the source after some
2062        * time. */
2063       if (data->current_time > source->bye_time &&
2064           data->current_time - source->bye_time > sess->stats.bye_timeout) {
2065         GST_DEBUG ("removing BYE source %08x", source->ssrc);
2066         remove = TRUE;
2067         byetimeout = TRUE;
2068       }
2069     }
2070     /* sources that were inactive for more than 5 times the deterministic reporting
2071      * interval get timed out. the min timeout is 5 seconds. */
2072     if (data->current_time > source->last_activity) {
2073       interval = MAX (data->interval * 5, 5 * GST_SECOND);
2074       if (data->current_time - source->last_activity > interval) {
2075         GST_DEBUG ("removing timeout source %08x, last %" GST_TIME_FORMAT,
2076             source->ssrc, GST_TIME_ARGS (source->last_activity));
2077         remove = TRUE;
2078       }
2079     }
2080   }
2081
2082   /* senders that did not send for a long time become a receiver, this also
2083    * holds for our own source. */
2084   if (is_sender) {
2085     if (data->current_time > source->last_rtp_activity) {
2086       interval = MAX (data->interval * 2, 5 * GST_SECOND);
2087       if (data->current_time - source->last_rtp_activity > interval) {
2088         GST_DEBUG ("sender source %08x timed out and became receiver, last %"
2089             GST_TIME_FORMAT, source->ssrc,
2090             GST_TIME_ARGS (source->last_rtp_activity));
2091         source->is_sender = FALSE;
2092         sess->stats.sender_sources--;
2093       }
2094     }
2095   }
2096
2097   if (remove) {
2098     sess->total_sources--;
2099     if (is_sender)
2100       sess->stats.sender_sources--;
2101     if (is_active)
2102       sess->stats.active_sources--;
2103
2104     if (byetimeout)
2105       on_bye_timeout (sess, source);
2106     else
2107       on_timeout (sess, source);
2108   }
2109   return remove;
2110 }
2111
2112 static void
2113 session_sdes (RTPSession * sess, ReportData * data)
2114 {
2115   GstRTCPPacket *packet = &data->packet;
2116   guint8 *sdes_data;
2117   guint sdes_len;
2118
2119   /* add SDES packet */
2120   gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_SDES, packet);
2121
2122   gst_rtcp_packet_sdes_add_item (packet, sess->source->ssrc);
2123
2124   rtp_source_get_sdes (sess->source, GST_RTCP_SDES_CNAME, &sdes_data,
2125       &sdes_len);
2126   gst_rtcp_packet_sdes_add_entry (packet, GST_RTCP_SDES_CNAME, sdes_len,
2127       sdes_data);
2128
2129   /* other SDES items must only be added at regular intervals and only when the
2130    * user requests to since it might be a privacy problem */
2131 #if 0
2132   gst_rtcp_packet_sdes_add_entry (&packet, GST_RTCP_SDES_NAME,
2133       strlen (sess->name), (guint8 *) sess->name);
2134   gst_rtcp_packet_sdes_add_entry (&packet, GST_RTCP_SDES_TOOL,
2135       strlen (sess->tool), (guint8 *) sess->tool);
2136 #endif
2137
2138   data->has_sdes = TRUE;
2139 }
2140
2141 /* schedule a BYE packet */
2142 static void
2143 session_bye (RTPSession * sess, ReportData * data)
2144 {
2145   GstRTCPPacket *packet = &data->packet;
2146
2147   /* open packet */
2148   session_start_rtcp (sess, data);
2149
2150   /* add SDES */
2151   session_sdes (sess, data);
2152
2153   /* add a BYE packet */
2154   gst_rtcp_buffer_add_packet (data->rtcp, GST_RTCP_TYPE_BYE, packet);
2155   gst_rtcp_packet_bye_add_ssrc (packet, sess->source->ssrc);
2156   if (sess->bye_reason)
2157     gst_rtcp_packet_bye_set_reason (packet, sess->bye_reason);
2158
2159   /* we have a BYE packet now */
2160   data->is_bye = TRUE;
2161 }
2162
2163 static gboolean
2164 is_rtcp_time (RTPSession * sess, GstClockTime current_time, ReportData * data)
2165 {
2166   GstClockTime new_send_time, elapsed;
2167   gboolean result;
2168
2169   /* no need to check yet */
2170   if (sess->next_rtcp_check_time > current_time) {
2171     GST_DEBUG ("no check time yet, next %" GST_TIME_FORMAT " > now %"
2172         GST_TIME_FORMAT, GST_TIME_ARGS (sess->next_rtcp_check_time),
2173         GST_TIME_ARGS (current_time));
2174     return FALSE;
2175   }
2176
2177   /* get elapsed time since we last reported */
2178   elapsed = current_time - sess->last_rtcp_send_time;
2179
2180   /* perform forward reconsideration */
2181   new_send_time = rtp_stats_add_rtcp_jitter (&sess->stats, data->interval);
2182
2183   GST_DEBUG ("forward reconsideration %" GST_TIME_FORMAT ", elapsed %"
2184       GST_TIME_FORMAT, GST_TIME_ARGS (new_send_time), GST_TIME_ARGS (elapsed));
2185
2186   new_send_time += sess->last_rtcp_send_time;
2187
2188   /* check if reconsideration */
2189   if (current_time < new_send_time) {
2190     GST_DEBUG ("reconsider RTCP for %" GST_TIME_FORMAT,
2191         GST_TIME_ARGS (new_send_time));
2192     result = FALSE;
2193     /* store new check time */
2194     sess->next_rtcp_check_time = new_send_time;
2195   } else {
2196     result = TRUE;
2197     new_send_time = calculate_rtcp_interval (sess, FALSE, FALSE);
2198
2199     GST_DEBUG ("can send RTCP now, next interval %" GST_TIME_FORMAT,
2200         GST_TIME_ARGS (new_send_time));
2201     sess->next_rtcp_check_time = current_time + new_send_time;
2202   }
2203   return result;
2204 }
2205
2206 /**
2207  * rtp_session_on_timeout:
2208  * @sess: an #RTPSession
2209  * @current_time: the current system time
2210  * @ntpnstime: the current NTP time in nanoseconds
2211  *
2212  * Perform maintenance actions after the timeout obtained with
2213  * rtp_session_next_timeout() expired.
2214  *
2215  * This function will perform timeouts of receivers and senders, send a BYE
2216  * packet or generate RTCP packets with current session stats.
2217  *
2218  * This function can call the #RTPSessionSendRTCP callback, possibly multiple
2219  * times, for each packet that should be processed.
2220  *
2221  * Returns: a #GstFlowReturn.
2222  */
2223 GstFlowReturn
2224 rtp_session_on_timeout (RTPSession * sess, GstClockTime current_time,
2225     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 (current_time), GST_TIME_ARGS (ntpnstime));
2235
2236   data.sess = sess;
2237   data.rtcp = NULL;
2238   data.current_time = current_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, current_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.current_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 < current_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 }