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->packets = g_queue_new ();
  72
  73   src->stats.cycles = -1;
  74   src->stats.jitter = 0;
  75   src->stats.transit = -1;
  76   src->stats.curr_sr = 0;
  77   src->stats.curr_rr = 0;
  78 }
  79
  80 static void
  81 rtp_source_finalize (GObject * object)
  82 {
  83   RTPSource *src;
  84   GstBuffer *buffer;
  85
  86   src = RTP_SOURCE_CAST (object);
  87
  88   while ((buffer = g_queue_pop_head (src->packets)))
  89     gst_buffer_unref (buffer);
  90   g_queue_free (src->packets);
  91
  92   G_OBJECT_CLASS (rtp_source_parent_class)->finalize (object);
  93 }
  94
  95 /**
  96  * rtp_source_new:
  97  * @ssrc: an SSRC
  98  *
  99  * Create a #RTPSource with @ssrc.
 100  *
 101  * Returns: a new #RTPSource. Use g_object_unref() after usage.
 102  */
 103 RTPSource *
 104 rtp_source_new (guint32 ssrc)
 105 {
 106   RTPSource *src;
 107
 108   src = g_object_new (RTP_TYPE_SOURCE, NULL);
 109   src->ssrc = ssrc;
 110
 111   return src;
 112 }
 113
 114 /**
 115  * rtp_source_set_callbacks:
 116  * @src: an #RTPSource
 117  * @cb: callback functions
 118  * @user_data: user data
 119  *
 120  * Set the callbacks for the source.
 121  */
 122 void
 123 rtp_source_set_callbacks (RTPSource * src, RTPSourceCallbacks * cb,
 124     gpointer user_data)
 125 {
 126   g_return_if_fail (RTP_IS_SOURCE (src));
 127
 128   src->callbacks.push_rtp = cb->push_rtp;
 129   src->callbacks.clock_rate = cb->clock_rate;
 130   src->user_data = user_data;
 131 }
 132
 133 /**
 134  * rtp_source_set_as_csrc:
 135  * @src: an #RTPSource
 136  *
 137  * Configure @src as a CSRC, this will validate the RTpSource.
 138  */
 139 void
 140 rtp_source_set_as_csrc (RTPSource * src)
 141 {
 142   g_return_if_fail (RTP_IS_SOURCE (src));
 143
 144   src->validated = TRUE;
 145   src->is_csrc = TRUE;
 146 }
 147
 148 /**
 149  * rtp_source_set_rtp_from:
 150  * @src: an #RTPSource
 151  * @address: the RTP address to set
 152  *
 153  * Set that @src is receiving RTP packets from @address. This is used for
 154  * collistion checking.
 155  */
 156 void
 157 rtp_source_set_rtp_from (RTPSource * src, GstNetAddress * address)
 158 {
 159   g_return_if_fail (RTP_IS_SOURCE (src));
 160
 161   src->have_rtp_from = TRUE;
 162   memcpy (&src->rtp_from, address, sizeof (GstNetAddress));
 163 }
 164
 165 /**
 166  * rtp_source_set_rtcp_from:
 167  * @src: an #RTPSource
 168  * @address: the RTCP address to set
 169  *
 170  * Set that @src is receiving RTCP packets from @address. This is used for
 171  * collistion checking.
 172  */
 173 void
 174 rtp_source_set_rtcp_from (RTPSource * src, GstNetAddress * address)
 175 {
 176   g_return_if_fail (RTP_IS_SOURCE (src));
 177
 178   src->have_rtcp_from = TRUE;
 179   memcpy (&src->rtcp_from, address, sizeof (GstNetAddress));
 180 }
 181
 182 static GstFlowReturn
 183 push_packet (RTPSource * src, GstBuffer * buffer)
 184 {
 185   GstFlowReturn ret = GST_FLOW_OK;
 186
 187   /* push queued packets first if any */
 188   while (!g_queue_is_empty (src->packets)) {
 189     GstBuffer *buffer = GST_BUFFER_CAST (g_queue_pop_head (src->packets));
 190
 191     GST_DEBUG ("pushing queued packet");
 192     if (src->callbacks.push_rtp)
 193       src->callbacks.push_rtp (src, buffer, src->user_data);
 194     else
 195       gst_buffer_unref (buffer);
 196   }
 197   GST_DEBUG ("pushing new packet");
 198   /* push packet */
 199   if (src->callbacks.push_rtp)
 200     ret = src->callbacks.push_rtp (src, buffer, src->user_data);
 201   else
 202     gst_buffer_unref (buffer);
 203
 204   return ret;
 205 }
 206
 207 static gint
 208 get_clock_rate (RTPSource * src, guint8 payload)
 209 {
 210   if (payload != src->payload) {
 211     gint clock_rate = -1;
 212
 213     if (src->callbacks.clock_rate)
 214       clock_rate = src->callbacks.clock_rate (src, payload, src->user_data);
 215
 216     GST_DEBUG ("new payload %d, got clock-rate %d", payload, clock_rate);
 217
 218     src->clock_rate = clock_rate;
 219     src->payload = payload;
 220   }
 221   return src->clock_rate;
 222 }
 223
 224 static void
 225 calculate_jitter (RTPSource * src, GstBuffer * buffer,
 226     RTPArrivalStats * arrival)
 227 {
 228   GstClockTime current;
 229   guint32 rtparrival, transit, rtptime;
 230   gint32 diff;
 231   gint clock_rate;
 232   guint8 pt;
 233
 234   /* get arrival time */
 235   if ((current = arrival->time) == GST_CLOCK_TIME_NONE)
 236     goto no_time;
 237
 238   pt = gst_rtp_buffer_get_payload_type (buffer);
 239
 240   /* get clockrate */
 241   if ((clock_rate = get_clock_rate (src, pt)) == -1)
 242     goto no_clock_rate;
 243
 244   rtptime = gst_rtp_buffer_get_timestamp (buffer);
 245
 246   /* convert arrival time to RTP timestamp units */
 247   rtparrival = gst_util_uint64_scale_int (current, clock_rate, GST_SECOND);
 248
 249   /* transit time is difference with RTP timestamp */
 250   transit = rtparrival - rtptime;
 251
 252   /* get ABS diff with previous transit time */
 253   if (src->stats.transit != -1) {
 254     if (transit > src->stats.transit)
 255       diff = transit - src->stats.transit;
 256     else
 257       diff = src->stats.transit - transit;
 258   } else
 259     diff = 0;
 260
 261   src->stats.transit = transit;
 262
 263   /* update jitter, the value we store is scaled up so we can keep precision. */
 264   src->stats.jitter += diff - ((src->stats.jitter + 8) >> 4);
 265
 266   src->stats.prev_rtptime = src->stats.last_rtptime;
 267   src->stats.last_rtptime = rtparrival;
 268
 269   GST_DEBUG ("rtparrival %u, rtptime %u, clock-rate %d, diff %d, jitter: %u",
 270       rtparrival, rtptime, clock_rate, diff, src->stats.jitter);
 271
 272   return;
 273
 274   /* ERRORS */
 275 no_time:
 276   {
 277     GST_WARNING ("cannot get current time");
 278     return;
 279   }
 280 no_clock_rate:
 281   {
 282     GST_WARNING ("cannot get clock-rate for pt %d", pt);
 283     return;
 284   }
 285 }
 286
 287 static void
 288 init_seq (RTPSource * src, guint16 seq)
 289 {
 290   src->stats.base_seq = seq;
 291   src->stats.max_seq = seq;
 292   src->stats.bad_seq = RTP_SEQ_MOD + 1; /* so seq == bad_seq is false */
 293   src->stats.cycles = 0;
 294   src->stats.packets_received = 0;
 295   src->stats.octets_received = 0;
 296   src->stats.bytes_received = 0;
 297   src->stats.prev_received = 0;
 298   src->stats.prev_expected = 0;
 299
 300   GST_DEBUG ("base_seq %d", seq);
 301 }
 302
 303 /**
 304  * rtp_source_process_rtp:
 305  * @src: an #RTPSource
 306  * @buffer: an RTP buffer
 307  *
 308  * Let @src handle the incomming RTP @buffer.
 309  *
 310  * Returns: a #GstFlowReturn.
 311  */
 312 GstFlowReturn
 313 rtp_source_process_rtp (RTPSource * src, GstBuffer * buffer,
 314     RTPArrivalStats * arrival)
 315 {
 316   GstFlowReturn result = GST_FLOW_OK;
 317   guint16 seqnr, udelta;
 318   RTPSourceStats *stats;
 319
 320   g_return_val_if_fail (RTP_IS_SOURCE (src), GST_FLOW_ERROR);
 321   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
 322
 323   stats = &src->stats;
 324
 325   seqnr = gst_rtp_buffer_get_seq (buffer);
 326
 327   if (stats->cycles == -1) {
 328     GST_DEBUG ("received first buffer");
 329     /* first time we heard of this source */
 330     init_seq (src, seqnr);
 331     src->stats.max_seq = seqnr - 1;
 332     src->probation = RTP_DEFAULT_PROBATION;
 333   }
 334
 335   udelta = seqnr - stats->max_seq;
 336
 337   /* if we are still on probation, check seqnum */
 338   if (src->probation) {
 339     guint16 expected;
 340
 341     expected = src->stats.max_seq + 1;
 342
 343     /* when in probation, we require consecutive seqnums */
 344     if (seqnr == expected) {
 345       /* expected packet */
 346       GST_DEBUG ("probation: seqnr %d == expected %d", seqnr, expected);
 347       src->probation--;
 348       src->stats.max_seq = seqnr;
 349       if (src->probation == 0) {
 350         GST_DEBUG ("probation done!", src->probation);
 351         init_seq (src, seqnr);
 352       } else {
 353         GstBuffer *q;
 354
 355         GST_DEBUG ("probation %d: queue buffer", src->probation);
 356         /* when still in probation, keep packets in a list. */
 357         g_queue_push_tail (src->packets, buffer);
 358         /* remove packets from queue if there are too many */
 359         while (g_queue_get_length (src->packets) > RTP_MAX_PROBATION_LEN) {
 360           q = g_queue_pop_head (src->packets);
 361           gst_object_unref (q);
 362         }
 363         goto done;
 364       }
 365     } else {
 366       GST_DEBUG ("probation: seqnr %d != expected %d", seqnr, expected);
 367       src->probation = RTP_DEFAULT_PROBATION;
 368       src->stats.max_seq = seqnr;
 369       goto done;
 370     }
 371   } else if (udelta < RTP_MAX_DROPOUT) {
 372     /* in order, with permissible gap */
 373     if (seqnr < stats->max_seq) {
 374       /* sequence number wrapped - count another 64K cycle. */
 375       stats->cycles += RTP_SEQ_MOD;
 376     }
 377     stats->max_seq = seqnr;
 378   } else if (udelta <= RTP_SEQ_MOD - RTP_MAX_MISORDER) {
 379     /* the sequence number made a very large jump */
 380     if (seqnr == stats->bad_seq) {
 381       /* two sequential packets -- assume that the other side
 382        * restarted without telling us so just re-sync
 383        * (i.e., pretend this was the first packet).  */
 384       init_seq (src, seqnr);
 385     } else {
 386       /* unacceptable jump */
 387       stats->bad_seq = (seqnr + 1) & (RTP_SEQ_MOD - 1);
 388       goto bad_sequence;
 389     }
 390   } else {
 391     /* duplicate or reordered packet, will be filtered by jitterbuffer. */
 392   }
 393
 394   src->stats.octets_received += arrival->payload_len;
 395   src->stats.bytes_received += arrival->bytes;
 396   src->stats.packets_received++;
 397   /* the source that sent the packet must be a sender */
 398   src->is_sender = TRUE;
 399   src->validated = TRUE;
 400
 401   GST_DEBUG ("seq %d, PC: %" G_GUINT64_FORMAT ", OC: %" G_GUINT64_FORMAT,
 402       seqnr, src->stats.packets_received, src->stats.octets_received);
 403
 404   /* calculate jitter */
 405   calculate_jitter (src, buffer, arrival);
 406
 407   /* we're ready to push the RTP packet now */
 408   result = push_packet (src, buffer);
 409
 410 done:
 411   return result;
 412
 413   /* ERRORS */
 414 bad_sequence:
 415   {
 416     GST_WARNING ("unacceptable seqnum received");
 417     return GST_FLOW_OK;
 418   }
 419 }
 420
 421 /**
 422  * rtp_source_process_bye:
 423  * @src: an #RTPSource
 424  * @reason: the reason for leaving
 425  *
 426  * Notify @src that a BYE packet has been received. This will make the source
 427  * inactive.
 428  */
 429 void
 430 rtp_source_process_bye (RTPSource * src, const gchar * reason)
 431 {
 432   g_return_if_fail (RTP_IS_SOURCE (src));
 433
 434   GST_DEBUG ("marking SSRC %08x as BYE, reason: %s", src->ssrc,
 435       GST_STR_NULL (reason));
 436
 437   /* copy the reason and mark as received_bye */
 438   g_free (src->bye_reason);
 439   src->bye_reason = g_strdup (reason);
 440   src->received_bye = TRUE;
 441 }
 442
 443 /**
 444  * rtp_source_send_rtp:
 445  * @src: an #RTPSource
 446  * @buffer: an RTP buffer
 447  *
 448  * Send an RTP @buffer originating from @src. This will make @src a sender.
 449  *
 450  * Returns: a #GstFlowReturn.
 451  */
 452 GstFlowReturn
 453 rtp_source_send_rtp (RTPSource * src, GstBuffer * buffer)
 454 {
 455   GstFlowReturn result = GST_FLOW_OK;
 456   guint len;
 457
 458   g_return_val_if_fail (RTP_IS_SOURCE (src), GST_FLOW_ERROR);
 459   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
 460
 461   len = gst_rtp_buffer_get_payload_len (buffer);
 462
 463   /* we are a sender now */
 464   src->is_sender = TRUE;
 465
 466   /* update stats for the SR */
 467   src->stats.packets_sent++;
 468   src->stats.octets_sent += len;
 469
 470
 471   /* push packet */
 472   if (src->callbacks.push_rtp) {
 473     GST_DEBUG ("pushing RTP packet %u", src->stats.packets_sent);
 474     result = src->callbacks.push_rtp (src, buffer, src->user_data);
 475   } else {
 476     GST_DEBUG ("no callback installed");
 477     gst_buffer_unref (buffer);
 478   }
 479
 480   return result;
 481 }
 482
 483 /**
 484  * rtp_source_process_sr:
 485  * @src: an #RTPSource
 486  * @ntptime: the NTP time
 487  * @rtptime: the RTP time
 488  * @packet_count: the packet count
 489  * @octet_count: the octect count
 490  * @time: time of packet arrival
 491  *
 492  * Update the sender report in @src.
 493  */
 494 void
 495 rtp_source_process_sr (RTPSource * src, guint64 ntptime, guint32 rtptime,
 496     guint32 packet_count, guint32 octet_count, GstClockTime time)
 497 {
 498   RTPSenderReport *curr;
 499   gint curridx;
 500
 501   g_return_if_fail (RTP_IS_SOURCE (src));
 502
 503   GST_DEBUG ("got SR packet: SSRC %08x, NTP %08x:%08x, RTP %u, PC %u, OC %u",
 504       src->ssrc, ntptime >> 32, ntptime & 0xffffffff, rtptime, packet_count,
 505       octet_count);
 506
 507   curridx = src->stats.curr_sr ^ 1;
 508   curr = &src->stats.sr[curridx];
 509
 510   /* this is a sender now */
 511   src->is_sender = TRUE;
 512
 513   /* update current */
 514   curr->is_valid = TRUE;
 515   curr->ntptime = ntptime;
 516   curr->rtptime = rtptime;
 517   curr->packet_count = packet_count;
 518   curr->octet_count = octet_count;
 519   curr->time = time;
 520
 521   /* make current */
 522   src->stats.curr_sr = curridx;
 523 }
 524
 525 /**
 526  * rtp_source_process_rb:
 527  * @src: an #RTPSource
 528  * @fractionlost: fraction lost since last SR/RR
 529  * @packetslost: the cumululative number of packets lost
 530  * @exthighestseq: the extended last sequence number received
 531  * @jitter: the interarrival jitter
 532  * @lsr: the last SR packet from this source
 533  * @dlsr: the delay since last SR packet
 534  *
 535  * Update the report block in @src.
 536  */
 537 void
 538 rtp_source_process_rb (RTPSource * src, guint8 fractionlost, gint32 packetslost,
 539     guint32 exthighestseq, guint32 jitter, guint32 lsr, guint32 dlsr)
 540 {
 541   RTPReceiverReport *curr;
 542   gint curridx;
 543
 544   g_return_if_fail (RTP_IS_SOURCE (src));
 545
 546   GST_DEBUG ("got RB packet %d: SSRC %08x, FL %u"
 547       ", PL %u, HS %u, JITTER %u, LSR %08x, DLSR %08x", src->ssrc, fractionlost,
 548       packetslost, exthighestseq, jitter, lsr, dlsr);
 549
 550   curridx = src->stats.curr_rr ^ 1;
 551   curr = &src->stats.rr[curridx];
 552
 553   /* update current */
 554   curr->is_valid = TRUE;
 555   curr->fractionlost = fractionlost;
 556   curr->packetslost = packetslost;
 557   curr->exthighestseq = exthighestseq;
 558   curr->jitter = jitter;
 559   curr->lsr = lsr;
 560   curr->dlsr = dlsr;
 561
 562   /* make current */
 563   src->stats.curr_rr = curridx;
 564 }
 565
 566 /**
 567  * rtp_source_get_last_sr:
 568  * @src: an #RTPSource
 569  * @ntptime: the NTP time
 570  * @rtptime: the RTP time
 571  * @packet_count: the packet count
 572  * @octet_count: the octect count
 573  * @time: time of packet arrival
 574  *
 575  * Get the values of the last sender report as set with rtp_source_process_sr().
 576  *
 577  * Returns: %TRUE if there was a valid SR report.
 578  */
 579 gboolean
 580 rtp_source_get_last_sr (RTPSource * src, guint64 * ntptime, guint32 * rtptime,
 581     guint32 * packet_count, guint32 * octet_count, GstClockTime * time)
 582 {
 583   RTPSenderReport *curr;
 584
 585   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 586
 587   curr = &src->stats.sr[src->stats.curr_sr];
 588   if (!curr->is_valid)
 589     return FALSE;
 590
 591   if (ntptime)
 592     *ntptime = curr->ntptime;
 593   if (rtptime)
 594     *rtptime = curr->rtptime;
 595   if (packet_count)
 596     *packet_count = curr->packet_count;
 597   if (octet_count)
 598     *octet_count = curr->octet_count;
 599   if (time)
 600     *time = curr->time;
 601
 602   return TRUE;
 603 }
 604
 605 /**
 606  * rtp_source_get_last_rb:
 607  * @src: an #RTPSource
 608  * @fractionlost: fraction lost since last SR/RR
 609  * @packetslost: the cumululative number of packets lost
 610  * @exthighestseq: the extended last sequence number received
 611  * @jitter: the interarrival jitter
 612  * @lsr: the last SR packet from this source
 613  * @dlsr: the delay since last SR packet
 614  *
 615  * Get the values of the last RB report set with rtp_source_process_rb().
 616  *
 617  * Returns: %TRUE if there was a valid SB report.
 618  */
 619 gboolean
 620 rtp_source_get_last_rb (RTPSource * src, guint8 * fractionlost,
 621     gint32 * packetslost, guint32 * exthighestseq, guint32 * jitter,
 622     guint32 * lsr, guint32 * dlsr)
 623 {
 624   RTPReceiverReport *curr;
 625
 626   g_return_val_if_fail (RTP_IS_SOURCE (src), FALSE);
 627
 628   curr = &src->stats.rr[src->stats.curr_rr];
 629   if (!curr->is_valid)
 630     return FALSE;
 631
 632   if (fractionlost)
 633     *fractionlost = curr->fractionlost;
 634   if (packetslost)
 635     *packetslost = curr->packetslost;
 636   if (exthighestseq)
 637     *exthighestseq = curr->exthighestseq;
 638   if (jitter)
 639     *jitter = curr->jitter;
 640   if (lsr)
 641     *lsr = curr->lsr;
 642   if (dlsr)
 643     *dlsr = curr->dlsr;
 644
 645   return TRUE;
 646 }