subprojects/gst-plugins-good/gst/rtp/gstrtpulpfecdec.c

   1 /* GStreamer plugin for forward error correction
   2  * Copyright (C) 2017 Pexip
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Lesser General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2.1 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  * Lesser General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Lesser General Public
  15  * License along with this library; if not, write to the Free Software
  16  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
  17  *
  18  * Author: Mikhail Fludkov <misha@pexip.com>
  19  */
  20
  21 /**
  22  * SECTION:element-rtpulpfecdec
  23  * @short_description: Generic RTP Forward Error Correction (FEC) decoder
  24  * @title: rtpulpfecdec
  25  *
  26  * Generic Forward Error Correction (FEC) decoder for Uneven Level
  27  * Protection (ULP) as described in RFC 5109.
  28  *
  29  * It differs from the RFC in one important way, it multiplexes the
  30  * FEC packets in the same sequence number as media packets. This is to be
  31  * compatible with libwebrtc as using in Google Chrome and with Microsoft
  32  * Lync / Skype for Business.
  33  *
  34  * This element will work in combination with an upstream #GstRtpStorage
  35  * element and attempt to recover packets declared lost through custom
  36  * 'GstRTPPacketLost' events, usually emitted by #GstRtpJitterBuffer.
  37  *
  38  * If no storage is provided using the #GstRtpUlpFecDec:storage
  39  * property, it will try to get it from an element upstream.
  40  *
  41  * Additionally, the payload types of the protection packets *must* be
  42  * provided to this element via its #GstRtpUlpFecDec:pt property.
  43  *
  44  * When using #GstRtpBin, this element should be inserted through the
  45  * #GstRtpBin::request-fec-decoder signal.
  46  *
  47  * ## Example pipeline
  48  *
  49  * |[
  50  * gst-launch-1.0 udpsrc port=8888 caps="application/x-rtp, payload=96, clock-rate=90000" ! rtpstorage size-time=220000000 ! rtpssrcdemux ! application/x-rtp, payload=96, clock-rate=90000, media=video, encoding-name=H264 ! rtpjitterbuffer do-lost=1 latency=200 !  rtpulpfecdec pt=122 ! rtph264depay ! avdec_h264 ! videoconvert ! autovideosink
  51  * ]| This example will receive a stream with FEC and try to reconstruct the packets.
  52  *
  53  * Example programs are available at
  54  * <https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/blob/master/examples/src/bin/rtpfecserver.rs>
  55  * and
  56  * <https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/blob/master/examples/src/bin/rtpfecclient.rs>
  57  *
  58  * See also: #GstRtpUlpFecEnc, #GstRtpBin, #GstRtpStorage
  59  * Since: 1.14
  60  */
  61
  62 #include <gst/rtp/gstrtpbuffer.h>
  63 #include <gst/rtp/gstrtp-enumtypes.h>
  64
  65 #include "gstrtpelements.h"
  66 #include "rtpulpfeccommon.h"
  67 #include "gstrtpulpfecdec.h"
  68
  69 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  70     GST_PAD_SINK,
  71     GST_PAD_ALWAYS,
  72     GST_STATIC_CAPS ("application/x-rtp")
  73     );
  74
  75 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  76     GST_PAD_SRC,
  77     GST_PAD_ALWAYS,
  78     GST_STATIC_CAPS ("application/x-rtp")
  79     );
  80
  81 enum
  82 {
  83   PROP_0,
  84   PROP_PT,
  85   PROP_STORAGE,
  86   PROP_RECOVERED,
  87   PROP_UNRECOVERED,
  88   PROP_PASSTHROUGH,
  89   N_PROPERTIES
  90 };
  91
  92 #define DEFAULT_FEC_PT 0
  93 #define DEFAULT_PASSTHROUGH FALSE
  94
  95 static GParamSpec *klass_properties[N_PROPERTIES] = { NULL, };
  96
  97 GST_DEBUG_CATEGORY (gst_rtp_ulpfec_dec_debug);
  98 #define GST_CAT_DEFAULT (gst_rtp_ulpfec_dec_debug)
  99
 100 G_DEFINE_TYPE (GstRtpUlpFecDec, gst_rtp_ulpfec_dec, GST_TYPE_ELEMENT);
 101 GST_ELEMENT_REGISTER_DEFINE_WITH_CODE (rtpulpfecdec, "rtpulpfecdec",
 102     GST_RANK_NONE, GST_TYPE_RTP_ULPFEC_DEC, rtp_element_init (plugin));
 103
 104 #define RTP_FEC_MAP_INFO_NTH(dec, data) (&g_array_index (\
 105     ((GstRtpUlpFecDec *)dec)->info_arr, \
 106     RtpUlpFecMapInfo, \
 107     GPOINTER_TO_UINT(data)))
 108
 109 static gint
 110 _compare_fec_map_info (gconstpointer a, gconstpointer b, gpointer userdata)
 111 {
 112   guint16 aseq =
 113       gst_rtp_buffer_get_seq (&RTP_FEC_MAP_INFO_NTH (userdata, a)->rtp);
 114   guint16 bseq =
 115       gst_rtp_buffer_get_seq (&RTP_FEC_MAP_INFO_NTH (userdata, b)->rtp);
 116   return gst_rtp_buffer_compare_seqnum (bseq, aseq);
 117 }
 118
 119 static void
 120 gst_rtp_ulpfec_dec_start (GstRtpUlpFecDec * self, GstBufferList * buflist,
 121     guint8 fec_pt, guint16 lost_seq)
 122 {
 123   guint fec_packets = 0;
 124   gsize i;
 125
 126   g_assert (NULL == self->info_media);
 127   g_assert (0 == self->info_fec->len);
 128   g_assert (0 == self->info_arr->len);
 129
 130   g_array_set_size (self->info_arr, gst_buffer_list_length (buflist));
 131
 132   for (i = 0;
 133       i < gst_buffer_list_length (buflist) && !self->lost_packet_from_storage;
 134       ++i) {
 135     GstBuffer *buffer = gst_buffer_list_get (buflist, i);
 136     RtpUlpFecMapInfo *info = RTP_FEC_MAP_INFO_NTH (self, i);
 137
 138     if (!rtp_ulpfec_map_info_map (gst_buffer_ref (buffer), info))
 139       g_assert_not_reached ();
 140
 141     if (fec_pt == gst_rtp_buffer_get_payload_type (&info->rtp)) {
 142       GST_DEBUG_RTP_PACKET (self, "rtp header (fec)", &info->rtp);
 143
 144       ++fec_packets;
 145       if (rtp_ulpfec_buffer_is_valid (&info->rtp)) {
 146         GST_DEBUG_FEC_PACKET (self, &info->rtp);
 147         g_ptr_array_add (self->info_fec, GUINT_TO_POINTER (i));
 148       }
 149     } else {
 150       GST_LOG_RTP_PACKET (self, "rtp header (incoming)", &info->rtp);
 151
 152       if (lost_seq == gst_rtp_buffer_get_seq (&info->rtp)) {
 153         GST_DEBUG_OBJECT (self, "Received lost packet from the storage");
 154         g_list_free (self->info_media);
 155         self->info_media = NULL;
 156         self->lost_packet_from_storage = TRUE;
 157       }
 158       self->info_media =
 159           g_list_insert_sorted_with_data (self->info_media,
 160           GUINT_TO_POINTER (i), _compare_fec_map_info, self);
 161     }
 162   }
 163   if (!self->lost_packet_from_storage) {
 164     self->fec_packets_received += fec_packets;
 165     self->fec_packets_rejected += fec_packets - self->info_fec->len;
 166   }
 167 }
 168
 169 static void
 170 gst_rtp_ulpfec_dec_stop (GstRtpUlpFecDec * self)
 171 {
 172   g_array_set_size (self->info_arr, 0);
 173   g_ptr_array_set_size (self->info_fec, 0);
 174   g_list_free (self->info_media);
 175   self->info_media = NULL;
 176   self->lost_packet_from_storage = FALSE;
 177   self->lost_packet_returned = FALSE;
 178 }
 179
 180 static guint64
 181 gst_rtp_ulpfec_dec_get_media_buffers_mask (GstRtpUlpFecDec * self,
 182     guint16 fec_seq_base)
 183 {
 184   guint64 mask = 0;
 185   GList *it;
 186
 187   for (it = self->info_media; it; it = it->next) {
 188     RtpUlpFecMapInfo *info = RTP_FEC_MAP_INFO_NTH (self, it->data);
 189     mask |=
 190         rtp_ulpfec_packet_mask_from_seqnum (gst_rtp_buffer_get_seq (&info->rtp),
 191         fec_seq_base, TRUE);
 192   }
 193   return mask;
 194 }
 195
 196 static gboolean
 197 gst_rtp_ulpfec_dec_is_recovered_pt_valid (GstRtpUlpFecDec * self, gint media_pt,
 198     guint8 recovered_pt)
 199 {
 200   GList *it;
 201   if (media_pt == recovered_pt)
 202     return TRUE;
 203
 204   for (it = self->info_media; it; it = it->next) {
 205     RtpUlpFecMapInfo *info = RTP_FEC_MAP_INFO_NTH (self, it->data);
 206     if (gst_rtp_buffer_get_payload_type (&info->rtp) == recovered_pt)
 207       return TRUE;
 208   }
 209   return FALSE;
 210 }
 211
 212 static GstBuffer *
 213 gst_rtp_ulpfec_dec_recover_from_fec (GstRtpUlpFecDec * self,
 214     RtpUlpFecMapInfo * info_fec, guint32 ssrc, gint media_pt, guint16 seq,
 215     guint8 * dst_pt)
 216 {
 217   guint64 fec_mask = rtp_ulpfec_buffer_get_mask (&info_fec->rtp);
 218   gboolean fec_mask_long = rtp_ulpfec_buffer_get_fechdr (&info_fec->rtp)->L;
 219   guint16 fec_seq_base = rtp_ulpfec_buffer_get_seq_base (&info_fec->rtp);
 220   GstBuffer *ret;
 221   GList *it;
 222
 223   g_array_set_size (self->scratch_buf, 0);
 224   rtp_buffer_to_ulpfec_bitstring (&info_fec->rtp, self->scratch_buf, TRUE,
 225       fec_mask_long);
 226
 227   for (it = self->info_media; it; it = it->next) {
 228     RtpUlpFecMapInfo *info = RTP_FEC_MAP_INFO_NTH (self, it->data);
 229     guint64 packet_mask =
 230         rtp_ulpfec_packet_mask_from_seqnum (gst_rtp_buffer_get_seq (&info->rtp),
 231         fec_seq_base, TRUE);
 232
 233     if (fec_mask & packet_mask) {
 234       fec_mask ^= packet_mask;
 235       rtp_buffer_to_ulpfec_bitstring (&info->rtp, self->scratch_buf, FALSE,
 236           fec_mask_long);
 237     }
 238   }
 239
 240   ret =
 241       rtp_ulpfec_bitstring_to_media_rtp_buffer (self->scratch_buf,
 242       fec_mask_long, ssrc, seq);
 243   if (ret) {
 244     /* We are about to put recovered packet back in self->info_media to be able
 245      * to reuse it later for recovery of other packets
 246      **/
 247     gint i = self->info_arr->len;
 248     RtpUlpFecMapInfo *info;
 249     guint8 recovered_pt;
 250
 251     g_array_set_size (self->info_arr, self->info_arr->len + 1);
 252     info = RTP_FEC_MAP_INFO_NTH (self, i);
 253
 254     if (!rtp_ulpfec_map_info_map (gst_buffer_ref (ret), info)) {
 255       GST_WARNING_OBJECT (self, "Invalid recovered packet");
 256       goto recovered_packet_invalid;
 257     }
 258
 259     recovered_pt = gst_rtp_buffer_get_payload_type (&info->rtp);
 260     if (!gst_rtp_ulpfec_dec_is_recovered_pt_valid (self, media_pt,
 261             recovered_pt)) {
 262       GST_WARNING_OBJECT (self,
 263           "Recovered packet has unexpected payload type (%u)", recovered_pt);
 264       goto recovered_packet_invalid;
 265     }
 266
 267     GST_DEBUG_RTP_PACKET (self, "rtp header (recovered)", &info->rtp);
 268     self->info_media =
 269         g_list_insert_sorted_with_data (self->info_media, GUINT_TO_POINTER (i),
 270         _compare_fec_map_info, self);
 271     *dst_pt = recovered_pt;
 272   }
 273   return ret;
 274
 275 recovered_packet_invalid:
 276   g_array_set_size (self->info_arr, self->info_arr->len - 1);
 277   gst_buffer_unref (ret);
 278   return NULL;
 279 }
 280
 281 static GstBuffer *
 282 gst_rtp_ulpfec_dec_recover_from_storage (GstRtpUlpFecDec * self,
 283     guint8 * dst_pt, guint16 * dst_seq)
 284 {
 285   RtpUlpFecMapInfo *info;
 286
 287   if (self->lost_packet_returned)
 288     return NULL;
 289
 290   g_assert (g_list_length (self->info_media) == 1);
 291
 292   info = RTP_FEC_MAP_INFO_NTH (self, self->info_media->data);
 293   *dst_seq = gst_rtp_buffer_get_seq (&info->rtp);
 294   *dst_pt = gst_rtp_buffer_get_payload_type (&info->rtp);
 295   self->lost_packet_returned = TRUE;
 296   GST_DEBUG_RTP_PACKET (self, "rtp header (recovered)", &info->rtp);
 297   return gst_buffer_ref (info->rtp.buffer);
 298 }
 299
 300 /* __has_builtin only works with clang, so test compiler version for gcc */
 301 /* Intel compiler and MSVC probably have their own things as well */
 302 /* TODO: make sure we use builtin for clang as well */
 303 #if defined(__GNUC__) && __GNUC__ >= 4
 304 #define rtp_ulpfec_ctz64 __builtin_ctzll
 305 #else
 306 static inline gint
 307 rtp_ulpfec_ctz64_inline (guint64 mask)
 308 {
 309   gint nth_bit = 0;
 310
 311   do {
 312     if ((mask & 1))
 313       return nth_bit;
 314     mask = mask >> 1;
 315   } while (++nth_bit < 64);
 316
 317   return -1;                    /* should not be reached, since mask must not be 0 */
 318 }
 319
 320 #define rtp_ulpfec_ctz64 rtp_ulpfec_ctz64_inline
 321 #endif
 322
 323 static GstBuffer *
 324 gst_rtp_ulpfec_dec_recover (GstRtpUlpFecDec * self, guint32 ssrc, gint media_pt,
 325     guint8 * dst_pt, guint16 * dst_seq)
 326 {
 327   guint64 media_mask = 0;
 328   gint media_mask_seq_base = -1;
 329   gsize i;
 330
 331   if (self->lost_packet_from_storage)
 332     return gst_rtp_ulpfec_dec_recover_from_storage (self, dst_pt, dst_seq);
 333
 334   /* Looking for a FEC packet which can be used for recovery */
 335   for (i = 0; i < self->info_fec->len; ++i) {
 336     RtpUlpFecMapInfo *info = RTP_FEC_MAP_INFO_NTH (self,
 337         g_ptr_array_index (self->info_fec, i));
 338     guint16 seq_base = rtp_ulpfec_buffer_get_seq_base (&info->rtp);
 339     guint64 fec_mask = rtp_ulpfec_buffer_get_mask (&info->rtp);
 340     guint64 missing_packets_mask;
 341
 342     if (media_mask_seq_base != (gint) seq_base) {
 343       media_mask_seq_base = seq_base;
 344       media_mask = gst_rtp_ulpfec_dec_get_media_buffers_mask (self, seq_base);
 345     }
 346
 347     /* media_mask has 1s if packet exist.
 348      * fec_mask is the mask of protected packets
 349      * The statement below excludes existing packets from the protected. So
 350      * we are left with 1s only for missing packets which can be recovered
 351      * by this FEC packet. */
 352     missing_packets_mask = fec_mask & (~media_mask);
 353
 354     /* Do we have any 1s? Checking if current FEC packet can be used for recovery */
 355     if (0 != missing_packets_mask) {
 356       guint trailing_zeros = rtp_ulpfec_ctz64 (missing_packets_mask);
 357
 358       /* Is it the only 1 in the mask? Checking if we lacking single packet in
 359        * that case FEC packet can be used for recovery */
 360       if (missing_packets_mask == (G_GUINT64_CONSTANT (1) << trailing_zeros)) {
 361         GstBuffer *ret;
 362
 363         *dst_seq =
 364             seq_base + (RTP_ULPFEC_SEQ_BASE_OFFSET_MAX (TRUE) - trailing_zeros);
 365         ret =
 366             gst_rtp_ulpfec_dec_recover_from_fec (self, info, ssrc, media_pt,
 367             *dst_seq, dst_pt);
 368         if (ret)
 369           return ret;
 370       }
 371     }
 372   }
 373   return NULL;
 374 }
 375
 376 static GstFlowReturn
 377 gst_rtp_ulpfec_dec_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
 378 {
 379   GstRtpUlpFecDec *self = GST_RTP_ULPFEC_DEC (parent);
 380
 381   if (G_LIKELY (GST_FLOW_OK == self->chain_return_val)) {
 382     GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 383     gboolean passthrough;
 384     buf = gst_buffer_make_writable (buf);
 385
 386     if (G_UNLIKELY (self->unset_discont_flag)) {
 387       self->unset_discont_flag = FALSE;
 388       GST_BUFFER_FLAG_UNSET (buf, GST_BUFFER_FLAG_DISCONT);
 389     }
 390
 391     GST_OBJECT_LOCK (self);
 392     if (G_UNLIKELY (self->needs_discont)) {
 393       self->needs_discont = FALSE;
 394       GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT);
 395     }
 396     passthrough = self->passthrough;
 397     GST_OBJECT_UNLOCK (self);
 398
 399     gst_rtp_buffer_map (buf, GST_MAP_WRITE, &rtp);
 400     if (passthrough) {
 401       self->next_seqnum = gst_rtp_buffer_get_seq (&rtp) + 1;
 402     } else {
 403       gst_rtp_buffer_set_seq (&rtp, self->next_seqnum++);
 404     }
 405     gst_rtp_buffer_unmap (&rtp);
 406
 407     return gst_pad_push (self->srcpad, buf);
 408   }
 409
 410   gst_buffer_unref (buf);
 411   return self->chain_return_val;
 412 }
 413
 414 static gboolean
 415 gst_rtp_ulpfec_dec_handle_packet_loss (GstRtpUlpFecDec * self, guint16 seqnum,
 416     GstClockTime timestamp, GstClockTime duration)
 417 {
 418   gint caps_pt = self->have_caps_pt ? self->caps_pt : -1;
 419   gboolean ret = TRUE;
 420   GstBufferList *buflist =
 421       rtp_storage_get_packets_for_recovery (self->storage, self->fec_pt,
 422       self->caps_ssrc, seqnum);
 423
 424   if (buflist) {
 425     GstBuffer *recovered_buffer = NULL;
 426     guint16 recovered_seq = 0;
 427     guint8 recovered_pt = 0;
 428
 429     gst_rtp_ulpfec_dec_start (self, buflist, self->fec_pt, seqnum);
 430
 431     while (NULL != (recovered_buffer =
 432             gst_rtp_ulpfec_dec_recover (self, self->caps_ssrc, caps_pt,
 433                 &recovered_pt, &recovered_seq))) {
 434       if (seqnum == recovered_seq) {
 435         GstBuffer *sent_buffer;
 436         GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 437
 438         recovered_buffer = gst_buffer_make_writable (recovered_buffer);
 439         GST_BUFFER_PTS (recovered_buffer) = timestamp;
 440         /* GST_BUFFER_DURATION (recovered_buffer) = duration;
 441          * JB does not set the duration, so we will not too */
 442
 443         if (!self->lost_packet_from_storage)
 444           rtp_storage_put_recovered_packet (self->storage,
 445               recovered_buffer, recovered_pt, self->caps_ssrc, recovered_seq);
 446
 447         GST_DEBUG_OBJECT (self,
 448             "Pushing recovered packet ssrc=0x%08x seq=%u %" GST_PTR_FORMAT,
 449             self->caps_ssrc, seqnum, recovered_buffer);
 450
 451         sent_buffer = gst_buffer_copy_deep (recovered_buffer);
 452
 453         if (self->lost_packet_from_storage)
 454           gst_buffer_unref (recovered_buffer);
 455
 456         gst_rtp_buffer_map (sent_buffer, GST_MAP_WRITE, &rtp);
 457         gst_rtp_buffer_set_seq (&rtp, self->next_seqnum++);
 458         gst_rtp_buffer_unmap (&rtp);
 459
 460         GST_OBJECT_LOCK (self);
 461         if (G_UNLIKELY (self->needs_discont)) {
 462           self->needs_discont = FALSE;
 463           GST_BUFFER_FLAG_SET (sent_buffer, GST_BUFFER_FLAG_DISCONT);
 464         }
 465         GST_OBJECT_UNLOCK (self);
 466
 467         ret = FALSE;
 468         self->unset_discont_flag = TRUE;
 469         self->chain_return_val = gst_pad_push (self->srcpad, sent_buffer);
 470         break;
 471       }
 472
 473       if (!self->lost_packet_from_storage) {
 474         rtp_storage_put_recovered_packet (self->storage,
 475             recovered_buffer, recovered_pt, self->caps_ssrc, recovered_seq);
 476       } else {
 477         gst_buffer_unref (recovered_buffer);
 478       }
 479     }
 480
 481     gst_rtp_ulpfec_dec_stop (self);
 482     gst_buffer_list_unref (buflist);
 483   }
 484
 485   GST_DEBUG_OBJECT (self, "Packet lost ssrc=0x%08x seq=%u", self->caps_ssrc,
 486       seqnum);
 487
 488   return ret;
 489 }
 490
 491 static gboolean
 492 gst_rtp_ulpfec_dec_handle_sink_event (GstPad * pad, GstObject * parent,
 493     GstEvent * event)
 494 {
 495   GstRtpUlpFecDec *self = GST_RTP_ULPFEC_DEC (parent);
 496   gboolean forward = TRUE;
 497
 498   GST_LOG_OBJECT (self, "Received event %" GST_PTR_FORMAT, event);
 499
 500   if (GST_FLOW_OK == self->chain_return_val &&
 501       GST_EVENT_CUSTOM_DOWNSTREAM == GST_EVENT_TYPE (event) &&
 502       gst_event_has_name (event, "GstRTPPacketLost")) {
 503     guint seqnum;
 504     GstClockTime timestamp, duration;
 505     GstStructure *s;
 506     gboolean passthrough;
 507
 508     GST_OBJECT_LOCK (self);
 509     passthrough = self->passthrough;
 510     GST_OBJECT_UNLOCK (self);
 511
 512     if (passthrough) {
 513       GST_TRACE_OBJECT (self,
 514           "in passthrough mode, ignoring packet loss event");
 515       forward = TRUE;
 516       goto out;
 517     }
 518
 519     event = gst_event_make_writable (event);
 520     s = gst_event_writable_structure (event);
 521
 522     g_assert (self->have_caps_ssrc);
 523
 524     if (self->storage == NULL) {
 525       GstQuery *q = gst_query_new_custom (GST_QUERY_CUSTOM,
 526           gst_structure_new_empty ("GstRtpStorage"));
 527
 528       if (gst_pad_peer_query (self->sinkpad, q)) {
 529         const GstStructure *s = gst_query_get_structure (q);
 530
 531         if (gst_structure_has_field_typed (s, "storage", G_TYPE_OBJECT)) {
 532           gst_structure_get (s, "storage", G_TYPE_OBJECT, &self->storage, NULL);
 533         }
 534       }
 535       gst_query_unref (q);
 536     }
 537
 538     if (self->storage == NULL) {
 539       GST_ELEMENT_WARNING (self, STREAM, FAILED, ("Internal storage not found"),
 540           ("You need to add rtpstorage element upstream from rtpulpfecdec."));
 541       return FALSE;
 542     }
 543
 544     if (!gst_structure_get (s,
 545             "seqnum", G_TYPE_UINT, &seqnum,
 546             "timestamp", G_TYPE_UINT64, &timestamp,
 547             "duration", G_TYPE_UINT64, &duration, NULL))
 548       g_assert_not_reached ();
 549
 550     forward =
 551         gst_rtp_ulpfec_dec_handle_packet_loss (self, seqnum, timestamp,
 552         duration);
 553
 554     if (forward) {
 555       gst_structure_remove_field (s, "seqnum");
 556       gst_structure_set (s, "might-have-been-fec", G_TYPE_BOOLEAN, TRUE, NULL);
 557       ++self->packets_unrecovered;
 558     } else {
 559       ++self->packets_recovered;
 560     }
 561
 562     GST_DEBUG_OBJECT (self, "Unrecovered / Recovered: %lu / %lu",
 563         (gulong) self->packets_unrecovered, (gulong) self->packets_recovered);
 564   } else if (GST_EVENT_CAPS == GST_EVENT_TYPE (event)) {
 565     GstCaps *caps;
 566     gboolean have_caps_pt = FALSE;
 567     gboolean have_caps_ssrc = FALSE;
 568     guint caps_ssrc = 0;
 569     gint caps_pt = 0;
 570
 571     gst_event_parse_caps (event, &caps);
 572     have_caps_ssrc =
 573         gst_structure_get_uint (gst_caps_get_structure (caps, 0), "ssrc",
 574         &caps_ssrc);
 575     have_caps_pt =
 576         gst_structure_get_int (gst_caps_get_structure (caps, 0), "payload",
 577         &caps_pt);
 578
 579     if (self->have_caps_ssrc != have_caps_ssrc || self->caps_ssrc != caps_ssrc)
 580       GST_DEBUG_OBJECT (self, "SSRC changed %u, 0x%08x -> %u, 0x%08x",
 581           self->have_caps_ssrc, self->caps_ssrc, have_caps_ssrc, caps_ssrc);
 582     if (self->have_caps_pt != have_caps_pt || self->caps_pt != caps_pt)
 583       GST_DEBUG_OBJECT (self, "PT changed %u, %u -> %u, %u",
 584           self->have_caps_pt, self->caps_pt, have_caps_pt, caps_pt);
 585
 586     self->have_caps_ssrc = have_caps_ssrc;
 587     self->have_caps_pt = have_caps_pt;
 588     self->caps_ssrc = caps_ssrc;
 589     self->caps_pt = caps_pt;
 590   }
 591
 592 out:
 593   if (forward)
 594     return gst_pad_push_event (self->srcpad, event);
 595   gst_event_unref (event);
 596   return TRUE;
 597 }
 598
 599 static void
 600 gst_rtp_ulpfec_dec_init (GstRtpUlpFecDec * self)
 601 {
 602   self->srcpad = gst_pad_new_from_static_template (&srctemplate, "src");
 603   self->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 604   GST_PAD_SET_PROXY_CAPS (self->sinkpad);
 605   GST_PAD_SET_PROXY_ALLOCATION (self->sinkpad);
 606   gst_pad_set_chain_function (self->sinkpad,
 607       GST_DEBUG_FUNCPTR (gst_rtp_ulpfec_dec_chain));
 608   gst_pad_set_event_function (self->sinkpad,
 609       GST_DEBUG_FUNCPTR (gst_rtp_ulpfec_dec_handle_sink_event));
 610
 611   gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
 612   gst_element_add_pad (GST_ELEMENT (self), self->sinkpad);
 613
 614   self->fec_pt = DEFAULT_FEC_PT;
 615   self->passthrough = DEFAULT_PASSTHROUGH;
 616
 617   self->next_seqnum = g_random_int_range (0, G_MAXINT16);
 618
 619   self->chain_return_val = GST_FLOW_OK;
 620   self->have_caps_ssrc = FALSE;
 621   self->caps_ssrc = 0;
 622   self->info_fec = g_ptr_array_new ();
 623   self->info_arr = g_array_new (FALSE, TRUE, sizeof (RtpUlpFecMapInfo));
 624   g_array_set_clear_func (self->info_arr,
 625       (GDestroyNotify) rtp_ulpfec_map_info_unmap);
 626   self->scratch_buf = g_array_new (FALSE, TRUE, sizeof (guint8));
 627 }
 628
 629 static void
 630 gst_rtp_ulpfec_dec_dispose (GObject * obj)
 631 {
 632   GstRtpUlpFecDec *self = GST_RTP_ULPFEC_DEC (obj);
 633
 634   GST_INFO_OBJECT (self,
 635       " ssrc=0x%08x pt=%u"
 636       " packets_recovered=%" G_GSIZE_FORMAT
 637       " packets_unrecovered=%" G_GSIZE_FORMAT,
 638       self->caps_ssrc, self->caps_pt,
 639       self->packets_recovered, self->packets_unrecovered);
 640
 641   if (self->storage)
 642     g_object_unref (self->storage);
 643
 644   g_assert (NULL == self->info_media);
 645   g_assert (0 == self->info_fec->len);
 646   g_assert (0 == self->info_arr->len);
 647
 648   if (self->fec_packets_received) {
 649     GST_INFO_OBJECT (self,
 650         " fec_packets_received=%" G_GSIZE_FORMAT
 651         " fec_packets_rejected=%" G_GSIZE_FORMAT
 652         " packets_rejected=%" G_GSIZE_FORMAT,
 653         self->fec_packets_received,
 654         self->fec_packets_rejected, self->packets_rejected);
 655   }
 656
 657   g_ptr_array_free (self->info_fec, TRUE);
 658   g_array_free (self->info_arr, TRUE);
 659   g_array_free (self->scratch_buf, TRUE);
 660
 661   G_OBJECT_CLASS (gst_rtp_ulpfec_dec_parent_class)->dispose (obj);
 662 }
 663
 664 static void
 665 gst_rtp_ulpfec_dec_set_property (GObject * object, guint prop_id,
 666     const GValue * value, GParamSpec * pspec)
 667 {
 668   GstRtpUlpFecDec *self = GST_RTP_ULPFEC_DEC (object);
 669
 670   switch (prop_id) {
 671     case PROP_PT:
 672       self->fec_pt = g_value_get_uint (value);
 673       break;
 674     case PROP_STORAGE:
 675       if (self->storage)
 676         g_object_unref (self->storage);
 677       self->storage = g_value_get_object (value);
 678       if (self->storage)
 679         g_object_ref (self->storage);
 680       break;
 681     case PROP_PASSTHROUGH:{
 682       gboolean newval = g_value_get_boolean (value);
 683       GST_OBJECT_LOCK (self);
 684       /* if we changing into non-passthrough mode, then the sequence numbers may
 685        * be completely different and we need to advertise that with a discont */
 686       GST_INFO_OBJECT (self, "passthrough changing from %u to %u",
 687           self->passthrough, newval);
 688       if (self->passthrough && !newval) {
 689         self->needs_discont = TRUE;
 690       }
 691       self->passthrough = newval;
 692       GST_OBJECT_UNLOCK (self);
 693       break;
 694     }
 695     default:
 696       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 697       break;
 698   }
 699 }
 700
 701 static void
 702 gst_rtp_ulpfec_dec_get_property (GObject * object, guint prop_id,
 703     GValue * value, GParamSpec * pspec)
 704 {
 705   GstRtpUlpFecDec *self = GST_RTP_ULPFEC_DEC (object);
 706
 707   switch (prop_id) {
 708     case PROP_PT:
 709       g_value_set_uint (value, self->fec_pt);
 710       break;
 711     case PROP_STORAGE:
 712       g_value_set_object (value, self->storage);
 713       break;
 714     case PROP_RECOVERED:
 715       g_value_set_uint (value, (guint) self->packets_recovered);
 716       break;
 717     case PROP_UNRECOVERED:
 718       g_value_set_uint (value, (guint) self->packets_unrecovered);
 719       break;
 720     case PROP_PASSTHROUGH:
 721       g_value_set_boolean (value, self->passthrough);
 722       break;
 723     default:
 724       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 725       break;
 726   }
 727 }
 728
 729 static void
 730 gst_rtp_ulpfec_dec_class_init (GstRtpUlpFecDecClass * klass)
 731 {
 732   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 733   GstElementClass *element_class = GST_ELEMENT_CLASS (klass);
 734
 735   GST_DEBUG_CATEGORY_INIT (gst_rtp_ulpfec_dec_debug,
 736       "rtpulpfecdec", 0, "RTP FEC Decoder");
 737
 738   gst_element_class_add_pad_template (element_class,
 739       gst_static_pad_template_get (&srctemplate));
 740   gst_element_class_add_pad_template (element_class,
 741       gst_static_pad_template_get (&sinktemplate));
 742
 743   gst_element_class_set_static_metadata (element_class,
 744       "RTP FEC Decoder",
 745       "Codec/Depayloader/Network/RTP",
 746       "Decodes RTP FEC (RFC5109)", "Mikhail Fludkov <misha@pexip.com>");
 747
 748   gobject_class->set_property =
 749       GST_DEBUG_FUNCPTR (gst_rtp_ulpfec_dec_set_property);
 750   gobject_class->get_property =
 751       GST_DEBUG_FUNCPTR (gst_rtp_ulpfec_dec_get_property);
 752   gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_rtp_ulpfec_dec_dispose);
 753
 754   klass_properties[PROP_PT] = g_param_spec_uint ("pt", "pt",
 755       "FEC packets payload type", 0, 127,
 756       DEFAULT_FEC_PT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
 757   klass_properties[PROP_STORAGE] =
 758       g_param_spec_object ("storage", "RTP storage", "RTP storage",
 759       G_TYPE_OBJECT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
 760   klass_properties[PROP_RECOVERED] =
 761       g_param_spec_uint ("recovered", "recovered",
 762       "The number of recovered packets", 0, G_MAXUINT, 0,
 763       G_PARAM_READABLE | G_PARAM_STATIC_STRINGS);
 764   klass_properties[PROP_UNRECOVERED] =
 765       g_param_spec_uint ("unrecovered", "unrecovered",
 766       "The number of unrecovered packets", 0, G_MAXUINT, 0,
 767       G_PARAM_READABLE | G_PARAM_STATIC_STRINGS);
 768   /**
 769    * GstRtpUlpFecDec:passthrough:
 770    *
 771    * Whether to push data through without any modification.  If passthrough is
 772    * enabled, then no packets will ever be recovered.
 773    *
 774    * Since: 1.22
 775    */
 776   klass_properties[PROP_PASSTHROUGH] =
 777       g_param_spec_boolean ("passthrough", "Passthrough",
 778       "Whether to passthrough all data as-is without modification and "
 779       "never attempt to recover packets", DEFAULT_PASSTHROUGH,
 780       G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
 781
 782   g_object_class_install_properties (gobject_class, N_PROPERTIES,
 783       klass_properties);
 784
 785   g_assert (rtp_ulpfec_ctz64 (G_GUINT64_CONSTANT (0x1)) == 0);
 786   g_assert (rtp_ulpfec_ctz64 (G_GUINT64_CONSTANT (0x8000000000000000)) == 63);
 787 }