gst/rtp/gstrtpulpfecenc.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-rtpulpfecenc
  23  * @short_description: Generic RTP Forward Error Correction (FEC) encoder
  24  * @title: rtpulpfecenc
  25  *
  26  * Generic Forward Error Correction (FEC) encoder using 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  * Be warned that after using this element, it is no longer possible to know if
  35  * there is a gap in the media stream based on the sequence numbers as the FEC
  36  * packets become interleaved with the media packets.
  37  *
  38  * This element will insert protection packets in any RTP stream, which
  39  * can then be used on the receiving side to recover lost packets.
  40  *
  41  * This element rewrites packets' seqnums, which means that when combined
  42  * with retransmission elements such as #GstRtpRtxSend, it *must* be
  43  * placed upstream of those, otherwise retransmission requests will request
  44  * incorrect seqnums.
  45  *
  46  * A payload type for the protection packets *must* be specified, different
  47  * from the payload type of the protected packets, with the GstRtpUlpFecEnc:pt
  48  * property.
  49  *
  50  * The marker bit of RTP packets is used to determine sets of packets to
  51  * protect as a unit, in order to modulate the level of protection, this
  52  * behaviour can be disabled with GstRtpUlpFecEnc:multipacket, but should
  53  * be left enabled for video streams.
  54  *
  55  * The level of protection can be configured with two properties,
  56  * #GstRtpUlpFecEnc:percentage and #GstRtpUlpFecEnc:percentage-important,
  57  * the element will determine which percentage to use for a given set of
  58  * packets based on the presence of the #GST_BUFFER_FLAG_NON_DROPPABLE
  59  * flag, upstream payloaders are expected to set this flag on "important"
  60  * packets such as those making up a keyframe.
  61  *
  62  * The percentage is expressed not in terms of bytes, but in terms of
  63  * packets, this for implementation convenience. The drawback with this
  64  * approach is that when using a percentage different from 100 %, and a
  65  * low bitrate, entire frames may be contained in a single packet, leading
  66  * to some packets not being protected, thus lowering the overall recovery
  67  * rate on the receiving side.
  68  *
  69  * When using #GstRtpBin, this element should be inserted through the
  70  * #GstRtpBin::request-fec-encoder signal.
  71  *
  72  *
  73  * <refsect2>
  74  * <title>Example pipeline</title>
  75  * |[
  76  * gst-launch-1.0 videotestsrc ! x264enc ! video/x-h264, profile=baseline ! rtph264pay pt=96 ! rtpulpfecenc percentage=100 pt=122 ! udpsink port=8888
  77  * ]| This example will receive a stream with FEC and try to reconstruct the packets.
  78  *
  79  * Example programs are available at
  80  * <ulink url="https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/blob/master/examples/src/bin/rtpfecserver.rs">rtpfecserver.rs</ulink>
  81  * and
  82  * <ulink url="https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/blob/master/examples/src/bin/rtpfecclient.rs">rtpfecclient.rs</ulink>
  83  * </refsect2>
  84  *
  85  * See also: #GstRtpUlpFecDec, #GstRtpBin
  86  * Since: 1.14
  87  */
  88
  89 #include <gst/rtp/gstrtp-enumtypes.h>
  90 #include <gst/rtp/gstrtpbuffer.h>
  91 #include <string.h>
  92
  93 #include "rtpulpfeccommon.h"
  94 #include "gstrtpulpfecenc.h"
  95
  96 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  97     GST_PAD_SINK,
  98     GST_PAD_ALWAYS,
  99     GST_STATIC_CAPS ("application/x-rtp"));
 100
 101 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
 102     GST_PAD_SRC,
 103     GST_PAD_ALWAYS,
 104     GST_STATIC_CAPS ("application/x-rtp"));
 105
 106 #define UNDEF_PT                255
 107
 108 #define DEFAULT_PT              UNDEF_PT
 109 #define DEFAULT_PCT             0
 110 #define DEFAULT_PCT_IMPORTANT   0
 111 #define DEFAULT_MULTIPACKET     TRUE
 112
 113 #define PACKETS_BUF_MAX_LENGTH  (RTP_ULPFEC_PROTECTED_PACKETS_MAX(TRUE))
 114
 115 GST_DEBUG_CATEGORY (gst_rtp_ulpfec_enc_debug);
 116 #define GST_CAT_DEFAULT (gst_rtp_ulpfec_enc_debug)
 117
 118 G_DEFINE_TYPE (GstRtpUlpFecEnc, gst_rtp_ulpfec_enc, GST_TYPE_ELEMENT);
 119
 120 enum
 121 {
 122   PROP_0,
 123   PROP_PT,
 124   PROP_MULTIPACKET,
 125   PROP_PROTECTED,
 126   PROP_PERCENTAGE,
 127   PROP_PERCENTAGE_IMPORTANT,
 128 };
 129
 130 #define RTP_FEC_MAP_INFO_NTH(ctx, data) (&g_array_index (\
 131     ((GstRtpUlpFecEncStreamCtx *)ctx)->info_arr, \
 132     RtpUlpFecMapInfo, \
 133     GPOINTER_TO_UINT(data)))
 134
 135 static void
 136 gst_rtp_ulpfec_enc_stream_ctx_start (GstRtpUlpFecEncStreamCtx * ctx,
 137     GQueue * packets, guint fec_packets)
 138 {
 139   GList *it = packets->tail;
 140   guint i;
 141
 142   g_array_set_size (ctx->info_arr, packets->length);
 143
 144   for (i = 0; i < packets->length; ++i) {
 145     GstBuffer *buffer = it->data;
 146     RtpUlpFecMapInfo *info = RTP_FEC_MAP_INFO_NTH (ctx, i);
 147
 148     if (!rtp_ulpfec_map_info_map (gst_buffer_ref (buffer), info))
 149       g_assert_not_reached ();
 150
 151     GST_LOG_RTP_PACKET (ctx->parent, "rtp header (incoming)", &info->rtp);
 152
 153     it = g_list_previous (it);
 154   }
 155
 156   ctx->fec_packets = fec_packets;
 157   ctx->fec_packet_idx = 0;
 158 }
 159
 160 static void
 161 gst_rtp_ulpfec_enc_stream_ctx_stop (GstRtpUlpFecEncStreamCtx * ctx)
 162 {
 163   g_array_set_size (ctx->info_arr, 0);
 164   g_array_set_size (ctx->scratch_buf, 0);
 165
 166   ctx->fec_packets = 0;
 167   ctx->fec_packet_idx = 0;
 168 }
 169
 170 static void
 171     gst_rtp_ulpfec_enc_stream_ctx_get_protection_parameters
 172     (GstRtpUlpFecEncStreamCtx * ctx, guint16 * dst_seq_base, guint64 * dst_mask,
 173     guint * dst_start, guint * dst_end)
 174 {
 175   guint media_packets = ctx->info_arr->len;
 176   guint start = ctx->fec_packet_idx * media_packets / ctx->fec_packets;
 177   guint end =
 178       ((ctx->fec_packet_idx + 1) * media_packets + ctx->fec_packets -
 179       1) / ctx->fec_packets - 1;
 180   guint len = end - start + 1;
 181   guint64 mask = 0;
 182   guint16 seq_base = 0;
 183   guint i;
 184
 185   len = MIN (len, RTP_ULPFEC_PROTECTED_PACKETS_MAX (TRUE));
 186   end = start + len - 1;
 187
 188   for (i = start; i <= end; ++i) {
 189     RtpUlpFecMapInfo *info = RTP_FEC_MAP_INFO_NTH (ctx, i);
 190     guint16 seq = gst_rtp_buffer_get_seq (&info->rtp);
 191
 192     if (mask) {
 193       gint diff = gst_rtp_buffer_compare_seqnum (seq_base, seq);
 194       if (diff < 0) {
 195         seq_base = seq;
 196         mask = mask >> (-diff);
 197       }
 198       mask |= rtp_ulpfec_packet_mask_from_seqnum (seq, seq_base, TRUE);
 199     } else {
 200       seq_base = seq;
 201       mask = rtp_ulpfec_packet_mask_from_seqnum (seq, seq_base, TRUE);
 202     }
 203   }
 204
 205   *dst_start = start;
 206   *dst_end = end;
 207   *dst_mask = mask;
 208   *dst_seq_base = seq_base;
 209 }
 210
 211 static GstBuffer *
 212 gst_rtp_ulpfec_enc_stream_ctx_protect (GstRtpUlpFecEncStreamCtx * ctx,
 213     guint8 pt, guint16 seq, guint32 timestamp, guint32 ssrc)
 214 {
 215   guint end = 0;
 216   guint start = 0;
 217   guint64 fec_mask = 0;
 218   guint16 seq_base = 0;
 219   GstBuffer *ret;
 220   guint64 tmp_mask;
 221   gboolean fec_mask_long;
 222   guint i;
 223
 224   if (ctx->fec_packet_idx >= ctx->fec_packets)
 225     return NULL;
 226
 227   g_array_set_size (ctx->scratch_buf, 0);
 228   gst_rtp_ulpfec_enc_stream_ctx_get_protection_parameters (ctx, &seq_base,
 229       &fec_mask, &start, &end);
 230
 231   tmp_mask = fec_mask;
 232   fec_mask_long = rtp_ulpfec_mask_is_long (fec_mask);
 233   for (i = start; i <= end; ++i) {
 234     RtpUlpFecMapInfo *info = RTP_FEC_MAP_INFO_NTH (ctx, i);
 235     guint64 packet_mask =
 236         rtp_ulpfec_packet_mask_from_seqnum (gst_rtp_buffer_get_seq (&info->rtp),
 237         seq_base,
 238         TRUE);
 239
 240     if (tmp_mask & packet_mask) {
 241       tmp_mask ^= packet_mask;
 242       rtp_buffer_to_ulpfec_bitstring (&info->rtp, ctx->scratch_buf, FALSE,
 243           fec_mask_long);
 244     }
 245   }
 246
 247   g_assert (tmp_mask == 0);
 248   ret =
 249       rtp_ulpfec_bitstring_to_fec_rtp_buffer (ctx->scratch_buf, seq_base,
 250       fec_mask_long, fec_mask, FALSE, pt, seq, timestamp, ssrc);
 251   ++ctx->fec_packet_idx;
 252   return ret;
 253 }
 254
 255 static void
 256 gst_rtp_ulpfec_enc_stream_ctx_report_budget (GstRtpUlpFecEncStreamCtx * ctx)
 257 {
 258   GST_TRACE_OBJECT (ctx->parent, "budget = %f budget_important = %f",
 259       ctx->budget, ctx->budget_important);
 260 }
 261
 262 static void
 263 gst_rtp_ulpfec_enc_stream_ctx_increment_budget (GstRtpUlpFecEncStreamCtx * ctx,
 264     GstBuffer * buffer)
 265 {
 266   if (ctx->percentage == 0 && ctx->percentage_important == 0) {
 267     if (ctx->budget > 0) {
 268       ctx->budget = 0;
 269       ctx->budget_important = 0;
 270     }
 271     if (ctx->budget < 0)
 272       ctx->budget += ctx->budget_inc;
 273
 274     return;
 275   }
 276   ctx->budget += ctx->budget_inc;
 277
 278   if (GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_NON_DROPPABLE)) {
 279     ctx->budget_important += ctx->budget_inc_important;
 280   }
 281
 282   gst_rtp_ulpfec_enc_stream_ctx_report_budget (ctx);
 283 }
 284
 285 static void
 286 gst_rtp_ulpfec_enc_stream_ctx_decrement_budget (GstRtpUlpFecEncStreamCtx * ctx,
 287     guint fec_packets_num)
 288 {
 289   if (ctx->budget_important >= 1.)
 290     ctx->budget_important -= fec_packets_num;
 291   ctx->budget -= fec_packets_num;
 292
 293   gst_rtp_ulpfec_enc_stream_ctx_report_budget (ctx);
 294 }
 295
 296 static guint
 297 gst_rtp_ulpfec_enc_stream_ctx_get_fec_packets_num (GstRtpUlpFecEncStreamCtx *
 298     ctx)
 299 {
 300   g_assert_cmpfloat (ctx->budget_important, >=, 0.);
 301
 302   if (ctx->budget_important >= 1.)
 303     return ctx->budget_important;
 304   return ctx->budget > 0. ? (guint) ctx->budget : 0;
 305 }
 306
 307 static void
 308 gst_rtp_ulpfec_enc_stream_ctx_free_packets_buf (GstRtpUlpFecEncStreamCtx * ctx)
 309 {
 310   while (ctx->packets_buf.length)
 311     gst_buffer_unref (g_queue_pop_tail (&ctx->packets_buf));
 312 }
 313
 314 static void
 315 gst_rtp_ulpfec_enc_stream_ctx_prepend_to_fec_buffer (GstRtpUlpFecEncStreamCtx *
 316     ctx, GstRTPBuffer * rtp, guint buf_max_size)
 317 {
 318   GList *new_head;
 319   if (ctx->packets_buf.length == buf_max_size) {
 320     new_head = g_queue_pop_tail_link (&ctx->packets_buf);
 321   } else {
 322     new_head = g_list_alloc ();
 323   }
 324
 325   gst_buffer_replace ((GstBuffer **) & new_head->data, rtp->buffer);
 326   g_queue_push_head_link (&ctx->packets_buf, new_head);
 327
 328   g_assert_cmpint (ctx->packets_buf.length, <=, buf_max_size);
 329 }
 330
 331 static GstFlowReturn
 332 gst_rtp_ulpfec_enc_stream_ctx_push_fec_packets (GstRtpUlpFecEncStreamCtx * ctx,
 333     guint8 pt, guint16 seq, guint32 timestamp, guint32 ssrc)
 334 {
 335   GstFlowReturn ret = GST_FLOW_OK;
 336   guint fec_packets_num =
 337       gst_rtp_ulpfec_enc_stream_ctx_get_fec_packets_num (ctx);
 338
 339   if (fec_packets_num) {
 340     guint fec_packets_pushed = 0;
 341     GstBuffer *latest_packet = ctx->packets_buf.head->data;
 342     GstBuffer *fec = NULL;
 343
 344     gst_rtp_ulpfec_enc_stream_ctx_start (ctx, &ctx->packets_buf,
 345         fec_packets_num);
 346
 347     while (NULL != (fec =
 348             gst_rtp_ulpfec_enc_stream_ctx_protect (ctx, pt,
 349                 seq + fec_packets_pushed, timestamp, ssrc))) {
 350       gst_buffer_copy_into (fec, latest_packet, GST_BUFFER_COPY_TIMESTAMPS, 0,
 351           -1);
 352
 353       ret = gst_pad_push (ctx->srcpad, fec);
 354       if (GST_FLOW_OK == ret)
 355         ++fec_packets_pushed;
 356       else
 357         break;
 358     }
 359
 360     gst_rtp_ulpfec_enc_stream_ctx_stop (ctx);
 361
 362     g_assert_cmpint (fec_packets_pushed, <=, fec_packets_num);
 363
 364     ctx->num_packets_protected += ctx->packets_buf.length;
 365     ctx->num_packets_fec += fec_packets_pushed;
 366     ctx->seqnum_offset += fec_packets_pushed;
 367     ctx->seqnum += fec_packets_pushed;
 368   }
 369
 370   gst_rtp_ulpfec_enc_stream_ctx_decrement_budget (ctx, fec_packets_num);
 371   return ret;
 372 }
 373
 374 static void
 375 gst_rtp_ulpfec_enc_stream_ctx_cache_packet (GstRtpUlpFecEncStreamCtx * ctx,
 376     GstRTPBuffer * rtp, gboolean * dst_empty_packet_buffer,
 377     gboolean * dst_push_fec)
 378 {
 379   if (ctx->multipacket) {
 380     gst_rtp_ulpfec_enc_stream_ctx_prepend_to_fec_buffer (ctx, rtp,
 381         PACKETS_BUF_MAX_LENGTH);
 382     gst_rtp_ulpfec_enc_stream_ctx_increment_budget (ctx, rtp->buffer);
 383
 384     *dst_empty_packet_buffer = gst_rtp_buffer_get_marker (rtp);
 385     *dst_push_fec = *dst_empty_packet_buffer;
 386   } else {
 387     gboolean push_fec;
 388
 389     gst_rtp_ulpfec_enc_stream_ctx_prepend_to_fec_buffer (ctx, rtp, 1);
 390
 391     push_fec = ctx->fec_nth == 0 ? FALSE :
 392         0 == (ctx->num_packets_received % ctx->fec_nth);
 393
 394     ctx->budget = push_fec ? 1 : 0;
 395     ctx->budget_important = 0;
 396
 397     *dst_push_fec = push_fec;
 398     *dst_empty_packet_buffer = FALSE;
 399   }
 400 }
 401
 402 static void
 403 gst_rtp_ulpfec_enc_stream_ctx_configure (GstRtpUlpFecEncStreamCtx * ctx,
 404     guint pt, guint percentage, guint percentage_important,
 405     gboolean multipacket)
 406 {
 407   ctx->pt = pt;
 408   ctx->percentage = percentage;
 409   ctx->percentage_important = percentage_important;
 410   ctx->multipacket = multipacket;
 411
 412   ctx->fec_nth = percentage ? 100 / percentage : 0;
 413   if (percentage) {
 414     ctx->budget_inc = percentage / 100.;
 415     ctx->budget_inc_important = percentage > percentage_important ?
 416         ctx->budget_inc : percentage_important / 100.;
 417   }
 418 /*
 419    else {
 420     ctx->budget_inc = 0.0;
 421   }
 422 */
 423   ctx->budget_inc_important = percentage > percentage_important ?
 424       ctx->budget_inc : percentage_important / 100.;
 425 }
 426
 427 static GstRtpUlpFecEncStreamCtx *
 428 gst_rtp_ulpfec_enc_stream_ctx_new (guint ssrc,
 429     GstElement * parent, GstPad * srcpad,
 430     guint pt, guint percentage, guint percentage_important,
 431     gboolean multipacket)
 432 {
 433   GstRtpUlpFecEncStreamCtx *ctx = g_new0 (GstRtpUlpFecEncStreamCtx, 1);
 434
 435   ctx->ssrc = ssrc;
 436   ctx->parent = parent;
 437   ctx->srcpad = srcpad;
 438
 439   ctx->seqnum = g_random_int_range (0, G_MAXUINT16 / 2);
 440
 441   ctx->info_arr = g_array_new (FALSE, TRUE, sizeof (RtpUlpFecMapInfo));
 442   g_array_set_clear_func (ctx->info_arr,
 443       (GDestroyNotify) rtp_ulpfec_map_info_unmap);
 444   ctx->parent = parent;
 445   ctx->scratch_buf = g_array_new (FALSE, TRUE, sizeof (guint8));
 446   gst_rtp_ulpfec_enc_stream_ctx_configure (ctx, pt,
 447       percentage, percentage_important, multipacket);
 448
 449   return ctx;
 450 }
 451
 452 static void
 453 gst_rtp_ulpfec_enc_stream_ctx_free (GstRtpUlpFecEncStreamCtx * ctx)
 454 {
 455   if (ctx->num_packets_received) {
 456     GST_INFO_OBJECT (ctx->parent, "Actual FEC overhead is %4.2f%% (%u/%u)\n",
 457         ctx->num_packets_fec * (double) 100. / ctx->num_packets_received,
 458         ctx->num_packets_fec, ctx->num_packets_received);
 459   }
 460   gst_rtp_ulpfec_enc_stream_ctx_free_packets_buf (ctx);
 461
 462   g_assert (0 == ctx->info_arr->len);
 463   g_array_free (ctx->info_arr, TRUE);
 464   g_array_free (ctx->scratch_buf, TRUE);
 465   g_slice_free1 (sizeof (GstRtpUlpFecEncStreamCtx), ctx);
 466 }
 467
 468 static GstFlowReturn
 469 gst_rtp_ulpfec_enc_stream_ctx_process (GstRtpUlpFecEncStreamCtx * ctx,
 470     GstBuffer * buffer)
 471 {
 472   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 473   GstFlowReturn ret;
 474   gboolean push_fec = FALSE;
 475   gboolean empty_packet_buffer = FALSE;
 476
 477   ctx->num_packets_received++;
 478
 479   if (ctx->seqnum_offset > 0) {
 480     buffer = gst_buffer_make_writable (buffer);
 481     if (!gst_rtp_buffer_map (buffer,
 482             GST_MAP_READWRITE | GST_RTP_BUFFER_MAP_FLAG_SKIP_PADDING, &rtp))
 483       g_assert_not_reached ();
 484     gst_rtp_buffer_set_seq (&rtp,
 485         gst_rtp_buffer_get_seq (&rtp) + ctx->seqnum_offset);
 486   } else {
 487     if (!gst_rtp_buffer_map (buffer,
 488             GST_MAP_READ | GST_RTP_BUFFER_MAP_FLAG_SKIP_PADDING, &rtp))
 489       g_assert_not_reached ();
 490   }
 491
 492   gst_rtp_ulpfec_enc_stream_ctx_cache_packet (ctx, &rtp, &empty_packet_buffer,
 493       &push_fec);
 494
 495   if (push_fec) {
 496     guint32 fec_timestamp = gst_rtp_buffer_get_timestamp (&rtp);
 497     guint32 fec_ssrc = gst_rtp_buffer_get_ssrc (&rtp);
 498     guint16 fec_seq = gst_rtp_buffer_get_seq (&rtp) + 1;
 499
 500     gst_rtp_buffer_unmap (&rtp);
 501
 502     ret = gst_pad_push (ctx->srcpad, buffer);
 503     if (GST_FLOW_OK == ret)
 504       ret =
 505           gst_rtp_ulpfec_enc_stream_ctx_push_fec_packets (ctx, ctx->pt, fec_seq,
 506           fec_timestamp, fec_ssrc);
 507   } else {
 508     gst_rtp_buffer_unmap (&rtp);
 509     ret = gst_pad_push (ctx->srcpad, buffer);
 510   }
 511
 512   if (empty_packet_buffer)
 513     gst_rtp_ulpfec_enc_stream_ctx_free_packets_buf (ctx);
 514
 515   return ret;
 516 }
 517
 518 static GstRtpUlpFecEncStreamCtx *
 519 gst_rtp_ulpfec_enc_aquire_ctx (GstRtpUlpFecEnc * fec, guint ssrc)
 520 {
 521   GstRtpUlpFecEncStreamCtx *ctx;
 522
 523   GST_OBJECT_LOCK (fec);
 524   ctx = g_hash_table_lookup (fec->ssrc_to_ctx, GUINT_TO_POINTER (ssrc));
 525   if (ctx == NULL) {
 526     ctx =
 527         gst_rtp_ulpfec_enc_stream_ctx_new (ssrc, GST_ELEMENT_CAST (fec),
 528         fec->srcpad, fec->pt, fec->percentage,
 529         fec->percentage_important, fec->multipacket);
 530     g_hash_table_insert (fec->ssrc_to_ctx, GUINT_TO_POINTER (ssrc), ctx);
 531   }
 532   GST_OBJECT_UNLOCK (fec);
 533
 534   return ctx;
 535 }
 536
 537 static GstFlowReturn
 538 gst_rtp_ulpfec_enc_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 539 {
 540   GstRtpUlpFecEnc *fec = GST_RTP_ULPFEC_ENC (parent);
 541   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 542   GstFlowReturn ret;
 543   guint ssrc = 0;
 544   GstRtpUlpFecEncStreamCtx *ctx;
 545
 546   if (fec->pt == UNDEF_PT)
 547     return gst_pad_push (fec->srcpad, buffer);
 548
 549   /* FIXME: avoid this additional mapping of the buffer to get the
 550      ssrc! */
 551   if (!gst_rtp_buffer_map (buffer,
 552           GST_MAP_READ | GST_RTP_BUFFER_MAP_FLAG_SKIP_PADDING, &rtp)) {
 553     g_assert_not_reached ();
 554   }
 555   ssrc = gst_rtp_buffer_get_ssrc (&rtp);
 556   gst_rtp_buffer_unmap (&rtp);
 557
 558   ctx = gst_rtp_ulpfec_enc_aquire_ctx (fec, ssrc);
 559
 560   ret = gst_rtp_ulpfec_enc_stream_ctx_process (ctx, buffer);
 561
 562   /* FIXME: does not work for mulitple ssrcs */
 563   fec->num_packets_protected = ctx->num_packets_protected;
 564
 565   return ret;
 566 }
 567
 568 static void
 569 gst_rtp_ulpfec_enc_configure_ctx (gpointer key, gpointer value,
 570     gpointer user_data)
 571 {
 572   GstRtpUlpFecEnc *fec = user_data;
 573   GstRtpUlpFecEncStreamCtx *ctx = value;
 574
 575   gst_rtp_ulpfec_enc_stream_ctx_configure (ctx, fec->pt,
 576       fec->percentage, fec->percentage_important, fec->multipacket);
 577 }
 578
 579 static void
 580 gst_rtp_ulpfec_enc_set_property (GObject * object, guint prop_id,
 581     const GValue * value, GParamSpec * pspec)
 582 {
 583   GstRtpUlpFecEnc *fec = GST_RTP_ULPFEC_ENC (object);
 584
 585   switch (prop_id) {
 586     case PROP_PT:
 587       fec->pt = g_value_get_uint (value);
 588       break;
 589     case PROP_MULTIPACKET:
 590       fec->multipacket = g_value_get_boolean (value);
 591       break;
 592     case PROP_PERCENTAGE:
 593       fec->percentage = g_value_get_uint (value);
 594       break;
 595     case PROP_PERCENTAGE_IMPORTANT:
 596       fec->percentage_important = g_value_get_uint (value);
 597       break;
 598     default:
 599       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 600       break;
 601   }
 602
 603   GST_OBJECT_LOCK (fec);
 604   g_hash_table_foreach (fec->ssrc_to_ctx, gst_rtp_ulpfec_enc_configure_ctx,
 605       fec);
 606   GST_OBJECT_UNLOCK (fec);
 607 }
 608
 609 static void
 610 gst_rtp_ulpfec_enc_get_property (GObject * object, guint prop_id,
 611     GValue * value, GParamSpec * pspec)
 612 {
 613   GstRtpUlpFecEnc *fec = GST_RTP_ULPFEC_ENC (object);
 614   switch (prop_id) {
 615     case PROP_PT:
 616       g_value_set_uint (value, fec->pt);
 617       break;
 618     case PROP_PROTECTED:
 619       g_value_set_uint (value, fec->num_packets_protected);
 620       break;
 621     case PROP_PERCENTAGE:
 622       g_value_set_uint (value, fec->percentage);
 623       break;
 624     case PROP_PERCENTAGE_IMPORTANT:
 625       g_value_set_uint (value, fec->percentage_important);
 626       break;
 627     case PROP_MULTIPACKET:
 628       g_value_set_boolean (value, fec->multipacket);
 629       break;
 630     default:
 631       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 632       break;
 633   }
 634 }
 635
 636 static void
 637 gst_rtp_ulpfec_enc_dispose (GObject * obj)
 638 {
 639   GstRtpUlpFecEnc *fec = GST_RTP_ULPFEC_ENC (obj);
 640
 641   g_hash_table_destroy (fec->ssrc_to_ctx);
 642
 643   G_OBJECT_CLASS (gst_rtp_ulpfec_enc_parent_class)->dispose (obj);
 644 }
 645
 646 static void
 647 gst_rtp_ulpfec_enc_init (GstRtpUlpFecEnc * fec)
 648 {
 649   fec->srcpad = gst_pad_new_from_static_template (&srctemplate, "src");
 650   gst_element_add_pad (GST_ELEMENT (fec), fec->srcpad);
 651
 652   fec->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 653   GST_PAD_SET_PROXY_CAPS (fec->sinkpad);
 654   GST_PAD_SET_PROXY_ALLOCATION (fec->sinkpad);
 655   gst_pad_set_chain_function (fec->sinkpad,
 656       GST_DEBUG_FUNCPTR (gst_rtp_ulpfec_enc_chain));
 657   gst_element_add_pad (GST_ELEMENT (fec), fec->sinkpad);
 658
 659   fec->ssrc_to_ctx = g_hash_table_new_full (NULL, NULL, NULL,
 660       (GDestroyNotify) gst_rtp_ulpfec_enc_stream_ctx_free);
 661 }
 662
 663 static void
 664 gst_rtp_ulpfec_enc_class_init (GstRtpUlpFecEncClass * klass)
 665 {
 666   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 667   GstElementClass *element_class = GST_ELEMENT_CLASS (klass);
 668
 669   GST_DEBUG_CATEGORY_INIT (gst_rtp_ulpfec_enc_debug, "rtpulpfecenc", 0,
 670       "FEC encoder element");
 671
 672   gst_element_class_add_pad_template (element_class,
 673       gst_static_pad_template_get (&srctemplate));
 674   gst_element_class_add_pad_template (element_class,
 675       gst_static_pad_template_get (&sinktemplate));
 676
 677   gst_element_class_set_static_metadata (element_class,
 678       "RTP FEC Encoder",
 679       "Codec/Payloader/Network/RTP",
 680       "Encodes RTP FEC (RFC5109)", "Mikhail Fludkov <misha@pexip.com>");
 681
 682   gobject_class->set_property =
 683       GST_DEBUG_FUNCPTR (gst_rtp_ulpfec_enc_set_property);
 684   gobject_class->get_property =
 685       GST_DEBUG_FUNCPTR (gst_rtp_ulpfec_enc_get_property);
 686   gobject_class->dispose = GST_DEBUG_FUNCPTR (gst_rtp_ulpfec_enc_dispose);
 687
 688   g_object_class_install_property (gobject_class, PROP_PT,
 689       g_param_spec_uint ("pt", "payload type",
 690           "The payload type of FEC packets", 0, 255, DEFAULT_PT,
 691           G_PARAM_CONSTRUCT | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 692
 693   g_object_class_install_property (gobject_class, PROP_MULTIPACKET,
 694       g_param_spec_boolean ("multipacket", "Multipacket",
 695           "Apply FEC on multiple packets", DEFAULT_MULTIPACKET,
 696           G_PARAM_CONSTRUCT | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 697
 698   g_object_class_install_property (gobject_class, PROP_PERCENTAGE,
 699       g_param_spec_uint ("percentage", "Percentage",
 700           "FEC overhead percentage for the whole stream", 0, 100, DEFAULT_PCT,
 701           G_PARAM_CONSTRUCT | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 702
 703   g_object_class_install_property (gobject_class, PROP_PERCENTAGE_IMPORTANT,
 704       g_param_spec_uint ("percentage-important", "Percentage important",
 705           "FEC overhead percentage for important packets",
 706           0, 100, DEFAULT_PCT_IMPORTANT,
 707           G_PARAM_CONSTRUCT | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 708
 709   g_object_class_install_property (gobject_class, PROP_PROTECTED,
 710       g_param_spec_uint ("protected", "Protected",
 711           "Count of protected packets", 0, G_MAXUINT32, 0,
 712           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 713 }