avformat: Add Pro-MPEG CoP #3-R2 FEC protocol

Pro-MPEG Code of Practice #3 release 2 forward error correction for rtp_mpegts streams

Signed-off-by: Vlad Tarca <vtarca@mobibase.com>
Signed-off-by: Michael Niedermayer <michael@niedermayer.cc>

4 files changed, 541 insertions, 3 deletions

diff --git a/libavformat/Makefile b/libavformat/Makefile
index c9defe7..e6bfc42 100644
--- a/libavformat/Makefile
+++ b/libavformat/Makefile
@@ -558,6 +558,7 @@ OBJS-$(CONFIG_MD5_PROTOCOL)              += md5proto.o
 OBJS-$(CONFIG_MMSH_PROTOCOL)             += mmsh.o mms.o asf.o
 OBJS-$(CONFIG_MMST_PROTOCOL)             += mmst.o mms.o asf.o
 OBJS-$(CONFIG_PIPE_PROTOCOL)             += file.o
+OBJS-$(CONFIG_PROMPEG_PROTOCOL)          += prompeg.o
 OBJS-$(CONFIG_RTMP_PROTOCOL)             += rtmpproto.o rtmppkt.o
 OBJS-$(CONFIG_RTMPE_PROTOCOL)            += rtmpproto.o rtmppkt.o
 OBJS-$(CONFIG_RTMPS_PROTOCOL)            += rtmpproto.o rtmppkt.o
diff --git a/libavformat/prompeg.c b/libavformat/prompeg.c
new file mode 100644
index 0000000..cc1baa4
--- /dev/null
+++ b/libavformat/prompeg.c
@@ -0,0 +1,481 @@
+/*
+ * Pro-MPEG Code of Practice #3 Release 2 FEC
+ * Copyright (c) 2016 Mobibase, France (http://www.mobibase.com)
+ *
+ * This file is part of FFmpeg.
+ *
+ * FFmpeg is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * FFmpeg is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with FFmpeg; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+/**
+ * @file
+ * Pro-MPEG Code of Practice #3 Release 2 FEC protocol
+ * @author Vlad Tarca <vlad.tarca@gmail.com>
+ */
+
+/*
+ * Reminder:
+
+ [RFC 2733] FEC Packet Structure
+
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |                         RTP Header                            |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |                         FEC Header                            |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |                         FEC Payload                           |
+   |                                                               |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+
+
+ [RFC 3550] RTP header
+
+    0                   1                   2                   3
+    0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |V=2|P|X|  CC   |M|     PT      |       sequence number         |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |                           timestamp                           |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |           synchronization source (SSRC) identifier            |
+   +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
+   |            contributing source (CSRC) identifiers             |
+   |                             ....                              |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+
+ [RFC 3550] RTP header extension (after CSRC)
+
+    0                   1                   2                   3
+    0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |      defined by profile       |           length              |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |                        header extension                       |
+   |                             ....                              |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+
+ [Pro-MPEG COP3] FEC Header
+
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |      SNBase low bits          |        length recovery        |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |E| PT recovery |                 mask                          |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |                          TS recovery                          |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |X|D|type |index|    offset     |      NA       |SNBase ext bits|
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+
+ */
+
+#include "libavutil/avstring.h"
+#include "libavutil/intreadwrite.h"
+#include "libavutil/opt.h"
+#include "libavutil/parseutils.h"
+#include "libavutil/random_seed.h"
+#include "avformat.h"
+#include "config.h"
+#include "url.h"
+
+#define PROMPEG_RTP_PT 0x60
+#define PROMPEG_FEC_COL 0x0
+#define PROMPEG_FEC_ROW 0x1
+
+typedef struct PrompegFec {
+    uint16_t sn;
+    uint32_t ts;
+    uint8_t *bitstring;
+} PrompegFec;
+
+typedef struct PrompegContext {
+    const AVClass *class;
+    URLContext *fec_col_hd, *fec_row_hd;
+    PrompegFec **fec_arr, **fec_col_tmp, **fec_col, *fec_row;
+    int ttl;
+    uint8_t l, d;
+    uint8_t *rtp_buf;
+    uint16_t rtp_col_sn, rtp_row_sn;
+    uint16_t length_recovery;
+    int packet_size;
+    int packet_idx, packet_idx_max;
+    int fec_arr_len;
+    int bitstring_size;
+    int rtp_buf_size;
+    int init;
+    int first;
+} PrompegContext;
+
+#define OFFSET(x) offsetof(PrompegContext, x)
+#define E AV_OPT_FLAG_ENCODING_PARAM
+
+static const AVOption options[] = {
+    { "ttl",   "Time to live (in milliseconds, multicast only)", OFFSET(ttl), AV_OPT_TYPE_INT, { .i64 = -1 }, -1, INT_MAX, .flags = E },
+    { "l", "FEC L", OFFSET(l), AV_OPT_TYPE_INT, { .i64 =  5 }, 4, 20, .flags = E },
+    { "d", "FEC D", OFFSET(d), AV_OPT_TYPE_INT, { .i64 =  5 }, 4, 20, .flags = E },
+    { NULL }
+};
+
+static const AVClass prompeg_class = {
+    .class_name = "prompeg",
+    .item_name  = av_default_item_name,
+    .option     = options,
+    .version    = LIBAVUTIL_VERSION_INT,
+};
+
+static void xor_fast(const uint8_t *in1, const uint8_t *in2, uint8_t *out, int size) {
+    int i, n, s;
+
+#if HAVE_FAST_64BIT
+    uint64_t v1, v2;
+
+    n = size / sizeof (uint64_t);
+    s = n * sizeof (uint64_t);
+
+    for (i = 0; i < n; i++) {
+        v1 = AV_RN64A(in1);
+        v2 = AV_RN64A(in2);
+        AV_WN64A(out, v1 ^ v2);
+        in1 += 8;
+        in2 += 8;
+        out += 8;
+    }
+#else
+    uint32_t v1, v2;
+
+    n = size / sizeof (uint32_t);
+    s = n * sizeof (uint32_t);
+
+    for (i = 0; i < n; i++) {
+        v1 = AV_RN32A(in1);
+        v2 = AV_RN32A(in2);
+        AV_WN32A(out, v1 ^ v2);
+        in1 += 4;
+        in2 += 4;
+        out += 4;
+    }
+#endif
+
+    n = size - s;
+
+    for (i = 0; i < n; i++) {
+        out[i] = in1[i] ^ in2[i];
+    }
+}
+
+static int prompeg_create_bitstring(URLContext *h, const uint8_t *buf, int size,
+        uint8_t **bitstring) {
+    PrompegContext *s = h->priv_data;
+    uint8_t *b;
+
+    if (size < 12 || (buf[0] & 0xc0) != 0x80 || (buf[1] & 0x7f) != 0x21) {
+        av_log(h, AV_LOG_ERROR, "Unsupported stream format (expected MPEG-TS over RTP)\n");
+        return AVERROR(EINVAL);
+    }
+    if (size != s->packet_size) {
+        av_log(h, AV_LOG_ERROR, "The RTP packet size must be constant (set pkt_size)\n");
+        return AVERROR(EINVAL);
+    }
+
+    *bitstring = av_malloc(s->bitstring_size);
+    if (!*bitstring) {
+        av_log(h, AV_LOG_ERROR, "Failed to allocate the bitstring buffer\n");
+        return AVERROR(ENOMEM);
+    }
+    b = *bitstring;
+
+    // P, X, CC
+    b[0] = buf[0] & 0x3f;
+    // M, PT
+    b[1] = buf[1];
+    // Timestamp
+    b[2] = buf[4];
+    b[3] = buf[5];
+    b[4] = buf[6];
+    b[5] = buf[7];
+    /*
+     * length_recovery: the unsigned network-ordered sum of lengths of CSRC,
+     * padding, extension and media payload
+     */
+    AV_WB16(b + 6, s->length_recovery);
+    // Payload
+    memcpy(b + 8, buf + 12, s->length_recovery);
+
+    return 0;
+}
+
+static int prompeg_write_fec(URLContext *h, PrompegFec *fec, uint8_t type) {
+    PrompegContext *s = h->priv_data;
+    URLContext *hd;
+    uint8_t *buf = s->rtp_buf; // zero-filled
+    uint8_t *b = fec->bitstring;
+    uint16_t sn;
+    int ret;
+
+    sn = type == PROMPEG_FEC_COL ? ++s->rtp_col_sn : ++s->rtp_row_sn;
+
+    // V, P, X, CC
+    buf[0] = 0x80 | (b[0] & 0x3f);
+    // M, PT
+    buf[1] = (b[1] & 0x80) | PROMPEG_RTP_PT;
+    // SN
+    AV_WB16(buf + 2, sn);
+    // TS
+    AV_WB32(buf + 4, fec->ts);
+    // CSRC=0
+    //AV_WB32(buf + 8, 0);
+    // SNBase low bits
+    AV_WB16(buf + 12, fec->sn);
+    // Length recovery
+    buf[14] = b[6];
+    buf[15] = b[7];
+    // E=1, PT recovery
+    buf[16] = 0x80 | b[1];
+    // Mask=0
+    //buf[17] = 0x0;
+    //buf[18] = 0x0;
+    //buf[19] = 0x0;
+    // TS recovery
+    buf[20] = b[2];
+    buf[21] = b[3];
+    buf[22] = b[4];
+    buf[23] = b[5];
+    // X=0, D, type=0, index=0
+    buf[24] = type == PROMPEG_FEC_COL ? 0x0 : 0x40;
+    // offset
+    buf[25] = type == PROMPEG_FEC_COL ? s->l : 0x1;
+    // NA
+    buf[26] = type == PROMPEG_FEC_COL ? s->d : s->l;
+    // SNBase ext bits=0
+    //buf[27] = 0x0;
+    // Payload
+    memcpy(buf + 28, b + 8, s->length_recovery);
+
+    hd = type == PROMPEG_FEC_COL ? s->fec_col_hd : s->fec_row_hd;
+    ret = ffurl_write(hd, buf, s->rtp_buf_size);
+    return ret;
+}
+
+static int prompeg_open(URLContext *h, const char *uri, int flags) {
+    PrompegContext *s = h->priv_data;
+    AVDictionary *udp_opts = NULL;
+    int rtp_port;
+    char hostname[256];
+    char buf[1024];
+
+    s->fec_col_hd = NULL;
+    s->fec_row_hd = NULL;
+
+    if (s->l * s->d > 100) {
+        av_log(h, AV_LOG_ERROR, "L * D must be <= 100\n");
+        return AVERROR(EINVAL);
+    }
+
+    av_url_split(NULL, 0, NULL, 0, hostname, sizeof (hostname), &rtp_port,
+            NULL, 0, uri);
+
+    if (rtp_port < 1 || rtp_port > UINT16_MAX - 4) {
+        av_log(h, AV_LOG_ERROR, "Invalid RTP base port %d\n", rtp_port);
+        return AVERROR(EINVAL);
+    }
+
+    if (s->ttl > 0) {
+        snprintf(buf, sizeof (buf), "%d", s->ttl);
+        av_dict_set(&udp_opts, "ttl", buf, 0);
+    }
+
+    ff_url_join(buf, sizeof (buf), "udp", NULL, hostname, rtp_port + 2, NULL);
+    if (ffurl_open_whitelist(&s->fec_col_hd, buf, flags, &h->interrupt_callback,
+            &udp_opts, h->protocol_whitelist, h->protocol_blacklist, h) < 0)
+        goto fail;
+    ff_url_join(buf, sizeof (buf), "udp", NULL, hostname, rtp_port + 4, NULL);
+    if (ffurl_open_whitelist(&s->fec_row_hd, buf, flags, &h->interrupt_callback,
+            &udp_opts, h->protocol_whitelist, h->protocol_blacklist, h) < 0)
+        goto fail;
+
+    h->max_packet_size = s->fec_col_hd->max_packet_size;
+    s->init = 1;
+
+    av_dict_free(&udp_opts);
+    av_log(h, AV_LOG_INFO, "ProMPEG CoP#3-R2 FEC L=%d D=%d\n", s->l, s->d);
+    return 0;
+
+fail:
+    ffurl_closep(&s->fec_col_hd);
+    ffurl_closep(&s->fec_row_hd);
+    av_dict_free(&udp_opts);
+    return AVERROR(EIO);
+}
+
+static int prompeg_init(URLContext *h, const uint8_t *buf, int size) {
+    PrompegContext *s = h->priv_data;
+    uint32_t seed;
+    int i;
+
+    s->fec_arr = NULL;
+    s->rtp_buf = NULL;
+
+    if (size < 12 || size > UINT16_MAX + 12) {
+        av_log(h, AV_LOG_ERROR, "Invalid RTP packet size\n");
+        return AVERROR_INVALIDDATA;
+    }
+
+    s->packet_idx = 0;
+    s->packet_idx_max = s->l * s->d;
+    s->packet_size = size;
+    s->length_recovery = size - 12;
+    s->rtp_buf_size = 28 + s->length_recovery; // 12 + 16: RTP + FEC headers
+    s->bitstring_size = 8 + s->length_recovery; // 8: P, X, CC, M, PT, SN, TS
+    s->fec_arr_len = 1 + 2 * s->l; // row + column tmp + column out
+
+    if (h->flags & AVFMT_FLAG_BITEXACT) {
+        s->rtp_col_sn = 0;
+        s->rtp_row_sn = 0;
+    } else {
+        seed = av_get_random_seed();
+        s->rtp_col_sn = seed & 0x0fff;
+        s->rtp_row_sn = (seed >> 16) & 0x0fff;
+    }
+
+    s->fec_arr = av_malloc_array(s->fec_arr_len, sizeof (PrompegFec*));
+    if (!s->fec_arr) {
+        goto fail;
+    }
+    for (i = 0; i < s->fec_arr_len; i++) {
+        s->fec_arr[i] = av_malloc(sizeof (PrompegFec));
+        if (!s->fec_arr[i]) {
+            goto fail;
+        }
+        s->fec_arr[i]->bitstring = av_malloc_array(s->bitstring_size, sizeof (uint8_t));
+        if (!s->fec_arr[i]->bitstring) {
+            av_freep(&s->fec_arr[i]);
+            goto fail;
+        }
+    }
+    s->fec_row = *s->fec_arr;
+    s->fec_col = s->fec_arr + 1;
+    s->fec_col_tmp = s->fec_arr + 1 + s->l;
+
+    s->rtp_buf = av_malloc_array(s->rtp_buf_size, sizeof (uint8_t));
+    if (!s->rtp_buf) {
+        goto fail;
+    }
+    memset(s->rtp_buf, 0, s->rtp_buf_size);
+
+    s->init = 0;
+    s->first = 1;
+
+    return 0;
+
+fail:
+    av_log(h, AV_LOG_ERROR, "Failed to allocate the FEC buffer\n");
+    return AVERROR(ENOMEM);
+}
+
+static int prompeg_write(URLContext *h, const uint8_t *buf, int size) {
+    PrompegContext *s = h->priv_data;
+    PrompegFec *fec_tmp;
+    uint8_t *bitstring = NULL;
+    int col_idx, col_out_idx, row_idx;
+    int ret, written = 0;
+
+    if (s->init && ((ret = prompeg_init(h, buf, size)) < 0))
+        goto end;
+
+    if ((ret = prompeg_create_bitstring(h, buf, size, &bitstring)) < 0)
+        goto end;
+
+    col_idx = s->packet_idx % s->l;
+    row_idx = s->packet_idx / s->l % s->d;
+
+    // FEC' (row) send block-aligned, xor
+    if (col_idx == 0) {
+        if (!s->first || s->packet_idx > 0) {
+            if ((ret = prompeg_write_fec(h, s->fec_row, PROMPEG_FEC_ROW)) < 0)
+                goto end;
+            written += ret;
+        }
+        memcpy(s->fec_row->bitstring, bitstring, s->bitstring_size);
+        s->fec_row->sn = AV_RB16(buf + 2);
+        s->fec_row->ts = AV_RB32(buf + 4);
+    } else {
+        xor_fast(s->fec_row->bitstring, bitstring, s->fec_row->bitstring,
+                s->bitstring_size);
+    }
+
+    // FEC (column) xor
+    if (row_idx == 0) {
+        if (!s->first) {
+            // swap fec_col and fec_col_tmp
+            fec_tmp = s->fec_col[col_idx];
+            s->fec_col[col_idx] = s->fec_col_tmp[col_idx];
+            s->fec_col_tmp[col_idx] = fec_tmp;
+        }
+        memcpy(s->fec_col_tmp[col_idx]->bitstring, bitstring, s->bitstring_size);
+        s->fec_col_tmp[col_idx]->sn = AV_RB16(buf + 2);
+        s->fec_col_tmp[col_idx]->ts = AV_RB32(buf + 4);
+    } else {
+        xor_fast(s->fec_col_tmp[col_idx]->bitstring, bitstring,
+                s->fec_col_tmp[col_idx]->bitstring, s->bitstring_size);
+    }
+
+    // FEC (column) send block-aligned
+    if (!s->first && s->packet_idx % s->d == 0) {
+        col_out_idx = s->packet_idx / s->l;
+        if ((ret = prompeg_write_fec(h, s->fec_col[col_out_idx], PROMPEG_FEC_COL)) < 0)
+            goto end;
+        written += ret;
+    }
+
+    if (++s->packet_idx >= s->packet_idx_max) {
+        s->packet_idx = 0;
+        if (s->first)
+            s->first = 0;
+    }
+
+    ret = written;
+
+end:
+    av_free(bitstring);
+    return ret;
+}
+
+static int prompeg_close(URLContext *h) {
+    PrompegContext *s = h->priv_data;
+    int i;
+
+    ffurl_closep(&s->fec_col_hd);
+    ffurl_closep(&s->fec_row_hd);
+
+    if (s->fec_arr) {
+        for (i = 0; i < s->fec_arr_len; i++) {
+            av_free(s->fec_arr[i]->bitstring);
+            av_freep(&s->fec_arr[i]);
+        }
+        av_freep(&s->fec_arr);
+    }
+    av_freep(&s->rtp_buf);
+
+    return 0;
+}
+
+const URLProtocol ff_prompeg_protocol = {
+    .name                      = "prompeg",
+    .url_open                  = prompeg_open,
+    .url_write                 = prompeg_write,
+    .url_close                 = prompeg_close,
+    .priv_data_size            = sizeof(PrompegContext),
+    .flags                     = URL_PROTOCOL_FLAG_NETWORK,
+    .priv_data_class           = &prompeg_class,
+};
diff --git a/libavformat/protocols.c b/libavformat/protocols.c
index b250f49..8d3555e 100644
--- a/libavformat/protocols.c
+++ b/libavformat/protocols.c
@@ -43,6 +43,7 @@ extern const URLProtocol ff_mmsh_protocol;
 extern const URLProtocol ff_mmst_protocol;
 extern const URLProtocol ff_md5_protocol;
 extern const URLProtocol ff_pipe_protocol;
+extern const URLProtocol ff_prompeg_protocol;
 extern const URLProtocol ff_rtmp_protocol;
 extern const URLProtocol ff_rtmpe_protocol;
 extern const URLProtocol ff_rtmps_protocol;
diff --git a/libavformat/rtpproto.c b/libavformat/rtpproto.c
index fa1dcb5..840a36d 100644
--- a/libavformat/rtpproto.c
+++ b/libavformat/rtpproto.c
@@ -44,7 +44,7 @@
 
 typedef struct RTPContext {
     const AVClass *class;
-    URLContext *rtp_hd, *rtcp_hd;
+    URLContext *rtp_hd, *rtcp_hd, *fec_hd;
     int rtp_fd, rtcp_fd, nb_ssm_include_addrs, nb_ssm_exclude_addrs;
     struct sockaddr_storage **ssm_include_addrs, **ssm_exclude_addrs;
     int write_to_source;
@@ -58,6 +58,7 @@ typedef struct RTPContext {
     int dscp;
     char *sources;
     char *block;
+    char *fec_options_str;
 } RTPContext;
 
 #define OFFSET(x) offsetof(RTPContext, x)
@@ -75,6 +76,7 @@ static const AVOption options[] = {
     { "dscp",               "DSCP class",                                                       OFFSET(dscp),            AV_OPT_TYPE_INT,    { .i64 = -1 },    -1, INT_MAX, .flags = D|E },
     { "sources",            "Source list",                                                      OFFSET(sources),         AV_OPT_TYPE_STRING, { .str = NULL },               .flags = D|E },
     { "block",              "Block list",                                                       OFFSET(block),           AV_OPT_TYPE_STRING, { .str = NULL },               .flags = D|E },
+    { "fec",                "FEC",                                                              OFFSET(fec_options_str), AV_OPT_TYPE_STRING, { .str = NULL },               .flags = E },
     { NULL }
 };
 
@@ -316,9 +318,11 @@ static void rtp_parse_addr_list(URLContext *h, char *buf,
 static int rtp_open(URLContext *h, const char *uri, int flags)
 {
     RTPContext *s = h->priv_data;
+    AVDictionary *fec_opts = NULL;
     int rtp_port;
     char hostname[256], include_sources[1024] = "", exclude_sources[1024] = "";
     char *sources = include_sources, *block = exclude_sources;
+    char *fec_protocol = NULL;
     char buf[1024];
     char path[1024];
     const char *p;
@@ -377,6 +381,31 @@ static int rtp_open(URLContext *h, const char *uri, int flags)
         }
     }
 
+    if (s->fec_options_str) {
+        p = s->fec_options_str;
+
+        if (!(fec_protocol = av_get_token(&p, "="))) {
+            av_log(h, AV_LOG_ERROR, "Failed to parse the FEC protocol value\n");
+            goto fail;
+        }
+        if (strcmp(fec_protocol, "prompeg")) {
+            av_log(h, AV_LOG_ERROR, "Unsupported FEC protocol %s\n", fec_protocol);
+            goto fail;
+        }
+
+        p = s->fec_options_str + strlen(fec_protocol);
+        while (*p && *p == '=') p++;
+
+        if (av_dict_parse_string(&fec_opts, p, "=", ":", 0) < 0) {
+            av_log(h, AV_LOG_ERROR, "Failed to parse the FEC options\n");
+            goto fail;
+        }
+        if (s->ttl > 0) {
+            snprintf(buf, sizeof (buf), "%d", s->ttl);
+            av_dict_set(&fec_opts, "ttl", buf, 0);
+        }
+    }
+
     for (i = 0; i < max_retry_count; i++) {
         build_udp_url(s, buf, sizeof(buf),
                       hostname, rtp_port, s->local_rtpport,
@@ -412,6 +441,14 @@ static int rtp_open(URLContext *h, const char *uri, int flags)
         break;
     }
 
+    s->fec_hd = NULL;
+    if (fec_protocol) {
+        ff_url_join(buf, sizeof(buf), fec_protocol, NULL, hostname, rtp_port, NULL);
+        if (ffurl_open_whitelist(&s->fec_hd, buf, flags, &h->interrupt_callback,
+                             &fec_opts, h->protocol_whitelist, h->protocol_blacklist, h) < 0)
+            goto fail;
+    }
+
     /* just to ease handle access. XXX: need to suppress direct handle
        access */
     s->rtp_fd = ffurl_get_file_handle(s->rtp_hd);
@@ -419,6 +456,10 @@ static int rtp_open(URLContext *h, const char *uri, int flags)
 
     h->max_packet_size = s->rtp_hd->max_packet_size;
     h->is_streamed = 1;
+
+    av_free(fec_protocol);
+    av_dict_free(&fec_opts);
+
     return 0;
 
  fail:
@@ -426,6 +467,9 @@ static int rtp_open(URLContext *h, const char *uri, int flags)
         ffurl_close(s->rtp_hd);
     if (s->rtcp_hd)
         ffurl_close(s->rtcp_hd);
+    ffurl_closep(&s->fec_hd);
+    av_free(fec_protocol);
+    av_dict_free(&fec_opts);
     return AVERROR(EIO);
 }
 
@@ -474,7 +518,7 @@ static int rtp_read(URLContext *h, uint8_t *buf, int size)
 static int rtp_write(URLContext *h, const uint8_t *buf, int size)
 {
     RTPContext *s = h->priv_data;
-    int ret;
+    int ret, ret_fec;
     URLContext *hd;
 
     if (size < 2)
@@ -543,7 +587,17 @@ static int rtp_write(URLContext *h, const uint8_t *buf, int size)
         hd = s->rtp_hd;
     }
 
-    ret = ffurl_write(hd, buf, size);
+    if ((ret = ffurl_write(hd, buf, size)) < 0) {
+        return ret;
+    }
+
+    if (s->fec_hd && !RTP_PT_IS_RTCP(buf[1])) {
+        if ((ret_fec = ffurl_write(s->fec_hd, buf, size)) < 0) {
+            av_log(h, AV_LOG_ERROR, "Failed to send FEC\n");
+            return ret_fec;
+        }
+    }
+
     return ret;
 }
 
@@ -561,6 +615,7 @@ static int rtp_close(URLContext *h)
 
     ffurl_close(s->rtp_hd);
     ffurl_close(s->rtcp_hd);
+    ffurl_closep(&s->fec_hd);
     return 0;
 }