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