/* GStreamer plugin for forward error correction

 * Copyright (C) 2017 Pexip

 *

 * This library 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.

 *

 * This library 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 this library; if not, write to the Free Software

 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA

 *

 * Author: Mikhail Fludkov <misha@pexip.com>

 */

#include "rtpstoragestream.h"

static RtpStorageItem *

rtp_storage_item_new (GstBuffer * buffer, guint8 pt, guint16 seq)

{

  RtpStorageItem *ret = g_slice_new0 (RtpStorageItem);

  ret->buffer = buffer;

  ret->pt = pt;

  ret->seq = seq;

  return ret;

}

static void

rtp_storage_item_free (RtpStorageItem * item)

{

  g_assert (item->buffer != NULL);

  gst_buffer_unref (item->buffer);

  g_slice_free (RtpStorageItem, item);

}

static gint

rtp_storage_item_compare (gconstpointer a, gconstpointer b, gpointer userdata)

{

  gint seq_diff = gst_rtp_buffer_compare_seqnum (

      ((RtpStorageItem const *) a)->seq, ((RtpStorageItem const *) b)->seq);

  if (seq_diff >= 0)

    return 0;

  return 1;

}

static void

rtp_storage_stream_resize (RtpStorageStream * stream, GstClockTime size_time)

{

  GList *it;

  guint i, too_old_buffers_num = 0;

  g_assert (GST_CLOCK_TIME_IS_VALID (stream->max_arrival_time));

  g_assert (GST_CLOCK_TIME_IS_VALID (size_time));

  g_assert_cmpint (size_time, >, 0);

  /* Iterating from oldest sequence numbers to newest */

  for (i = 0, it = stream->queue.tail; it; it = it->prev, ++i) {

    RtpStorageItem *item = it->data;

    GstClockTime arrival_time = GST_BUFFER_DTS_OR_PTS (item->buffer);

    if (GST_CLOCK_TIME_IS_VALID (arrival_time)) {

      if (stream->max_arrival_time - arrival_time > size_time) {

        too_old_buffers_num = i + 1;

      } else

        break;

    }

  }

  for (i = 0; i < too_old_buffers_num; ++i) {

    RtpStorageItem *item = g_queue_pop_tail (&stream->queue);

    rtp_storage_item_free (item);

  }

}

void

rtp_storage_stream_resize_and_add_item (RtpStorageStream * stream,

    GstClockTime size_time, GstBuffer * buffer, guint8 pt, guint16 seq)

{

  GstClockTime arrival_time = GST_BUFFER_DTS_OR_PTS (buffer);

  if (G_LIKELY (GST_CLOCK_TIME_IS_VALID (arrival_time))) {

    if (G_LIKELY (GST_CLOCK_TIME_IS_VALID (stream->max_arrival_time)))

      stream->max_arrival_time = MAX (stream->max_arrival_time, arrival_time);

    else

      stream->max_arrival_time = arrival_time;

    rtp_storage_stream_resize (stream, size_time);

    rtp_storage_stream_add_item (stream, buffer, pt, seq);

  } else {

    rtp_storage_stream_add_item (stream, buffer, pt, seq);

  }

}

RtpStorageStream *

rtp_storage_stream_new (guint32 ssrc)

{

  RtpStorageStream *ret = g_slice_new0 (RtpStorageStream);

  ret->max_arrival_time = GST_CLOCK_TIME_NONE;

  ret->ssrc = ssrc;

  g_mutex_init (&ret->stream_lock);

  return ret;

}

void

rtp_storage_stream_free (RtpStorageStream * stream)

{

  STREAM_LOCK (stream);

  while (stream->queue.length)

    rtp_storage_item_free (g_queue_pop_tail (&stream->queue));

  STREAM_UNLOCK (stream);

  g_mutex_clear (&stream->stream_lock);

  g_slice_free (RtpStorageStream, stream);

}

void

rtp_storage_stream_add_item (RtpStorageStream * stream, GstBuffer * buffer,

    guint8 pt, guint16 seq)

{

  RtpStorageItem *item = rtp_storage_item_new (buffer, pt, seq);

  GList *sibling = g_queue_find_custom (&stream->queue, item,

      (GCompareFunc) rtp_storage_item_compare);

  g_queue_insert_before (&stream->queue, sibling, item);

}

GstBufferList *

rtp_storage_stream_get_packets_for_recovery (RtpStorageStream * stream,

    guint8 pt_fec, guint16 lost_seq)

{

  guint ret_length = 0;

  GList *end = NULL;

  GList *start = NULL;

  gboolean saw_fec = TRUE;      /* To initialize the start pointer in the loop below */

  GList *it;

  /* Looking for media stream chunk with FEC packets at the end, which could

   * can have the lost packet. For example:

   *

   *   |#10 FEC|  |#9 FEC|  |#8| ... |#6|  |#5 FEC|  |#4 FEC|  |#3 FEC|  |#2|  |#1|  |#0|

   *

   * Say @lost_seq = 7. Want to return bufferlist with packets [#6 : #10]. Other

   * packets are not relevant for recovery of packet 7.

   *

   * Or the lost packet can be in the storage. In that case single packet is returned.

   * It can happen if:

   * - it could have arrived right after it was considered lost (more of a corner case)

   * - it was recovered together with the other lost packet (most likely)

   */

  for (it = stream->queue.tail; it; it = it->prev) {

    RtpStorageItem *item = it->data;

    gboolean found_end = FALSE;

    /* Is the buffer we lost in the storage? */

    if (item->seq == lost_seq) {

      start = it;

      end = it;

      ret_length = 1;

      break;

    }

    if (pt_fec == item->pt) {

      gint seq_diff = gst_rtp_buffer_compare_seqnum (lost_seq, item->seq);

      if (seq_diff >= 0) {

        if (it->prev) {

          gboolean media_next =

              pt_fec != ((RtpStorageItem *) it->prev->data)->pt;

          found_end = media_next;

        } else

          found_end = TRUE;

      }

      saw_fec = TRUE;

    } else if (saw_fec) {

      saw_fec = FALSE;

      start = it;

      ret_length = 0;

    }

    ++ret_length;

    if (found_end) {

      end = it;

      break;

    }

  }

  if (end && !start)

    start = end;

  if (start && end) {

    GstBufferList *ret = gst_buffer_list_new_sized (ret_length);

    GList *it;

    for (it = start; it != end->prev; it = it->prev)

      gst_buffer_list_add (ret,

          gst_buffer_ref (((RtpStorageItem *) it->data)->buffer));

    return ret;

  }

  return NULL;

}

GstBuffer *

rtp_storage_stream_get_redundant_packet (RtpStorageStream * stream,

    guint16 lost_seq)

{

  GList *it;

  for (it = stream->queue.head; it; it = it->next) {

    RtpStorageItem *item = it->data;

    if (item->seq == lost_seq)

      return gst_buffer_ref (item->buffer);

  }

  return NULL;

}