gst/rtpmanager/rtpsource.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., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19 #include <string.h>
  20
  21 #include <gst/rtp/gstrtpbuffer.h>
  22 #include <gst/rtp/gstrtcpbuffer.h>
  23
  24 #include "rtpsource.h"
  25
  26 GST_DEBUG_CATEGORY_STATIC (rtp_source_debug);
  27 #define GST_CAT_DEFAULT rtp_source_debug
  28
  29 #define RTP_MAX_PROBATION_LEN  32
  30
  31 /* signals and args */
  32 enum
  33 {
  34   LAST_SIGNAL
  35 };
  36
  37 #define DEFAULT_SSRC                 0
  38 #define DEFAULT_IS_CSRC              FALSE
  39 #define DEFAULT_IS_VALIDATED         FALSE
  40 #define DEFAULT_IS_SENDER            FALSE
  41 #define DEFAULT_SDES                 NULL
  42 #define DEFAULT_PROBATION            RTP_DEFAULT_PROBATION
  43
  44 enum
  45 {
  46   PROP_0,
  47   PROP_SSRC,
  48   PROP_IS_CSRC,
  49   PROP_IS_VALIDATED,
  50   PROP_IS_SENDER,
  51   PROP_SDES,
  52   PROP_STATS,
  53   PROP_PROBATION
  54 };
  55
  56 /* GObject vmethods */
  57 static void rtp_source_finalize (GObject * object);
  58 static void rtp_source_set_property (GObject * object, guint prop_id,
  59     const GValue * value, GParamSpec * pspec);
  60 static void rtp_source_get_property (GObject * object, guint prop_id,
  61     GValue * value, GParamSpec * pspec);
  62
  63 /* static guint rtp_source_signals[LAST_SIGNAL] = { 0 }; */
  64
  65 G_DEFINE_TYPE (RTPSource, rtp_source, G_TYPE_OBJECT);
  66
  67 static void
  68 rtp_source_class_init (RTPSourceClass * klass)
  69 {
  70   GObjectClass *gobject_class;
  71
  72   gobject_class = (GObjectClass *) klass;
  73
  74   gobject_class->finalize = rtp_source_finalize;
  75
  76   gobject_class->set_property = rtp_source_set_property;
  77   gobject_class->get_property = rtp_source_get_property;
  78
  79   g_object_class_install_property (gobject_class, PROP_SSRC,
  80       g_param_spec_uint ("ssrc", "SSRC",
  81           "The SSRC of this source", 0, G_MAXUINT, DEFAULT_SSRC,
  82           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
  83
  84   g_object_class_install_property (gobject_class, PROP_IS_CSRC,
  85       g_param_spec_boolean ("is-csrc", "Is CSRC",
  86           "If this SSRC is acting as a contributing source",
  87           DEFAULT_IS_CSRC, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
  88
  89   g_object_class_install_property (gobject_class, PROP_IS_VALIDATED,
  90       g_param_spec_boolean ("is-validated", "Is Validated",
  91           "If this SSRC is validated", DEFAULT_IS_VALIDATED,
  92           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
  93
  94   g_object_class_install_property (gobject_class, PROP_IS_SENDER,
  95       g_param_spec_boolean ("is-sender", "Is Sender",
  96           "If this SSRC is a sender", DEFAULT_IS_SENDER,
  97           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
  98
  99   /**
 100    * RTPSource::sdes
 101    *
 102    * The current SDES items of the source. Returns a structure with name
 103    * application/x-rtp-source-sdes and may contain the following fields:
 104    *
 105    *  'cname'       G_TYPE_STRING  : The canonical name
 106    *  'name'        G_TYPE_STRING  : The user name
 107    *  'email'       G_TYPE_STRING  : The user's electronic mail address
 108    *  'phone'       G_TYPE_STRING  : The user's phone number
 109    *  'location'    G_TYPE_STRING  : The geographic user location
 110    *  'tool'        G_TYPE_STRING  : The name of application or tool
 111    *  'note'        G_TYPE_STRING  : A notice about the source
 112    *
 113    *  Other fields may be present and these represent private items in
 114    *  the SDES where the field name is the prefix.
 115    */
 116   g_object_class_install_property (gobject_class, PROP_SDES,
 117       g_param_spec_boxed ("sdes", "SDES",
 118           "The SDES information for this source",
 119           GST_TYPE_STRUCTURE, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 120
 121   /**
 122    * RTPSource::stats
 123    *
 124    * This property returns a GstStructure named application/x-rtp-source-stats with
 125    * fields useful for statistics and diagnostics.
 126    *
 127    * Take note of each respective field's units:
 128    *
 129    * - NTP times are in the appropriate 32-bit or 64-bit fixed-point format
 130    *   starting from January 1, 1970 (except for timespans).
 131    * - RTP times are in clock rate units (i.e. clock rate = 1 second)
 132    *   starting at a random offset.
 133    * - For fields indicating packet loss, note that late packets are not considered lost,
 134    *   and duplicates are not taken into account. Hence, the loss may be negative
 135    *   if there are duplicates.
 136    *
 137    * The following fields are always present.
 138    *
 139    *  "ssrc"         G_TYPE_UINT     the SSRC of this source
 140    *  "internal"     G_TYPE_BOOLEAN  this source is a source of the session
 141    *  "validated"    G_TYPE_BOOLEAN  the source is validated
 142    *  "received-bye" G_TYPE_BOOLEAN  we received a BYE from this source
 143    *  "is-csrc"      G_TYPE_BOOLEAN  this source was found as CSRC
 144    *  "is-sender"    G_TYPE_BOOLEAN  this source is a sender
 145    *  "seqnum-base"  G_TYPE_INT      first seqnum if known
 146    *  "clock-rate"   G_TYPE_INT      the clock rate of the media
 147    *
 148    * The following fields are only present when known.
 149    *
 150    *  "rtp-from"     G_TYPE_STRING   where we received the last RTP packet from
 151    *  "rtcp-from"    G_TYPE_STRING   where we received the last RTCP packet from
 152    *
 153    * The following fields make sense for internal sources and will only increase
 154    * when "is-sender" is TRUE.
 155    *
 156    *  "octets-sent"  G_TYPE_UINT64   number of bytes we sent
 157    *  "packets-sent" G_TYPE_UINT64   number of packets we sent
 158    *
 159    * The following fields make sense for non-internal sources and will only
 160    * increase when "is-sender" is TRUE.
 161    *
 162    *  "octets-received"  G_TYPE_UINT64  total number of bytes received
 163    *  "packets-received" G_TYPE_UINT64  total number of packets received
 164    *
 165    * Following fields are updated when "is-sender" is TRUE.
 166    *
 167    *  "bitrate"      G_TYPE_UINT64   bitrate in bits per second
 168    *  "jitter"       G_TYPE_UINT     estimated jitter (in clock rate units)
 169    *  "packets-lost" G_TYPE_INT      estimated amount of packets lost
 170    *
 171    * The last SR report this source sent. This only updates when "is-sender" is
 172    * TRUE.
 173    *
 174    *  "have-sr"         G_TYPE_BOOLEAN  the source has sent SR
 175    *  "sr-ntptime"      G_TYPE_UINT64   NTP time of SR (in NTP Timestamp Format, 32.32 fixed point)
 176    *  "sr-rtptime"      G_TYPE_UINT     RTP time of SR (in clock rate units)
 177    *  "sr-octet-count"  G_TYPE_UINT     the number of bytes in the SR
 178    *  "sr-packet-count" G_TYPE_UINT     the number of packets in the SR
 179    *
 180    * The following fields are only present for non-internal sources and
 181    * represent the content of the last RB packet that was sent to this source.
 182    * These values are only updated when the source is sending.
 183    *
 184    *  "sent-rb"               G_TYPE_BOOLEAN  we have sent an RB
 185    *  "sent-rb-fractionlost"  G_TYPE_UINT     calculated lost fraction
 186    *  "sent-rb-packetslost"   G_TYPE_INT      lost packets
 187    *  "sent-rb-exthighestseq" G_TYPE_UINT     last seen seqnum
 188    *  "sent-rb-jitter"        G_TYPE_UINT     jitter (in clock rate units)
 189    *  "sent-rb-lsr"           G_TYPE_UINT     last SR time (in NTP Short Format, 16.16 fixed point)
 190    *  "sent-rb-dlsr"          G_TYPE_UINT     delay since last SR (in NTP Short Format, 16.16 fixed point)
 191    *
 192    * The following fields are only present for non-internal sources and
 193    * represents the last RB that this source sent. This is only updated
 194    * when the source is receiving data and sending RB blocks.
 195    *
 196    *  "have-rb"          G_TYPE_BOOLEAN  the source has sent RB
 197    *  "rb-fractionlost"  G_TYPE_UINT     lost fraction
 198    *  "rb-packetslost"   G_TYPE_INT      lost packets
 199    *  "rb-exthighestseq" G_TYPE_UINT     highest received seqnum
 200    *  "rb-jitter"        G_TYPE_UINT     reception jitter (in clock rate units)
 201    *  "rb-lsr"           G_TYPE_UINT     last SR time (in NTP Short Format, 16.16 fixed point)
 202    *  "rb-dlsr"          G_TYPE_UINT     delay since last SR (in NTP Short Format, 16.16 fixed point)
 203    *
 204    * The round trip of this source is calculated from the last RB
 205    * values and the reception time of the last RB packet. It is only present for
 206    * non-internal sources.
 207    *
 208    *  "rb-round-trip"    G_TYPE_UINT     the round-trip time (in NTP Short Format, 16.16 fixed point)
 209    *
 210    */
 211   g_object_class_install_property (gobject_class, PROP_STATS,
 212       g_param_spec_boxed ("stats", "Stats",
 213           "The stats of this source", GST_TYPE_STRUCTURE,
 214           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 215
 216   g_object_class_install_property (gobject_class, PROP_PROBATION,
 217       g_param_spec_uint ("probation", "Number of probations",
 218           "Consecutive packet sequence numbers to accept the source",
 219           0, G_MAXUINT, DEFAULT_PROBATION,
 220           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 221
 222   GST_DEBUG_CATEGORY_INIT (rtp_source_debug, "rtpsource", 0, "RTP Source");
 223 }
 224
 225 /**
 226  * rtp_source_reset:
 227  * @src: an #RTPSource
 228  *
 229  * Reset the stats of @src.
 230  */
 231 void
 232 rtp_source_reset (RTPSource * src)
 233 {
 234   src->marked_bye = FALSE;
 235   if (src->bye_reason)
 236     g_free (src->bye_reason);
 237   src->bye_reason = NULL;
 238   src->sent_bye = FALSE;
 239   g_hash_table_remove_all (src->reported_in_sr_of);
 240
 241   src->stats.cycles = -1;
 242   src->stats.jitter = 0;
 243   src->stats.transit = -1;
 244   src->stats.curr_sr = 0;
 245   src->stats.sr[0].is_valid = FALSE;
 246   src->stats.curr_rr = 0;
 247   src->stats.rr[0].is_valid = FALSE;
 248   src->stats.prev_rtptime = GST_CLOCK_TIME_NONE;
 249   src->stats.prev_rtcptime = GST_CLOCK_TIME_NONE;
 250   src->stats.last_rtptime = GST_CLOCK_TIME_NONE;
 251   src->stats.last_rtcptime = GST_CLOCK_TIME_NONE;
 252   g_array_set_size (src->nacks, 0);
 253
 254   src->stats.sent_pli_count = 0;
 255   src->stats.sent_fir_count = 0;
 256 }
 257
 258 static void
 259 rtp_source_init (RTPSource * src)
 260 {
 261   /* sources are initialy on probation until we receive enough valid RTP
 262    * packets or a valid RTCP packet */
 263   src->validated = FALSE;
 264   src->internal = FALSE;
 265   src->probation = DEFAULT_PROBATION;
 266   src->curr_probation = src->probation;
 267   src->closing = FALSE;
 268
 269   src->sdes = gst_structure_new_empty ("application/x-rtp-source-sdes");
 270
 271   src->payload = -1;
 272   src->clock_rate = -1;
 273   src->packets = g_queue_new ();
 274   src->seqnum_offset = -1;
 275   src->last_rtptime = -1;
 276
 277   src->retained_feedback = g_queue_new ();
 278   src->nacks = g_array_new (FALSE, FALSE, sizeof (guint32));
 279
 280   src->reported_in_sr_of = g_hash_table_new (g_direct_hash, g_direct_equal);
 281
 282   rtp_source_reset (src);
 283 }
 284
 285 void
 286 rtp_conflicting_address_free (RTPConflictingAddress * addr)
 287 {
 288   g_object_unref (addr->address);
 289   g_slice_free (RTPConflictingAddress, addr);
 290 }
 291
 292 static void
 293 rtp_source_finalize (GObject * object)
 294 {
 295   RTPSource *src;
 296
 297   src = RTP_SOURCE_CAST (object);
 298
 299   g_queue_foreach (src->packets, (GFunc) gst_buffer_unref, NULL);
 300   g_queue_free (src->packets);
 301
 302   gst_structure_free (src->sdes);
 303
 304   g_free (src->bye_reason);
 305
 306   gst_caps_replace (&src->caps, NULL);
 307
 308   g_list_free_full (src->conflicting_addresses,
 309       (GDestroyNotify) rtp_conflicting_address_free);
 310   g_queue_foreach (src->retained_feedback, (GFunc) gst_buffer_unref, NULL);
 311   g_queue_free (src->retained_feedback);
 312
 313   g_array_free (src->nacks, TRUE);
 314
 315   if (src->rtp_from)
 316     g_object_unref (src->rtp_from);
 317   if (src->rtcp_from)
 318     g_object_unref (src->rtcp_from);
 319
 320   g_hash_table_unref (src->reported_in_sr_of);
 321
 322   G_OBJECT_CLASS (rtp_source_parent_class)->finalize (object);
 323 }
 324
 325 static GstStructure *
 326 rtp_source_create_stats (RTPSource * src)
 327 {
 328   GstStructure *s;
 329   gboolean is_sender = src->is_sender;
 330   gboolean internal = src->internal;
 331   gchar *address_str;
 332   gboolean have_rb;
 333   guint8 fractionlost = 0;
 334   gint32 packetslost = 0;
 335   guint32 exthighestseq = 0;
 336   guint32 jitter = 0;
 337   guint32 lsr = 0;
 338   guint32 dlsr = 0;
 339   guint32 round_trip = 0;
 340   gboolean have_sr;
 341   GstClockTime time = 0;
 342   guint64 ntptime = 0;
 343   guint32 rtptime = 0;
 344   guint32 packet_count = 0;
 345   guint32 octet_count = 0;
 346
 347
 348   /* common data for all types of sources */
 349   s = gst_structure_new ("application/x-rtp-source-stats",
 350       "ssrc", G_TYPE_UINT, (guint) src->ssrc,
 351       "internal", G_TYPE_BOOLEAN, internal,
 352       "validated", G_TYPE_BOOLEAN, src->validated,
 353       "received-bye", G_TYPE_BOOLEAN, src->marked_bye,
 354       "is-csrc", G_TYPE_BOOLEAN, src->is_csrc,
 355       "is-sender", G_TYPE_BOOLEAN, is_sender,
 356       "seqnum-base", G_TYPE_INT, src->seqnum_offset,
 357       "clock-rate", G_TYPE_INT, src->clock_rate, NULL);
 358
 359   /* add address and port */
 360   if (src->rtp_from) {
 361     address_str = __g_socket_address_to_string (src->rtp_from);
 362     gst_structure_set (s, "rtp-from", G_TYPE_STRING, address_str, NULL);
 363     g_free (address_str);
 364   }
 365   if (src->rtcp_from) {
 366     address_str = __g_socket_address_to_string (src->rtcp_from);
 367     gst_structure_set (s, "rtcp-from", G_TYPE_STRING, address_str, NULL);
 368     g_free (address_str);
 369   }
 370
 371   gst_structure_set (s,
 372       "octets-sent", G_TYPE_UINT64, src->stats.octets_sent,
 373       "packets-sent", G_TYPE_UINT64, src->stats.packets_sent,
 374       "octets-received", G_TYPE_UINT64, src->stats.octets_received,
 375       "packets-received", G_TYPE_UINT64, src->stats.packets_received,
 376       "bitrate", G_TYPE_UINT64, src->bitrate,
 377       "packets-lost", G_TYPE_INT,
 378       (gint) rtp_stats_get_packets_lost (&src->stats), "jitter", G_TYPE_UINT,
 379       (guint) (src->stats.jitter >> 4),
 380       "sent-pli-count", G_TYPE_UINT, src->stats.sent_pli_count,
 381       "recv-pli-count", G_TYPE_UINT, src->stats.recv_pli_count,
 382       "sent-fir-count", G_TYPE_UINT, src->stats.sent_fir_count,
 383       "recv-fir-count", G_TYPE_UINT, src->stats.recv_fir_count, NULL);
 384
 385   /* get the last SR. */
 386   have_sr = rtp_source_get_last_sr (src, &time, &ntptime, &rtptime,
 387       &packet_count, &octet_count);
 388   gst_structure_set (s,
 389       "have-sr", G_TYPE_BOOLEAN, have_sr,
 390       "sr-ntptime", G_TYPE_UINT64, ntptime,
 391       "sr-rtptime", G_TYPE_UINT, (guint) rtptime,
 392       "sr-octet-count", G_TYPE_UINT, (guint) octet_count,
 393       "sr-packet-count", G_TYPE_UINT, (guint) packet_count, NULL);
 394
 395   if (!internal) {
 396     /* get the last RB we sent */
 397     gst_structure_set (s,
 398         "sent-rb", G_TYPE_BOOLEAN, src->last_rr.is_valid,
 399         "sent-rb-fractionlost", G_TYPE_UINT, (guint) src->last_rr.fractionlost,
 400         "sent-rb-packetslost", G_TYPE_INT, (gint) src->last_rr.packetslost,
 401         "sent-rb-exthighestseq", G_TYPE_UINT,
 402         (guint) src->last_rr.exthighestseq, "sent-rb-jitter", G_TYPE_UINT,
 403         (guint) src->last_rr.jitter, "sent-rb-lsr", G_TYPE_UINT,
 404         (guint) src->last_rr.lsr, "sent-rb-dlsr", G_TYPE_UINT,
 405         (guint) src->last_rr.dlsr, NULL);
 406
 407     /* get the last RB */
 408     have_rb = rtp_source_get_last_rb (src, &fractionlost, &packetslost,
 409         &exthighestseq, &jitter, &lsr, &dlsr, &round_trip);
 410
 411     gst_structure_set (s,
 412         "have-rb", G_TYPE_BOOLEAN, have_rb,
 413         "rb-fractionlost", G_TYPE_UINT, (guint) fractionlost,
 414         "rb-packetslost", G_TYPE_INT, (gint) packetslost,
 415         "rb-exthighestseq", G_TYPE_UINT, (guint) exthighestseq,
 416         "rb-jitter", G_TYPE_UINT, (guint) jitter,
 417         "rb-lsr", G_TYPE_UINT, (guint) lsr,
 418         "rb-dlsr", G_TYPE_UINT, (guint) dlsr,
 419         "rb-round-trip", G_TYPE_UINT, (guint) round_trip, NULL);
 420   }
 421
 422   return s;
 423 }
 424
 425 /**
 426  * rtp_source_get_sdes_struct:
 427  * @src: an #RTPSource
 428  *
 429  * Get the SDES from @src. See the SDES property for more details.
 430  *
 431  * Returns: %GstStructure of type "application/x-rtp-source-sdes". The result is
 432  * valid until the SDES items of @src are modified.
 433  */
 434 const GstStructure *
 435 rtp_source_get_sdes_struct (RTPSource * src)
 436 {
 437   g_return_val_if_fail (RTP_IS_SOURCE (src), NULL);
 438
 439   return src->sdes;
 440 }
 441
 442 static gboolean
 443 sdes_struct_compare_func (GQuark field_id, const GValue * value,
 444     gpointer user_data)
 445 {
 446   GstStructure *old;
 447   const gchar *field;
 448
 449   old = GST_STRUCTURE (user_data);
 450   field = g_quark_to_string (field_id);
 451
 452   if (!gst_structure_has_field (old, field))
 453     return FALSE;
 454
 455   g_assert (G_VALUE_HOLDS_STRING (value));
 456
 457   return strcmp (g_value_get_string (value), gst_structure_get_string (old,
 458           field)) == 0;
 459 }
 460
 461 /**
 462  * rtp_source_set_sdes_struct:
 463  * @src: an #RTPSource
 464  * @sdes: the SDES structure
 465  *
 466  * Store the @sdes in @src. @sdes must be a structure of type
 467  * "application/x-rtp-source-sdes", see the SDES property for more details.
 468  *
 469  * This function takes ownership of @sdes.
 470  *
 471  * Returns: %FALSE if the SDES was unchanged.
 472  */
 473 gboolean
 474 rtp_source_set_sdes_struct (RTPSource * src, GstStructure * sdes)
 475 {
 476   gboolean changed;
 477
 478   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 479   g_return_val_if_fail (strcmp (gst_structure_get_name (sdes),
 480           "application/x-rtp-source-sdes") == 0, FALSE);
 481
 482   changed = !gst_structure_foreach (sdes, sdes_struct_compare_func, src->sdes);
 483
 484   if (changed) {
 485     gst_structure_free (src->sdes);
 486     src->sdes = sdes;
 487   } else {
 488     gst_structure_free (sdes);
 489   }
 490   return changed;
 491 }
 492
 493 static void
 494 rtp_source_set_property (GObject * object, guint prop_id,
 495     const GValue * value, GParamSpec * pspec)
 496 {
 497   RTPSource *src;
 498
 499   src = RTP_SOURCE (object);
 500
 501   switch (prop_id) {
 502     case PROP_SSRC:
 503       src->ssrc = g_value_get_uint (value);
 504       break;
 505     case PROP_PROBATION:
 506       src->probation = g_value_get_uint (value);
 507       break;
 508     default:
 509       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 510       break;
 511   }
 512 }
 513
 514 static void
 515 rtp_source_get_property (GObject * object, guint prop_id,
 516     GValue * value, GParamSpec * pspec)
 517 {
 518   RTPSource *src;
 519
 520   src = RTP_SOURCE (object);
 521
 522   switch (prop_id) {
 523     case PROP_SSRC:
 524       g_value_set_uint (value, rtp_source_get_ssrc (src));
 525       break;
 526     case PROP_IS_CSRC:
 527       g_value_set_boolean (value, rtp_source_is_as_csrc (src));
 528       break;
 529     case PROP_IS_VALIDATED:
 530       g_value_set_boolean (value, rtp_source_is_validated (src));
 531       break;
 532     case PROP_IS_SENDER:
 533       g_value_set_boolean (value, rtp_source_is_sender (src));
 534       break;
 535     case PROP_SDES:
 536       g_value_set_boxed (value, rtp_source_get_sdes_struct (src));
 537       break;
 538     case PROP_STATS:
 539       g_value_take_boxed (value, rtp_source_create_stats (src));
 540       break;
 541     case PROP_PROBATION:
 542       g_value_set_uint (value, src->probation);
 543       break;
 544     default:
 545       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 546       break;
 547   }
 548 }
 549
 550 /**
 551  * rtp_source_new:
 552  * @ssrc: an SSRC
 553  *
 554  * Create a #RTPSource with @ssrc.
 555  *
 556  * Returns: a new #RTPSource. Use g_object_unref() after usage.
 557  */
 558 RTPSource *
 559 rtp_source_new (guint32 ssrc)
 560 {
 561   RTPSource *src;
 562
 563   src = g_object_new (RTP_TYPE_SOURCE, NULL);
 564   src->ssrc = ssrc;
 565
 566   return src;
 567 }
 568
 569 /**
 570  * rtp_source_set_callbacks:
 571  * @src: an #RTPSource
 572  * @cb: callback functions
 573  * @user_data: user data
 574  *
 575  * Set the callbacks for the source.
 576  */
 577 void
 578 rtp_source_set_callbacks (RTPSource * src, RTPSourceCallbacks * cb,
 579     gpointer user_data)
 580 {
 581   g_return_if_fail (RTP_IS_SOURCE (src));
 582
 583   src->callbacks.push_rtp = cb->push_rtp;
 584   src->callbacks.clock_rate = cb->clock_rate;
 585   src->user_data = user_data;
 586 }
 587
 588 /**
 589  * rtp_source_get_ssrc:
 590  * @src: an #RTPSource
 591  *
 592  * Get the SSRC of @source.
 593  *
 594  * Returns: the SSRC of src.
 595  */
 596 guint32
 597 rtp_source_get_ssrc (RTPSource * src)
 598 {
 599   guint32 result;
 600
 601   g_return_val_if_fail (RTP_IS_SOURCE (src), 0);
 602
 603   result = src->ssrc;
 604
 605   return result;
 606 }
 607
 608 /**
 609  * rtp_source_set_as_csrc:
 610  * @src: an #RTPSource
 611  *
 612  * Configure @src as a CSRC, this will also validate @src.
 613  */
 614 void
 615 rtp_source_set_as_csrc (RTPSource * src)
 616 {
 617   g_return_if_fail (RTP_IS_SOURCE (src));
 618
 619   src->validated = TRUE;
 620   src->is_csrc = TRUE;
 621 }
 622
 623 /**
 624  * rtp_source_is_as_csrc:
 625  * @src: an #RTPSource
 626  *
 627  * Check if @src is a contributing source.
 628  *
 629  * Returns: %TRUE if @src is acting as a contributing source.
 630  */
 631 gboolean
 632 rtp_source_is_as_csrc (RTPSource * src)
 633 {
 634   gboolean result;
 635
 636   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 637
 638   result = src->is_csrc;
 639
 640   return result;
 641 }
 642
 643 /**
 644  * rtp_source_is_active:
 645  * @src: an #RTPSource
 646  *
 647  * Check if @src is an active source. A source is active if it has been
 648  * validated and has not yet received a BYE packet
 649  *
 650  * Returns: %TRUE if @src is an qactive source.
 651  */
 652 gboolean
 653 rtp_source_is_active (RTPSource * src)
 654 {
 655   gboolean result;
 656
 657   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 658
 659   result = RTP_SOURCE_IS_ACTIVE (src);
 660
 661   return result;
 662 }
 663
 664 /**
 665  * rtp_source_is_validated:
 666  * @src: an #RTPSource
 667  *
 668  * Check if @src is a validated source.
 669  *
 670  * Returns: %TRUE if @src is a validated source.
 671  */
 672 gboolean
 673 rtp_source_is_validated (RTPSource * src)
 674 {
 675   gboolean result;
 676
 677   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 678
 679   result = src->validated;
 680
 681   return result;
 682 }
 683
 684 /**
 685  * rtp_source_is_sender:
 686  * @src: an #RTPSource
 687  *
 688  * Check if @src is a sending source.
 689  *
 690  * Returns: %TRUE if @src is a sending source.
 691  */
 692 gboolean
 693 rtp_source_is_sender (RTPSource * src)
 694 {
 695   gboolean result;
 696
 697   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 698
 699   result = RTP_SOURCE_IS_SENDER (src);
 700
 701   return result;
 702 }
 703
 704 /**
 705  * rtp_source_is_marked_bye:
 706  * @src: an #RTPSource
 707  *
 708  * Check if @src is marked as leaving the session with a BYE packet.
 709  *
 710  * Returns: %TRUE if @src has been marked BYE.
 711  */
 712 gboolean
 713 rtp_source_is_marked_bye (RTPSource * src)
 714 {
 715   gboolean result;
 716
 717   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 718
 719   result = RTP_SOURCE_IS_MARKED_BYE (src);
 720
 721   return result;
 722 }
 723
 724
 725 /**
 726  * rtp_source_get_bye_reason:
 727  * @src: an #RTPSource
 728  *
 729  * Get the BYE reason for @src. Check if the source is marked as leaving the
 730  * session with a BYE message first with rtp_source_is_marked_bye().
 731  *
 732  * Returns: The BYE reason or NULL when no reason was given or the source was
 733  * not marked BYE yet. g_free() after usage.
 734  */
 735 gchar *
 736 rtp_source_get_bye_reason (RTPSource * src)
 737 {
 738   gchar *result;
 739
 740   g_return_val_if_fail (RTP_IS_SOURCE (src), NULL);
 741
 742   result = g_strdup (src->bye_reason);
 743
 744   return result;
 745 }
 746
 747 /**
 748  * rtp_source_update_caps:
 749  * @src: an #RTPSource
 750  * @caps: a #GstCaps
 751  *
 752  * Parse @caps and store all relevant information in @source.
 753  */
 754 void
 755 rtp_source_update_caps (RTPSource * src, GstCaps * caps)
 756 {
 757   GstStructure *s;
 758   guint val;
 759   gint ival;
 760   gboolean rtx;
 761
 762   /* nothing changed, return */
 763   if (caps == NULL || src->caps == caps)
 764     return;
 765
 766   s = gst_caps_get_structure (caps, 0);
 767
 768   rtx = (gst_structure_get_uint (s, "rtx-ssrc", &val) && val == src->ssrc);
 769
 770   if (gst_structure_get_int (s, rtx ? "rtx-payload" : "payload", &ival))
 771     src->payload = ival;
 772   else
 773     src->payload = -1;
 774
 775   GST_DEBUG ("got %spayload %d", rtx ? "rtx " : "", src->payload);
 776
 777   if (gst_structure_get_int (s, "clock-rate", &ival))
 778     src->clock_rate = ival;
 779   else
 780     src->clock_rate = -1;
 781
 782   GST_DEBUG ("got clock-rate %d", src->clock_rate);
 783
 784   if (gst_structure_get_uint (s, rtx ? "rtx-seqnum-offset" : "seqnum-offset",
 785           &val))
 786     src->seqnum_offset = val;
 787   else
 788     src->seqnum_offset = -1;
 789
 790   GST_DEBUG ("got %sseqnum-offset %" G_GINT32_FORMAT, rtx ? "rtx " : "",
 791       src->seqnum_offset);
 792
 793   gst_caps_replace (&src->caps, caps);
 794 }
 795
 796 /**
 797  * rtp_source_set_rtp_from:
 798  * @src: an #RTPSource
 799  * @address: the RTP address to set
 800  *
 801  * Set that @src is receiving RTP packets from @address. This is used for
 802  * collistion checking.
 803  */
 804 void
 805 rtp_source_set_rtp_from (RTPSource * src, GSocketAddress * address)
 806 {
 807   g_return_if_fail (RTP_IS_SOURCE (src));
 808
 809   if (src->rtp_from)
 810     g_object_unref (src->rtp_from);
 811   src->rtp_from = G_SOCKET_ADDRESS (g_object_ref (address));
 812 }
 813
 814 /**
 815  * rtp_source_set_rtcp_from:
 816  * @src: an #RTPSource
 817  * @address: the RTCP address to set
 818  *
 819  * Set that @src is receiving RTCP packets from @address. This is used for
 820  * collistion checking.
 821  */
 822 void
 823 rtp_source_set_rtcp_from (RTPSource * src, GSocketAddress * address)
 824 {
 825   g_return_if_fail (RTP_IS_SOURCE (src));
 826
 827   if (src->rtcp_from)
 828     g_object_unref (src->rtcp_from);
 829   src->rtcp_from = G_SOCKET_ADDRESS (g_object_ref (address));
 830 }
 831
 832 static GstFlowReturn
 833 push_packet (RTPSource * src, GstBuffer * buffer)
 834 {
 835   GstFlowReturn ret = GST_FLOW_OK;
 836
 837   /* push queued packets first if any */
 838   while (!g_queue_is_empty (src->packets)) {
 839     GstBuffer *buffer = GST_BUFFER_CAST (g_queue_pop_head (src->packets));
 840
 841     GST_LOG ("pushing queued packet");
 842     if (src->callbacks.push_rtp)
 843       src->callbacks.push_rtp (src, buffer, src->user_data);
 844     else
 845       gst_buffer_unref (buffer);
 846   }
 847   GST_LOG ("pushing new packet");
 848   /* push packet */
 849   if (src->callbacks.push_rtp)
 850     ret = src->callbacks.push_rtp (src, buffer, src->user_data);
 851   else
 852     gst_buffer_unref (buffer);
 853
 854   return ret;
 855 }
 856
 857 static gint
 858 get_clock_rate (RTPSource * src, guint8 payload)
 859 {
 860   if (src->payload == -1) {
 861     /* first payload received, nothing was in the caps, lock on to this payload */
 862     src->payload = payload;
 863     GST_DEBUG ("first payload %d", payload);
 864   } else if (payload != src->payload) {
 865     /* we have a different payload than before, reset the clock-rate */
 866     GST_DEBUG ("new payload %d", payload);
 867     src->payload = payload;
 868     src->clock_rate = -1;
 869     src->stats.transit = -1;
 870   }
 871
 872   if (src->clock_rate == -1) {
 873     gint clock_rate = -1;
 874
 875     if (src->callbacks.clock_rate)
 876       clock_rate = src->callbacks.clock_rate (src, payload, src->user_data);
 877
 878     GST_DEBUG ("got clock-rate %d", clock_rate);
 879
 880     src->clock_rate = clock_rate;
 881   }
 882   return src->clock_rate;
 883 }
 884
 885 /* Jitter is the variation in the delay of received packets in a flow. It is
 886  * measured by comparing the interval when RTP packets were sent to the interval
 887  * at which they were received. For instance, if packet #1 and packet #2 leave
 888  * 50 milliseconds apart and arrive 60 milliseconds apart, then the jitter is 10
 889  * milliseconds. */
 890 static void
 891 calculate_jitter (RTPSource * src, RTPPacketInfo * pinfo)
 892 {
 893   GstClockTime running_time;
 894   guint32 rtparrival, transit, rtptime;
 895   gint32 diff;
 896   gint clock_rate;
 897   guint8 pt;
 898
 899   /* get arrival time */
 900   if ((running_time = pinfo->running_time) == GST_CLOCK_TIME_NONE)
 901     goto no_time;
 902
 903   pt = pinfo->pt;
 904
 905   GST_LOG ("SSRC %08x got payload %d", src->ssrc, pt);
 906
 907   /* get clockrate */
 908   if ((clock_rate = get_clock_rate (src, pt)) == -1)
 909     goto no_clock_rate;
 910
 911   rtptime = pinfo->rtptime;
 912
 913   /* convert arrival time to RTP timestamp units, truncate to 32 bits, we don't
 914    * care about the absolute value, just the difference. */
 915   rtparrival = gst_util_uint64_scale_int (running_time, clock_rate, GST_SECOND);
 916
 917   /* transit time is difference with RTP timestamp */
 918   transit = rtparrival - rtptime;
 919
 920   /* get ABS diff with previous transit time */
 921   if (src->stats.transit != -1) {
 922     if (transit > src->stats.transit)
 923       diff = transit - src->stats.transit;
 924     else
 925       diff = src->stats.transit - transit;
 926   } else
 927     diff = 0;
 928
 929   src->stats.transit = transit;
 930
 931   /* update jitter, the value we store is scaled up so we can keep precision. */
 932   src->stats.jitter += diff - ((src->stats.jitter + 8) >> 4);
 933
 934   src->stats.prev_rtptime = src->stats.last_rtptime;
 935   src->stats.last_rtptime = rtparrival;
 936
 937   GST_LOG ("rtparrival %u, rtptime %u, clock-rate %d, diff %d, jitter: %f",
 938       rtparrival, rtptime, clock_rate, diff, (src->stats.jitter) / 16.0);
 939
 940   return;
 941
 942   /* ERRORS */
 943 no_time:
 944   {
 945     GST_WARNING ("cannot get current running_time");
 946     return;
 947   }
 948 no_clock_rate:
 949   {
 950     GST_WARNING ("cannot get clock-rate for pt %d", pt);
 951     return;
 952   }
 953 }
 954
 955 static void
 956 init_seq (RTPSource * src, guint16 seq)
 957 {
 958   src->stats.base_seq = seq;
 959   src->stats.max_seq = seq;
 960   src->stats.bad_seq = RTP_SEQ_MOD + 1; /* so seq == bad_seq is false */
 961   src->stats.cycles = 0;
 962   src->stats.packets_received = 0;
 963   src->stats.octets_received = 0;
 964   src->stats.bytes_received = 0;
 965   src->stats.prev_received = 0;
 966   src->stats.prev_expected = 0;
 967   src->stats.recv_pli_count = 0;
 968   src->stats.recv_fir_count = 0;
 969
 970   GST_DEBUG ("base_seq %d", seq);
 971 }
 972
 973 #define BITRATE_INTERVAL (2 * GST_SECOND)
 974
 975 static void
 976 do_bitrate_estimation (RTPSource * src, GstClockTime running_time,
 977     guint64 * bytes_handled)
 978 {
 979   guint64 elapsed;
 980
 981   if (src->prev_rtime) {
 982     elapsed = running_time - src->prev_rtime;
 983
 984     if (elapsed > BITRATE_INTERVAL) {
 985       guint64 rate;
 986
 987       rate = gst_util_uint64_scale (*bytes_handled, 8 * GST_SECOND, elapsed);
 988
 989       GST_LOG ("Elapsed %" G_GUINT64_FORMAT ", bytes %" G_GUINT64_FORMAT
 990           ", rate %" G_GUINT64_FORMAT, elapsed, *bytes_handled, rate);
 991
 992       if (src->bitrate == 0)
 993         src->bitrate = rate;
 994       else
 995         src->bitrate = ((src->bitrate * 3) + rate) / 4;
 996
 997       src->prev_rtime = running_time;
 998       *bytes_handled = 0;
 999     }
1000   } else {
1001     GST_LOG ("Reset bitrate measurement");
1002     src->prev_rtime = running_time;
1003     src->bitrate = 0;
1004   }
1005 }
1006
1007 static gboolean
1008 update_receiver_stats (RTPSource * src, RTPPacketInfo * pinfo)
1009 {
1010   guint16 seqnr, expected;
1011   RTPSourceStats *stats;
1012   gint16 delta;
1013
1014   stats = &src->stats;
1015
1016   seqnr = pinfo->seqnum;
1017
1018   if (stats->cycles == -1) {
1019     GST_DEBUG ("received first packet");
1020     /* first time we heard of this source */
1021     init_seq (src, seqnr);
1022     src->stats.max_seq = seqnr - 1;
1023     src->curr_probation = src->probation;
1024   }
1025
1026   expected = src->stats.max_seq + 1;
1027   delta = gst_rtp_buffer_compare_seqnum (expected, seqnr);
1028
1029   /* if we are still on probation, check seqnum */
1030   if (src->curr_probation) {
1031     /* when in probation, we require consecutive seqnums */
1032     if (delta == 0) {
1033       /* expected packet */
1034       GST_DEBUG ("probation: seqnr %d == expected %d", seqnr, expected);
1035       src->curr_probation--;
1036       if (seqnr < stats->max_seq) {
1037         /* sequence number wrapped - count another 64K cycle. */
1038         stats->cycles += RTP_SEQ_MOD;
1039       }
1040       src->stats.max_seq = seqnr;
1041
1042       if (src->curr_probation == 0) {
1043         GST_DEBUG ("probation done!");
1044         init_seq (src, seqnr);
1045       } else {
1046         GstBuffer *q;
1047
1048         GST_DEBUG ("probation %d: queue packet", src->curr_probation);
1049         /* when still in probation, keep packets in a list. */
1050         g_queue_push_tail (src->packets, pinfo->data);
1051         pinfo->data = NULL;
1052         /* remove packets from queue if there are too many */
1053         while (g_queue_get_length (src->packets) > RTP_MAX_PROBATION_LEN) {
1054           q = g_queue_pop_head (src->packets);
1055           gst_buffer_unref (q);
1056         }
1057         goto done;
1058       }
1059     } else {
1060       /* unexpected seqnum in probation */
1061       goto probation_seqnum;
1062     }
1063   } else if (delta >= 0 && delta < RTP_MAX_DROPOUT) {
1064     /* Clear bad packets */
1065     stats->bad_seq = RTP_SEQ_MOD + 1;   /* so seq == bad_seq is false */
1066     g_queue_foreach (src->packets, (GFunc) gst_buffer_unref, NULL);
1067     g_queue_clear (src->packets);
1068
1069     /* in order, with permissible gap */
1070     if (seqnr < stats->max_seq) {
1071       /* sequence number wrapped - count another 64K cycle. */
1072       stats->cycles += RTP_SEQ_MOD;
1073     }
1074     stats->max_seq = seqnr;
1075   } else if (delta < -RTP_MAX_MISORDER || delta >= RTP_MAX_DROPOUT) {
1076     /* the sequence number made a very large jump */
1077     if (seqnr == stats->bad_seq && src->packets->head) {
1078       /* two sequential packets -- assume that the other side
1079        * restarted without telling us so just re-sync
1080        * (i.e., pretend this was the first packet).  */
1081       init_seq (src, seqnr);
1082     } else {
1083       /* unacceptable jump */
1084       stats->bad_seq = (seqnr + 1) & (RTP_SEQ_MOD - 1);
1085       g_queue_foreach (src->packets, (GFunc) gst_buffer_unref, NULL);
1086       g_queue_clear (src->packets);
1087       g_queue_push_tail (src->packets, pinfo->data);
1088       pinfo->data = NULL;
1089       goto bad_sequence;
1090     }
1091   } else {                      /* delta < 0 && delta >= -RTP_MAX_MISORDER */
1092     /* Clear bad packets */
1093     stats->bad_seq = RTP_SEQ_MOD + 1;   /* so seq == bad_seq is false */
1094     g_queue_foreach (src->packets, (GFunc) gst_buffer_unref, NULL);
1095     g_queue_clear (src->packets);
1096
1097     /* duplicate or reordered packet, will be filtered by jitterbuffer. */
1098     GST_WARNING ("duplicate or reordered packet (seqnr %u, expected %u)", seqnr,
1099         expected);
1100   }
1101
1102   src->stats.octets_received += pinfo->payload_len;
1103   src->stats.bytes_received += pinfo->bytes;
1104   src->stats.packets_received++;
1105   /* for the bitrate estimation */
1106   src->bytes_received += pinfo->payload_len;
1107
1108   GST_LOG ("seq %u, PC: %" G_GUINT64_FORMAT ", OC: %" G_GUINT64_FORMAT,
1109       seqnr, src->stats.packets_received, src->stats.octets_received);
1110
1111   return TRUE;
1112
1113   /* ERRORS */
1114 done:
1115   {
1116     return FALSE;
1117   }
1118 bad_sequence:
1119   {
1120     GST_WARNING ("unacceptable seqnum received");
1121     return FALSE;
1122   }
1123 probation_seqnum:
1124   {
1125     GST_WARNING ("probation: seqnr %d != expected %d", seqnr, expected);
1126     src->curr_probation = src->probation;
1127     src->stats.max_seq = seqnr;
1128     return FALSE;
1129   }
1130 }
1131
1132 /**
1133  * rtp_source_process_rtp:
1134  * @src: an #RTPSource
1135  * @pinfo: an #RTPPacketInfo
1136  *
1137  * Let @src handle the incomming RTP packet described in @pinfo.
1138  *
1139  * Returns: a #GstFlowReturn.
1140  */
1141 GstFlowReturn
1142 rtp_source_process_rtp (RTPSource * src, RTPPacketInfo * pinfo)
1143 {
1144   GstFlowReturn result;
1145
1146   g_return_val_if_fail (RTP_IS_SOURCE (src), GST_FLOW_ERROR);
1147   g_return_val_if_fail (pinfo != NULL, GST_FLOW_ERROR);
1148
1149   if (!update_receiver_stats (src, pinfo))
1150     return GST_FLOW_OK;
1151
1152   /* the source that sent the packet must be a sender */
1153   src->is_sender = TRUE;
1154   src->validated = TRUE;
1155
1156   do_bitrate_estimation (src, pinfo->running_time, &src->bytes_received);
1157
1158   /* calculate jitter for the stats */
1159   calculate_jitter (src, pinfo);
1160
1161   /* we're ready to push the RTP packet now */
1162   result = push_packet (src, pinfo->data);
1163   pinfo->data = NULL;
1164
1165   return result;
1166 }
1167
1168 /**
1169  * rtp_source_mark_bye:
1170  * @src: an #RTPSource
1171  * @reason: the reason for leaving
1172  *
1173  * Mark @src in the BYE state. This can happen when the source wants to
1174  * leave the sesssion or when a BYE packets has been received.
1175  *
1176  * This will make the source inactive.
1177  */
1178 void
1179 rtp_source_mark_bye (RTPSource * src, const gchar * reason)
1180 {
1181   g_return_if_fail (RTP_IS_SOURCE (src));
1182
1183   GST_DEBUG ("marking SSRC %08x as BYE, reason: %s", src->ssrc,
1184       GST_STR_NULL (reason));
1185
1186   /* copy the reason and mark as bye */
1187   g_free (src->bye_reason);
1188   src->bye_reason = g_strdup (reason);
1189   src->marked_bye = TRUE;
1190 }
1191
1192 /**
1193  * rtp_source_send_rtp:
1194  * @src: an #RTPSource
1195  * @data: an RTP buffer or a list of RTP buffers
1196  * @is_list: if @data is a buffer or list
1197  * @running_time: the running time of @data
1198  *
1199  * Send @data (an RTP buffer or list of buffers) originating from @src.
1200  * This will make @src a sender. This function takes ownership of @data and
1201  * modifies the SSRC in the RTP packet to that of @src when needed.
1202  *
1203  * Returns: a #GstFlowReturn.
1204  */
1205 GstFlowReturn
1206 rtp_source_send_rtp (RTPSource * src, RTPPacketInfo * pinfo)
1207 {
1208   GstFlowReturn result;
1209   GstClockTime running_time;
1210   guint32 rtptime;
1211   guint64 ext_rtptime;
1212   guint64 rt_diff, rtp_diff;
1213
1214   g_return_val_if_fail (RTP_IS_SOURCE (src), GST_FLOW_ERROR);
1215
1216   /* we are a sender now */
1217   src->is_sender = TRUE;
1218
1219   /* we are also a receiver of our packets */
1220   if (!update_receiver_stats (src, pinfo))
1221     return GST_FLOW_OK;
1222
1223   /* update stats for the SR */
1224   src->stats.packets_sent += pinfo->packets;
1225   src->stats.octets_sent += pinfo->payload_len;
1226   src->bytes_sent += pinfo->payload_len;
1227
1228   running_time = pinfo->running_time;
1229
1230   do_bitrate_estimation (src, running_time, &src->bytes_sent);
1231
1232   rtptime = pinfo->rtptime;
1233
1234   ext_rtptime = src->last_rtptime;
1235   ext_rtptime = gst_rtp_buffer_ext_timestamp (&ext_rtptime, rtptime);
1236
1237   GST_LOG ("SSRC %08x, RTP %" G_GUINT64_FORMAT ", running_time %"
1238       GST_TIME_FORMAT, src->ssrc, ext_rtptime, GST_TIME_ARGS (running_time));
1239
1240   if (ext_rtptime > src->last_rtptime) {
1241     rtp_diff = ext_rtptime - src->last_rtptime;
1242     rt_diff = running_time - src->last_rtime;
1243
1244     /* calc the diff so we can detect drift at the sender. This can also be used
1245      * to guestimate the clock rate if the NTP time is locked to the RTP
1246      * timestamps (as is the case when the capture device is providing the clock). */
1247     GST_LOG ("SSRC %08x, diff RTP %" G_GUINT64_FORMAT ", diff running_time %"
1248         GST_TIME_FORMAT, src->ssrc, rtp_diff, GST_TIME_ARGS (rt_diff));
1249   }
1250
1251   /* we keep track of the last received RTP timestamp and the corresponding
1252    * buffer running_time so that we can use this info when constructing SR reports */
1253   src->last_rtime = running_time;
1254   src->last_rtptime = ext_rtptime;
1255
1256   /* push packet */
1257   if (!src->callbacks.push_rtp)
1258     goto no_callback;
1259
1260   GST_LOG ("pushing RTP %s %" G_GUINT64_FORMAT,
1261       pinfo->is_list ? "list" : "packet", src->stats.packets_sent);
1262
1263   result = src->callbacks.push_rtp (src, pinfo->data, src->user_data);
1264   pinfo->data = NULL;
1265
1266   return result;
1267
1268   /* ERRORS */
1269 no_callback:
1270   {
1271     GST_WARNING ("no callback installed, dropping packet");
1272     return GST_FLOW_OK;
1273   }
1274 }
1275
1276 /**
1277  * rtp_source_process_sr:
1278  * @src: an #RTPSource
1279  * @time: time of packet arrival
1280  * @ntptime: the NTP time (in NTP Timestamp Format, 32.32 fixed point)
1281  * @rtptime: the RTP time (in clock rate units)
1282  * @packet_count: the packet count
1283  * @octet_count: the octet count
1284  *
1285  * Update the sender report in @src.
1286  */
1287 void
1288 rtp_source_process_sr (RTPSource * src, GstClockTime time, guint64 ntptime,
1289     guint32 rtptime, guint32 packet_count, guint32 octet_count)
1290 {
1291   RTPSenderReport *curr;
1292   gint curridx;
1293
1294   g_return_if_fail (RTP_IS_SOURCE (src));
1295
1296   GST_DEBUG ("got SR packet: SSRC %08x, NTP %08x:%08x, RTP %" G_GUINT32_FORMAT
1297       ", PC %" G_GUINT32_FORMAT ", OC %" G_GUINT32_FORMAT, src->ssrc,
1298       (guint32) (ntptime >> 32), (guint32) (ntptime & 0xffffffff), rtptime,
1299       packet_count, octet_count);
1300
1301   curridx = src->stats.curr_sr ^ 1;
1302   curr = &src->stats.sr[curridx];
1303
1304   /* this is a sender now */
1305   src->is_sender = TRUE;
1306
1307   /* update current */
1308   curr->is_valid = TRUE;
1309   curr->ntptime = ntptime;
1310   curr->rtptime = rtptime;
1311   curr->packet_count = packet_count;
1312   curr->octet_count = octet_count;
1313   curr->time = time;
1314
1315   /* make current */
1316   src->stats.curr_sr = curridx;
1317
1318   src->stats.prev_rtcptime = src->stats.last_rtcptime;
1319   src->stats.last_rtcptime = time;
1320 }
1321
1322 /**
1323  * rtp_source_process_rb:
1324  * @src: an #RTPSource
1325  * @ntpnstime: the current time in nanoseconds since 1970
1326  * @fractionlost: fraction lost since last SR/RR
1327  * @packetslost: the cumulative number of packets lost
1328  * @exthighestseq: the extended last sequence number received
1329  * @jitter: the interarrival jitter (in clock rate units)
1330  * @lsr: the time of the last SR packet on this source
1331  *   (in NTP Short Format, 16.16 fixed point)
1332  * @dlsr: the delay since the last SR packet
1333  *   (in NTP Short Format, 16.16 fixed point)
1334  *
1335  * Update the report block in @src.
1336  */
1337 void
1338 rtp_source_process_rb (RTPSource * src, guint64 ntpnstime,
1339     guint8 fractionlost, gint32 packetslost, guint32 exthighestseq,
1340     guint32 jitter, guint32 lsr, guint32 dlsr)
1341 {
1342   RTPReceiverReport *curr;
1343   gint curridx;
1344   guint32 ntp, A;
1345   guint64 f_ntp;
1346
1347   g_return_if_fail (RTP_IS_SOURCE (src));
1348
1349   GST_DEBUG ("got RB packet: SSRC %08x, FL %2x, PL %d, HS %" G_GUINT32_FORMAT
1350       ", jitter %" G_GUINT32_FORMAT ", LSR %04x:%04x, DLSR %04x:%04x",
1351       src->ssrc, fractionlost, packetslost, exthighestseq, jitter, lsr >> 16,
1352       lsr & 0xffff, dlsr >> 16, dlsr & 0xffff);
1353
1354   curridx = src->stats.curr_rr ^ 1;
1355   curr = &src->stats.rr[curridx];
1356
1357   /* update current */
1358   curr->is_valid = TRUE;
1359   curr->fractionlost = fractionlost;
1360   curr->packetslost = packetslost;
1361   curr->exthighestseq = exthighestseq;
1362   curr->jitter = jitter;
1363   curr->lsr = lsr;
1364   curr->dlsr = dlsr;
1365
1366   /* convert the NTP time in nanoseconds to 32.32 fixed point */
1367   f_ntp = gst_util_uint64_scale (ntpnstime, (1LL << 32), GST_SECOND);
1368   /* calculate round trip, round the time up */
1369   ntp = ((f_ntp + 0xffff) >> 16) & 0xffffffff;
1370
1371   A = dlsr + lsr;
1372   if (A > 0 && ntp > A)
1373     A = ntp - A;
1374   else
1375     A = 0;
1376   curr->round_trip = A;
1377
1378   GST_DEBUG ("NTP %04x:%04x, round trip %04x:%04x", ntp >> 16, ntp & 0xffff,
1379       A >> 16, A & 0xffff);
1380
1381   /* make current */
1382   src->stats.curr_rr = curridx;
1383 }
1384
1385 /**
1386  * rtp_source_get_new_sr:
1387  * @src: an #RTPSource
1388  * @ntpnstime: the current time in nanoseconds since 1970
1389  * @running_time: the current running_time of the pipeline
1390  * @ntptime: the NTP time (in NTP Timestamp Format, 32.32 fixed point)
1391  * @rtptime: the RTP time corresponding to @ntptime (in clock rate units)
1392  * @packet_count: the packet count
1393  * @octet_count: the octet count
1394  *
1395  * Get new values to put into a new SR report from this source.
1396  *
1397  * @running_time and @ntpnstime are captured at the same time and represent the
1398  * running time of the pipeline clock and the absolute current system time in
1399  * nanoseconds respectively. Together with the last running_time and RTP timestamp
1400  * we have observed in the source, we can generate @ntptime and @rtptime for an SR
1401  * packet. @ntptime is basically the fixed point representation of @ntpnstime
1402  * and @rtptime the associated RTP timestamp.
1403  *
1404  * Returns: %TRUE on success.
1405  */
1406 gboolean
1407 rtp_source_get_new_sr (RTPSource * src, guint64 ntpnstime,
1408     GstClockTime running_time, guint64 * ntptime, guint32 * rtptime,
1409     guint32 * packet_count, guint32 * octet_count)
1410 {
1411   guint64 t_rtp;
1412   guint64 t_current_ntp;
1413   GstClockTimeDiff diff;
1414
1415   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
1416
1417   /* We last saw a buffer with last_rtptime at last_rtime. Given a running_time
1418    * and an NTP time, we can scale the RTP timestamps so that they match the
1419    * given NTP time.  for scaling, we assume that the slope of the rtptime vs
1420    * running_time vs ntptime curve is close to 1, which is certainly
1421    * sufficient for the frequency at which we report SR and the rate we send
1422    * out RTP packets. */
1423   t_rtp = src->last_rtptime;
1424
1425   GST_DEBUG ("last_rtime %" GST_TIME_FORMAT ", last_rtptime %"
1426       G_GUINT64_FORMAT, GST_TIME_ARGS (src->last_rtime), t_rtp);
1427
1428   if (src->clock_rate != -1) {
1429     /* get the diff between the clock running_time and the buffer running_time.
1430      * This is the elapsed time, as measured against the pipeline clock, between
1431      * when the rtp timestamp was observed and the current running_time.
1432      *
1433      * We need to apply this diff to the RTP timestamp to get the RTP timestamp
1434      * for the given ntpnstime. */
1435     diff = GST_CLOCK_DIFF (src->last_rtime, running_time);
1436
1437     /* now translate the diff to RTP time, handle positive and negative cases.
1438      * If there is no diff, we already set rtptime correctly above. */
1439     if (diff > 0) {
1440       GST_DEBUG ("running_time %" GST_TIME_FORMAT ", diff %" GST_TIME_FORMAT,
1441           GST_TIME_ARGS (running_time), GST_TIME_ARGS (diff));
1442       t_rtp += gst_util_uint64_scale_int (diff, src->clock_rate, GST_SECOND);
1443     } else {
1444       diff = -diff;
1445       GST_DEBUG ("running_time %" GST_TIME_FORMAT ", diff -%" GST_TIME_FORMAT,
1446           GST_TIME_ARGS (running_time), GST_TIME_ARGS (diff));
1447       t_rtp -= gst_util_uint64_scale_int (diff, src->clock_rate, GST_SECOND);
1448     }
1449   } else {
1450     GST_WARNING ("no clock-rate, cannot interpolate rtp time");
1451   }
1452
1453   /* convert the NTP time in nanoseconds to 32.32 fixed point */
1454   t_current_ntp = gst_util_uint64_scale (ntpnstime, (1LL << 32), GST_SECOND);
1455
1456   GST_DEBUG ("NTP %08x:%08x, RTP %" G_GUINT32_FORMAT,
1457       (guint32) (t_current_ntp >> 32), (guint32) (t_current_ntp & 0xffffffff),
1458       (guint32) t_rtp);
1459
1460   if (ntptime)
1461     *ntptime = t_current_ntp;
1462   if (rtptime)
1463     *rtptime = t_rtp;
1464   if (packet_count)
1465     *packet_count = src->stats.packets_sent;
1466   if (octet_count)
1467     *octet_count = src->stats.octets_sent;
1468
1469   return TRUE;
1470 }
1471
1472 /**
1473  * rtp_source_get_new_rb:
1474  * @src: an #RTPSource
1475  * @time: the current time of the system clock
1476  * @fractionlost: fraction lost since last SR/RR
1477  * @packetslost: the cumulative number of packets lost
1478  * @exthighestseq: the extended last sequence number received
1479  * @jitter: the interarrival jitter (in clock rate units)
1480  * @lsr: the time of the last SR packet on this source
1481  *   (in NTP Short Format, 16.16 fixed point)
1482  * @dlsr: the delay since the last SR packet
1483  *   (in NTP Short Format, 16.16 fixed point)
1484  *
1485  * Get new values to put into a new report block from this source.
1486  *
1487  * Returns: %TRUE on success.
1488  */
1489 gboolean
1490 rtp_source_get_new_rb (RTPSource * src, GstClockTime time,
1491     guint8 * fractionlost, gint32 * packetslost, guint32 * exthighestseq,
1492     guint32 * jitter, guint32 * lsr, guint32 * dlsr)
1493 {
1494   RTPSourceStats *stats;
1495   guint64 extended_max, expected;
1496   guint64 expected_interval, received_interval, ntptime;
1497   gint64 lost, lost_interval;
1498   guint32 fraction, LSR, DLSR;
1499   GstClockTime sr_time;
1500
1501   stats = &src->stats;
1502
1503   extended_max = stats->cycles + stats->max_seq;
1504   expected = extended_max - stats->base_seq + 1;
1505
1506   GST_DEBUG ("ext_max %" G_GUINT64_FORMAT ", expected %" G_GUINT64_FORMAT
1507       ", received %" G_GUINT64_FORMAT ", base_seq %" G_GUINT32_FORMAT,
1508       extended_max, expected, stats->packets_received, stats->base_seq);
1509
1510   lost = expected - stats->packets_received;
1511   lost = CLAMP (lost, -0x800000, 0x7fffff);
1512
1513   expected_interval = expected - stats->prev_expected;
1514   stats->prev_expected = expected;
1515   received_interval = stats->packets_received - stats->prev_received;
1516   stats->prev_received = stats->packets_received;
1517
1518   lost_interval = expected_interval - received_interval;
1519
1520   if (expected_interval == 0 || lost_interval <= 0)
1521     fraction = 0;
1522   else
1523     fraction = (lost_interval << 8) / expected_interval;
1524
1525   GST_DEBUG ("add RR for SSRC %08x", src->ssrc);
1526   /* we scaled the jitter up for additional precision */
1527   GST_DEBUG ("fraction %" G_GUINT32_FORMAT ", lost %" G_GINT64_FORMAT
1528       ", extseq %" G_GUINT64_FORMAT ", jitter %d", fraction, lost,
1529       extended_max, stats->jitter >> 4);
1530
1531   if (rtp_source_get_last_sr (src, &sr_time, &ntptime, NULL, NULL, NULL)) {
1532     GstClockTime diff;
1533
1534     /* LSR is middle 32 bits of the last ntptime */
1535     LSR = (ntptime >> 16) & 0xffffffff;
1536     diff = time - sr_time;
1537     GST_DEBUG ("last SR time diff %" GST_TIME_FORMAT, GST_TIME_ARGS (diff));
1538     /* DLSR, delay since last SR is expressed in 1/65536 second units */
1539     DLSR = gst_util_uint64_scale_int (diff, 65536, GST_SECOND);
1540   } else {
1541     /* No valid SR received, LSR/DLSR are set to 0 then */
1542     GST_DEBUG ("no valid SR received");
1543     LSR = 0;
1544     DLSR = 0;
1545   }
1546   GST_DEBUG ("LSR %04x:%04x, DLSR %04x:%04x", LSR >> 16, LSR & 0xffff,
1547       DLSR >> 16, DLSR & 0xffff);
1548
1549   if (fractionlost)
1550     *fractionlost = fraction;
1551   if (packetslost)
1552     *packetslost = lost;
1553   if (exthighestseq)
1554     *exthighestseq = extended_max;
1555   if (jitter)
1556     *jitter = stats->jitter >> 4;
1557   if (lsr)
1558     *lsr = LSR;
1559   if (dlsr)
1560     *dlsr = DLSR;
1561
1562   return TRUE;
1563 }
1564
1565 /**
1566  * rtp_source_get_last_sr:
1567  * @src: an #RTPSource
1568  * @time: time of packet arrival
1569  * @ntptime: the NTP time (in NTP Timestamp Format, 32.32 fixed point)
1570  * @rtptime: the RTP time (in clock rate units)
1571  * @packet_count: the packet count
1572  * @octet_count: the octet count
1573  *
1574  * Get the values of the last sender report as set with rtp_source_process_sr().
1575  *
1576  * Returns: %TRUE if there was a valid SR report.
1577  */
1578 gboolean
1579 rtp_source_get_last_sr (RTPSource * src, GstClockTime * time, guint64 * ntptime,
1580     guint32 * rtptime, guint32 * packet_count, guint32 * octet_count)
1581 {
1582   RTPSenderReport *curr;
1583
1584   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
1585
1586   curr = &src->stats.sr[src->stats.curr_sr];
1587   if (!curr->is_valid)
1588     return FALSE;
1589
1590   if (ntptime)
1591     *ntptime = curr->ntptime;
1592   if (rtptime)
1593     *rtptime = curr->rtptime;
1594   if (packet_count)
1595     *packet_count = curr->packet_count;
1596   if (octet_count)
1597     *octet_count = curr->octet_count;
1598   if (time)
1599     *time = curr->time;
1600
1601   return TRUE;
1602 }
1603
1604 /**
1605  * rtp_source_get_last_rb:
1606  * @src: an #RTPSource
1607  * @fractionlost: fraction lost since last SR/RR
1608  * @packetslost: the cumulative number of packets lost
1609  * @exthighestseq: the extended last sequence number received
1610  * @jitter: the interarrival jitter (in clock rate units)
1611  * @lsr: the time of the last SR packet on this source
1612  *   (in NTP Short Format, 16.16 fixed point)
1613  * @dlsr: the delay since the last SR packet
1614  *   (in NTP Short Format, 16.16 fixed point)
1615  * @round_trip: the round-trip time
1616  *   (in NTP Short Format, 16.16 fixed point)
1617  *
1618  * Get the values of the last RB report set with rtp_source_process_rb().
1619  *
1620  * Returns: %TRUE if there was a valid SB report.
1621  */
1622 gboolean
1623 rtp_source_get_last_rb (RTPSource * src, guint8 * fractionlost,
1624     gint32 * packetslost, guint32 * exthighestseq, guint32 * jitter,
1625     guint32 * lsr, guint32 * dlsr, guint32 * round_trip)
1626 {
1627   RTPReceiverReport *curr;
1628
1629   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
1630
1631   curr = &src->stats.rr[src->stats.curr_rr];
1632   if (!curr->is_valid)
1633     return FALSE;
1634
1635   if (fractionlost)
1636     *fractionlost = curr->fractionlost;
1637   if (packetslost)
1638     *packetslost = curr->packetslost;
1639   if (exthighestseq)
1640     *exthighestseq = curr->exthighestseq;
1641   if (jitter)
1642     *jitter = curr->jitter;
1643   if (lsr)
1644     *lsr = curr->lsr;
1645   if (dlsr)
1646     *dlsr = curr->dlsr;
1647   if (round_trip)
1648     *round_trip = curr->round_trip;
1649
1650   return TRUE;
1651 }
1652
1653 gboolean
1654 find_conflicting_address (GList * conflicting_addresses,
1655     GSocketAddress * address, GstClockTime time)
1656 {
1657   GList *item;
1658
1659   for (item = conflicting_addresses; item; item = g_list_next (item)) {
1660     RTPConflictingAddress *known_conflict = item->data;
1661
1662     if (__g_socket_address_equal (address, known_conflict->address)) {
1663       known_conflict->time = time;
1664       return TRUE;
1665     }
1666   }
1667
1668   return FALSE;
1669 }
1670
1671 GList *
1672 add_conflicting_address (GList * conflicting_addresses,
1673     GSocketAddress * address, GstClockTime time)
1674 {
1675   RTPConflictingAddress *new_conflict;
1676
1677   new_conflict = g_slice_new (RTPConflictingAddress);
1678
1679   new_conflict->address = G_SOCKET_ADDRESS (g_object_ref (address));
1680   new_conflict->time = time;
1681
1682   return g_list_prepend (conflicting_addresses, new_conflict);
1683 }
1684
1685 GList *
1686 timeout_conflicting_addresses (GList * conflicting_addresses,
1687     GstClockTime current_time)
1688 {
1689   GList *item;
1690   /* "a relatively long time" -- RFC 3550 section 8.2 */
1691   const GstClockTime collision_timeout =
1692       RTP_STATS_MIN_INTERVAL * GST_SECOND * 10;
1693
1694   item = g_list_first (conflicting_addresses);
1695   while (item) {
1696     RTPConflictingAddress *known_conflict = item->data;
1697     GList *next_item = g_list_next (item);
1698
1699     if (known_conflict->time < current_time - collision_timeout) {
1700       gchar *buf;
1701
1702       conflicting_addresses = g_list_delete_link (conflicting_addresses, item);
1703       buf = __g_socket_address_to_string (known_conflict->address);
1704       GST_DEBUG ("collision %p timed out: %s", known_conflict, buf);
1705       g_free (buf);
1706       rtp_conflicting_address_free (known_conflict);
1707     }
1708     item = next_item;
1709   }
1710
1711   return conflicting_addresses;
1712 }
1713
1714 /**
1715  * rtp_source_find_conflicting_address:
1716  * @src: The source the packet came in
1717  * @address: address to check for
1718  * @time: The time when the packet that is possibly in conflict arrived
1719  *
1720  * Checks if an address which has a conflict is already known. If it is
1721  * a known conflict, remember the time
1722  *
1723  * Returns: TRUE if it was a known conflict, FALSE otherwise
1724  */
1725 gboolean
1726 rtp_source_find_conflicting_address (RTPSource * src, GSocketAddress * address,
1727     GstClockTime time)
1728 {
1729   return find_conflicting_address (src->conflicting_addresses, address, time);
1730 }
1731
1732 /**
1733  * rtp_source_add_conflicting_address:
1734  * @src: The source the packet came in
1735  * @address: address to remember
1736  * @time: The time when the packet that is in conflict arrived
1737  *
1738  * Adds a new conflict address
1739  */
1740 void
1741 rtp_source_add_conflicting_address (RTPSource * src,
1742     GSocketAddress * address, GstClockTime time)
1743 {
1744   src->conflicting_addresses =
1745       add_conflicting_address (src->conflicting_addresses, address, time);
1746 }
1747
1748 /**
1749  * rtp_source_timeout:
1750  * @src: The #RTPSource
1751  * @current_time: The current time
1752  * @feedback_retention_window: The running time before which retained feedback
1753  * packets have to be discarded
1754  *
1755  * This is processed on each RTCP interval. It times out old collisions.
1756  * It also times out old retained feedback packets
1757  */
1758 void
1759 rtp_source_timeout (RTPSource * src, GstClockTime current_time,
1760     GstClockTime feedback_retention_window)
1761 {
1762   GstRTCPPacket *pkt;
1763
1764   src->conflicting_addresses =
1765       timeout_conflicting_addresses (src->conflicting_addresses, current_time);
1766
1767   /* Time out AVPF packets that are older than the desired length */
1768   while ((pkt = g_queue_peek_tail (src->retained_feedback)) &&
1769       GST_BUFFER_PTS (pkt) < feedback_retention_window)
1770     gst_buffer_unref (g_queue_pop_tail (src->retained_feedback));
1771 }
1772
1773 static gint
1774 compare_buffers (gconstpointer a, gconstpointer b, gpointer user_data)
1775 {
1776   const GstBuffer *bufa = a;
1777   const GstBuffer *bufb = b;
1778
1779   return GST_BUFFER_PTS (bufa) - GST_BUFFER_PTS (bufb);
1780 }
1781
1782 void
1783 rtp_source_retain_rtcp_packet (RTPSource * src, GstRTCPPacket * packet,
1784     GstClockTime running_time)
1785 {
1786   GstBuffer *buffer;
1787
1788   buffer = gst_buffer_copy_region (packet->rtcp->buffer, GST_BUFFER_COPY_MEMORY,
1789       packet->offset, (gst_rtcp_packet_get_length (packet) + 1) * 4);
1790
1791   GST_BUFFER_PTS (buffer) = running_time;
1792
1793   g_queue_insert_sorted (src->retained_feedback, buffer, compare_buffers, NULL);
1794 }
1795
1796 gboolean
1797 rtp_source_has_retained (RTPSource * src, GCompareFunc func, gconstpointer data)
1798 {
1799   if (g_queue_find_custom (src->retained_feedback, data, func))
1800     return TRUE;
1801   else
1802     return FALSE;
1803 }
1804
1805 /**
1806  * rtp_source_register_nack:
1807  * @src: The #RTPSource
1808  * @seqnum: a seqnum
1809  *
1810  * Register that @seqnum has not been received from @src.
1811  */
1812 void
1813 rtp_source_register_nack (RTPSource * src, guint16 seqnum)
1814 {
1815   guint i, len;
1816   guint32 dword = seqnum << 16;
1817   gint diff = 16;
1818
1819   len = src->nacks->len;
1820   for (i = 0; i < len; i++) {
1821     guint32 tdword;
1822     guint16 tseq;
1823
1824     tdword = g_array_index (src->nacks, guint32, i);
1825     tseq = tdword >> 16;
1826
1827     diff = gst_rtp_buffer_compare_seqnum (tseq, seqnum);
1828     if (diff < 16)
1829       break;
1830   }
1831   /* we already have this seqnum */
1832   if (diff == 0)
1833     return;
1834   /* it comes before the recorded seqnum, FIXME, we could merge it
1835    * if not to far away */
1836   if (diff < 0) {
1837     GST_DEBUG ("insert NACK #%u at %u", seqnum, i);
1838     g_array_insert_val (src->nacks, i, dword);
1839   } else if (diff < 16) {
1840     /* we can merge it */
1841     dword = g_array_index (src->nacks, guint32, i);
1842     dword |= 1 << (diff - 1);
1843     GST_DEBUG ("merge NACK #%u at %u with NACK #%u -> 0x%08x", seqnum, i,
1844         dword >> 16, dword);
1845     g_array_index (src->nacks, guint32, i) = dword;
1846   } else {
1847     GST_DEBUG ("append NACK #%u", seqnum);
1848     g_array_append_val (src->nacks, dword);
1849   }
1850   src->send_nack = TRUE;
1851 }
1852
1853 /**
1854  * rtp_source_get_nacks:
1855  * @src: The #RTPSource
1856  * @n_nacks: result number of nacks
1857  *
1858  * Get the registered NACKS since the last rtp_source_clear_nacks().
1859  *
1860  * Returns: an array of @n_nacks seqnum values.
1861  */
1862 guint32 *
1863 rtp_source_get_nacks (RTPSource * src, guint * n_nacks)
1864 {
1865   if (n_nacks)
1866     *n_nacks = src->nacks->len;
1867
1868   return (guint32 *) src->nacks->data;
1869 }
1870
1871 void
1872 rtp_source_clear_nacks (RTPSource * src)
1873 {
1874   g_array_set_size (src->nacks, 0);
1875   src->send_nack = FALSE;
1876 }