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