subprojects/gst-plugins-bad/gst/rist/gstristrtxsend.c

   1 /* RIST Retransmission sender element for GStreamer
   2  *
   3  * gsristprtxsend.c:
   4  *
   5  * Copyright (C) 2013-2019 Collabora Ltd.
   6  *   @author Julien Isorce <julien.isorce@collabora.co.uk>
   7  *           Nicoas Dufresne <nicolas.dufresne@collabora.com>
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Library General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Library General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Library General Public
  20  * License along with this library; if not, write to the
  21  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  22  * Boston, MA 02110-1301, USA.
  23  */
  24
  25 /**
  26  * SECTION:element-ristrtxsend
  27  * @title: ristrtxsend
  28  * @see_also: ristrtxreceive
  29  *
  30  * This elements replies to custom events 'GstRTPRetransmissionRequest' and
  31  * when available sends in RIST form the lost packet. This element is intented
  32  * to be used by ristsink element.
  33  */
  34
  35 #ifdef HAVE_CONFIG_H
  36 #include "config.h"
  37 #endif
  38
  39 #include <gst/gst.h>
  40 #include <gst/rtp/gstrtpbuffer.h>
  41 #include <gst/base/gstdataqueue.h>
  42
  43 #include "gstrist.h"
  44
  45 GST_DEBUG_CATEGORY_STATIC (gst_rist_rtx_send_debug);
  46 #define GST_CAT_DEFAULT gst_rist_rtx_send_debug
  47
  48 #define DEFAULT_MAX_SIZE_TIME    0
  49 #define DEFAULT_MAX_SIZE_PACKETS 100
  50
  51 enum
  52 {
  53   PROP_0,
  54   PROP_MAX_SIZE_TIME,
  55   PROP_MAX_SIZE_PACKETS,
  56   PROP_NUM_RTX_REQUESTS,
  57   PROP_NUM_RTX_PACKETS,
  58 };
  59
  60 static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",
  61     GST_PAD_SRC,
  62     GST_PAD_ALWAYS,
  63     GST_STATIC_CAPS ("application/x-rtp")
  64     );
  65
  66 static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink",
  67     GST_PAD_SINK,
  68     GST_PAD_ALWAYS,
  69     GST_STATIC_CAPS ("application/x-rtp, " "clock-rate = (int) [1, MAX]")
  70     );
  71
  72 struct _GstRistRtxSend
  73 {
  74   GstElement element;
  75
  76   /* pad */
  77   GstPad *sinkpad;
  78   GstPad *srcpad;
  79
  80   /* rtp packets that will be pushed out */
  81   GstDataQueue *queue;
  82
  83   /* ssrc -> SSRCRtxData */
  84   GHashTable *ssrc_data;
  85   /* rtx ssrc -> master ssrc */
  86   GHashTable *rtx_ssrcs;
  87
  88   /* buffering control properties */
  89   guint max_size_time;
  90   guint max_size_packets;
  91
  92   /* statistics */
  93   guint num_rtx_requests;
  94   guint num_rtx_packets;
  95 };
  96
  97 static gboolean gst_rist_rtx_send_queue_check_full (GstDataQueue * queue,
  98     guint visible, guint bytes, guint64 time, gpointer checkdata);
  99
 100 static gboolean gst_rist_rtx_send_src_event (GstPad * pad, GstObject * parent,
 101     GstEvent * event);
 102 static gboolean gst_rist_rtx_send_sink_event (GstPad * pad, GstObject * parent,
 103     GstEvent * event);
 104 static GstFlowReturn gst_rist_rtx_send_chain (GstPad * pad, GstObject * parent,
 105     GstBuffer * buffer);
 106 static GstFlowReturn gst_rist_rtx_send_chain_list (GstPad * pad,
 107     GstObject * parent, GstBufferList * list);
 108
 109 static void gst_rist_rtx_send_src_loop (GstRistRtxSend * rtx);
 110 static gboolean gst_rist_rtx_send_activate_mode (GstPad * pad,
 111     GstObject * parent, GstPadMode mode, gboolean active);
 112
 113 static GstStateChangeReturn gst_rist_rtx_send_change_state (GstElement *
 114     element, GstStateChange transition);
 115
 116 static void gst_rist_rtx_send_set_property (GObject * object, guint prop_id,
 117     const GValue * value, GParamSpec * pspec);
 118 static void gst_rist_rtx_send_get_property (GObject * object, guint prop_id,
 119     GValue * value, GParamSpec * pspec);
 120 static void gst_rist_rtx_send_finalize (GObject * object);
 121
 122 G_DEFINE_TYPE_WITH_CODE (GstRistRtxSend, gst_rist_rtx_send, GST_TYPE_ELEMENT,
 123     GST_DEBUG_CATEGORY_INIT (gst_rist_rtx_send_debug, "ristrtxsend", 0,
 124         "RIST retransmission sender"));
 125 GST_ELEMENT_REGISTER_DEFINE (ristrtxsend, "ristrtxsend", GST_RANK_NONE,
 126     GST_TYPE_RIST_RTX_SEND);
 127
 128 typedef struct
 129 {
 130   guint32 extseqnum;
 131   guint32 timestamp;
 132   GstBuffer *buffer;
 133 } BufferQueueItem;
 134
 135 static void
 136 buffer_queue_item_free (BufferQueueItem * item)
 137 {
 138   gst_buffer_unref (item->buffer);
 139   g_slice_free (BufferQueueItem, item);
 140 }
 141
 142 typedef struct
 143 {
 144   guint32 rtx_ssrc;
 145   guint16 seqnum_base, next_seqnum;
 146   gint clock_rate;
 147
 148   /* history of rtp packets */
 149   GSequence *queue;
 150   guint32 max_extseqnum;
 151
 152   /* current rtcp app seqnum extension */
 153   gboolean has_seqnum_ext;
 154   guint16 seqnum_ext;
 155 } SSRCRtxData;
 156
 157 static SSRCRtxData *
 158 ssrc_rtx_data_new (guint32 rtx_ssrc)
 159 {
 160   SSRCRtxData *data = g_slice_new0 (SSRCRtxData);
 161
 162   data->rtx_ssrc = rtx_ssrc;
 163   data->next_seqnum = data->seqnum_base = g_random_int_range (0, G_MAXUINT16);
 164   data->queue = g_sequence_new ((GDestroyNotify) buffer_queue_item_free);
 165   data->max_extseqnum = -1;
 166
 167   return data;
 168 }
 169
 170 static void
 171 ssrc_rtx_data_free (SSRCRtxData * data)
 172 {
 173   g_sequence_free (data->queue);
 174   g_slice_free (SSRCRtxData, data);
 175 }
 176
 177 static void
 178 gst_rist_rtx_send_class_init (GstRistRtxSendClass * klass)
 179 {
 180   GObjectClass *gobject_class;
 181   GstElementClass *gstelement_class;
 182
 183   gobject_class = (GObjectClass *) klass;
 184   gstelement_class = (GstElementClass *) klass;
 185
 186   gobject_class->get_property = gst_rist_rtx_send_get_property;
 187   gobject_class->set_property = gst_rist_rtx_send_set_property;
 188   gobject_class->finalize = gst_rist_rtx_send_finalize;
 189
 190   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_TIME,
 191       g_param_spec_uint ("max-size-time", "Max Size Time",
 192           "Amount of ms to queue (0 = unlimited)", 0, G_MAXUINT,
 193           DEFAULT_MAX_SIZE_TIME, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 194
 195   g_object_class_install_property (gobject_class, PROP_MAX_SIZE_PACKETS,
 196       g_param_spec_uint ("max-size-packets", "Max Size Packets",
 197           "Amount of packets to queue (0 = unlimited)", 0, G_MAXINT16,
 198           DEFAULT_MAX_SIZE_PACKETS,
 199           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 200
 201   g_object_class_install_property (gobject_class, PROP_NUM_RTX_REQUESTS,
 202       g_param_spec_uint ("num-rtx-requests", "Num RTX Requests",
 203           "Number of retransmission events received", 0, G_MAXUINT,
 204           0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 205
 206   g_object_class_install_property (gobject_class, PROP_NUM_RTX_PACKETS,
 207       g_param_spec_uint ("num-rtx-packets", "Num RTX Packets",
 208           " Number of retransmission packets sent", 0, G_MAXUINT,
 209           0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 210
 211   gst_element_class_add_static_pad_template (gstelement_class, &src_factory);
 212   gst_element_class_add_static_pad_template (gstelement_class, &sink_factory);
 213
 214   gst_element_class_set_static_metadata (gstelement_class,
 215       "RIST Retransmission Sender", "Codec",
 216       "Retransmit RTP packets when needed, according to VSF TR-06-1",
 217       "Nicolas Dufresne <nicolas.dufresne@collabora.com>");
 218
 219   gstelement_class->change_state =
 220       GST_DEBUG_FUNCPTR (gst_rist_rtx_send_change_state);
 221 }
 222
 223 static void
 224 gst_rist_rtx_send_reset (GstRistRtxSend * rtx)
 225 {
 226   GST_OBJECT_LOCK (rtx);
 227   gst_data_queue_flush (rtx->queue);
 228   g_hash_table_remove_all (rtx->ssrc_data);
 229   g_hash_table_remove_all (rtx->rtx_ssrcs);
 230   rtx->num_rtx_requests = 0;
 231   rtx->num_rtx_packets = 0;
 232   GST_OBJECT_UNLOCK (rtx);
 233 }
 234
 235 static void
 236 gst_rist_rtx_send_finalize (GObject * object)
 237 {
 238   GstRistRtxSend *rtx = GST_RIST_RTX_SEND (object);
 239
 240   g_hash_table_unref (rtx->ssrc_data);
 241   g_hash_table_unref (rtx->rtx_ssrcs);
 242   g_object_unref (rtx->queue);
 243
 244   G_OBJECT_CLASS (gst_rist_rtx_send_parent_class)->finalize (object);
 245 }
 246
 247 static void
 248 gst_rist_rtx_send_init (GstRistRtxSend * rtx)
 249 {
 250   GstElementClass *klass = GST_ELEMENT_GET_CLASS (rtx);
 251
 252   rtx->srcpad =
 253       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
 254           "src"), "src");
 255   GST_PAD_SET_PROXY_CAPS (rtx->srcpad);
 256   GST_PAD_SET_PROXY_ALLOCATION (rtx->srcpad);
 257   gst_pad_set_event_function (rtx->srcpad,
 258       GST_DEBUG_FUNCPTR (gst_rist_rtx_send_src_event));
 259   gst_pad_set_activatemode_function (rtx->srcpad,
 260       GST_DEBUG_FUNCPTR (gst_rist_rtx_send_activate_mode));
 261   gst_element_add_pad (GST_ELEMENT (rtx), rtx->srcpad);
 262
 263   rtx->sinkpad =
 264       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
 265           "sink"), "sink");
 266   GST_PAD_SET_PROXY_CAPS (rtx->sinkpad);
 267   GST_PAD_SET_PROXY_ALLOCATION (rtx->sinkpad);
 268   gst_pad_set_event_function (rtx->sinkpad,
 269       GST_DEBUG_FUNCPTR (gst_rist_rtx_send_sink_event));
 270   gst_pad_set_chain_function (rtx->sinkpad,
 271       GST_DEBUG_FUNCPTR (gst_rist_rtx_send_chain));
 272   gst_pad_set_chain_list_function (rtx->sinkpad,
 273       GST_DEBUG_FUNCPTR (gst_rist_rtx_send_chain_list));
 274   gst_element_add_pad (GST_ELEMENT (rtx), rtx->sinkpad);
 275
 276   rtx->queue = gst_data_queue_new (gst_rist_rtx_send_queue_check_full, NULL,
 277       NULL, rtx);
 278   rtx->ssrc_data = g_hash_table_new_full (g_direct_hash, g_direct_equal,
 279       NULL, (GDestroyNotify) ssrc_rtx_data_free);
 280   rtx->rtx_ssrcs = g_hash_table_new (g_direct_hash, g_direct_equal);
 281
 282   rtx->max_size_time = DEFAULT_MAX_SIZE_TIME;
 283   rtx->max_size_packets = DEFAULT_MAX_SIZE_PACKETS;
 284 }
 285
 286 static void
 287 gst_rist_rtx_send_set_flushing (GstRistRtxSend * rtx, gboolean flush)
 288 {
 289   GST_OBJECT_LOCK (rtx);
 290   gst_data_queue_set_flushing (rtx->queue, flush);
 291   gst_data_queue_flush (rtx->queue);
 292   GST_OBJECT_UNLOCK (rtx);
 293 }
 294
 295 static gboolean
 296 gst_rist_rtx_send_queue_check_full (GstDataQueue * queue,
 297     guint visible, guint bytes, guint64 time, gpointer checkdata)
 298 {
 299   return FALSE;
 300 }
 301
 302 static void
 303 gst_rtp_rtx_data_queue_item_free (gpointer item)
 304 {
 305   GstDataQueueItem *data = item;
 306   if (data->object)
 307     gst_mini_object_unref (data->object);
 308   g_slice_free (GstDataQueueItem, data);
 309 }
 310
 311 static gboolean
 312 gst_rist_rtx_send_push_out (GstRistRtxSend * rtx, gpointer object)
 313 {
 314   GstDataQueueItem *data;
 315   gboolean success;
 316
 317   data = g_slice_new0 (GstDataQueueItem);
 318   data->object = GST_MINI_OBJECT (object);
 319   data->size = 1;
 320   data->duration = 1;
 321   data->visible = TRUE;
 322   data->destroy = gst_rtp_rtx_data_queue_item_free;
 323
 324   success = gst_data_queue_push (rtx->queue, data);
 325
 326   if (!success)
 327     data->destroy (data);
 328
 329   return success;
 330 }
 331
 332 static SSRCRtxData *
 333 gst_rist_rtx_send_get_ssrc_data (GstRistRtxSend * rtx, guint32 ssrc)
 334 {
 335   SSRCRtxData *data;
 336   guint32 rtx_ssrc = 0;
 337
 338   data = g_hash_table_lookup (rtx->ssrc_data, GUINT_TO_POINTER (ssrc));
 339   if (!data) {
 340     /* See 5.3.2 Retransmitted Packets, original packet have SSRC LSB set to
 341      * 0, while RTX packet have LSB set to 1 */
 342     rtx_ssrc = ssrc + 1;
 343     data = ssrc_rtx_data_new (rtx_ssrc);
 344     g_hash_table_insert (rtx->ssrc_data, GUINT_TO_POINTER (ssrc), data);
 345     g_hash_table_insert (rtx->rtx_ssrcs, GUINT_TO_POINTER (rtx_ssrc),
 346         GUINT_TO_POINTER (ssrc));
 347   }
 348
 349   return data;
 350 }
 351
 352 /*
 353  * see RIST TR-06-1 5.3.2 Retransmitted Packets
 354  *
 355  * RIST simply resend the packet verbatim, with SSRC+1, the defaults SSRC always
 356  * have the LSB set to 0, so we can differentiate the retransmission and the
 357  * normal packet.
 358  */
 359 static GstBuffer *
 360 gst_rtp_rist_buffer_new (GstRistRtxSend * rtx, GstBuffer * buffer, guint32 ssrc)
 361 {
 362   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 363
 364   buffer = gst_buffer_copy_deep (buffer);
 365   gst_rtp_buffer_map (buffer, GST_MAP_WRITE, &rtp);
 366   gst_rtp_buffer_set_ssrc (&rtp, ssrc + 1);
 367   gst_rtp_buffer_unmap (&rtp);
 368
 369   return buffer;
 370 }
 371
 372 static gint
 373 buffer_queue_items_cmp (BufferQueueItem * a, BufferQueueItem * b,
 374     gpointer user_data)
 375 {
 376   /* gst_rtp_buffer_compare_seqnum returns the opposite of what we want,
 377    * it returns negative when seqnum1 > seqnum2 and we want negative
 378    * when b > a, i.e. a is smaller, so it comes first in the sequence */
 379   return a->extseqnum - b->extseqnum;
 380 }
 381
 382 static gboolean
 383 gst_rist_rtx_send_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
 384 {
 385   GstRistRtxSend *rtx = GST_RIST_RTX_SEND (parent);
 386
 387   switch (GST_EVENT_TYPE (event)) {
 388     case GST_EVENT_CUSTOM_UPSTREAM:
 389     {
 390       const GstStructure *s = gst_event_get_structure (event);
 391
 392       /* This event usually comes from the downstream gstrtpsession */
 393       if (gst_structure_has_name (s, "GstRTPRetransmissionRequest")) {
 394         guint seqnum = 0;
 395         guint ssrc = 0;
 396         GstBuffer *rtx_buf = NULL;
 397
 398         /* retrieve seqnum of the packet that need to be retransmitted */
 399         if (!gst_structure_get_uint (s, "seqnum", &seqnum))
 400           seqnum = -1;
 401
 402         /* retrieve ssrc of the packet that need to be retransmitted */
 403         if (!gst_structure_get_uint (s, "ssrc", &ssrc))
 404           ssrc = -1;
 405
 406         GST_DEBUG_OBJECT (rtx, "got rtx request for seqnum: %u, ssrc: %X",
 407             seqnum, ssrc);
 408
 409         GST_OBJECT_LOCK (rtx);
 410         /* check if request is for us */
 411         if (g_hash_table_contains (rtx->ssrc_data, GUINT_TO_POINTER (ssrc))) {
 412           SSRCRtxData *data;
 413           GSequenceIter *iter;
 414           BufferQueueItem search_item;
 415           guint32 extseqnum;
 416
 417           /* update statistics */
 418           ++rtx->num_rtx_requests;
 419
 420           data = gst_rist_rtx_send_get_ssrc_data (rtx, ssrc);
 421
 422
 423           if (data->has_seqnum_ext) {
 424             extseqnum = data->seqnum_ext << 16 | seqnum;
 425           } else {
 426             guint32 max_extseqnum = data->max_extseqnum;
 427             extseqnum = gst_rist_rtp_ext_seq (&max_extseqnum, seqnum);
 428           }
 429
 430           search_item.extseqnum = extseqnum;
 431           iter = g_sequence_lookup (data->queue, &search_item,
 432               (GCompareDataFunc) buffer_queue_items_cmp, NULL);
 433           if (iter) {
 434             BufferQueueItem *item = g_sequence_get (iter);
 435             GST_LOG_OBJECT (rtx, "found %u (%u:%u)", item->extseqnum,
 436                 item->extseqnum >> 16, item->extseqnum & 0xFFFF);
 437             rtx_buf = gst_rtp_rist_buffer_new (rtx, item->buffer, ssrc);
 438           }
 439 #ifndef GST_DISABLE_DEBUG
 440           else {
 441             BufferQueueItem *item = NULL;
 442
 443             iter = g_sequence_get_begin_iter (data->queue);
 444             if (!g_sequence_iter_is_end (iter))
 445               item = g_sequence_get (iter);
 446
 447             if (item && extseqnum < item->extseqnum) {
 448               GST_DEBUG_OBJECT (rtx, "requested seqnum %u has already been "
 449                   "removed from the rtx queue; the first available is %u",
 450                   seqnum, item->extseqnum);
 451             } else {
 452               GST_WARNING_OBJECT (rtx, "requested seqnum %u has not been "
 453                   "transmitted yet in the original stream; either the remote end "
 454                   "is not configured correctly, or the source is too slow",
 455                   seqnum);
 456             }
 457 #endif
 458           }
 459         }
 460         GST_OBJECT_UNLOCK (rtx);
 461
 462         if (rtx_buf)
 463           gst_rist_rtx_send_push_out (rtx, rtx_buf);
 464
 465         gst_event_unref (event);
 466         return TRUE;
 467       }
 468       break;
 469     }
 470     default:
 471       break;
 472   }
 473
 474   return gst_pad_event_default (pad, parent, event);
 475 }
 476
 477 static gboolean
 478 gst_rist_rtx_send_sink_event (GstPad * pad, GstObject * parent,
 479     GstEvent * event)
 480 {
 481   GstRistRtxSend *rtx = GST_RIST_RTX_SEND (parent);
 482
 483   switch (GST_EVENT_TYPE (event)) {
 484     case GST_EVENT_FLUSH_START:
 485       gst_pad_push_event (rtx->srcpad, event);
 486       gst_rist_rtx_send_set_flushing (rtx, TRUE);
 487       gst_pad_pause_task (rtx->srcpad);
 488       return TRUE;
 489     case GST_EVENT_FLUSH_STOP:
 490       gst_pad_push_event (rtx->srcpad, event);
 491       gst_rist_rtx_send_set_flushing (rtx, FALSE);
 492       gst_pad_start_task (rtx->srcpad,
 493           (GstTaskFunction) gst_rist_rtx_send_src_loop, rtx, NULL);
 494       return TRUE;
 495     case GST_EVENT_EOS:
 496       GST_INFO_OBJECT (rtx, "Got EOS - enqueueing it");
 497       gst_rist_rtx_send_push_out (rtx, event);
 498       return TRUE;
 499     case GST_EVENT_CAPS:
 500     {
 501       GstCaps *caps;
 502       GstStructure *s;
 503       guint ssrc;
 504       gint payload;
 505       SSRCRtxData *data;
 506
 507       gst_event_parse_caps (event, &caps);
 508
 509       s = gst_caps_get_structure (caps, 0);
 510       if (!gst_structure_get_uint (s, "ssrc", &ssrc))
 511         ssrc = -1;
 512       if (!gst_structure_get_int (s, "payload", &payload))
 513         payload = -1;
 514
 515       if (payload == -1)
 516         GST_WARNING_OBJECT (rtx, "No payload in caps");
 517
 518       GST_OBJECT_LOCK (rtx);
 519       data = gst_rist_rtx_send_get_ssrc_data (rtx, ssrc);
 520
 521       GST_DEBUG_OBJECT (rtx,
 522           "got caps for payload: %d->%d, ssrc: %u : %" GST_PTR_FORMAT,
 523           payload, ssrc, data->rtx_ssrc, caps);
 524
 525       gst_structure_get_int (s, "clock-rate", &data->clock_rate);
 526
 527       /* The session might need to know the RTX ssrc */
 528       caps = gst_caps_copy (caps);
 529       gst_caps_set_simple (caps, "rtx-ssrc", G_TYPE_UINT, data->rtx_ssrc,
 530           "rtx-seqnum-offset", G_TYPE_UINT, data->seqnum_base, NULL);
 531
 532       GST_DEBUG_OBJECT (rtx, "got clock-rate from caps: %d for ssrc: %u",
 533           data->clock_rate, ssrc);
 534       GST_OBJECT_UNLOCK (rtx);
 535
 536       gst_event_unref (event);
 537       event = gst_event_new_caps (caps);
 538       gst_caps_unref (caps);
 539       break;
 540     }
 541     default:
 542       break;
 543   }
 544
 545   return gst_pad_event_default (pad, parent, event);
 546 }
 547
 548 /* like rtp_jitter_buffer_get_ts_diff() */
 549 static guint32
 550 gst_rist_rtx_send_get_ts_diff (SSRCRtxData * data)
 551 {
 552   guint64 high_ts, low_ts;
 553   BufferQueueItem *high_buf, *low_buf;
 554   guint32 result;
 555
 556   high_buf =
 557       g_sequence_get (g_sequence_iter_prev (g_sequence_get_end_iter
 558           (data->queue)));
 559   low_buf = g_sequence_get (g_sequence_get_begin_iter (data->queue));
 560
 561   if (!high_buf || !low_buf || high_buf == low_buf)
 562     return 0;
 563
 564   high_ts = high_buf->timestamp;
 565   low_ts = low_buf->timestamp;
 566
 567   /* it needs to work if ts wraps */
 568   if (high_ts >= low_ts) {
 569     result = (guint32) (high_ts - low_ts);
 570   } else {
 571     result = (guint32) (high_ts + G_MAXUINT32 + 1 - low_ts);
 572   }
 573
 574   /* return value in ms instead of clock ticks */
 575   return (guint32) gst_util_uint64_scale_int (result, 1000, data->clock_rate);
 576 }
 577
 578 /* Must be called with lock */
 579 static void
 580 process_buffer (GstRistRtxSend * rtx, GstBuffer * buffer)
 581 {
 582   GstRTPBuffer rtp = GST_RTP_BUFFER_INIT;
 583   BufferQueueItem *item;
 584   SSRCRtxData *data;
 585   guint16 seqnum;
 586   guint32 ssrc, rtptime;
 587   guint16 bits;
 588   gpointer extdata;
 589   guint extlen;
 590   gboolean has_seqnum_ext = FALSE;
 591   guint32 extseqnum;
 592
 593   /* read the information we want from the buffer */
 594   gst_rtp_buffer_map (buffer, GST_MAP_READ, &rtp);
 595   seqnum = gst_rtp_buffer_get_seq (&rtp);
 596   ssrc = gst_rtp_buffer_get_ssrc (&rtp);
 597   rtptime = gst_rtp_buffer_get_timestamp (&rtp);
 598   if (gst_rtp_buffer_get_extension_data (&rtp, &bits, &extdata, &extlen)) {
 599     /* Has header extension */
 600     has_seqnum_ext = (bits >> 14) & 1;  /* E */
 601     if (extlen != 1)
 602       has_seqnum_ext = FALSE;
 603     if (has_seqnum_ext)
 604       extseqnum = GST_READ_UINT16_BE (extdata) << 16 | seqnum;
 605   }
 606   gst_rtp_buffer_unmap (&rtp);
 607
 608   GST_TRACE_OBJECT (rtx, "Processing buffer seqnum: %u, ssrc: %X", seqnum,
 609       ssrc);
 610
 611   data = gst_rist_rtx_send_get_ssrc_data (rtx, ssrc);
 612
 613   if (has_seqnum_ext)
 614     data->max_extseqnum = MAX (data->max_extseqnum, extseqnum);
 615   else
 616     extseqnum = gst_rist_rtp_ext_seq (&data->max_extseqnum, seqnum);
 617
 618   /* add current rtp buffer to queue history */
 619   item = g_slice_new0 (BufferQueueItem);
 620   item->extseqnum = extseqnum;
 621   item->timestamp = rtptime;
 622   item->buffer = gst_buffer_ref (buffer);
 623   g_sequence_append (data->queue, item);
 624
 625   /* remove oldest packets from history if they are too many */
 626   if (rtx->max_size_packets) {
 627     while (g_sequence_get_length (data->queue) > rtx->max_size_packets)
 628       g_sequence_remove (g_sequence_get_begin_iter (data->queue));
 629   }
 630   if (rtx->max_size_time) {
 631     while (gst_rist_rtx_send_get_ts_diff (data) > rtx->max_size_time)
 632       g_sequence_remove (g_sequence_get_begin_iter (data->queue));
 633   }
 634 }
 635
 636 static GstFlowReturn
 637 gst_rist_rtx_send_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 638 {
 639   GstRistRtxSend *rtx = GST_RIST_RTX_SEND (parent);
 640   GstFlowReturn ret;
 641
 642   GST_OBJECT_LOCK (rtx);
 643   process_buffer (rtx, buffer);
 644   GST_OBJECT_UNLOCK (rtx);
 645   ret = gst_pad_push (rtx->srcpad, buffer);
 646
 647   return ret;
 648 }
 649
 650 static gboolean
 651 process_buffer_from_list (GstBuffer ** buffer, guint idx, gpointer user_data)
 652 {
 653   process_buffer (user_data, *buffer);
 654   return TRUE;
 655 }
 656
 657 static GstFlowReturn
 658 gst_rist_rtx_send_chain_list (GstPad * pad, GstObject * parent,
 659     GstBufferList * list)
 660 {
 661   GstRistRtxSend *rtx = GST_RIST_RTX_SEND (parent);
 662   GstFlowReturn ret;
 663
 664   GST_OBJECT_LOCK (rtx);
 665   gst_buffer_list_foreach (list, process_buffer_from_list, rtx);
 666   GST_OBJECT_UNLOCK (rtx);
 667
 668   ret = gst_pad_push_list (rtx->srcpad, list);
 669
 670   return ret;
 671 }
 672
 673 static void
 674 gst_rist_rtx_send_src_loop (GstRistRtxSend * rtx)
 675 {
 676   GstDataQueueItem *data;
 677
 678   if (gst_data_queue_pop (rtx->queue, &data)) {
 679     GST_LOG_OBJECT (rtx, "pushing rtx buffer %p", data->object);
 680
 681     if (G_LIKELY (GST_IS_BUFFER (data->object))) {
 682       GST_OBJECT_LOCK (rtx);
 683       /* Update statistics just before pushing. */
 684       rtx->num_rtx_packets++;
 685       GST_OBJECT_UNLOCK (rtx);
 686
 687       gst_pad_push (rtx->srcpad, GST_BUFFER (data->object));
 688     } else if (GST_IS_EVENT (data->object)) {
 689       gst_pad_push_event (rtx->srcpad, GST_EVENT (data->object));
 690
 691       /* after EOS, we should not send any more buffers,
 692        * even if there are more requests coming in */
 693       if (GST_EVENT_TYPE (data->object) == GST_EVENT_EOS) {
 694         gst_rist_rtx_send_set_flushing (rtx, TRUE);
 695       }
 696     } else {
 697       g_assert_not_reached ();
 698     }
 699
 700     data->object = NULL;        /* we no longer own that object */
 701     data->destroy (data);
 702   } else {
 703     GST_LOG_OBJECT (rtx, "flushing");
 704     gst_pad_pause_task (rtx->srcpad);
 705   }
 706 }
 707
 708 static gboolean
 709 gst_rist_rtx_send_activate_mode (GstPad * pad, GstObject * parent,
 710     GstPadMode mode, gboolean active)
 711 {
 712   GstRistRtxSend *rtx = GST_RIST_RTX_SEND (parent);
 713   gboolean ret = FALSE;
 714
 715   switch (mode) {
 716     case GST_PAD_MODE_PUSH:
 717       if (active) {
 718         gst_rist_rtx_send_set_flushing (rtx, FALSE);
 719         ret = gst_pad_start_task (rtx->srcpad,
 720             (GstTaskFunction) gst_rist_rtx_send_src_loop, rtx, NULL);
 721       } else {
 722         gst_rist_rtx_send_set_flushing (rtx, TRUE);
 723         ret = gst_pad_stop_task (rtx->srcpad);
 724       }
 725       GST_INFO_OBJECT (rtx, "activate_mode: active %d, ret %d", active, ret);
 726       break;
 727     default:
 728       break;
 729   }
 730   return ret;
 731 }
 732
 733 static void
 734 gst_rist_rtx_send_get_property (GObject * object,
 735     guint prop_id, GValue * value, GParamSpec * pspec)
 736 {
 737   GstRistRtxSend *rtx = GST_RIST_RTX_SEND (object);
 738
 739   switch (prop_id) {
 740     case PROP_MAX_SIZE_TIME:
 741       GST_OBJECT_LOCK (rtx);
 742       g_value_set_uint (value, rtx->max_size_time);
 743       GST_OBJECT_UNLOCK (rtx);
 744       break;
 745     case PROP_MAX_SIZE_PACKETS:
 746       GST_OBJECT_LOCK (rtx);
 747       g_value_set_uint (value, rtx->max_size_packets);
 748       GST_OBJECT_UNLOCK (rtx);
 749       break;
 750     case PROP_NUM_RTX_REQUESTS:
 751       GST_OBJECT_LOCK (rtx);
 752       g_value_set_uint (value, rtx->num_rtx_requests);
 753       GST_OBJECT_UNLOCK (rtx);
 754       break;
 755     case PROP_NUM_RTX_PACKETS:
 756       GST_OBJECT_LOCK (rtx);
 757       g_value_set_uint (value, rtx->num_rtx_packets);
 758       GST_OBJECT_UNLOCK (rtx);
 759       break;
 760     default:
 761       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 762       break;
 763   }
 764 }
 765
 766 static void
 767 gst_rist_rtx_send_set_property (GObject * object,
 768     guint prop_id, const GValue * value, GParamSpec * pspec)
 769 {
 770   GstRistRtxSend *rtx = GST_RIST_RTX_SEND (object);
 771
 772   switch (prop_id) {
 773     case PROP_MAX_SIZE_TIME:
 774       GST_OBJECT_LOCK (rtx);
 775       rtx->max_size_time = g_value_get_uint (value);
 776       GST_OBJECT_UNLOCK (rtx);
 777       break;
 778     case PROP_MAX_SIZE_PACKETS:
 779       GST_OBJECT_LOCK (rtx);
 780       rtx->max_size_packets = g_value_get_uint (value);
 781       GST_OBJECT_UNLOCK (rtx);
 782       break;
 783     default:
 784       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 785       break;
 786   }
 787 }
 788
 789 static GstStateChangeReturn
 790 gst_rist_rtx_send_change_state (GstElement * element, GstStateChange transition)
 791 {
 792   GstStateChangeReturn ret;
 793   GstRistRtxSend *rtx = GST_RIST_RTX_SEND (element);
 794
 795   ret =
 796       GST_ELEMENT_CLASS (gst_rist_rtx_send_parent_class)->change_state (element,
 797       transition);
 798
 799   switch (transition) {
 800     case GST_STATE_CHANGE_PAUSED_TO_READY:
 801       gst_rist_rtx_send_reset (rtx);
 802       break;
 803     default:
 804       break;
 805   }
 806
 807   return ret;
 808 }
 809
 810 void
 811 gst_rist_rtx_send_set_extseqnum (GstRistRtxSend * rtx, guint32 ssrc,
 812     guint16 seqnum_ext)
 813 {
 814   SSRCRtxData *data;
 815
 816   GST_OBJECT_LOCK (rtx);
 817   data = g_hash_table_lookup (rtx->ssrc_data, GUINT_TO_POINTER (ssrc));
 818
 819   if (data) {
 820     data->has_seqnum_ext = TRUE;
 821     data->seqnum_ext = seqnum_ext;
 822   }
 823   GST_OBJECT_UNLOCK (rtx);
 824 }
 825
 826 void
 827 gst_rist_rtx_send_clear_extseqnum (GstRistRtxSend * rtx, guint32 ssrc)
 828 {
 829   SSRCRtxData *data;
 830
 831   GST_OBJECT_LOCK (rtx);
 832   data = g_hash_table_lookup (rtx->ssrc_data, GUINT_TO_POINTER (ssrc));
 833
 834   if (data)
 835     data->has_seqnum_ext = FALSE;
 836   GST_OBJECT_UNLOCK (rtx);
 837 }