gst/rtpmanager/rtpsource.c

   1 /* GStreamer
   2  * Copyright (C) <2007> Wim Taymans <wim@fluendo.com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  17  * Boston, MA 02111-1307, USA.
  18  */
  19 #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 enum
  38 {
  39   PROP_0
  40 };
  41
  42 /* GObject vmethods */
  43 static void rtp_source_finalize (GObject * object);
  44
  45 /* static guint rtp_source_signals[LAST_SIGNAL] = { 0 }; */
  46
  47 G_DEFINE_TYPE (RTPSource, rtp_source, G_TYPE_OBJECT);
  48
  49 static void
  50 rtp_source_class_init (RTPSourceClass * klass)
  51 {
  52   GObjectClass *gobject_class;
  53
  54   gobject_class = (GObjectClass *) klass;
  55
  56   gobject_class->finalize = rtp_source_finalize;
  57
  58   GST_DEBUG_CATEGORY_INIT (rtp_source_debug, "rtpsource", 0, "RTP Source");
  59 }
  60
  61 static void
  62 rtp_source_init (RTPSource * src)
  63 {
  64   /* sources are initialy on probation until we receive enough valid RTP
  65    * packets or a valid RTCP packet */
  66   src->validated = FALSE;
  67   src->probation = RTP_DEFAULT_PROBATION;
  68
  69   src->payload = 0;
  70   src->clock_rate = -1;
  71   src->clock_base = -1;
  72   src->skew_base_ntpnstime = -1;
  73   src->ext_rtptime = -1;
  74   src->prev_ext_rtptime = -1;
  75   src->packets = g_queue_new ();
  76   src->seqnum_base = -1;
  77
  78   src->stats.cycles = -1;
  79   src->stats.jitter = 0;
  80   src->stats.transit = -1;
  81   src->stats.curr_sr = 0;
  82   src->stats.curr_rr = 0;
  83 }
  84
  85 static void
  86 rtp_source_finalize (GObject * object)
  87 {
  88   RTPSource *src;
  89   GstBuffer *buffer;
  90
  91   src = RTP_SOURCE_CAST (object);
  92
  93   while ((buffer = g_queue_pop_head (src->packets)))
  94     gst_buffer_unref (buffer);
  95   g_queue_free (src->packets);
  96
  97   G_OBJECT_CLASS (rtp_source_parent_class)->finalize (object);
  98 }
  99
 100 /**
 101  * rtp_source_new:
 102  * @ssrc: an SSRC
 103  *
 104  * Create a #RTPSource with @ssrc.
 105  *
 106  * Returns: a new #RTPSource. Use g_object_unref() after usage.
 107  */
 108 RTPSource *
 109 rtp_source_new (guint32 ssrc)
 110 {
 111   RTPSource *src;
 112
 113   src = g_object_new (RTP_TYPE_SOURCE, NULL);
 114   src->ssrc = ssrc;
 115
 116   return src;
 117 }
 118
 119 /**
 120  * rtp_source_update_caps:
 121  * @src: an #RTPSource
 122  * @caps: a #GstCaps
 123  *
 124  * Parse @caps and store all relevant information in @source.
 125  */
 126 void
 127 rtp_source_update_caps (RTPSource * src, GstCaps * caps)
 128 {
 129   GstStructure *s;
 130   guint val;
 131   gint ival;
 132
 133   /* nothing changed, return */
 134   if (src->caps == caps)
 135     return;
 136
 137   s = gst_caps_get_structure (caps, 0);
 138
 139   if (gst_structure_get_int (s, "payload", &ival))
 140     src->payload = ival;
 141   GST_DEBUG ("got payload %d", src->payload);
 142
 143   gst_structure_get_int (s, "clock-rate", &src->clock_rate);
 144   GST_DEBUG ("got clock-rate %d", src->clock_rate);
 145
 146   if (gst_structure_get_uint (s, "clock-base", &val))
 147     src->clock_base = val;
 148   GST_DEBUG ("got clock-base %" G_GINT64_FORMAT, src->clock_base);
 149
 150   if (gst_structure_get_uint (s, "seqnum-base", &val))
 151     src->seqnum_base = val;
 152   GST_DEBUG ("got seqnum-base %" G_GINT32_FORMAT, src->seqnum_base);
 153
 154   gst_caps_replace (&src->caps, caps);
 155 }
 156
 157 /**
 158  * rtp_source_set_callbacks:
 159  * @src: an #RTPSource
 160  * @cb: callback functions
 161  * @user_data: user data
 162  *
 163  * Set the callbacks for the source.
 164  */
 165 void
 166 rtp_source_set_callbacks (RTPSource * src, RTPSourceCallbacks * cb,
 167     gpointer user_data)
 168 {
 169   g_return_if_fail (RTP_IS_SOURCE (src));
 170
 171   src->callbacks.push_rtp = cb->push_rtp;
 172   src->callbacks.clock_rate = cb->clock_rate;
 173   src->user_data = user_data;
 174 }
 175
 176 /**
 177  * rtp_source_set_as_csrc:
 178  * @src: an #RTPSource
 179  *
 180  * Configure @src as a CSRC, this will validate the RTpSource.
 181  */
 182 void
 183 rtp_source_set_as_csrc (RTPSource * src)
 184 {
 185   g_return_if_fail (RTP_IS_SOURCE (src));
 186
 187   src->validated = TRUE;
 188   src->is_csrc = TRUE;
 189 }
 190
 191 /**
 192  * rtp_source_set_rtp_from:
 193  * @src: an #RTPSource
 194  * @address: the RTP address to set
 195  *
 196  * Set that @src is receiving RTP packets from @address. This is used for
 197  * collistion checking.
 198  */
 199 void
 200 rtp_source_set_rtp_from (RTPSource * src, GstNetAddress * address)
 201 {
 202   g_return_if_fail (RTP_IS_SOURCE (src));
 203
 204   src->have_rtp_from = TRUE;
 205   memcpy (&src->rtp_from, address, sizeof (GstNetAddress));
 206 }
 207
 208 /**
 209  * rtp_source_set_rtcp_from:
 210  * @src: an #RTPSource
 211  * @address: the RTCP address to set
 212  *
 213  * Set that @src is receiving RTCP packets from @address. This is used for
 214  * collistion checking.
 215  */
 216 void
 217 rtp_source_set_rtcp_from (RTPSource * src, GstNetAddress * address)
 218 {
 219   g_return_if_fail (RTP_IS_SOURCE (src));
 220
 221   src->have_rtcp_from = TRUE;
 222   memcpy (&src->rtcp_from, address, sizeof (GstNetAddress));
 223 }
 224
 225 static GstFlowReturn
 226 push_packet (RTPSource * src, GstBuffer * buffer)
 227 {
 228   GstFlowReturn ret = GST_FLOW_OK;
 229
 230   /* push queued packets first if any */
 231   while (!g_queue_is_empty (src->packets)) {
 232     GstBuffer *buffer = GST_BUFFER_CAST (g_queue_pop_head (src->packets));
 233
 234     GST_DEBUG ("pushing queued packet");
 235     if (src->callbacks.push_rtp)
 236       src->callbacks.push_rtp (src, buffer, src->user_data);
 237     else
 238       gst_buffer_unref (buffer);
 239   }
 240   GST_DEBUG ("pushing new packet");
 241   /* push packet */
 242   if (src->callbacks.push_rtp)
 243     ret = src->callbacks.push_rtp (src, buffer, src->user_data);
 244   else
 245     gst_buffer_unref (buffer);
 246
 247   return ret;
 248 }
 249
 250 static gint
 251 get_clock_rate (RTPSource * src, guint8 payload)
 252 {
 253   if (src->clock_rate == -1) {
 254     gint clock_rate = -1;
 255
 256     if (src->callbacks.clock_rate)
 257       clock_rate = src->callbacks.clock_rate (src, payload, src->user_data);
 258
 259     GST_DEBUG ("new payload %d, got clock-rate %d", payload, clock_rate);
 260
 261     src->clock_rate = clock_rate;
 262   }
 263   src->payload = payload;
 264
 265   return src->clock_rate;
 266 }
 267
 268 static void
 269 calculate_jitter (RTPSource * src, GstBuffer * buffer,
 270     RTPArrivalStats * arrival)
 271 {
 272   guint64 ntpnstime;
 273   guint32 rtparrival, transit, rtptime;
 274   guint64 ext_rtptime;
 275   gint32 diff;
 276   gint clock_rate;
 277   guint8 pt;
 278   guint64 rtpdiff, ntpdiff;
 279   gint64 skew;
 280
 281   /* get arrival time */
 282   if ((ntpnstime = arrival->ntpnstime) == GST_CLOCK_TIME_NONE)
 283     goto no_time;
 284
 285   pt = gst_rtp_buffer_get_payload_type (buffer);
 286
 287   /* get clockrate */
 288   if ((clock_rate = get_clock_rate (src, pt)) == -1)
 289     goto no_clock_rate;
 290
 291   rtptime = gst_rtp_buffer_get_timestamp (buffer);
 292
 293   /* convert to extended timestamp right away */
 294   ext_rtptime = gst_rtp_buffer_ext_timestamp (&src->ext_rtptime, rtptime);
 295
 296   /* no clock-base, take first rtptime as base */
 297   if (src->clock_base == -1) {
 298     GST_DEBUG ("using clock-base of %" G_GUINT32_FORMAT, rtptime);
 299     src->clock_base = rtptime;
 300   }
 301
 302   if (src->skew_base_ntpnstime == -1) {
 303     /* lock on first observed NTP and RTP time, they should increment in-sync or
 304      * we have a clock skew. */
 305     GST_DEBUG ("using base_ntpnstime of %" GST_TIME_FORMAT,
 306         GST_TIME_ARGS (ntpnstime));
 307     src->skew_base_ntpnstime = ntpnstime;
 308     src->skew_base_rtptime = rtptime;
 309     src->prev_ext_rtptime = ext_rtptime;
 310     src->avg_skew = 0;
 311   } else if (src->prev_ext_rtptime < ext_rtptime) {
 312     /* get elapsed rtptime but only when the previous rtptime was stricly smaller
 313      * than the new one. */
 314     rtpdiff = ext_rtptime - src->skew_base_rtptime;
 315     /* get NTP diff and convert to RTP time, this is always positive */
 316     ntpdiff = ntpnstime - src->skew_base_ntpnstime;
 317     ntpdiff = gst_util_uint64_scale_int (ntpdiff, clock_rate, GST_SECOND);
 318
 319     /* see how the NTP and RTP relate any deviation from 0 means that they drift
 320      * out of sync and we must compensate. */
 321     skew = ntpdiff - rtpdiff;
 322     /* average out the skew to get a smooth value. */
 323     src->avg_skew = (31 * src->avg_skew + skew) / 32;
 324
 325     GST_DEBUG ("skew %" G_GINT64_FORMAT ", avg %" G_GINT64_FORMAT, skew,
 326         src->avg_skew);
 327     if (src->avg_skew != 0) {
 328       guint32 timestamp;
 329
 330       /* patch the buffer RTP timestamp with the skew */
 331       GST_DEBUG ("adjusting timestamp %" G_GINT64_FORMAT, src->avg_skew);
 332       timestamp = gst_rtp_buffer_get_timestamp (buffer);
 333       timestamp += src->avg_skew;
 334       gst_rtp_buffer_set_timestamp (buffer, timestamp);
 335     }
 336     /* store previous extended timestamp */
 337     src->prev_ext_rtptime = ext_rtptime;
 338   }
 339
 340   /* convert arrival time to RTP timestamp units, truncate to 32 bits, we don't
 341    * care about the absolute value, just the difference. */
 342   rtparrival = gst_util_uint64_scale_int (ntpnstime, clock_rate, GST_SECOND);
 343
 344   /* transit time is difference with RTP timestamp */
 345   transit = rtparrival - rtptime;
 346
 347   /* get ABS diff with previous transit time */
 348   if (src->stats.transit != -1) {
 349     if (transit > src->stats.transit)
 350       diff = transit - src->stats.transit;
 351     else
 352       diff = src->stats.transit - transit;
 353   } else
 354     diff = 0;
 355
 356   src->stats.transit = transit;
 357
 358   /* update jitter, the value we store is scaled up so we can keep precision. */
 359   src->stats.jitter += diff - ((src->stats.jitter + 8) >> 4);
 360
 361   src->stats.prev_rtptime = src->stats.last_rtptime;
 362   src->stats.last_rtptime = rtparrival;
 363
 364   GST_DEBUG ("rtparrival %u, rtptime %u, clock-rate %d, diff %d, jitter: %f",
 365       rtparrival, rtptime, clock_rate, diff, (src->stats.jitter) / 16.0);
 366
 367   return;
 368
 369   /* ERRORS */
 370 no_time:
 371   {
 372     GST_WARNING ("cannot get current time");
 373     return;
 374   }
 375 no_clock_rate:
 376   {
 377     GST_WARNING ("cannot get clock-rate for pt %d", pt);
 378     return;
 379   }
 380 }
 381
 382 static void
 383 init_seq (RTPSource * src, guint16 seq)
 384 {
 385   src->stats.base_seq = seq;
 386   src->stats.max_seq = seq;
 387   src->stats.bad_seq = RTP_SEQ_MOD + 1; /* so seq == bad_seq is false */
 388   src->stats.cycles = 0;
 389   src->stats.packets_received = 0;
 390   src->stats.octets_received = 0;
 391   src->stats.bytes_received = 0;
 392   src->stats.prev_received = 0;
 393   src->stats.prev_expected = 0;
 394
 395   GST_DEBUG ("base_seq %d", seq);
 396 }
 397
 398 /**
 399  * rtp_source_process_rtp:
 400  * @src: an #RTPSource
 401  * @buffer: an RTP buffer
 402  *
 403  * Let @src handle the incomming RTP @buffer.
 404  *
 405  * Returns: a #GstFlowReturn.
 406  */
 407 GstFlowReturn
 408 rtp_source_process_rtp (RTPSource * src, GstBuffer * buffer,
 409     RTPArrivalStats * arrival)
 410 {
 411   GstFlowReturn result = GST_FLOW_OK;
 412   guint16 seqnr, udelta;
 413   RTPSourceStats *stats;
 414
 415   g_return_val_if_fail (RTP_IS_SOURCE (src), GST_FLOW_ERROR);
 416   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
 417
 418   stats = &src->stats;
 419
 420   seqnr = gst_rtp_buffer_get_seq (buffer);
 421
 422   rtp_source_update_caps (src, GST_BUFFER_CAPS (buffer));
 423
 424   if (stats->cycles == -1) {
 425     GST_DEBUG ("received first buffer");
 426     /* first time we heard of this source */
 427     init_seq (src, seqnr);
 428     src->stats.max_seq = seqnr - 1;
 429     src->probation = RTP_DEFAULT_PROBATION;
 430   }
 431
 432   udelta = seqnr - stats->max_seq;
 433
 434   /* if we are still on probation, check seqnum */
 435   if (src->probation) {
 436     guint16 expected;
 437
 438     expected = src->stats.max_seq + 1;
 439
 440     /* when in probation, we require consecutive seqnums */
 441     if (seqnr == expected) {
 442       /* expected packet */
 443       GST_DEBUG ("probation: seqnr %d == expected %d", seqnr, expected);
 444       src->probation--;
 445       src->stats.max_seq = seqnr;
 446       if (src->probation == 0) {
 447         GST_DEBUG ("probation done!");
 448         init_seq (src, seqnr);
 449       } else {
 450         GstBuffer *q;
 451
 452         GST_DEBUG ("probation %d: queue buffer", src->probation);
 453         /* when still in probation, keep packets in a list. */
 454         g_queue_push_tail (src->packets, buffer);
 455         /* remove packets from queue if there are too many */
 456         while (g_queue_get_length (src->packets) > RTP_MAX_PROBATION_LEN) {
 457           q = g_queue_pop_head (src->packets);
 458           gst_object_unref (q);
 459         }
 460         goto done;
 461       }
 462     } else {
 463       GST_DEBUG ("probation: seqnr %d != expected %d", seqnr, expected);
 464       src->probation = RTP_DEFAULT_PROBATION;
 465       src->stats.max_seq = seqnr;
 466       goto done;
 467     }
 468   } else if (udelta < RTP_MAX_DROPOUT) {
 469     /* in order, with permissible gap */
 470     if (seqnr < stats->max_seq) {
 471       /* sequence number wrapped - count another 64K cycle. */
 472       stats->cycles += RTP_SEQ_MOD;
 473     }
 474     stats->max_seq = seqnr;
 475   } else if (udelta <= RTP_SEQ_MOD - RTP_MAX_MISORDER) {
 476     /* the sequence number made a very large jump */
 477     if (seqnr == stats->bad_seq) {
 478       /* two sequential packets -- assume that the other side
 479        * restarted without telling us so just re-sync
 480        * (i.e., pretend this was the first packet).  */
 481       init_seq (src, seqnr);
 482     } else {
 483       /* unacceptable jump */
 484       stats->bad_seq = (seqnr + 1) & (RTP_SEQ_MOD - 1);
 485       goto bad_sequence;
 486     }
 487   } else {
 488     /* duplicate or reordered packet, will be filtered by jitterbuffer. */
 489     GST_WARNING ("duplicate or reordered packet");
 490   }
 491
 492   src->stats.octets_received += arrival->payload_len;
 493   src->stats.bytes_received += arrival->bytes;
 494   src->stats.packets_received++;
 495   /* the source that sent the packet must be a sender */
 496   src->is_sender = TRUE;
 497   src->validated = TRUE;
 498
 499   GST_DEBUG ("seq %d, PC: %" G_GUINT64_FORMAT ", OC: %" G_GUINT64_FORMAT,
 500       seqnr, src->stats.packets_received, src->stats.octets_received);
 501
 502   /* calculate jitter and perform skew correction */
 503   calculate_jitter (src, buffer, arrival);
 504
 505   /* we're ready to push the RTP packet now */
 506   result = push_packet (src, buffer);
 507
 508 done:
 509   return result;
 510
 511   /* ERRORS */
 512 bad_sequence:
 513   {
 514     GST_WARNING ("unacceptable seqnum received");
 515     return GST_FLOW_OK;
 516   }
 517 }
 518
 519 /**
 520  * rtp_source_process_bye:
 521  * @src: an #RTPSource
 522  * @reason: the reason for leaving
 523  *
 524  * Notify @src that a BYE packet has been received. This will make the source
 525  * inactive.
 526  */
 527 void
 528 rtp_source_process_bye (RTPSource * src, const gchar * reason)
 529 {
 530   g_return_if_fail (RTP_IS_SOURCE (src));
 531
 532   GST_DEBUG ("marking SSRC %08x as BYE, reason: %s", src->ssrc,
 533       GST_STR_NULL (reason));
 534
 535   /* copy the reason and mark as received_bye */
 536   g_free (src->bye_reason);
 537   src->bye_reason = g_strdup (reason);
 538   src->received_bye = TRUE;
 539 }
 540
 541 /**
 542  * rtp_source_send_rtp:
 543  * @src: an #RTPSource
 544  * @buffer: an RTP buffer
 545  * @ntpnstime: the NTP time when this buffer was captured in nanoseconds
 546  *
 547  * Send an RTP @buffer originating from @src. This will make @src a sender.
 548  * This function takes ownership of @buffer and modifies the SSRC in the RTP
 549  * packet to that of @src when needed.
 550  *
 551  * Returns: a #GstFlowReturn.
 552  */
 553 GstFlowReturn
 554 rtp_source_send_rtp (RTPSource * src, GstBuffer * buffer, guint64 ntpnstime)
 555 {
 556   GstFlowReturn result = GST_FLOW_OK;
 557   guint len;
 558
 559   g_return_val_if_fail (RTP_IS_SOURCE (src), GST_FLOW_ERROR);
 560   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
 561
 562   len = gst_rtp_buffer_get_payload_len (buffer);
 563
 564   rtp_source_update_caps (src, GST_BUFFER_CAPS (buffer));
 565
 566   /* we are a sender now */
 567   src->is_sender = TRUE;
 568
 569   /* update stats for the SR */
 570   src->stats.packets_sent++;
 571   src->stats.octets_sent += len;
 572
 573   /* we keep track of the last received RTP timestamp and the corresponding
 574    * NTP timestamp so that we can use this info when constructing SR reports */
 575   src->last_rtptime = gst_rtp_buffer_get_timestamp (buffer);
 576   src->last_ntpnstime = ntpnstime;
 577
 578   /* push packet */
 579   if (src->callbacks.push_rtp) {
 580     guint32 ssrc;
 581
 582     ssrc = gst_rtp_buffer_get_ssrc (buffer);
 583     if (ssrc != src->ssrc) {
 584       /* the SSRC of the packet is not correct, make a writable buffer and
 585        * update the SSRC. This could involve a complete copy of the packet when
 586        * it is not writable. Usually the payloader will use caps negotiation to
 587        * get the correct SSRC. */
 588       buffer = gst_buffer_make_writable (buffer);
 589
 590       GST_DEBUG ("updating SSRC from %08x to %08x", ssrc, src->ssrc);
 591       gst_rtp_buffer_set_ssrc (buffer, src->ssrc);
 592     }
 593     GST_DEBUG ("pushing RTP packet %" G_GUINT64_FORMAT,
 594         src->stats.packets_sent);
 595     result = src->callbacks.push_rtp (src, buffer, src->user_data);
 596   } else {
 597     GST_DEBUG ("no callback installed");
 598     gst_buffer_unref (buffer);
 599   }
 600
 601   return result;
 602 }
 603
 604 /**
 605  * rtp_source_process_sr:
 606  * @src: an #RTPSource
 607  * @time: time of packet arrival
 608  * @ntptime: the NTP time
 609  * @rtptime: the RTP time
 610  * @packet_count: the packet count
 611  * @octet_count: the octect count
 612  *
 613  * Update the sender report in @src.
 614  */
 615 void
 616 rtp_source_process_sr (RTPSource * src, GstClockTime time, guint64 ntptime,
 617     guint32 rtptime, guint32 packet_count, guint32 octet_count)
 618 {
 619   RTPSenderReport *curr;
 620   gint curridx;
 621
 622   g_return_if_fail (RTP_IS_SOURCE (src));
 623
 624   GST_DEBUG ("got SR packet: SSRC %08x, NTP %08x:%08x, RTP %" G_GUINT32_FORMAT
 625       ", PC %" G_GUINT32_FORMAT ", OC %" G_GUINT32_FORMAT, src->ssrc,
 626       (guint32) (ntptime >> 32), (guint32) (ntptime & 0xffffffff), rtptime,
 627       packet_count, octet_count);
 628
 629   curridx = src->stats.curr_sr ^ 1;
 630   curr = &src->stats.sr[curridx];
 631
 632   /* this is a sender now */
 633   src->is_sender = TRUE;
 634
 635   /* update current */
 636   curr->is_valid = TRUE;
 637   curr->ntptime = ntptime;
 638   curr->rtptime = rtptime;
 639   curr->packet_count = packet_count;
 640   curr->octet_count = octet_count;
 641   curr->time = time;
 642
 643   /* make current */
 644   src->stats.curr_sr = curridx;
 645 }
 646
 647 /**
 648  * rtp_source_process_rb:
 649  * @src: an #RTPSource
 650  * @time: the current time in nanoseconds since 1970
 651  * @fractionlost: fraction lost since last SR/RR
 652  * @packetslost: the cumululative number of packets lost
 653  * @exthighestseq: the extended last sequence number received
 654  * @jitter: the interarrival jitter
 655  * @lsr: the last SR packet from this source
 656  * @dlsr: the delay since last SR packet
 657  *
 658  * Update the report block in @src.
 659  */
 660 void
 661 rtp_source_process_rb (RTPSource * src, GstClockTime time, guint8 fractionlost,
 662     gint32 packetslost, guint32 exthighestseq, guint32 jitter, guint32 lsr,
 663     guint32 dlsr)
 664 {
 665   RTPReceiverReport *curr;
 666   gint curridx;
 667   guint32 ntp, A;
 668
 669   g_return_if_fail (RTP_IS_SOURCE (src));
 670
 671   GST_DEBUG ("got RB packet: SSRC %08x, FL %2x, PL %d, HS %" G_GUINT32_FORMAT
 672       ", jitter %" G_GUINT32_FORMAT ", LSR %04x:%04x, DLSR %04x:%04x",
 673       src->ssrc, fractionlost, packetslost, exthighestseq, jitter, lsr >> 16,
 674       lsr & 0xffff, dlsr >> 16, dlsr & 0xffff);
 675
 676   curridx = src->stats.curr_rr ^ 1;
 677   curr = &src->stats.rr[curridx];
 678
 679   /* update current */
 680   curr->is_valid = TRUE;
 681   curr->fractionlost = fractionlost;
 682   curr->packetslost = packetslost;
 683   curr->exthighestseq = exthighestseq;
 684   curr->jitter = jitter;
 685   curr->lsr = lsr;
 686   curr->dlsr = dlsr;
 687
 688   /* calculate round trip */
 689   ntp = (gst_rtcp_unix_to_ntp (time) >> 16) & 0xffffffff;
 690   A = ntp - dlsr;
 691   A -= lsr;
 692   curr->round_trip = A;
 693
 694   GST_DEBUG ("NTP %04x:%04x, round trip %04x:%04x", ntp >> 16, ntp & 0xffff,
 695       A >> 16, A & 0xffff);
 696
 697   /* make current */
 698   src->stats.curr_rr = curridx;
 699 }
 700
 701 /**
 702  * rtp_source_get_new_sr:
 703  * @src: an #RTPSource
 704  * @time: the current time in nanoseconds since 1970
 705  * @ntptime: the NTP time
 706  * @rtptime: the RTP time
 707  * @packet_count: the packet count
 708  * @octet_count: the octect count
 709  *
 710  * Get new values to put into a new SR report from this source.
 711  *
 712  * Returns: %TRUE on success.
 713  */
 714 gboolean
 715 rtp_source_get_new_sr (RTPSource * src, GstClockTime ntpnstime,
 716     guint64 * ntptime, guint32 * rtptime, guint32 * packet_count,
 717     guint32 * octet_count)
 718 {
 719   guint32 t_rtp;
 720   guint64 t_current_ntp;
 721   GstClockTimeDiff diff;
 722
 723   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 724
 725   /* use the sync params to interpollate the date->time member to rtptime. We
 726    * use the last sent timestamp and rtptime as reference points. We assume
 727    * that the slope of the rtptime vs timestamp curve is 1, which is certainly
 728    * sufficient for the frequency at which we report SR and the rate we send
 729    * out RTP packets. */
 730   t_rtp = src->last_rtptime;
 731
 732   GST_DEBUG ("last_ntpnstime %" GST_TIME_FORMAT ", last_rtptime %"
 733       G_GUINT32_FORMAT, GST_TIME_ARGS (src->last_ntpnstime), t_rtp);
 734
 735   if (src->clock_rate != -1) {
 736     /* get the diff with the SR time */
 737     diff = GST_CLOCK_DIFF (src->last_ntpnstime, ntpnstime);
 738
 739     /* now translate the diff to RTP time, handle positive and negative cases.
 740      * If there is no diff, we already set rtptime correctly above. */
 741     if (diff > 0) {
 742       GST_DEBUG ("ntpnstime %" GST_TIME_FORMAT ", diff %" GST_TIME_FORMAT,
 743           GST_TIME_ARGS (ntpnstime), GST_TIME_ARGS (diff));
 744       t_rtp += gst_util_uint64_scale_int (diff, src->clock_rate, GST_SECOND);
 745     } else {
 746       diff = -diff;
 747       GST_DEBUG ("ntpnstime %" GST_TIME_FORMAT ", diff -%" GST_TIME_FORMAT,
 748           GST_TIME_ARGS (ntpnstime), GST_TIME_ARGS (diff));
 749       t_rtp -= gst_util_uint64_scale_int (diff, src->clock_rate, GST_SECOND);
 750     }
 751   } else {
 752     GST_WARNING ("no clock-rate, cannot interpollate rtp time");
 753   }
 754
 755   t_current_ntp = gst_util_uint64_scale (ntpnstime, (1LL << 32), GST_SECOND);
 756
 757   GST_DEBUG ("NTP %08x:%08x, RTP %" G_GUINT32_FORMAT,
 758       (guint32) (t_current_ntp >> 32), (guint32) (t_current_ntp & 0xffffffff),
 759       t_rtp);
 760
 761   if (ntptime)
 762     *ntptime = t_current_ntp;
 763   if (rtptime)
 764     *rtptime = t_rtp;
 765   if (packet_count)
 766     *packet_count = src->stats.packets_sent;
 767   if (octet_count)
 768     *octet_count = src->stats.octets_sent;
 769
 770   return TRUE;
 771 }
 772
 773 /**
 774  * rtp_source_get_new_rb:
 775  * @src: an #RTPSource
 776  * @time: the current time in nanoseconds since 1970
 777  * @fractionlost: fraction lost since last SR/RR
 778  * @packetslost: the cumululative number of packets lost
 779  * @exthighestseq: the extended last sequence number received
 780  * @jitter: the interarrival jitter
 781  * @lsr: the last SR packet from this source
 782  * @dlsr: the delay since last SR packet
 783  *
 784  * Get the values of the last RB report set with rtp_source_process_rb().
 785  *
 786  * Returns: %TRUE on success.
 787  */
 788 gboolean
 789 rtp_source_get_new_rb (RTPSource * src, GstClockTime time,
 790     guint8 * fractionlost, gint32 * packetslost, guint32 * exthighestseq,
 791     guint32 * jitter, guint32 * lsr, guint32 * dlsr)
 792 {
 793   RTPSourceStats *stats;
 794   guint64 extended_max, expected;
 795   guint64 expected_interval, received_interval, ntptime;
 796   gint64 lost, lost_interval;
 797   guint32 fraction, LSR, DLSR;
 798   GstClockTime sr_time;
 799
 800   stats = &src->stats;
 801
 802   extended_max = stats->cycles + stats->max_seq;
 803   expected = extended_max - stats->base_seq + 1;
 804
 805   GST_DEBUG ("ext_max %" G_GUINT64_FORMAT ", expected %" G_GUINT64_FORMAT
 806       ", received %" G_GUINT64_FORMAT ", base_seq %" G_GUINT32_FORMAT,
 807       extended_max, expected, stats->packets_received, stats->base_seq);
 808
 809   lost = expected - stats->packets_received;
 810   lost = CLAMP (lost, -0x800000, 0x7fffff);
 811
 812   expected_interval = expected - stats->prev_expected;
 813   stats->prev_expected = expected;
 814   received_interval = stats->packets_received - stats->prev_received;
 815   stats->prev_received = stats->packets_received;
 816
 817   lost_interval = expected_interval - received_interval;
 818
 819   if (expected_interval == 0 || lost_interval <= 0)
 820     fraction = 0;
 821   else
 822     fraction = (lost_interval << 8) / expected_interval;
 823
 824   GST_DEBUG ("add RR for SSRC %08x", src->ssrc);
 825   /* we scaled the jitter up for additional precision */
 826   GST_DEBUG ("fraction %" G_GUINT32_FORMAT ", lost %" G_GINT64_FORMAT
 827       ", extseq %" G_GUINT64_FORMAT ", jitter %d", fraction, lost,
 828       extended_max, stats->jitter >> 4);
 829
 830   if (rtp_source_get_last_sr (src, &sr_time, &ntptime, NULL, NULL, NULL)) {
 831     GstClockTime diff;
 832
 833     /* LSR is middle 32 bits of the last ntptime */
 834     LSR = (ntptime >> 16) & 0xffffffff;
 835     diff = time - sr_time;
 836     GST_DEBUG ("last SR time diff %" GST_TIME_FORMAT, GST_TIME_ARGS (diff));
 837     /* DLSR, delay since last SR is expressed in 1/65536 second units */
 838     DLSR = gst_util_uint64_scale_int (diff, 65536, GST_SECOND);
 839   } else {
 840     /* No valid SR received, LSR/DLSR are set to 0 then */
 841     GST_DEBUG ("no valid SR received");
 842     LSR = 0;
 843     DLSR = 0;
 844   }
 845   GST_DEBUG ("LSR %04x:%04x, DLSR %04x:%04x", LSR >> 16, LSR & 0xffff,
 846       DLSR >> 16, DLSR & 0xffff);
 847
 848   if (fractionlost)
 849     *fractionlost = fraction;
 850   if (packetslost)
 851     *packetslost = lost;
 852   if (exthighestseq)
 853     *exthighestseq = extended_max;
 854   if (jitter)
 855     *jitter = stats->jitter >> 4;
 856   if (lsr)
 857     *lsr = LSR;
 858   if (dlsr)
 859     *dlsr = DLSR;
 860
 861   return TRUE;
 862 }
 863
 864 /**
 865  * rtp_source_get_last_sr:
 866  * @src: an #RTPSource
 867  * @time: time of packet arrival
 868  * @ntptime: the NTP time
 869  * @rtptime: the RTP time
 870  * @packet_count: the packet count
 871  * @octet_count: the octect count
 872  *
 873  * Get the values of the last sender report as set with rtp_source_process_sr().
 874  *
 875  * Returns: %TRUE if there was a valid SR report.
 876  */
 877 gboolean
 878 rtp_source_get_last_sr (RTPSource * src, GstClockTime * time, guint64 * ntptime,
 879     guint32 * rtptime, guint32 * packet_count, guint32 * octet_count)
 880 {
 881   RTPSenderReport *curr;
 882
 883   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 884
 885   curr = &src->stats.sr[src->stats.curr_sr];
 886   if (!curr->is_valid)
 887     return FALSE;
 888
 889   if (ntptime)
 890     *ntptime = curr->ntptime;
 891   if (rtptime)
 892     *rtptime = curr->rtptime;
 893   if (packet_count)
 894     *packet_count = curr->packet_count;
 895   if (octet_count)
 896     *octet_count = curr->octet_count;
 897   if (time)
 898     *time = curr->time;
 899
 900   return TRUE;
 901 }
 902
 903 /**
 904  * rtp_source_get_last_rb:
 905  * @src: an #RTPSource
 906  * @fractionlost: fraction lost since last SR/RR
 907  * @packetslost: the cumululative number of packets lost
 908  * @exthighestseq: the extended last sequence number received
 909  * @jitter: the interarrival jitter
 910  * @lsr: the last SR packet from this source
 911  * @dlsr: the delay since last SR packet
 912  *
 913  * Get the values of the last RB report set with rtp_source_process_rb().
 914  *
 915  * Returns: %TRUE if there was a valid SB report.
 916  */
 917 gboolean
 918 rtp_source_get_last_rb (RTPSource * src, guint8 * fractionlost,
 919     gint32 * packetslost, guint32 * exthighestseq, guint32 * jitter,
 920     guint32 * lsr, guint32 * dlsr)
 921 {
 922   RTPReceiverReport *curr;
 923
 924   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 925
 926   curr = &src->stats.rr[src->stats.curr_rr];
 927   if (!curr->is_valid)
 928     return FALSE;
 929
 930   if (fractionlost)
 931     *fractionlost = curr->fractionlost;
 932   if (packetslost)
 933     *packetslost = curr->packetslost;
 934   if (exthighestseq)
 935     *exthighestseq = curr->exthighestseq;
 936   if (jitter)
 937     *jitter = curr->jitter;
 938   if (lsr)
 939     *lsr = curr->lsr;
 940   if (dlsr)
 941     *dlsr = curr->dlsr;
 942
 943   return TRUE;
 944 }