rtpsource: Do not try to push NULL buffers
[platform/upstream/gst-plugins-good.git] / 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  * @src: The #RTPSource
1807  * @seqnum: a seqnum
1808  *
1809  * Register that @seqnum has not been received from @src.
1810  */
1811 void
1812 rtp_source_register_nack (RTPSource * src, guint16 seqnum)
1813 {
1814   guint i, len;
1815   guint32 dword = seqnum << 16;
1816   gint diff = 16;
1817
1818   len = src->nacks->len;
1819   for (i = 0; i < len; i++) {
1820     guint32 tdword;
1821     guint16 tseq;
1822
1823     tdword = g_array_index (src->nacks, guint32, i);
1824     tseq = tdword >> 16;
1825
1826     diff = gst_rtp_buffer_compare_seqnum (tseq, seqnum);
1827     if (diff < 16)
1828       break;
1829   }
1830   /* we already have this seqnum */
1831   if (diff == 0)
1832     return;
1833   /* it comes before the recorded seqnum, FIXME, we could merge it
1834    * if not to far away */
1835   if (diff < 0) {
1836     GST_DEBUG ("insert NACK #%u at %u", seqnum, i);
1837     g_array_insert_val (src->nacks, i, dword);
1838   } else if (diff < 16) {
1839     /* we can merge it */
1840     dword = g_array_index (src->nacks, guint32, i);
1841     dword |= 1 << (diff - 1);
1842     GST_DEBUG ("merge NACK #%u at %u with NACK #%u -> 0x%08x", seqnum, i,
1843         dword >> 16, dword);
1844     g_array_index (src->nacks, guint32, i) = dword;
1845   } else {
1846     GST_DEBUG ("append NACK #%u", seqnum);
1847     g_array_append_val (src->nacks, dword);
1848   }
1849   src->send_nack = TRUE;
1850 }
1851
1852 /**
1853  * @src: The #RTPSource
1854  * @n_nacks: result number of nacks
1855  *
1856  * Get the registered NACKS since the last rtp_source_clear_nacks().
1857  *
1858  * Returns: an array of @n_nacks seqnum values.
1859  */
1860 guint32 *
1861 rtp_source_get_nacks (RTPSource * src, guint * n_nacks)
1862 {
1863   if (n_nacks)
1864     *n_nacks = src->nacks->len;
1865
1866   return (guint32 *) src->nacks->data;
1867 }
1868
1869 void
1870 rtp_source_clear_nacks (RTPSource * src)
1871 {
1872   g_array_set_size (src->nacks, 0);
1873   src->send_nack = FALSE;
1874 }