gst/rtpmanager/gstrtprtxreceive.c

   1 /* RTP Retransmission receiver element for GStreamer
   2  *
   3  * gstrtprtxreceive.c:
   4  *
   5  * Copyright (C) 2013 Collabora Ltd.
   6  *   @author Julien Isorce <julien.isorce@collabora.co.uk>
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Library General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Library General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Library General Public
  19  * License along with this library; if not, write to the
  20  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  21  * Boston, MA 02110-1301, USA.
  22  */
  23
  24 /**
  25  * SECTION:element-rtprtxreceive
  26  * @see_also: rtprtxsend, rtpsession, rtpjitterbuffer
  27  *
  28  * rtprtxreceive listens to the retransmission events from the
  29  * downstream rtpjitterbuffer and remembers the SSRC (ssrc1) of the stream and
  30  * the sequence number that was requested. When it receives a packet with
  31  * a sequence number equal to one of the ones stored and with a different SSRC,
  32  * it identifies the new SSRC (ssrc2) as the retransmission stream of ssrc1.
  33  * From this point on, it replaces ssrc2 with ssrc1 in all packets of the
  34  * ssrc2 stream and flags them as retransmissions, so that rtpjitterbuffer
  35  * can reconstruct the original stream.
  36  *
  37  * This algorithm is implemented as specified in RFC 4588.
  38  *
  39  * This element is meant to be used with rtprtxsend on the sender side.
  40  * See #GstRtpRtxSend
  41  *
  42  * Below you can see some examples that illustrate how rtprtxreceive and
  43  * rtprtxsend fit among the other rtp elements and how they work internally.
  44  * Normally, hoewever, you should avoid using such pipelines and use
  45  * rtpbin instead, with its #GstRtpBin::request-aux-sender and
  46  * #GstRtpBin::request-aux-receiver signals. See #GstRtpBin.
  47  *
  48  * # Example pipelines
  49  * |[
  50  * gst-launch-1.0 rtpsession name=rtpsession rtp-profile=avpf \
  51  *     audiotestsrc is-live=true ! opusenc ! rtpopuspay pt=96 ! \
  52  *         rtprtxsend payload-type-map="application/x-rtp-pt-map,96=(uint)97" ! \
  53  *         rtpsession.send_rtp_sink \
  54  *     rtpsession.send_rtp_src ! identity drop-probability=0.01 ! \
  55  *         udpsink host="127.0.0.1" port=5000 \
  56  *     udpsrc port=5001 ! rtpsession.recv_rtcp_sink \
  57  *     rtpsession.send_rtcp_src ! udpsink host="127.0.0.1" port=5002 \
  58  *         sync=false async=false
  59  * ]| Send audio stream through port 5000 (5001 and 5002 are just the rtcp
  60  * link with the receiver)
  61  * |[
  62  * gst-launch-1.0 rtpsession name=rtpsession rtp-profile=avpf \
  63  *     udpsrc port=5000 caps="application/x-rtp,media=(string)audio,clock-rate=(int)48000,encoding-name=(string)OPUS,payload=(int)96" ! \
  64  *         rtpsession.recv_rtp_sink \
  65  *     rtpsession.recv_rtp_src ! \
  66  *         rtprtxreceive payload-type-map="application/x-rtp-pt-map,96=(uint)97" ! \
  67  *         rtpssrcdemux ! rtpjitterbuffer do-retransmission=true ! \
  68  *         rtpopusdepay ! opusdec ! audioconvert ! audioresample ! autoaudiosink \
  69  *     rtpsession.send_rtcp_src ! \
  70  *         udpsink host="127.0.0.1" port=5001 sync=false async=false \
  71  *     udpsrc port=5002 ! rtpsession.recv_rtcp_sink
  72  * ]| Receive audio stream from port 5000 (5001 and 5002 are just the rtcp
  73  * link with the sender)
  74  *
  75  * In this example we can see a simple streaming of an OPUS stream with some
  76  * of the packets being artificially dropped by the identity element.
  77  * Thanks to retransmission, you should still hear a clear sound when setting
  78  * drop-probability to something greater than 0.
  79  *
  80  * Internally, the rtpjitterbuffer will generate a custom upstream event,
  81  * GstRTPRetransmissionRequest, when it detects that one packet is missing.
  82  * Then this request is translated to a FB NACK in the rtcp link by rtpsession.
  83  * Finally the rtpsession of the sender side will re-convert it in a
  84  * GstRTPRetransmissionRequest that will be handled by rtprtxsend. rtprtxsend
  85  * will then re-send the missing packet with a new srrc and a different payload
  86  * type (here, 97), but with the same original sequence number. On the receiver
  87  * side, rtprtxreceive will associate this new stream with the original and
  88  * forward the retransmission packets to rtpjitterbuffer with the original
  89  * ssrc and payload type.
  90  *
  91  * |[
  92  * gst-launch-1.0 rtpsession name=rtpsession rtp-profile=avpf \
  93  *     audiotestsrc is-live=true ! opusenc ! rtpopuspay pt=97 seqnum-offset=1 ! \
  94  *         rtprtxsend payload-type-map="application/x-rtp-pt-map,97=(uint)99" ! \
  95  *         funnel name=f ! rtpsession.send_rtp_sink \
  96  *     audiotestsrc freq=660.0 is-live=true ! opusenc ! \
  97  *         rtpopuspay pt=97 seqnum-offset=100 ! \
  98  *         rtprtxsend payload-type-map="application/x-rtp-pt-map,97=(uint)99" ! \
  99  *         f. \
 100  *     rtpsession.send_rtp_src ! identity drop-probability=0.01 ! \
 101  *         udpsink host="127.0.0.1" port=5000 \
 102  *     udpsrc port=5001 ! rtpsession.recv_rtcp_sink \
 103  *     rtpsession.send_rtcp_src ! udpsink host="127.0.0.1" port=5002 \
 104  *         sync=false async=false
 105  * ]| Send two audio streams to port 5000.
 106  * |[
 107  * gst-launch-1.0 rtpsession name=rtpsession rtp-profile=avpf \
 108  *     udpsrc port=5000 caps="application/x-rtp,media=(string)audio,clock-rate=(int)48000,encoding-name=(string)OPUS,payload=(int)97" ! \
 109  *         rtpsession.recv_rtp_sink \
 110  *     rtpsession.recv_rtp_src ! \
 111  *         rtprtxreceive payload-type-map="application/x-rtp-pt-map,97=(uint)99" ! \
 112  *         rtpssrcdemux name=demux \
 113  *     demux. ! queue ! rtpjitterbuffer do-retransmission=true ! rtpopusdepay ! \
 114  *         opusdec ! audioconvert ! autoaudiosink \
 115  *     demux. ! queue ! rtpjitterbuffer do-retransmission=true ! rtpopusdepay ! \
 116  *         opusdec ! audioconvert ! autoaudiosink \
 117  *     udpsrc port=5002 ! rtpsession.recv_rtcp_sink \
 118  *     rtpsession.send_rtcp_src ! udpsink host="127.0.0.1" port=5001 \
 119  *         sync=false async=false
 120  * ]| Receive two audio streams from port 5000.
 121  *
 122  * In this example we are streaming two streams of the same type through the
 123  * same port. They, however, are using a different SSRC (ssrc is randomly
 124  * generated on each payloader - rtpopuspay in this example), so they can be
 125  * identified and demultiplexed by rtpssrcdemux on the receiver side. This is
 126  * an example of SSRC-multiplexing.
 127  *
 128  * It is important here to use a different starting sequence number
 129  * (seqnum-offset), since this is the only means of identification that
 130  * rtprtxreceive uses the very first time to identify retransmission streams.
 131  * It is an error, according to RFC4588 to have two retransmission requests for
 132  * packets belonging to two different streams but with the same sequence number.
 133  * Note that the default seqnum-offset value (-1, which means random) would
 134  * work just fine, but it is overriden here for illustration purposes.
 135  */
 136
 137 #ifdef HAVE_CONFIG_H
 138 #include "config.h"
 139 #endif
 140
 141 #include <gst/gst.h>
 142 #include <gst/rtp/gstrtpbuffer.h>
 143 #include <string.h>
 144 #include <stdlib.h>
 145
 146 #include "gstrtprtxreceive.h"
 147
 148 #define ASSOC_TIMEOUT (GST_SECOND)
 149
 150 GST_DEBUG_CATEGORY_STATIC (gst_rtp_rtx_receive_debug);
 151 #define GST_CAT_DEFAULT gst_rtp_rtx_receive_debug
 152
 153 enum
 154 {
 155   PROP_0,
 156   PROP_PAYLOAD_TYPE_MAP,
 157   PROP_NUM_RTX_REQUESTS,
 158   PROP_NUM_RTX_PACKETS,
 159   PROP_NUM_RTX_ASSOC_PACKETS
 160 };
 161
 162 static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",
 163     GST_PAD_SRC,
 164     GST_PAD_ALWAYS,
 165     GST_STATIC_CAPS ("application/x-rtp")
 166     );
 167
 168 static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink",
 169     GST_PAD_SINK,
 170     GST_PAD_ALWAYS,
 171     GST_STATIC_CAPS ("application/x-rtp")
 172     );
 173
 174 static gboolean gst_rtp_rtx_receive_src_event (GstPad * pad, GstObject * parent,
 175     GstEvent * event);
 176 static GstFlowReturn gst_rtp_rtx_receive_chain (GstPad * pad,
 177     GstObject * parent, GstBuffer * buffer);
 178
 179 static GstStateChangeReturn gst_rtp_rtx_receive_change_state (GstElement *
 180     element, GstStateChange transition);
 181
 182 static void gst_rtp_rtx_receive_set_property (GObject * object, guint prop_id,
 183     const GValue * value, GParamSpec * pspec);
 184 static void gst_rtp_rtx_receive_get_property (GObject * object, guint prop_id,
 185     GValue * value, GParamSpec * pspec);
 186 static void gst_rtp_rtx_receive_finalize (GObject * object);
 187
 188 G_DEFINE_TYPE (GstRtpRtxReceive, gst_rtp_rtx_receive, GST_TYPE_ELEMENT);
 189
 190 static void
 191 gst_rtp_rtx_receive_class_init (GstRtpRtxReceiveClass * klass)
 192 {
 193   GObjectClass *gobject_class;
 194   GstElementClass *gstelement_class;
 195
 196   gobject_class = (GObjectClass *) klass;
 197   gstelement_class = (GstElementClass *) klass;
 198
 199   gobject_class->get_property = gst_rtp_rtx_receive_get_property;
 200   gobject_class->set_property = gst_rtp_rtx_receive_set_property;
 201   gobject_class->finalize = gst_rtp_rtx_receive_finalize;
 202
 203   g_object_class_install_property (gobject_class, PROP_PAYLOAD_TYPE_MAP,
 204       g_param_spec_boxed ("payload-type-map", "Payload Type Map",
 205           "Map of original payload types to their retransmission payload types",
 206           GST_TYPE_STRUCTURE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 207
 208   g_object_class_install_property (gobject_class, PROP_NUM_RTX_REQUESTS,
 209       g_param_spec_uint ("num-rtx-requests", "Num RTX Requests",
 210           "Number of retransmission events received", 0, G_MAXUINT,
 211           0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 212
 213   g_object_class_install_property (gobject_class, PROP_NUM_RTX_PACKETS,
 214       g_param_spec_uint ("num-rtx-packets", "Num RTX Packets",
 215           " Number of retransmission packets received", 0, G_MAXUINT,
 216           0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 217
 218   g_object_class_install_property (gobject_class, PROP_NUM_RTX_ASSOC_PACKETS,
 219       g_param_spec_uint ("num-rtx-assoc-packets",
 220           "Num RTX Associated Packets", "Number of retransmission packets "
 221           "correctly associated with retransmission requests", 0, G_MAXUINT,
 222           0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 223
 224   gst_element_class_add_static_pad_template (gstelement_class, &src_factory);
 225   gst_element_class_add_static_pad_template (gstelement_class, &sink_factory);
 226
 227   gst_element_class_set_static_metadata (gstelement_class,
 228       "RTP Retransmission receiver", "Codec",
 229       "Receive retransmitted RTP packets according to RFC4588",
 230       "Julien Isorce <julien.isorce@collabora.co.uk>");
 231
 232   gstelement_class->change_state =
 233       GST_DEBUG_FUNCPTR (gst_rtp_rtx_receive_change_state);
 234 }
 235
 236 static void
 237 gst_rtp_rtx_receive_reset (GstRtpRtxReceive * rtx)
 238 {
 239   GST_OBJECT_LOCK (rtx);
 240   g_hash_table_remove_all (rtx->ssrc2_ssrc1_map);
 241   g_hash_table_remove_all (rtx->seqnum_ssrc1_map);
 242   rtx->num_rtx_requests = 0;
 243   rtx->num_rtx_packets = 0;
 244   rtx->num_rtx_assoc_packets = 0;
 245   GST_OBJECT_UNLOCK (rtx);
 246 }
 247
 248 static void
 249 gst_rtp_rtx_receive_finalize (GObject * object)
 250 {
 251   GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (object);
 252
 253   g_hash_table_unref (rtx->ssrc2_ssrc1_map);
 254   g_hash_table_unref (rtx->seqnum_ssrc1_map);
 255   g_hash_table_unref (rtx->rtx_pt_map);
 256   if (rtx->rtx_pt_map_structure)
 257     gst_structure_free (rtx->rtx_pt_map_structure);
 258
 259   G_OBJECT_CLASS (gst_rtp_rtx_receive_parent_class)->finalize (object);
 260 }
 261
 262 typedef struct
 263 {
 264   guint32 ssrc;
 265   GstClockTime time;
 266 } SsrcAssoc;
 267
 268 static SsrcAssoc *
 269 ssrc_assoc_new (guint32 ssrc, GstClockTime time)
 270 {
 271   SsrcAssoc *assoc = g_slice_new (SsrcAssoc);
 272
 273   assoc->ssrc = ssrc;
 274   assoc->time = time;
 275
 276   return assoc;
 277 }
 278
 279 static void
 280 ssrc_assoc_free (SsrcAssoc * assoc)
 281 {
 282   g_slice_free (SsrcAssoc, assoc);
 283 }
 284
 285 static void
 286 gst_rtp_rtx_receive_init (GstRtpRtxReceive * rtx)
 287 {
 288   GstElementClass *klass = GST_ELEMENT_GET_CLASS (rtx);
 289
 290   rtx->srcpad =
 291       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
 292           "src"), "src");
 293   GST_PAD_SET_PROXY_CAPS (rtx->srcpad);
 294   GST_PAD_SET_PROXY_ALLOCATION (rtx->srcpad);
 295   gst_pad_set_event_function (rtx->srcpad,
 296       GST_DEBUG_FUNCPTR (gst_rtp_rtx_receive_src_event));
 297   gst_element_add_pad (GST_ELEMENT (rtx), rtx->srcpad);
 298
 299   rtx->sinkpad =
 300       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
 301           "sink"), "sink");
 302   GST_PAD_SET_PROXY_CAPS (rtx->sinkpad);
 303   GST_PAD_SET_PROXY_ALLOCATION (rtx->sinkpad);
 304   gst_pad_set_chain_function (rtx->sinkpad,
 305       GST_DEBUG_FUNCPTR (gst_rtp_rtx_receive_chain));
 306   gst_element_add_pad (GST_ELEMENT (rtx), rtx->sinkpad);
 307
 308   rtx->ssrc2_ssrc1_map = g_hash_table_new (g_direct_hash, g_direct_equal);
 309   rtx->seqnum_ssrc1_map = g_hash_table_new_full (g_direct_hash, g_direct_equal,
 310       NULL, (GDestroyNotify) ssrc_assoc_free);
 311
 312   rtx->rtx_pt_map = g_hash_table_new (g_direct_hash, g_direct_equal);
 313 }
 314
 315 static gboolean
 316 gst_rtp_rtx_receive_src_event (GstPad * pad, GstObject * parent,
 317     GstEvent * event)
 318 {
 319   GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (parent);
 320   gboolean res;
 321
 322   switch (GST_EVENT_TYPE (event)) {
 323     case GST_EVENT_CUSTOM_UPSTREAM:
 324     {
 325       const GstStructure *s = gst_event_get_structure (event);
 326
 327       /* This event usually comes from the downstream gstrtpjitterbuffer */
 328       if (gst_structure_has_name (s, "GstRTPRetransmissionRequest")) {
 329         guint seqnum = 0;
 330         guint ssrc = 0;
 331         gpointer ssrc2 = 0;
 332
 333         /* retrieve seqnum of the packet that need to be retransmitted */
 334         if (!gst_structure_get_uint (s, "seqnum", &seqnum))
 335           seqnum = -1;
 336
 337         /* retrieve ssrc of the packet that need to be retransmitted
 338          * it's useful when reconstructing the original packet from the rtx packet */
 339         if (!gst_structure_get_uint (s, "ssrc", &ssrc))
 340           ssrc = -1;
 341
 342         GST_DEBUG_OBJECT (rtx, "got rtx request for seqnum: %u, ssrc: %X",
 343             seqnum, ssrc);
 344
 345         GST_OBJECT_LOCK (rtx);
 346
 347         /* increase number of seen requests for our statistics */
 348         ++rtx->num_rtx_requests;
 349
 350         /* First, we lookup in our map to see if we have already associate this
 351          * master stream ssrc with its retransmitted stream.
 352          * Every ssrc are unique so we can use the same hash table
 353          * for both retrieving the ssrc1 from ssrc2 and also ssrc2 from ssrc1
 354          */
 355         if (g_hash_table_lookup_extended (rtx->ssrc2_ssrc1_map,
 356                 GUINT_TO_POINTER (ssrc), NULL, &ssrc2)
 357             && GPOINTER_TO_UINT (ssrc2) != GPOINTER_TO_UINT (ssrc)) {
 358           GST_TRACE_OBJECT (rtx, "Retransmited stream %X already associated "
 359               "to its master, %X", GPOINTER_TO_UINT (ssrc2), ssrc);
 360         } else {
 361           SsrcAssoc *assoc;
 362
 363           /* not already associated but also we have to check that we have not
 364            * already considered this request.
 365            */
 366           if (g_hash_table_lookup_extended (rtx->seqnum_ssrc1_map,
 367                   GUINT_TO_POINTER (seqnum), NULL, (gpointer *) & assoc)) {
 368             if (assoc->ssrc == ssrc) {
 369               /* same seqnum, same ssrc */
 370
 371               /* do nothing because we have already considered this request
 372                * The jitter may be too impatient of the rtx packet has been
 373                * lost too.
 374                * It does not mean we reject the event, we still want to forward
 375                * the request to the gstrtpsession to be translater into a FB NACK
 376                */
 377               GST_LOG_OBJECT (rtx, "Duplicate request: seqnum: %u, ssrc: %X",
 378                   seqnum, ssrc);
 379             } else {
 380               /* same seqnum, different ssrc */
 381
 382               /* If the association attempt is larger than ASSOC_TIMEOUT,
 383                * then we give up on it, and try this one.
 384                */
 385               if (!GST_CLOCK_TIME_IS_VALID (rtx->last_time) ||
 386                   !GST_CLOCK_TIME_IS_VALID (assoc->time) ||
 387                   assoc->time + ASSOC_TIMEOUT < rtx->last_time) {
 388                 /* From RFC 4588:
 389                  * the receiver MUST NOT have two outstanding requests for the
 390                  * same packet sequence number in two different original streams
 391                  * before the association is resolved. Otherwise it's impossible
 392                  * to associate a rtx stream and its master stream
 393                  */
 394
 395                 /* remove seqnum in order to reuse the spot */
 396                 g_hash_table_remove (rtx->seqnum_ssrc1_map,
 397                     GUINT_TO_POINTER (seqnum));
 398                 goto retransmit;
 399               } else {
 400                 GST_INFO_OBJECT (rtx, "rejecting request for seqnum %u"
 401                     " of master stream %X; there is already a pending request "
 402                     "for the same seqnum on ssrc %X that has not expired",
 403                     seqnum, ssrc, assoc->ssrc);
 404
 405                 /* do not forward the event as we are rejecting this request */
 406                 GST_OBJECT_UNLOCK (rtx);
 407                 gst_event_unref (event);
 408                 return TRUE;
 409               }
 410             }
 411           } else {
 412           retransmit:
 413             /* the request has not been already considered
 414              * insert it for the first time */
 415             g_hash_table_insert (rtx->seqnum_ssrc1_map,
 416                 GUINT_TO_POINTER (seqnum),
 417                 ssrc_assoc_new (ssrc, rtx->last_time));
 418           }
 419         }
 420
 421         GST_DEBUG_OBJECT (rtx, "packet number %u of master stream %X"
 422             " needs to be retransmitted", seqnum, ssrc);
 423
 424         GST_OBJECT_UNLOCK (rtx);
 425       }
 426
 427       /* Transfer event upstream so that the request can acutally by translated
 428        * through gstrtpsession through the network */
 429       res = gst_pad_event_default (pad, parent, event);
 430       break;
 431     }
 432     default:
 433       res = gst_pad_event_default (pad, parent, event);
 434       break;
 435   }
 436   return res;
 437 }
 438
 439 /* Copy fixed header and extension. Replace current ssrc by ssrc1,
 440  * remove OSN and replace current seq num by OSN.
 441  * Copy memory to avoid to manually copy each rtp buffer field.
 442  */
 443 static GstBuffer *
 444 _gst_rtp_buffer_new_from_rtx (GstRTPBuffer * rtp, guint32 ssrc1,
 445     guint16 orign_seqnum, guint8 origin_payload_type)
 446 {
 447   GstMemory *mem = NULL;
 448   GstRTPBuffer new_rtp = GST_RTP_BUFFER_INIT;
 449   GstBuffer *new_buffer = gst_buffer_new ();
 450   GstMapInfo map;
 451   guint payload_len = 0;
 452
 453   /* copy fixed header */
 454   mem = gst_memory_copy (rtp->map[0].memory,
 455       (guint8 *) rtp->data[0] - rtp->map[0].data, rtp->size[0]);
 456   gst_buffer_append_memory (new_buffer, mem);
 457
 458   /* copy extension if any */
 459   if (rtp->size[1]) {
 460     mem = gst_memory_copy (rtp->map[1].memory,
 461         (guint8 *) rtp->data[1] - rtp->map[1].data, rtp->size[1]);
 462     gst_buffer_append_memory (new_buffer, mem);
 463   }
 464
 465   /* copy payload and remove OSN */
 466   payload_len = rtp->size[2] - 2;
 467   mem = gst_allocator_alloc (NULL, payload_len, NULL);
 468
 469   gst_memory_map (mem, &map, GST_MAP_WRITE);
 470   if (rtp->size[2])
 471     memcpy (map.data, (guint8 *) rtp->data[2] + 2, payload_len);
 472   gst_memory_unmap (mem, &map);
 473   gst_buffer_append_memory (new_buffer, mem);
 474
 475   /* the sender always constructs rtx packets without padding,
 476    * But the receiver can still receive rtx packets with padding.
 477    * So just copy it.
 478    */
 479   if (rtp->size[3]) {
 480     guint pad_len = rtp->size[3];
 481
 482     mem = gst_allocator_alloc (NULL, pad_len, NULL);
 483
 484     gst_memory_map (mem, &map, GST_MAP_WRITE);
 485     map.data[pad_len - 1] = pad_len;
 486     gst_memory_unmap (mem, &map);
 487
 488     gst_buffer_append_memory (new_buffer, mem);
 489   }
 490
 491   /* set ssrc and seq num */
 492   gst_rtp_buffer_map (new_buffer, GST_MAP_WRITE, &new_rtp);
 493   gst_rtp_buffer_set_ssrc (&new_rtp, ssrc1);
 494   gst_rtp_buffer_set_seq (&new_rtp, orign_seqnum);
 495   gst_rtp_buffer_set_payload_type (&new_rtp, origin_payload_type);
 496   gst_rtp_buffer_unmap (&new_rtp);
 497
 498   gst_buffer_copy_into (new_buffer, rtp->buffer,
 499       GST_BUFFER_COPY_FLAGS | GST_BUFFER_COPY_TIMESTAMPS, 0, -1);
 500   GST_BUFFER_FLAG_SET (new_buffer, GST_RTP_BUFFER_FLAG_RETRANSMISSION);
 501
 502   return new_buffer;
 503 }
 504
 505 static GstFlowReturn
 506 gst_rtp_rtx_receive_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 507 {
 508   GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (parent);
 509   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 510   GstFlowReturn ret = GST_FLOW_OK;
 511   GstBuffer *new_buffer = NULL;
 512   guint32 ssrc = 0;
 513   gpointer ssrc1 = 0;
 514   guint32 ssrc2 = 0;
 515   guint16 seqnum = 0;
 516   guint16 orign_seqnum = 0;
 517   guint8 payload_type = 0;
 518   gpointer payload = NULL;
 519   guint8 origin_payload_type = 0;
 520   gboolean is_rtx;
 521   gboolean drop = FALSE;
 522
 523   /* map current rtp packet to parse its header */
 524   if (!gst_rtp_buffer_map (buffer, GST_MAP_READ, &rtp))
 525     goto invalid_buffer;
 526
 527   ssrc = gst_rtp_buffer_get_ssrc (&rtp);
 528   seqnum = gst_rtp_buffer_get_seq (&rtp);
 529   payload_type = gst_rtp_buffer_get_payload_type (&rtp);
 530
 531   /* check if we have a retransmission packet (this information comes from SDP) */
 532   GST_OBJECT_LOCK (rtx);
 533
 534   is_rtx =
 535       g_hash_table_lookup_extended (rtx->rtx_pt_map,
 536       GUINT_TO_POINTER (payload_type), NULL, NULL);
 537
 538   if (is_rtx) {
 539     payload = gst_rtp_buffer_get_payload (&rtp);
 540
 541     if (!payload || gst_rtp_buffer_get_payload_len (&rtp) < 2) {
 542       GST_OBJECT_UNLOCK (rtx);
 543       gst_rtp_buffer_unmap (&rtp);
 544       goto invalid_buffer;
 545     }
 546   }
 547
 548   rtx->last_time = GST_BUFFER_PTS (buffer);
 549
 550   if (g_hash_table_size (rtx->seqnum_ssrc1_map) > 0) {
 551     GHashTableIter iter;
 552     gpointer key, value;
 553
 554     g_hash_table_iter_init (&iter, rtx->seqnum_ssrc1_map);
 555     while (g_hash_table_iter_next (&iter, &key, &value)) {
 556       SsrcAssoc *assoc = value;
 557
 558       /* remove association request if it is too old */
 559       if (GST_CLOCK_TIME_IS_VALID (rtx->last_time) &&
 560           GST_CLOCK_TIME_IS_VALID (assoc->time) &&
 561           assoc->time + ASSOC_TIMEOUT < rtx->last_time) {
 562         g_hash_table_iter_remove (&iter);
 563       }
 564     }
 565   }
 566
 567   /* if the current packet is from a retransmission stream */
 568   if (is_rtx) {
 569     /* increase our statistic */
 570     ++rtx->num_rtx_packets;
 571
 572     /* read OSN in the rtx payload */
 573     orign_seqnum = GST_READ_UINT16_BE (gst_rtp_buffer_get_payload (&rtp));
 574     origin_payload_type =
 575         GPOINTER_TO_UINT (g_hash_table_lookup (rtx->rtx_pt_map,
 576             GUINT_TO_POINTER (payload_type)));
 577
 578     GST_DEBUG_OBJECT (rtx, "Got rtx packet: rtx seqnum %u, rtx ssrc %X, "
 579         "rtx pt %u, orig seqnum %u, orig pt %u", seqnum, ssrc, payload_type,
 580         orign_seqnum, origin_payload_type);
 581
 582     /* first we check if we already have associated this retransmission stream
 583      * to a master stream */
 584     if (g_hash_table_lookup_extended (rtx->ssrc2_ssrc1_map,
 585             GUINT_TO_POINTER (ssrc), NULL, &ssrc1)) {
 586       GST_TRACE_OBJECT (rtx,
 587           "packet is from retransmission stream %X already associated to "
 588           "master stream %X", ssrc, GPOINTER_TO_UINT (ssrc1));
 589       ssrc2 = ssrc;
 590     } else {
 591       SsrcAssoc *assoc;
 592
 593       /* the current retransmitted packet has its rtx stream not already
 594        * associated to a master stream, so retrieve it from our request
 595        * history */
 596       if (g_hash_table_lookup_extended (rtx->seqnum_ssrc1_map,
 597               GUINT_TO_POINTER (orign_seqnum), NULL, (gpointer *) & assoc)) {
 598         GST_LOG_OBJECT (rtx,
 599             "associating retransmitted stream %X to master stream %X thanks "
 600             "to rtx packet %u (orig seqnum %u)", ssrc, assoc->ssrc, seqnum,
 601             orign_seqnum);
 602         ssrc1 = GUINT_TO_POINTER (assoc->ssrc);
 603         ssrc2 = ssrc;
 604
 605         /* just put a guard */
 606         if (GPOINTER_TO_UINT (ssrc1) == ssrc2)
 607           GST_WARNING_OBJECT (rtx, "RTX receiver ssrc2_ssrc1_map bad state, "
 608               "master and rtx SSRCs are the same (%X)\n", ssrc);
 609
 610         /* free the spot so that this seqnum can be used to do another
 611          * association */
 612         g_hash_table_remove (rtx->seqnum_ssrc1_map,
 613             GUINT_TO_POINTER (orign_seqnum));
 614
 615         /* actually do the association between rtx stream and master stream */
 616         g_hash_table_insert (rtx->ssrc2_ssrc1_map, GUINT_TO_POINTER (ssrc2),
 617             ssrc1);
 618
 619         /* also do the association between master stream and rtx stream
 620          * every ssrc are unique so we can use the same hash table
 621          * for both retrieving the ssrc1 from ssrc2 and also ssrc2 from ssrc1
 622          */
 623         g_hash_table_insert (rtx->ssrc2_ssrc1_map, ssrc1,
 624             GUINT_TO_POINTER (ssrc2));
 625
 626       } else {
 627         /* we are not able to associate this rtx packet with a master stream */
 628         GST_INFO_OBJECT (rtx,
 629             "dropping rtx packet %u because its orig seqnum (%u) is not in our"
 630             " pending retransmission requests", seqnum, orign_seqnum);
 631         drop = TRUE;
 632       }
 633     }
 634   }
 635
 636   /* if not dropped the packet was successfully associated */
 637   if (is_rtx && !drop)
 638     ++rtx->num_rtx_assoc_packets;
 639
 640   GST_OBJECT_UNLOCK (rtx);
 641
 642   /* just drop the packet if the association could not have been made */
 643   if (drop) {
 644     gst_rtp_buffer_unmap (&rtp);
 645     gst_buffer_unref (buffer);
 646     return GST_FLOW_OK;
 647   }
 648
 649   /* create the retransmission packet */
 650   if (is_rtx)
 651     new_buffer =
 652         _gst_rtp_buffer_new_from_rtx (&rtp, GPOINTER_TO_UINT (ssrc1),
 653         orign_seqnum, origin_payload_type);
 654
 655   gst_rtp_buffer_unmap (&rtp);
 656
 657   /* push the packet */
 658   if (is_rtx) {
 659     gst_buffer_unref (buffer);
 660     GST_LOG_OBJECT (rtx, "pushing packet seqnum:%u from restransmission "
 661         "stream ssrc: %X (master ssrc %X)", orign_seqnum, ssrc2,
 662         GPOINTER_TO_UINT (ssrc1));
 663     ret = gst_pad_push (rtx->srcpad, new_buffer);
 664   } else {
 665     GST_TRACE_OBJECT (rtx, "pushing packet seqnum:%u from master stream "
 666         "ssrc: %X", seqnum, ssrc);
 667     ret = gst_pad_push (rtx->srcpad, buffer);
 668   }
 669
 670   return ret;
 671
 672 invalid_buffer:
 673   {
 674     GST_ELEMENT_WARNING (rtx, STREAM, DECODE, (NULL),
 675         ("Received invalid RTP payload, dropping"));
 676     gst_buffer_unref (buffer);
 677     return GST_FLOW_OK;
 678   }
 679 }
 680
 681 static void
 682 gst_rtp_rtx_receive_get_property (GObject * object,
 683     guint prop_id, GValue * value, GParamSpec * pspec)
 684 {
 685   GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (object);
 686
 687   switch (prop_id) {
 688     case PROP_PAYLOAD_TYPE_MAP:
 689       GST_OBJECT_LOCK (rtx);
 690       g_value_set_boxed (value, rtx->rtx_pt_map_structure);
 691       GST_OBJECT_UNLOCK (rtx);
 692       break;
 693     case PROP_NUM_RTX_REQUESTS:
 694       GST_OBJECT_LOCK (rtx);
 695       g_value_set_uint (value, rtx->num_rtx_requests);
 696       GST_OBJECT_UNLOCK (rtx);
 697       break;
 698     case PROP_NUM_RTX_PACKETS:
 699       GST_OBJECT_LOCK (rtx);
 700       g_value_set_uint (value, rtx->num_rtx_packets);
 701       GST_OBJECT_UNLOCK (rtx);
 702       break;
 703     case PROP_NUM_RTX_ASSOC_PACKETS:
 704       GST_OBJECT_LOCK (rtx);
 705       g_value_set_uint (value, rtx->num_rtx_assoc_packets);
 706       GST_OBJECT_UNLOCK (rtx);
 707       break;
 708     default:
 709       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 710       break;
 711   }
 712 }
 713
 714 static gboolean
 715 structure_to_hash_table_inv (GQuark field_id, const GValue * value,
 716     gpointer hash)
 717 {
 718   const gchar *field_str;
 719   guint field_uint;
 720   guint value_uint;
 721
 722   field_str = g_quark_to_string (field_id);
 723   field_uint = atoi (field_str);
 724   value_uint = g_value_get_uint (value);
 725   g_hash_table_insert ((GHashTable *) hash, GUINT_TO_POINTER (value_uint),
 726       GUINT_TO_POINTER (field_uint));
 727
 728   return TRUE;
 729 }
 730
 731 static void
 732 gst_rtp_rtx_receive_set_property (GObject * object,
 733     guint prop_id, const GValue * value, GParamSpec * pspec)
 734 {
 735   GstRtpRtxReceive *rtx = GST_RTP_RTX_RECEIVE (object);
 736
 737   switch (prop_id) {
 738     case PROP_PAYLOAD_TYPE_MAP:
 739       GST_OBJECT_LOCK (rtx);
 740       if (rtx->rtx_pt_map_structure)
 741         gst_structure_free (rtx->rtx_pt_map_structure);
 742       rtx->rtx_pt_map_structure = g_value_dup_boxed (value);
 743       g_hash_table_remove_all (rtx->rtx_pt_map);
 744       gst_structure_foreach (rtx->rtx_pt_map_structure,
 745           structure_to_hash_table_inv, rtx->rtx_pt_map);
 746       GST_OBJECT_UNLOCK (rtx);
 747       break;
 748     default:
 749       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 750       break;
 751   }
 752 }
 753
 754 static GstStateChangeReturn
 755 gst_rtp_rtx_receive_change_state (GstElement * element,
 756     GstStateChange transition)
 757 {
 758   GstStateChangeReturn ret;
 759   GstRtpRtxReceive *rtx;
 760
 761   rtx = GST_RTP_RTX_RECEIVE (element);
 762
 763   switch (transition) {
 764     default:
 765       break;
 766   }
 767
 768   ret =
 769       GST_ELEMENT_CLASS (gst_rtp_rtx_receive_parent_class)->change_state
 770       (element, transition);
 771
 772   switch (transition) {
 773     case GST_STATE_CHANGE_PAUSED_TO_READY:
 774       gst_rtp_rtx_receive_reset (rtx);
 775       break;
 776     default:
 777       break;
 778   }
 779
 780   return ret;
 781 }
 782
 783 gboolean
 784 gst_rtp_rtx_receive_plugin_init (GstPlugin * plugin)
 785 {
 786   GST_DEBUG_CATEGORY_INIT (gst_rtp_rtx_receive_debug, "rtprtxreceive", 0,
 787       "rtp retransmission receiver");
 788
 789   return gst_element_register (plugin, "rtprtxreceive", GST_RANK_NONE,
 790       GST_TYPE_RTP_RTX_RECEIVE);
 791 }